Repository: cloudspannerecosystem/harbourbridge
Pull request #248 (2021-12-28 10:40:48)

Refactoring 1: Cleanup profiles and move conversion helpers methods to utils
This PR refactors 2 things:

- Move helper functions from `conversion/conversion.go` to `common/utils/utils.go`:
Currently, a lot of helper functions are stashed inside conversion/conversion.go which are also used by other packages. Often, other packages import conversion just to use these helper functions.
This is also required so that source profile does not need to import conversion.go and cyclic imports issue can be resolved. This will come in handy when we refactor the code to pass source and target profiles throughout.

- Breakout source and target profiles code from cmd folder: 
The profile related files are moved into a new package called `profiles`. The folder `cmd/` only has the subcommand files. The cmd directory imports the conversion package and profiles package. Conversion imports just the profiles package.
This refactoring is necessary to solve the cyclic dependency issue as well.

Deep1998: We have this kind of repetition in multiple places like cmd/cmd.go, common/constants/constants.go, conversion/conversion.go. But if we were to resolve this, we can go 2 ways about it. 
To avoid confusion it is a good practice to keep folder name and package name same. 

- With just common/utils.go, we would change the package name to `package common` which seems very generic for util functions. 
- Another alternative is to get rid of the common folder, and just have `constants/constants.go` and `utils/utils.go` in the root dir.
I'm leaning towards keeping the repetition since its happening in multiple places, but if were to change it, the latter seems better.
bharadwaj-aditya: lets go with the structure defined here and figure out if we want to change at a later stage then. Approving

Patch:
diff --git a/cmd/cmd.go b/cmd/cmd.go
index ebd4a7c..9a08c76 100644
--- a/cmd/cmd.go
+++ b/cmd/cmd.go
@@ -20,6 +20,7 @@ import (
 	"fmt"
 	"time"
 
+	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/conversion"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 )
@@ -37,7 +38,7 @@ var (
 // 2. Create database (if schemaOnly is set to false)
 // 3. Run data conversion (if schemaOnly is set to false)
 // 4. Generate report
-func CommandLine(ctx context.Context, driver, targetDb, dbURI string, dataOnly, schemaOnly, skipForeignKeys bool, schemaSampleSize int64, sessionJSON string, ioHelper *conversion.IOStreams, outputFilePrefix string, now time.Time) error {
+func CommandLine(ctx context.Context, driver, targetDb, dbURI string, dataOnly, schemaOnly, skipForeignKeys bool, schemaSampleSize int64, sessionJSON string, ioHelper *utils.IOStreams, outputFilePrefix string, now time.Time) error {
 	var conv *internal.Conv
 	var err error
 	if !dataOnly {
@@ -64,9 +65,9 @@ func CommandLine(ctx context.Context, driver, targetDb, dbURI string, dataOnly,
 			return err
 		}
 	}
-	adminClient, err := conversion.NewDatabaseAdminClient(ctx)
+	adminClient, err := utils.NewDatabaseAdminClient(ctx)
 	if err != nil {
-		return fmt.Errorf("can't create admin client: %w", conversion.AnalyzeError(err, dbURI))
+		return fmt.Errorf("can't create admin client: %w", utils.AnalyzeError(err, dbURI))
 	}
 	defer adminClient.Close()
 	err = conversion.CreateOrUpdateDatabase(ctx, adminClient, dbURI, conv, ioHelper.Out)
@@ -74,7 +75,7 @@ func CommandLine(ctx context.Context, driver, targetDb, dbURI string, dataOnly,
 		return fmt.Errorf("can't create/update database: %v", err)
 	}
 
-	client, err := conversion.GetClient(ctx, dbURI)
+	client, err := utils.GetClient(ctx, dbURI)
 	if err != nil {
 		return fmt.Errorf("can't create client for db %s: %v", dbURI, err)
 	}
@@ -90,7 +91,7 @@ func CommandLine(ctx context.Context, driver, targetDb, dbURI string, dataOnly,
 			return fmt.Errorf("can't perform update schema on db %s with foreign keys: %v", dbURI, err)
 		}
 	}
-	banner := conversion.GetBanner(now, dbURI)
+	banner := utils.GetBanner(now, dbURI)
 	conversion.Report(driver, bw.DroppedRowsByTable(), ioHelper.BytesRead, banner, conv, outputFilePrefix+reportFile, ioHelper.Out)
 	conversion.WriteBadData(bw, conv, banner, outputFilePrefix+badDataFile, ioHelper.Out)
 	return nil
diff --git a/cmd/common.go b/cmd/common.go
deleted file mode 100644
index 8e0d9f3..0000000
--- a/cmd/common.go
+++ /dev/null
@@ -1,111 +0,0 @@
-package cmd
-
-import (
-	"context"
-	"encoding/csv"
-	"fmt"
-	"os"
-	"strings"
-	"time"
-
-	"github.com/cloudspannerecosystem/harbourbridge/conversion"
-)
-
-// Parses input string `s` as a map of key-value pairs. It's expected that the
-// input string is of the form "key1=value1,key2=value2,..." etc. Return error
-// otherwise.
-func parseProfile(s string) (map[string]string, error) {
-	params := make(map[string]string)
-	if len(s) == 0 {
-		return params, nil
-	}
-
-	// We use CSV reader to parse key=value pairs separated by a comma to
-	// handle the case where a value may contain a comma within a quote. We
-	// expect exactly one record to be returned.
-	r := csv.NewReader(strings.NewReader(s))
-	r.Comma = ','
-	r.TrimLeadingSpace = true
-	records, err := r.ReadAll()
-	if err != nil {
-		return params, err
-	}
-	if len(records) > 1 {
-		return params, fmt.Errorf("contains invalid newline characters")
-	}
-
-	for _, kv := range records[0] {
-		s := strings.Split(strings.TrimSpace(kv), "=")
-		if len(s) != 2 {
-			return params, fmt.Errorf("invalid key=value pair (expected format: key1=value1): %v", kv)
-		}
-		if _, ok := params[s[0]]; ok {
-			return params, fmt.Errorf("duplicate key found: %v", s[0])
-		}
-		params[s[0]] = s[1]
-	}
-	return params, nil
-}
-
-func getResourceIds(ctx context.Context, targetProfile TargetProfile, now time.Time, driverName string, out *os.File) (string, string, string, error) {
-	var err error
-	project := targetProfile.conn.sp.project
-	if project == "" {
-		project, err = conversion.GetProject()
-		if err != nil {
-			return "", "", "", fmt.Errorf("can't get project: %v", err)
-		}
-	}
-	fmt.Println("Using Google Cloud project:", project)
-
-	instance := targetProfile.conn.sp.instance
-	if instance == "" {
-		instance, err = conversion.GetInstance(ctx, project, out)
-		if err != nil {
-			return "", "", "", fmt.Errorf("can't get instance: %v", err)
-		}
-	}
-	fmt.Println("Using Cloud Spanner instance:", instance)
-	conversion.PrintPermissionsWarning(driverName, out)
-
-	dbName := targetProfile.conn.sp.dbname
-	if dbName == "" {
-		dbName, err = conversion.GetDatabaseName(driverName, now)
-		if err != nil {
-			return "", "", "", fmt.Errorf("can't get database name: %v", err)
-		}
-	}
-	return project, instance, dbName, err
-}
-
-func getSQLConnectionStr(sourceProfile SourceProfile) string {
-	sqlConnectionStr := ""
-	if sourceProfile.ty == SourceProfileTypeConnection {
-		switch sourceProfile.conn.ty {
-		case SourceProfileConnectionTypeMySQL:
-			connParams := sourceProfile.conn.mysql
-			return conversion.GetMYSQLConnectionStr(connParams.host, connParams.port, connParams.user, connParams.pwd, connParams.db)
-		case SourceProfileConnectionTypePostgreSQL:
-			connParams := sourceProfile.conn.pg
-			return conversion.GetPGSQLConnectionStr(connParams.host, connParams.port, connParams.user, connParams.pwd, connParams.db)
-		case SourceProfileConnectionTypeDynamoDB:
-			// For DynamoDB, client provided by aws-sdk reads connection credentials from env variables only.
-			// Thus, there is no need to create sqlConnectionStr for the same. We instead set the env variables
-			// programmatically if not set.
-			return ""
-		}
-	}
-	return sqlConnectionStr
-}
-
-func getSchemaSampleSize(sourceProfile SourceProfile) int64 {
-	schemaSampleSize := int64(100000)
-	if sourceProfile.ty == SourceProfileTypeConnection {
-		if sourceProfile.conn.ty == SourceProfileConnectionTypeDynamoDB {
-			if sourceProfile.conn.dydb.schemaSampleSize != 0 {
-				schemaSampleSize = sourceProfile.conn.dydb.schemaSampleSize
-			}
-		}
-	}
-	return schemaSampleSize
-}
diff --git a/cmd/data.go b/cmd/data.go
index 0b40afd..29eebe8 100644
--- a/cmd/data.go
+++ b/cmd/data.go
@@ -8,8 +8,10 @@ import (
 	"path"
 	"time"
 
+	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/conversion"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/profiles"
 	"github.com/google/subcommands"
 )
 
@@ -64,7 +66,7 @@ func (cmd *DataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface
 		}
 	}()
 
-	sourceProfile, err := NewSourceProfile(cmd.sourceProfile, cmd.source)
+	sourceProfile, err := profiles.NewSourceProfile(cmd.sourceProfile, cmd.source)
 	if err != nil {
 		return subcommands.ExitUsageError
 	}
@@ -73,23 +75,23 @@ func (cmd *DataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface
 		return subcommands.ExitUsageError
 	}
 
-	targetProfile, err := NewTargetProfile(cmd.targetProfile)
+	targetProfile, err := profiles.NewTargetProfile(cmd.targetProfile)
 	if err != nil {
 		return subcommands.ExitUsageError
 	}
 	targetDb := targetProfile.ToLegacyTargetDb()
 
 	dumpFilePath := ""
-	if sourceProfile.ty == SourceProfileTypeFile && (sourceProfile.file.format == "" || sourceProfile.file.format == "dump") {
-		dumpFilePath = sourceProfile.file.path
+	if sourceProfile.Ty == profiles.SourceProfileTypeFile && (sourceProfile.File.Format == "" || sourceProfile.File.Format == "dump") {
+		dumpFilePath = sourceProfile.File.Path
 	}
-	ioHelper := conversion.NewIOStreams(driverName, dumpFilePath)
+	ioHelper := utils.NewIOStreams(driverName, dumpFilePath)
 	if ioHelper.SeekableIn != nil {
 		defer ioHelper.In.Close()
 	}
 
 	now := time.Now()
-	project, instance, dbName, err := getResourceIds(ctx, targetProfile, now, driverName, ioHelper.Out)
+	project, instance, dbName, err := profiles.GetResourceIds(ctx, targetProfile, now, driverName, ioHelper.Out)
 	if err != nil {
 		return subcommands.ExitUsageError
 	}
@@ -110,13 +112,13 @@ func (cmd *DataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface
 		return subcommands.ExitUsageError
 	}
 
-	adminClient, err := conversion.NewDatabaseAdminClient(ctx)
+	adminClient, err := utils.NewDatabaseAdminClient(ctx)
 	if err != nil {
-		err = fmt.Errorf("can't create admin client: %w", conversion.AnalyzeError(err, dbURI))
+		err = fmt.Errorf("can't create admin client: %w", utils.AnalyzeError(err, dbURI))
 		return subcommands.ExitFailure
 	}
 	defer adminClient.Close()
-	client, err := conversion.GetClient(ctx, dbURI)
+	client, err := utils.GetClient(ctx, dbURI)
 	if err != nil {
 		err = fmt.Errorf("can't create client for db %s: %v", dbURI, err)
 		return subcommands.ExitFailure
@@ -129,7 +131,7 @@ func (cmd *DataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface
 		return subcommands.ExitFailure
 	}
 
-	bw, err := conversion.DataConv(driverName, getSQLConnectionStr(sourceProfile), &ioHelper, client, conv, true, getSchemaSampleSize(sourceProfile))
+	bw, err := conversion.DataConv(driverName, profiles.GetSQLConnectionStr(sourceProfile), &ioHelper, client, conv, true, profiles.GetSchemaSampleSize(sourceProfile))
 	if err != nil {
 		err = fmt.Errorf("can't finish data conversion for db %s: %v", dbURI, err)
 		return subcommands.ExitFailure
@@ -140,7 +142,7 @@ func (cmd *DataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interface
 			return subcommands.ExitFailure
 		}
 	}
-	banner := conversion.GetBanner(now, dbURI)
+	banner := utils.GetBanner(now, dbURI)
 	conversion.Report(driverName, bw.DroppedRowsByTable(), ioHelper.BytesRead, banner, conv, cmd.filePrefix+reportFile, ioHelper.Out)
 	conversion.WriteBadData(bw, conv, banner, cmd.filePrefix+badDataFile, ioHelper.Out)
 	return subcommands.ExitSuccess
diff --git a/cmd/schema.go b/cmd/schema.go
index d1e9d44..d04f814 100644
--- a/cmd/schema.go
+++ b/cmd/schema.go
@@ -8,8 +8,10 @@ import (
 	"path"
 	"time"
 
+	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/conversion"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/profiles"
 	"github.com/google/subcommands"
 )
 
@@ -60,7 +62,7 @@ func (cmd *SchemaCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interfa
 		}
 	}()
 
-	sourceProfile, err := NewSourceProfile(cmd.sourceProfile, cmd.source)
+	sourceProfile, err := profiles.NewSourceProfile(cmd.sourceProfile, cmd.source)
 	if err != nil {
 		return subcommands.ExitUsageError
 	}
@@ -69,24 +71,24 @@ func (cmd *SchemaCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interfa
 		return subcommands.ExitUsageError
 	}
 
-	targetProfile, err := NewTargetProfile(cmd.targetProfile)
+	targetProfile, err := profiles.NewTargetProfile(cmd.targetProfile)
 	if err != nil {
 		return subcommands.ExitUsageError
 	}
 	targetDb := targetProfile.ToLegacyTargetDb()
 
 	dumpFilePath := ""
-	if sourceProfile.ty == SourceProfileTypeFile && (sourceProfile.file.format == "" || sourceProfile.file.format == "dump") {
-		dumpFilePath = sourceProfile.file.path
+	if sourceProfile.Ty == profiles.SourceProfileTypeFile && (sourceProfile.File.Format == "" || sourceProfile.File.Format == "dump") {
+		dumpFilePath = sourceProfile.File.Path
 	}
-	ioHelper := conversion.NewIOStreams(driverName, dumpFilePath)
+	ioHelper := utils.NewIOStreams(driverName, dumpFilePath)
 	if ioHelper.SeekableIn != nil {
 		defer ioHelper.In.Close()
 	}
 
 	// If filePrefix not explicitly set, use generated dbName.
 	if cmd.filePrefix == "" {
-		dbName, err := conversion.GetDatabaseName(driverName, time.Now())
+		dbName, err := utils.GetDatabaseName(driverName, time.Now())
 		if err != nil {
 			err = fmt.Errorf("can't generate database name for prefix: %v", err)
 			return subcommands.ExitFailure
@@ -95,7 +97,7 @@ func (cmd *SchemaCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...interfa
 	}
 
 	var conv *internal.Conv
-	conv, err = conversion.SchemaConv(driverName, getSQLConnectionStr(sourceProfile), targetDb, &ioHelper, getSchemaSampleSize(sourceProfile))
+	conv, err = conversion.SchemaConv(driverName, profiles.GetSQLConnectionStr(sourceProfile), targetDb, &ioHelper, profiles.GetSchemaSampleSize(sourceProfile))
 	if err != nil {
 		return subcommands.ExitFailure
 	}
diff --git a/cmd/schema_and_data.go b/cmd/schema_and_data.go
index 47657e3..ee89eda 100644
--- a/cmd/schema_and_data.go
+++ b/cmd/schema_and_data.go
@@ -8,8 +8,10 @@ import (
 	"path"
 	"time"
 
+	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/conversion"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/profiles"
 	"github.com/google/subcommands"
 )
 
@@ -62,7 +64,7 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
 		}
 	}()
 
-	sourceProfile, err := NewSourceProfile(cmd.sourceProfile, cmd.source)
+	sourceProfile, err := profiles.NewSourceProfile(cmd.sourceProfile, cmd.source)
 	if err != nil {
 		return subcommands.ExitUsageError
 	}
@@ -71,17 +73,17 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
 		return subcommands.ExitUsageError
 	}
 
-	targetProfile, err := NewTargetProfile(cmd.targetProfile)
+	targetProfile, err := profiles.NewTargetProfile(cmd.targetProfile)
 	if err != nil {
 		return subcommands.ExitUsageError
 	}
 	targetDb := targetProfile.ToLegacyTargetDb()
 
 	dumpFilePath := ""
-	if sourceProfile.ty == SourceProfileTypeFile && (sourceProfile.file.format == "" || sourceProfile.file.format == "dump") {
-		dumpFilePath = sourceProfile.file.path
+	if sourceProfile.Ty == profiles.SourceProfileTypeFile && (sourceProfile.File.Format == "" || sourceProfile.File.Format == "dump") {
+		dumpFilePath = sourceProfile.File.Path
 	}
-	ioHelper := conversion.NewIOStreams(driverName, dumpFilePath)
+	ioHelper := utils.NewIOStreams(driverName, dumpFilePath)
 	if ioHelper.SeekableIn != nil {
 		defer ioHelper.In.Close()
 	}
@@ -90,15 +92,15 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
 
 	// If filePrefix not explicitly set, use dbName as prefix.
 	if cmd.filePrefix == "" {
-		dbName, err := conversion.GetDatabaseName(driverName, now)
+		dbName, err := utils.GetDatabaseName(driverName, now)
 		if err != nil {
 			panic(fmt.Errorf("can't generate database name for prefix: %v", err))
 		}
 		cmd.filePrefix = dbName + "."
 	}
 
-	sqlConnectionStr := getSQLConnectionStr(sourceProfile)
-	schemaSampleSize := getSchemaSampleSize(sourceProfile)
+	sqlConnectionStr := profiles.GetSQLConnectionStr(sourceProfile)
+	schemaSampleSize := profiles.GetSchemaSampleSize(sourceProfile)
 	var conv *internal.Conv
 	conv, err = conversion.SchemaConv(driverName, sqlConnectionStr, targetDb, &ioHelper, schemaSampleSize)
 	if err != nil {
@@ -109,19 +111,19 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
 	conversion.WriteSessionFile(conv, cmd.filePrefix+sessionFile, ioHelper.Out)
 	conversion.Report(driverName, nil, ioHelper.BytesRead, "", conv, cmd.filePrefix+reportFile, ioHelper.Out)
 
-	project, instance, dbName, err := getResourceIds(ctx, targetProfile, now, driverName, ioHelper.Out)
+	project, instance, dbName, err := profiles.GetResourceIds(ctx, targetProfile, now, driverName, ioHelper.Out)
 	if err != nil {
 		return subcommands.ExitUsageError
 	}
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", project, instance, dbName)
 
-	adminClient, err := conversion.NewDatabaseAdminClient(ctx)
+	adminClient, err := utils.NewDatabaseAdminClient(ctx)
 	if err != nil {
-		err = fmt.Errorf("can't create admin client: %w", conversion.AnalyzeError(err, dbURI))
+		err = fmt.Errorf("can't create admin client: %w", utils.AnalyzeError(err, dbURI))
 		return subcommands.ExitFailure
 	}
 	defer adminClient.Close()
-	client, err := conversion.GetClient(ctx, dbURI)
+	client, err := utils.GetClient(ctx, dbURI)
 	if err != nil {
 		err = fmt.Errorf("can't create client for db %s: %v", dbURI, err)
 		return subcommands.ExitFailure
@@ -145,7 +147,7 @@ func (cmd *SchemaAndDataCmd) Execute(ctx context.Context, f *flag.FlagSet, _ ...
 			return subcommands.ExitFailure
 		}
 	}
-	banner := conversion.GetBanner(now, dbURI)
+	banner := utils.GetBanner(now, dbURI)
 	conversion.Report(driverName, bw.DroppedRowsByTable(), ioHelper.BytesRead, banner, conv, cmd.filePrefix+reportFile, ioHelper.Out)
 	conversion.WriteBadData(bw, conv, banner, cmd.filePrefix+badDataFile, ioHelper.Out)
 	return subcommands.ExitSuccess
diff --git a/common/utils/utils.go b/common/utils/utils.go
new file mode 100644
index 0000000..7568bec
--- /dev/null
+++ b/common/utils/utils.go
@@ -0,0 +1,379 @@
+// Package utils contains common helper functions used across multiple other packages.
+// Utils should not import any harbourbridge packages.
+package utils
+
+import (
+	"bufio"
+	"context"
+	"crypto/rand"
+	"fmt"
+	"io"
+	"io/ioutil"
+	"log"
+	"net/url"
+	"os"
+	"os/exec"
+	"strings"
+	"syscall"
+	"time"
+
+	sp "cloud.google.com/go/spanner"
+	database "cloud.google.com/go/spanner/admin/database/apiv1"
+	instance "cloud.google.com/go/spanner/admin/instance/apiv1"
+	"cloud.google.com/go/storage"
+	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
+	"golang.org/x/crypto/ssh/terminal"
+	"google.golang.org/api/iterator"
+	"google.golang.org/api/option"
+	instancepb "google.golang.org/genproto/googleapis/spanner/admin/instance/v1"
+)
+
+// IOStreams is a struct that contains the file descriptor for dumpFile.
+type IOStreams struct {
+	In, SeekableIn, Out *os.File
+	BytesRead           int64
+}
+
+// NewIOStreams returns a new IOStreams struct such that input stream is set
+// to open file descriptor for dumpFile if driver is PGDUMP or MYSQLDUMP.
+// Input stream defaults to stdin. Output stream is always set to stdout.
+func NewIOStreams(driver string, dumpFile string) IOStreams {
+	io := IOStreams{In: os.Stdin, Out: os.Stdout}
+	u, err := url.Parse(dumpFile)
+	if err != nil {
+		fmt.Printf("parseFilePath: unable parse file path for dumpfile %s", dumpFile)
+		log.Fatal(err)
+	}
+	if (driver == constants.PGDUMP || driver == constants.MYSQLDUMP) && dumpFile != "" {
+		fmt.Printf("\nLoading dump file from path: %s\n", dumpFile)
+		var f *os.File
+		var err error
+		if u.Scheme == "gs" {
+			bucketName := u.Host
+			filePath := u.Path[1:] // removes "/" from beginning of path
+			f, err = downloadFromGCS(bucketName, filePath)
+		} else {
+			f, err = os.Open(dumpFile)
+		}
+		if err != nil {
+			fmt.Printf("\nError reading dump file: %v err:%v\n", dumpFile, err)
+			log.Fatal(err)
+		}
+		io.In = f
+	}
+	return io
+}
+
+// downloadFromGCS returns the dump file that is downloaded from GCS
+func downloadFromGCS(bucketName string, filePath string) (*os.File, error) {
+	ctx := context.Background()
+
+	client, err := storage.NewClient(ctx)
+	if err != nil {
+		fmt.Printf("Failed to create GCS client for bucket %q", bucketName)
+		log.Fatal(err)
+	}
+	defer client.Close()
+
+	bucket := client.Bucket(bucketName)
+	rc, err := bucket.Object(filePath).NewReader(ctx)
+	if err != nil {
+		fmt.Printf("readFile: unable to open file from bucket %q, file %q: %v", bucketName, filePath, err)
+		log.Fatal(err)
+		return nil, err
+	}
+	defer rc.Close()
+	r := bufio.NewReader(rc)
+
+	tmpfile, err := ioutil.TempFile("", "harbourbridge.gcs.data")
+	if err != nil {
+		fmt.Printf("saveFile: unable to open temporary file to save dump file from GCS bucket %v", err)
+		log.Fatal(err)
+		return nil, err
+	}
+	syscall.Unlink(tmpfile.Name()) // File will be deleted when this process exits.
+
+	fmt.Printf("\nDownloading dump file from GCS bucket %s, path %s\n", bucketName, filePath)
+	buffer := make([]byte, 1024)
+	for {
+		// read a chunk
+		n, err := r.Read(buffer[:cap(buffer)])
+
+		if err != nil && err != io.EOF {
+			fmt.Printf("readFile: unable to read entire dump file from bucket %s, file %s: %v", bucketName, filePath, err)
+			log.Fatal(err)
+			return nil, err
+		}
+		if n == 0 && err == io.EOF {
+			break
+		}
+
+		// write a chunk
+		if _, err = tmpfile.Write(buffer[:n]); err != nil {
+			fmt.Printf("saveFile: unable to save read data from bucket %s, file %s: %v", bucketName, filePath, err)
+			log.Fatal(err)
+		}
+	}
+
+	return tmpfile, nil
+}
+
+// GetProject returns the cloud project we should use for accessing Spanner.
+// Use environment variable GCLOUD_PROJECT if it is set.
+// Otherwise, use the default project returned from gcloud.
+func GetProject() (string, error) {
+	project := os.Getenv("GCLOUD_PROJECT")
+	if project != "" {
+		return project, nil
+	}
+	cmd := exec.Command("gcloud", "config", "list", "--format", "value(core.project)")
+	out, err := cmd.CombinedOutput()
+	if err != nil {
+		return "", fmt.Errorf("call to gcloud to get project failed: %w", err)
+	}
+	project = strings.TrimSpace(string(out))
+	return project, nil
+}
+
+// GetInstance returns the Spanner instance we should use for creating DBs.
+// If the user specified instance (via flag 'instance') then use that.
+// Otherwise try to deduce the instance using gcloud.
+func GetInstance(ctx context.Context, project string, out *os.File) (string, error) {
+	l, err := getInstances(ctx, project)
+	if err != nil {
+		return "", err
+	}
+	if len(l) == 0 {
+		fmt.Fprintf(out, "Could not find any Spanner instances for project %s\n", project)
+		return "", fmt.Errorf("no Spanner instances for %s", project)
+	}
+
+	// Note: we could ask for user input to select/confirm which Spanner
+	// instance to use, but that interacts poorly with piping pg_dump/mysqldump data
+	// to the tool via stdin.
+	if len(l) == 1 {
+		fmt.Fprintf(out, "Using only available Spanner instance: %s\n", l[0])
+		return l[0], nil
+	}
+	fmt.Fprintf(out, "Available Spanner instances:\n")
+	for i, x := range l {
+		fmt.Fprintf(out, " %d) %s\n", i+1, x)
+	}
+	fmt.Fprintf(out, "Please pick one of the available instances and set the flag '--instance'\n\n")
+	return "", fmt.Errorf("auto-selection of instance failed: project %s has more than one Spanner instance. "+
+		"Please use the flag '--instance' to select an instance", project)
+}
+
+func getInstances(ctx context.Context, project string) ([]string, error) {
+	instanceClient, err := instance.NewInstanceAdminClient(ctx)
+	if err != nil {
+		return nil, AnalyzeError(err, fmt.Sprintf("projects/%s", project))
+	}
+	it := instanceClient.ListInstances(ctx, &instancepb.ListInstancesRequest{Parent: fmt.Sprintf("projects/%s", project)})
+	var l []string
+	for {
+		resp, err := it.Next()
+		if err == iterator.Done {
+			break
+		}
+		if err != nil {
+			return nil, AnalyzeError(err, fmt.Sprintf("projects/%s", project))
+		}
+		l = append(l, strings.TrimPrefix(resp.Name, fmt.Sprintf("projects/%s/instances/", project)))
+	}
+	return l, nil
+}
+
+func GetPassword() string {
+	fmt.Print("Enter Password: ")
+	bytePassword, err := terminal.ReadPassword(int(syscall.Stdin))
+	if err != nil {
+		fmt.Println("\nCoudln't read password")
+		return ""
+	}
+	fmt.Printf("\n")
+	return strings.TrimSpace(string(bytePassword))
+}
+
+// GetDatabaseName generates database name with driver_date prefix.
+func GetDatabaseName(driver string, now time.Time) (string, error) {
+	return generateName(fmt.Sprintf("%s_%s", driver, now.Format("2006-01-02")))
+}
+
+func generateName(prefix string) (string, error) {
+	b := make([]byte, 4)
+	_, err := rand.Read(b)
+	if err != nil {
+		return "", fmt.Errorf("error generating name: %w", err)
+
+	}
+	return fmt.Sprintf("%s_%x-%x", prefix, b[0:2], b[2:4]), nil
+}
+
+// parseURI parses an unknown URI string that could be a database, instance or project URI.
+func parseURI(URI string) (project, instance, dbName string) {
+	project, instance, dbName = "", "", ""
+	if strings.Contains(URI, "databases") {
+		project, instance, dbName = ParseDbURI(URI)
+	} else if strings.Contains(URI, "instances") {
+		project, instance = parseInstanceURI(URI)
+	} else if strings.Contains(URI, "projects") {
+		project = parseProjectURI(URI)
+	}
+	return
+}
+
+func ParseDbURI(dbURI string) (project, instance, dbName string) {
+	split := strings.Split(dbURI, "/databases/")
+	project, instance = parseInstanceURI(split[0])
+	dbName = split[1]
+	return
+}
+
+func parseInstanceURI(instanceURI string) (project, instance string) {
+	split := strings.Split(instanceURI, "/instances/")
+	project = parseProjectURI(split[0])
+	instance = split[1]
+	return
+}
+
+func parseProjectURI(projectURI string) (project string) {
+	split := strings.Split(projectURI, "/")
+	project = split[1]
+	return
+}
+
+// AnalyzeError inspects an error returned from Cloud Spanner and adds information
+// about potential root causes e.g. authentication issues.
+func AnalyzeError(err error, URI string) error {
+	project, instance, _ := parseURI(URI)
+	e := strings.ToLower(err.Error())
+	if ContainsAny(e, []string{"unauthenticated", "cannot fetch token", "default credentials"}) {
+		return fmt.Errorf("%w."+`
+Possible cause: credentials are mis-configured. Do you need to run
+
+  gcloud auth application-default login
+
+or configure environment variable GOOGLE_APPLICATION_CREDENTIALS.
+See https://cloud.google.com/docs/authentication/getting-started`, err)
+	}
+	if ContainsAny(e, []string{"instance not found"}) && instance != "" {
+		return fmt.Errorf("%w.\n"+`
+Possible cause: Spanner instance specified via instance option does not exist.
+Please check that '%s' is correct and that it is a valid Spanner
+instance for project %s`, err, instance, project)
+	}
+	return err
+}
+
+// PrintPermissionsWarning prints permission warning.
+func PrintPermissionsWarning(driver string, out *os.File) {
+	fmt.Fprintf(out,
+		`
+WARNING: Please check that permissions for this Spanner instance are
+appropriate. Spanner manages access control at the database level, and the
+database created by HarbourBridge will inherit default permissions from this
+instance. All data written to Spanner will be visible to anyone who can
+access the created database. Note that `+driver+` table-level and row-level
+ACLs are dropped during conversion since they are not supported by Spanner.
+
+`)
+}
+
+func ContainsAny(s string, l []string) bool {
+	for _, a := range l {
+		if strings.Contains(s, a) {
+			return true
+		}
+	}
+	return false
+}
+
+func GetFileSize(f *os.File) (int64, error) {
+	info, err := f.Stat()
+	if err != nil {
+		return 0, fmt.Errorf("can't stat file: %w", err)
+	}
+	return info.Size(), nil
+}
+
+// SetupLogFile configures the file used for logs.
+// By default we just drop logs on the floor. To enable them (e.g. to debug
+// Cloud Spanner client library issues), set logfile to a non-empty filename.
+// Note: this tool itself doesn't generate logs, but some of the libraries it
+// uses do. If we don't set the log file, we see a number of unhelpful and
+// unactionable logs spamming stdout, which is annoying and confusing.
+func SetupLogFile() (*os.File, error) {
+	// To enable debug logs, set logfile to a non-empty filename.
+	logfile := ""
+	if logfile == "" {
+		log.SetOutput(ioutil.Discard)
+		return nil, nil
+	}
+	f, err := os.Create(logfile)
+	if err != nil {
+		return nil, err
+	}
+	log.SetOutput(f)
+	return f, nil
+}
+
+// Close closes file.
+func Close(f *os.File) {
+	if f != nil {
+		f.Close()
+	}
+}
+
+func PrintSeekError(driver string, err error, out *os.File) {
+	fmt.Fprintf(out, "\nCan't get seekable input file: %v\n", err)
+	fmt.Fprintf(out, "Likely cause: not enough space in %s.\n", os.TempDir())
+	fmt.Fprintf(out, "Try writing "+driver+" output to a file first i.e.\n")
+	fmt.Fprintf(out, " "+driver+" > tmpfile\n")
+	fmt.Fprintf(out, "  harbourbridge < tmpfile\n")
+}
+
+// NewSpannerClient returns a new Spanner client.
+// It respects SPANNER_API_ENDPOINT.
+func NewSpannerClient(ctx context.Context, db string) (*sp.Client, error) {
+	if endpoint := os.Getenv("SPANNER_API_ENDPOINT"); endpoint != "" {
+		return sp.NewClient(ctx, db, option.WithEndpoint(endpoint))
+	}
+	return sp.NewClient(ctx, db)
+}
+
+// GetClient returns a new Spanner client.  It uses the background context.
+func GetClient(ctx context.Context, db string) (*sp.Client, error) {
+	return NewSpannerClient(ctx, db)
+}
+
+// NewDatabaseAdminClient returns a new db-admin client.
+// It respects SPANNER_API_ENDPOINT.
+func NewDatabaseAdminClient(ctx context.Context) (*database.DatabaseAdminClient, error) {
+	if endpoint := os.Getenv("SPANNER_API_ENDPOINT"); endpoint != "" {
+		return database.NewDatabaseAdminClient(ctx, option.WithEndpoint(endpoint))
+	}
+	return database.NewDatabaseAdminClient(ctx)
+}
+
+// NewInstanceAdminClient returns a new instance-admin client.
+// It respects SPANNER_API_ENDPOINT.
+func NewInstanceAdminClient(ctx context.Context) (*instance.InstanceAdminClient, error) {
+	if endpoint := os.Getenv("SPANNER_API_ENDPOINT"); endpoint != "" {
+		return instance.NewInstanceAdminClient(ctx, option.WithEndpoint(endpoint))
+	}
+	return instance.NewInstanceAdminClient(ctx)
+}
+
+func SumMapValues(m map[string]int64) int64 {
+	n := int64(0)
+	for _, c := range m {
+		n += c
+	}
+	return n
+}
+
+// GetBanner prints banner message after command line process is finished.
+func GetBanner(now time.Time, db string) string {
+	return fmt.Sprintf("Generated at %s for db %s\n\n", now.Format("2006-01-02 15:04:05"), db)
+}
diff --git a/conversion/conversion.go b/conversion/conversion.go
index f4c2975..bf16858 100644
--- a/conversion/conversion.go
+++ b/conversion/conversion.go
@@ -25,16 +25,12 @@ package conversion
 import (
 	"bufio"
 	"context"
-	"crypto/rand"
 	"database/sql"
 	"encoding/json"
 	"fmt"
 	"io"
 	"io/ioutil"
-	"log"
-	"net/url"
 	"os"
-	"os/exec"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -43,20 +39,15 @@ import (
 
 	sp "cloud.google.com/go/spanner"
 	database "cloud.google.com/go/spanner/admin/database/apiv1"
-	instance "cloud.google.com/go/spanner/admin/instance/apiv1"
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/session"
 	dydb "github.com/aws/aws-sdk-go/service/dynamodb"
-	"golang.org/x/crypto/ssh/terminal"
-	"google.golang.org/api/iterator"
-	"google.golang.org/api/option"
 	adminpb "google.golang.org/genproto/googleapis/spanner/admin/database/v1"
-	instancepb "google.golang.org/genproto/googleapis/spanner/admin/instance/v1"
-
-	"cloud.google.com/go/storage"
 
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
+	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
+	"github.com/cloudspannerecosystem/harbourbridge/profiles"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/dynamodb"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/mysql"
@@ -82,7 +73,7 @@ var (
 //  - Driver is DynamoDB or a dump file mode.
 //  - This function is called as part of the legacy global CLI flag mode. (This string is constructed from env variables later on)
 // When using source-profile, the sqlConnectionStr is constructed from the input params.
-func SchemaConv(driver, sqlConnectionStr, targetDb string, ioHelper *IOStreams, schemaSampleSize int64) (*internal.Conv, error) {
+func SchemaConv(driver, sqlConnectionStr, targetDb string, ioHelper *utils.IOStreams, schemaSampleSize int64) (*internal.Conv, error) {
 	switch driver {
 	case constants.POSTGRES, constants.MYSQL, constants.DYNAMODB:
 		return schemaFromDatabase(driver, sqlConnectionStr, targetDb, schemaSampleSize)
@@ -99,7 +90,7 @@ func SchemaConv(driver, sqlConnectionStr, targetDb string, ioHelper *IOStreams,
 //  - Driver is DynamoDB or a dump file mode.
 //  - This function is called as part of the legacy global CLI flag mode. (This string is constructed from env variables later on)
 // When using source-profile, the sqlConnectionStr and schemaSampleSize are constructed from the input params.
-func DataConv(driver, sqlConnectionStr string, ioHelper *IOStreams, client *sp.Client, conv *internal.Conv, dataOnly bool, schemaSampleSize int64) (*spanner.BatchWriter, error) {
+func DataConv(driver, sqlConnectionStr string, ioHelper *utils.IOStreams, client *sp.Client, conv *internal.Conv, dataOnly bool, schemaSampleSize int64) (*spanner.BatchWriter, error) {
 	config := spanner.BatchWriterConfig{
 		BytesLimit: 100 * 1000 * 1000,
 		WriteLimit: 40,
@@ -125,14 +116,14 @@ func connectionConfig(driver string, sqlConnectionStr string) (interface{}, erro
 		// If empty, this is called as part of the legacy mode witih global CLI flags.
 		// When using source-profile mode is used, the sqlConnectionStr is already populated.
 		if sqlConnectionStr == "" {
-			return generatePGSQLConnectionStr()
+			return profiles.GeneratePGSQLConnectionStr()
 		}
 		return sqlConnectionStr, nil
 	case constants.MYSQL:
 		// If empty, this is called as part of the legacy mode witih global CLI flags.
 		// When using source-profile mode is used, the sqlConnectionStr is already populated.
 		if sqlConnectionStr == "" {
-			return generateMYSQLConnectionStr()
+			return profiles.GenerateMYSQLConnectionStr()
 		}
 		return sqlConnectionStr, nil
 	case constants.DYNAMODB:
@@ -142,46 +133,6 @@ func connectionConfig(driver string, sqlConnectionStr string) (interface{}, erro
 	}
 }
 
-func generatePGSQLConnectionStr() (string, error) {
-	server := os.Getenv("PGHOST")
-	port := os.Getenv("PGPORT")
-	user := os.Getenv("PGUSER")
-	dbname := os.Getenv("PGDATABASE")
-	if server == "" || port == "" || user == "" || dbname == "" {
-		fmt.Printf("Please specify host, port, user and database using PGHOST, PGPORT, PGUSER and PGDATABASE environment variables\n")
-		return "", fmt.Errorf("could not connect to source database")
-	}
-	password := os.Getenv("PGPASSWORD")
-	if password == "" {
-		password = GetPassword()
-	}
-	return GetPGSQLConnectionStr(server, port, user, password, dbname), nil
-}
-
-func GetPGSQLConnectionStr(server, port, user, password, dbname string) string {
-	return fmt.Sprintf("host=%s port=%s user=%s password=%s dbname=%s sslmode=disable", server, port, user, password, dbname)
-}
-
-func generateMYSQLConnectionStr() (string, error) {
-	server := os.Getenv("MYSQLHOST")
-	port := os.Getenv("MYSQLPORT")
-	user := os.Getenv("MYSQLUSER")
-	dbname := os.Getenv("MYSQLDATABASE")
-	if server == "" || port == "" || user == "" || dbname == "" {
-		fmt.Printf("Please specify host, port, user and database using MYSQLHOST, MYSQLPORT, MYSQLUSER and MYSQLDATABASE environment variables\n")
-		return "", fmt.Errorf("could not connect to source database")
-	}
-	password := os.Getenv("MYSQLPWD")
-	if password == "" {
-		password = GetPassword()
-	}
-	return GetMYSQLConnectionStr(server, port, user, password, dbname), nil
-}
-
-func GetMYSQLConnectionStr(server, port, user, password, dbname string) string {
-	return fmt.Sprintf("%s:%s@tcp(%s:%s)/%s", user, password, server, port, dbname)
-}
-
 func getDbNameFromSQLConnectionStr(driver, sqlConnectionStr string) string {
 	switch driver {
 	case constants.POSTGRES:
@@ -244,100 +195,10 @@ func getDynamoDBClientConfig() (*aws.Config, error) {
 	return &cfg, nil
 }
 
-// IOStreams is a struct that contains the file descriptor for dumpFile.
-type IOStreams struct {
-	In, SeekableIn, Out *os.File
-	BytesRead           int64
-}
-
-// downloadFromGCS returns the dump file that is downloaded from GCS
-func downloadFromGCS(bucketName string, filePath string) (*os.File, error) {
-	ctx := context.Background()
-
-	client, err := storage.NewClient(ctx)
-	if err != nil {
-		fmt.Printf("Failed to create GCS client for bucket %q", bucketName)
-		log.Fatal(err)
-	}
-	defer client.Close()
-
-	bucket := client.Bucket(bucketName)
-	rc, err := bucket.Object(filePath).NewReader(ctx)
-	if err != nil {
-		fmt.Printf("readFile: unable to open file from bucket %q, file %q: %v", bucketName, filePath, err)
-		log.Fatal(err)
-		return nil, err
-	}
-	defer rc.Close()
-	r := bufio.NewReader(rc)
-
-	tmpfile, err := ioutil.TempFile("", "harbourbridge.gcs.data")
-	if err != nil {
-		fmt.Printf("saveFile: unable to open temporary file to save dump file from GCS bucket %v", err)
-		log.Fatal(err)
-		return nil, err
-	}
-	syscall.Unlink(tmpfile.Name()) // File will be deleted when this process exits.
-
-	fmt.Printf("\nDownloading dump file from GCS bucket %s, path %s\n", bucketName, filePath)
-	buffer := make([]byte, 1024)
-	for {
-		// read a chunk
-		n, err := r.Read(buffer[:cap(buffer)])
-
-		if err != nil && err != io.EOF {
-			fmt.Printf("readFile: unable to read entire dump file from bucket %s, file %s: %v", bucketName, filePath, err)
-			log.Fatal(err)
-			return nil, err
-		}
-		if n == 0 && err == io.EOF {
-			break
-		}
-
-		// write a chunk
-		if _, err = tmpfile.Write(buffer[:n]); err != nil {
-			fmt.Printf("saveFile: unable to save read data from bucket %s, file %s: %v", bucketName, filePath, err)
-			log.Fatal(err)
-		}
-	}
-
-	return tmpfile, nil
-}
-
-// NewIOStreams returns a new IOStreams struct such that input stream is set
-// to open file descriptor for dumpFile if driver is PGDUMP or MYSQLDUMP.
-// Input stream defaults to stdin. Output stream is always set to stdout.
-func NewIOStreams(driver string, dumpFile string) IOStreams {
-	io := IOStreams{In: os.Stdin, Out: os.Stdout}
-	u, err := url.Parse(dumpFile)
-	if err != nil {
-		fmt.Printf("parseFilePath: unable parse file path for dumpfile %s", dumpFile)
-		log.Fatal(err)
-	}
-	if (driver == constants.PGDUMP || driver == constants.MYSQLDUMP) && dumpFile != "" {
-		fmt.Printf("\nLoading dump file from path: %s\n", dumpFile)
-		var f *os.File
-		var err error
-		if u.Scheme == "gs" {
-			bucketName := u.Host
-			filePath := u.Path[1:] // removes "/" from beginning of path
-			f, err = downloadFromGCS(bucketName, filePath)
-		} else {
-			f, err = os.Open(dumpFile)
-		}
-		if err != nil {
-			fmt.Printf("\nError reading dump file: %v err:%v\n", dumpFile, err)
-			log.Fatal(err)
-		}
-		io.In = f
-	}
-	return io
-}
-
-func schemaFromDump(driver string, targetDb string, ioHelper *IOStreams) (*internal.Conv, error) {
+func schemaFromDump(driver string, targetDb string, ioHelper *utils.IOStreams) (*internal.Conv, error) {
 	f, n, err := getSeekable(ioHelper.In)
 	if err != nil {
-		printSeekError(driver, err, ioHelper.Out)
+		utils.PrintSeekError(driver, err, ioHelper.Out)
 		return nil, fmt.Errorf("can't get seekable input file")
 	}
 	ioHelper.SeekableIn = f
@@ -357,7 +218,7 @@ func schemaFromDump(driver string, targetDb string, ioHelper *IOStreams) (*inter
 	return conv, nil
 }
 
-func dataFromDump(driver string, config spanner.BatchWriterConfig, ioHelper *IOStreams, client *sp.Client, conv *internal.Conv, dataOnly bool) (*spanner.BatchWriter, error) {
+func dataFromDump(driver string, config spanner.BatchWriterConfig, ioHelper *utils.IOStreams, client *sp.Client, conv *internal.Conv, dataOnly bool) (*spanner.BatchWriter, error) {
 	// TODO: refactor of the way we handle getSeekable
 	// to avoid the code duplication here
 	if !dataOnly {
@@ -371,7 +232,7 @@ func dataFromDump(driver string, config spanner.BatchWriterConfig, ioHelper *IOS
 		// changes in showing progress for data migration.
 		f, n, err := getSeekable(ioHelper.In)
 		if err != nil {
-			printSeekError(driver, err, ioHelper.Out)
+			utils.PrintSeekError(driver, err, ioHelper.Out)
 			return nil, fmt.Errorf("can't get seekable input file")
 		}
 		ioHelper.SeekableIn = f
@@ -445,7 +306,7 @@ func Report(driver string, badWrites map[string]int64, BytesRead int64, banner s
 func getSeekable(f *os.File) (*os.File, int64, error) {
 	_, err := f.Seek(0, 0)
 	if err == nil { // Stdin is seekable, let's just use that. This happens when you run 'cmd < file'.
-		n, err := getSize(f)
+		n, err := utils.GetFileSize(f)
 		return f, n, err
 	}
 	internal.VerbosePrintln("Creating a tmp file with a copy of stdin because stdin is not seekable.")
@@ -468,7 +329,7 @@ func getSeekable(f *os.File) (*os.File, int64, error) {
 	if err != nil {
 		return nil, 0, fmt.Errorf("can't reset file offset: %w", err)
 	}
-	n, _ := getSize(fcopy)
+	n, _ := utils.GetFileSize(fcopy)
 	return fcopy, n, nil
 }
 
@@ -488,7 +349,7 @@ func VerifyDb(ctx context.Context, adminClient *database.DatabaseAdminClient, db
 func CheckExistingDb(ctx context.Context, adminClient *database.DatabaseAdminClient, dbURI string) (bool, error) {
 	_, err := adminClient.GetDatabase(ctx, &adminpb.GetDatabaseRequest{Name: dbURI})
 	if err != nil {
-		if containsAny(strings.ToLower(err.Error()), []string{"database not found"}) {
+		if utils.ContainsAny(strings.ToLower(err.Error()), []string{"database not found"}) {
 			return false, nil
 		}
 		return false, fmt.Errorf("can't get database info: %s", err)
@@ -534,7 +395,7 @@ func CreateOrUpdateDatabase(ctx context.Context, adminClient *database.DatabaseA
 // Spanner instance to use, generates a new Spanner DB name,
 // and call into the Spanner admin interface to create the new DB.
 func CreateDatabase(ctx context.Context, adminClient *database.DatabaseAdminClient, dbURI string, conv *internal.Conv, out *os.File) error {
-	project, instance, dbName := parseDbURI(dbURI)
+	project, instance, dbName := utils.ParseDbURI(dbURI)
 	fmt.Fprintf(out, "Creating new database %s in instance %s with default permissions ... \n", dbName, instance)
 	// The schema we send to Spanner excludes comments (since Cloud
 	// Spanner DDL doesn't accept them), and protects table and col names
@@ -557,10 +418,10 @@ func CreateDatabase(ctx context.Context, adminClient *database.DatabaseAdminClie
 
 	op, err := adminClient.CreateDatabase(ctx, req)
 	if err != nil {
-		return fmt.Errorf("can't build CreateDatabaseRequest: %w", AnalyzeError(err, dbURI))
+		return fmt.Errorf("can't build CreateDatabaseRequest: %w", utils.AnalyzeError(err, dbURI))
 	}
 	if _, err := op.Wait(ctx); err != nil {
-		return fmt.Errorf("createDatabase call failed: %w", AnalyzeError(err, dbURI))
+		return fmt.Errorf("createDatabase call failed: %w", utils.AnalyzeError(err, dbURI))
 	}
 	fmt.Fprintf(out, "Created database successfully.\n")
 
@@ -585,48 +446,15 @@ func UpdateDatabase(ctx context.Context, adminClient *database.DatabaseAdminClie
 	}
 	op, err := adminClient.UpdateDatabaseDdl(ctx, req)
 	if err != nil {
-		return fmt.Errorf("can't build UpdateDatabaseDdlRequest: %w", AnalyzeError(err, dbURI))
+		return fmt.Errorf("can't build UpdateDatabaseDdlRequest: %w", utils.AnalyzeError(err, dbURI))
 	}
 	if err := op.Wait(ctx); err != nil {
-		return fmt.Errorf("UpdateDatabaseDdl call failed: %w", AnalyzeError(err, dbURI))
+		return fmt.Errorf("UpdateDatabaseDdl call failed: %w", utils.AnalyzeError(err, dbURI))
 	}
 	fmt.Fprintf(out, "Updated schema successfully.\n")
 	return nil
 }
 
-// parseURI parses an unknown URI string that could be a database, instance or project URI.
-func parseURI(URI string) (project, instance, dbName string) {
-	project, instance, dbName = "", "", ""
-	if strings.Contains(URI, "databases") {
-		project, instance, dbName = parseDbURI(URI)
-	} else if strings.Contains(URI, "instances") {
-		project, instance = parseInstanceURI(URI)
-	} else if strings.Contains(URI, "projects") {
-		project = parseProjectURI(URI)
-	}
-	return
-}
-
-func parseDbURI(dbURI string) (project, instance, dbName string) {
-	split := strings.Split(dbURI, "/databases/")
-	project, instance = parseInstanceURI(split[0])
-	dbName = split[1]
-	return
-}
-
-func parseInstanceURI(instanceURI string) (project, instance string) {
-	split := strings.Split(instanceURI, "/instances/")
-	project = parseProjectURI(split[0])
-	instance = split[1]
-	return
-}
-
-func parseProjectURI(projectURI string) (project string) {
-	split := strings.Split(projectURI, "/")
-	project = split[1]
-	return
-}
-
 // UpdateDDLForeignKeys updates the Spanner database with foreign key
 // constraints using ALTER TABLE statements.
 func UpdateDDLForeignKeys(ctx context.Context, adminClient *database.DatabaseAdminClient, dbURI string, conv *internal.Conv, out *os.File) error {
@@ -698,72 +526,6 @@ Recommended value is between 20-30.`)
 	return nil
 }
 
-// GetProject returns the cloud project we should use for accessing Spanner.
-// Use environment variable GCLOUD_PROJECT if it is set.
-// Otherwise, use the default project returned from gcloud.
-func GetProject() (string, error) {
-	project := os.Getenv("GCLOUD_PROJECT")
-	if project != "" {
-		return project, nil
-	}
-	cmd := exec.Command("gcloud", "config", "list", "--format", "value(core.project)")
-	out, err := cmd.CombinedOutput()
-	if err != nil {
-		return "", fmt.Errorf("call to gcloud to get project failed: %w", err)
-	}
-	project = strings.TrimSpace(string(out))
-	return project, nil
-}
-
-// GetInstance returns the Spanner instance we should use for creating DBs.
-// If the user specified instance (via flag 'instance') then use that.
-// Otherwise try to deduce the instance using gcloud.
-func GetInstance(ctx context.Context, project string, out *os.File) (string, error) {
-	l, err := getInstances(ctx, project)
-	if err != nil {
-		return "", err
-	}
-	if len(l) == 0 {
-		fmt.Fprintf(out, "Could not find any Spanner instances for project %s\n", project)
-		return "", fmt.Errorf("no Spanner instances for %s", project)
-	}
-
-	// Note: we could ask for user input to select/confirm which Spanner
-	// instance to use, but that interacts poorly with piping pg_dump/mysqldump data
-	// to the tool via stdin.
-	if len(l) == 1 {
-		fmt.Fprintf(out, "Using only available Spanner instance: %s\n", l[0])
-		return l[0], nil
-	}
-	fmt.Fprintf(out, "Available Spanner instances:\n")
-	for i, x := range l {
-		fmt.Fprintf(out, " %d) %s\n", i+1, x)
-	}
-	fmt.Fprintf(out, "Please pick one of the available instances and set the flag '--instance'\n\n")
-	return "", fmt.Errorf("auto-selection of instance failed: project %s has more than one Spanner instance. "+
-		"Please use the flag '--instance' to select an instance", project)
-}
-
-func getInstances(ctx context.Context, project string) ([]string, error) {
-	instanceClient, err := instance.NewInstanceAdminClient(ctx)
-	if err != nil {
-		return nil, AnalyzeError(err, fmt.Sprintf("projects/%s", project))
-	}
-	it := instanceClient.ListInstances(ctx, &instancepb.ListInstancesRequest{Parent: fmt.Sprintf("projects/%s", project)})
-	var l []string
-	for {
-		resp, err := it.Next()
-		if err == iterator.Done {
-			break
-		}
-		if err != nil {
-			return nil, AnalyzeError(err, fmt.Sprintf("projects/%s", project))
-		}
-		l = append(l, strings.TrimPrefix(resp.Name, fmt.Sprintf("projects/%s/instances/", project)))
-	}
-	return l, nil
-}
-
 // WriteSchemaFile writes DDL statements in a file. It includes CREATE TABLE
 // statements and ALTER TABLE statements to add foreign keys.
 // The parameter name should end with a .txt.
@@ -878,7 +640,7 @@ func ReadSessionFile(conv *internal.Conv, sessionJSON string) error {
 // to file 'name'.
 func WriteBadData(bw *spanner.BatchWriter, conv *internal.Conv, banner, name string, out *os.File) {
 	badConversions := conv.BadRows()
-	badWrites := sum(bw.DroppedRowsByTable())
+	badWrites := utils.SumMapValues(bw.DroppedRowsByTable())
 	if badConversions == 0 && badWrites == 0 {
 		os.Remove(name) // Cleanup bad-data file from previous run.
 		return
@@ -923,167 +685,6 @@ func WriteBadData(bw *spanner.BatchWriter, conv *internal.Conv, banner, name str
 	fmt.Fprintf(out, "See file '%s' for details of bad rows\n", name)
 }
 
-// GetDatabaseName generates database name with driver_date prefix.
-func GetDatabaseName(driver string, now time.Time) (string, error) {
-	return generateName(fmt.Sprintf("%s_%s", driver, now.Format("2006-01-02")))
-}
-
-func GetPassword() string {
-	fmt.Print("Enter Password: ")
-	bytePassword, err := terminal.ReadPassword(int(syscall.Stdin))
-	if err != nil {
-		fmt.Println("\nCoudln't read password")
-		return ""
-	}
-	fmt.Printf("\n")
-	return strings.TrimSpace(string(bytePassword))
-}
-
-// AnalyzeError inspects an error returned from Cloud Spanner and adds information
-// about potential root causes e.g. authentication issues.
-func AnalyzeError(err error, URI string) error {
-	project, instance, _ := parseURI(URI)
-	e := strings.ToLower(err.Error())
-	if containsAny(e, []string{"unauthenticated", "cannot fetch token", "default credentials"}) {
-		return fmt.Errorf("%w."+`
-Possible cause: credentials are mis-configured. Do you need to run
-
-  gcloud auth application-default login
-
-or configure environment variable GOOGLE_APPLICATION_CREDENTIALS.
-See https://cloud.google.com/docs/authentication/getting-started`, err)
-	}
-	if containsAny(e, []string{"instance not found"}) && instance != "" {
-		return fmt.Errorf("%w.\n"+`
-Possible cause: Spanner instance specified via instance option does not exist.
-Please check that '%s' is correct and that it is a valid Spanner
-instance for project %s`, err, instance, project)
-	}
-	return err
-}
-
-// PrintPermissionsWarning prints permission warning.
-func PrintPermissionsWarning(driver string, out *os.File) {
-	fmt.Fprintf(out,
-		`
-WARNING: Please check that permissions for this Spanner instance are
-appropriate. Spanner manages access control at the database level, and the
-database created by HarbourBridge will inherit default permissions from this
-instance. All data written to Spanner will be visible to anyone who can
-access the created database. Note that `+driver+` table-level and row-level
-ACLs are dropped during conversion since they are not supported by Spanner.
-
-`)
-}
-
-func printSeekError(driver string, err error, out *os.File) {
-	fmt.Fprintf(out, "\nCan't get seekable input file: %v\n", err)
-	fmt.Fprintf(out, "Likely cause: not enough space in %s.\n", os.TempDir())
-	fmt.Fprintf(out, "Try writing "+driver+" output to a file first i.e.\n")
-	fmt.Fprintf(out, " "+driver+" > tmpfile\n")
-	fmt.Fprintf(out, "  harbourbridge < tmpfile\n")
-}
-
-func containsAny(s string, l []string) bool {
-	for _, a := range l {
-		if strings.Contains(s, a) {
-			return true
-		}
-	}
-	return false
-}
-
-func generateName(prefix string) (string, error) {
-	b := make([]byte, 4)
-	_, err := rand.Read(b)
-	if err != nil {
-		return "", fmt.Errorf("error generating name: %w", err)
-
-	}
-	return fmt.Sprintf("%s_%x-%x", prefix, b[0:2], b[2:4]), nil
-}
-
-// NewSpannerClient returns a new Spanner client.
-// It respects SPANNER_API_ENDPOINT.
-func NewSpannerClient(ctx context.Context, db string) (*sp.Client, error) {
-	if endpoint := os.Getenv("SPANNER_API_ENDPOINT"); endpoint != "" {
-		return sp.NewClient(ctx, db, option.WithEndpoint(endpoint))
-	}
-	return sp.NewClient(ctx, db)
-}
-
-// GetClient returns a new Spanner client.  It uses the background context.
-func GetClient(ctx context.Context, db string) (*sp.Client, error) {
-	return NewSpannerClient(ctx, db)
-}
-
-// NewDatabaseAdminClient returns a new db-admin client.
-// It respects SPANNER_API_ENDPOINT.
-func NewDatabaseAdminClient(ctx context.Context) (*database.DatabaseAdminClient, error) {
-	if endpoint := os.Getenv("SPANNER_API_ENDPOINT"); endpoint != "" {
-		return database.NewDatabaseAdminClient(ctx, option.WithEndpoint(endpoint))
-	}
-	return database.NewDatabaseAdminClient(ctx)
-}
-
-// NewInstanceAdminClient returns a new instance-admin client.
-// It respects SPANNER_API_ENDPOINT.
-func NewInstanceAdminClient(ctx context.Context) (*instance.InstanceAdminClient, error) {
-	if endpoint := os.Getenv("SPANNER_API_ENDPOINT"); endpoint != "" {
-		return instance.NewInstanceAdminClient(ctx, option.WithEndpoint(endpoint))
-	}
-	return instance.NewInstanceAdminClient(ctx)
-}
-
-func getSize(f *os.File) (int64, error) {
-	info, err := f.Stat()
-	if err != nil {
-		return 0, fmt.Errorf("can't stat file: %w", err)
-	}
-	return info.Size(), nil
-}
-
-// SetupLogFile configures the file used for logs.
-// By default we just drop logs on the floor. To enable them (e.g. to debug
-// Cloud Spanner client library issues), set logfile to a non-empty filename.
-// Note: this tool itself doesn't generate logs, but some of the libraries it
-// uses do. If we don't set the log file, we see a number of unhelpful and
-// unactionable logs spamming stdout, which is annoying and confusing.
-func SetupLogFile() (*os.File, error) {
-	// To enable debug logs, set logfile to a non-empty filename.
-	logfile := ""
-	if logfile == "" {
-		log.SetOutput(ioutil.Discard)
-		return nil, nil
-	}
-	f, err := os.Create(logfile)
-	if err != nil {
-		return nil, err
-	}
-	log.SetOutput(f)
-	return f, nil
-}
-
-// Close closes file.
-func Close(f *os.File) {
-	if f != nil {
-		f.Close()
-	}
-}
-
-func sum(m map[string]int64) int64 {
-	n := int64(0)
-	for _, c := range m {
-		n += c
-	}
-	return n
-}
-
-// GetBanner prints banner message after command line process is finished.
-func GetBanner(now time.Time, db string) string {
-	return fmt.Sprintf("Generated at %s for db %s\n\n", now.Format("2006-01-02 15:04:05"), db)
-}
-
 // ProcessDump invokes process dump function from a sql package based on driver selected.
 func ProcessDump(driver string, conv *internal.Conv, r *internal.Reader) error {
 	switch driver {
diff --git a/main.go b/main.go
index ffcfb8f..1f90c49 100644
--- a/main.go
+++ b/main.go
@@ -30,7 +30,7 @@ import (
 
 	"github.com/cloudspannerecosystem/harbourbridge/cmd"
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
-	"github.com/cloudspannerecosystem/harbourbridge/conversion"
+	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/web"
 	"github.com/google/subcommands"
@@ -91,12 +91,12 @@ Sample usage:
 
 func main() {
 	ctx := context.Background()
-	lf, err := conversion.SetupLogFile()
+	lf, err := utils.SetupLogFile()
 	if err != nil {
 		fmt.Printf("\nCan't set up log file: %v\n", err)
 		panic(fmt.Errorf("can't set up log file"))
 	}
-	defer conversion.Close(lf)
+	defer utils.Close(lf)
 
 	// TODO: Remove this check and always run HB in subcommands mode once
 	// global command line mode is deprecated. We can also enable support for
@@ -149,11 +149,11 @@ func main() {
 	}
 	fmt.Printf("Using driver (source DB): %s target-db: %s\n", driverName, targetDb)
 
-	ioHelper := conversion.NewIOStreams(driverName, dumpFilePath)
+	ioHelper := utils.NewIOStreams(driverName, dumpFilePath)
 
 	var project, instance string
 	if !schemaOnly {
-		project, err = conversion.GetProject()
+		project, err = utils.GetProject()
 		if err != nil {
 			fmt.Printf("\nCan't get project: %v\n", err)
 			panic(fmt.Errorf("can't get project"))
@@ -162,20 +162,20 @@ func main() {
 
 		instance = instanceOverride
 		if instance == "" {
-			instance, err = conversion.GetInstance(ctx, project, ioHelper.Out)
+			instance, err = utils.GetInstance(ctx, project, ioHelper.Out)
 			if err != nil {
 				fmt.Printf("\nCan't get instance: %v\n", err)
 				panic(fmt.Errorf("can't get instance"))
 			}
 		}
 		fmt.Println("Using Cloud Spanner instance:", instance)
-		conversion.PrintPermissionsWarning(driverName, ioHelper.Out)
+		utils.PrintPermissionsWarning(driverName, ioHelper.Out)
 	}
 
 	now := time.Now()
 	dbName := dbNameOverride
 	if dbName == "" {
-		dbName, err = conversion.GetDatabaseName(driverName, now)
+		dbName, err = utils.GetDatabaseName(driverName, now)
 		if err != nil {
 			fmt.Printf("\nCan't get database name: %v\n", err)
 			panic(fmt.Errorf("can't get database name"))
diff --git a/profiles/common.go b/profiles/common.go
new file mode 100644
index 0000000..5bd4832
--- /dev/null
+++ b/profiles/common.go
@@ -0,0 +1,151 @@
+package profiles
+
+import (
+	"context"
+	"encoding/csv"
+	"fmt"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
+)
+
+// Parses input string `s` as a map of key-value pairs. It's expected that the
+// input string is of the form "key1=value1,key2=value2,..." etc. Return error
+// otherwise.
+func parseProfile(s string) (map[string]string, error) {
+	params := make(map[string]string)
+	if len(s) == 0 {
+		return params, nil
+	}
+
+	// We use CSV reader to parse key=value pairs separated by a comma to
+	// handle the case where a value may contain a comma within a quote. We
+	// expect exactly one record to be returned.
+	r := csv.NewReader(strings.NewReader(s))
+	r.Comma = ','
+	r.TrimLeadingSpace = true
+	records, err := r.ReadAll()
+	if err != nil {
+		return params, err
+	}
+	if len(records) > 1 {
+		return params, fmt.Errorf("contains invalid newline characters")
+	}
+
+	for _, kv := range records[0] {
+		s := strings.Split(strings.TrimSpace(kv), "=")
+		if len(s) != 2 {
+			return params, fmt.Errorf("invalid key=value pair (expected format: key1=value1): %v", kv)
+		}
+		if _, ok := params[s[0]]; ok {
+			return params, fmt.Errorf("duplicate key found: %v", s[0])
+		}
+		params[s[0]] = s[1]
+	}
+	return params, nil
+}
+
+func GetResourceIds(ctx context.Context, targetProfile TargetProfile, now time.Time, driverName string, out *os.File) (string, string, string, error) {
+	var err error
+	project := targetProfile.conn.sp.project
+	if project == "" {
+		project, err = utils.GetProject()
+		if err != nil {
+			return "", "", "", fmt.Errorf("can't get project: %v", err)
+		}
+	}
+	fmt.Println("Using Google Cloud project:", project)
+
+	instance := targetProfile.conn.sp.instance
+	if instance == "" {
+		instance, err = utils.GetInstance(ctx, project, out)
+		if err != nil {
+			return "", "", "", fmt.Errorf("can't get instance: %v", err)
+		}
+	}
+	fmt.Println("Using Cloud Spanner instance:", instance)
+	utils.PrintPermissionsWarning(driverName, out)
+
+	dbName := targetProfile.conn.sp.dbname
+	if dbName == "" {
+		dbName, err = utils.GetDatabaseName(driverName, now)
+		if err != nil {
+			return "", "", "", fmt.Errorf("can't get database name: %v", err)
+		}
+	}
+	return project, instance, dbName, err
+}
+
+func GetSQLConnectionStr(sourceProfile SourceProfile) string {
+	sqlConnectionStr := ""
+	if sourceProfile.Ty == SourceProfileTypeConnection {
+		switch sourceProfile.Conn.Ty {
+		case SourceProfileConnectionTypeMySQL:
+			connParams := sourceProfile.Conn.Mysql
+			return getMYSQLConnectionStr(connParams.host, connParams.port, connParams.user, connParams.pwd, connParams.db)
+		case SourceProfileConnectionTypePostgreSQL:
+			connParams := sourceProfile.Conn.Pg
+			return getPGSQLConnectionStr(connParams.host, connParams.port, connParams.user, connParams.pwd, connParams.db)
+		case SourceProfileConnectionTypeDynamoDB:
+			// For DynamoDB, client provided by aws-sdk reads connection credentials from env variables only.
+			// Thus, there is no need to create sqlConnectionStr for the same. We instead set the env variables
+			// programmatically if not set.
+			return ""
+		}
+	}
+	return sqlConnectionStr
+}
+
+func GeneratePGSQLConnectionStr() (string, error) {
+	server := os.Getenv("PGHOST")
+	port := os.Getenv("PGPORT")
+	user := os.Getenv("PGUSER")
+	dbname := os.Getenv("PGDATABASE")
+	if server == "" || port == "" || user == "" || dbname == "" {
+		fmt.Printf("Please specify host, port, user and database using PGHOST, PGPORT, PGUSER and PGDATABASE environment variables\n")
+		return "", fmt.Errorf("could not connect to source database")
+	}
+	password := os.Getenv("PGPASSWORD")
+	if password == "" {
+		password = utils.GetPassword()
+	}
+	return getPGSQLConnectionStr(server, port, user, password, dbname), nil
+}
+
+func getPGSQLConnectionStr(server, port, user, password, dbname string) string {
+	return fmt.Sprintf("host=%s port=%s user=%s password=%s dbname=%s sslmode=disable", server, port, user, password, dbname)
+}
+
+func GenerateMYSQLConnectionStr() (string, error) {
+	server := os.Getenv("MYSQLHOST")
+	port := os.Getenv("MYSQLPORT")
+	user := os.Getenv("MYSQLUSER")
+	dbname := os.Getenv("MYSQLDATABASE")
+	if server == "" || port == "" || user == "" || dbname == "" {
+		fmt.Printf("Please specify host, port, user and database using MYSQLHOST, MYSQLPORT, MYSQLUSER and MYSQLDATABASE environment variables\n")
+		return "", fmt.Errorf("could not connect to source database")
+	}
+	password := os.Getenv("MYSQLPWD")
+	if password == "" {
+		password = utils.GetPassword()
+	}
+	return getMYSQLConnectionStr(server, port, user, password, dbname), nil
+}
+
+func getMYSQLConnectionStr(server, port, user, password, dbname string) string {
+	return fmt.Sprintf("%s:%s@tcp(%s:%s)/%s", user, password, server, port, dbname)
+}
+
+func GetSchemaSampleSize(sourceProfile SourceProfile) int64 {
+	schemaSampleSize := int64(100000)
+	if sourceProfile.Ty == SourceProfileTypeConnection {
+		if sourceProfile.Conn.Ty == SourceProfileConnectionTypeDynamoDB {
+			if sourceProfile.Conn.Dydb.schemaSampleSize != 0 {
+				schemaSampleSize = sourceProfile.Conn.Dydb.schemaSampleSize
+			}
+		}
+	}
+	return schemaSampleSize
+}
diff --git a/cmd/source_profile.go b/profiles/source_profile.go
similarity index 91%
rename from cmd/source_profile.go
rename to profiles/source_profile.go
index 84d09cd..10f3368 100644
--- a/cmd/source_profile.go
+++ b/profiles/source_profile.go
@@ -1,4 +1,4 @@
-package cmd
+package profiles
 
 import (
 	"fmt"
@@ -7,7 +7,7 @@ import (
 	"strings"
 
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
-	"github.com/cloudspannerecosystem/harbourbridge/conversion"
+	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 )
 
 type SourceProfileType int
@@ -20,21 +20,21 @@ const (
 )
 
 type SourceProfileFile struct {
-	path   string
-	format string
+	Path   string
+	Format string
 }
 
 func NewSourceProfileFile(params map[string]string) SourceProfileFile {
 	profile := SourceProfileFile{}
 	if !filePipedToStdin() {
-		profile.path = params["file"]
+		profile.Path = params["file"]
 	}
 	if format, ok := params["format"]; ok {
-		profile.format = format
+		profile.Format = format
 		// TODO: Add check that format takes values from ["dump", "csv", "avro", ... etc]
 	} else {
 		fmt.Printf("source-profile format defaulting to `dump`\n")
-		profile.format = "dump"
+		profile.Format = "dump"
 	}
 	return profile
 }
@@ -101,7 +101,7 @@ func NewSourceProfileConnectionMySQL(params map[string]string) (SourceProfileCon
 		mysql.port = "3306"
 	}
 	if mysql.pwd == "" {
-		mysql.pwd = conversion.GetPassword()
+		mysql.pwd = utils.GetPassword()
 	}
 
 	return mysql, nil
@@ -154,7 +154,7 @@ func NewSourceProfileConnectionPostgreSQL(params map[string]string) (SourceProfi
 		pg.port = "5432"
 	}
 	if pg.pwd == "" {
-		pg.pwd = conversion.GetPassword()
+		pg.pwd = utils.GetPassword()
 	}
 
 	return pg, nil
@@ -200,10 +200,10 @@ func NewSourceProfileConnectionDynamoDB(params map[string]string) (SourceProfile
 }
 
 type SourceProfileConnection struct {
-	ty    SourceProfileConnectionType
-	mysql SourceProfileConnectionMySQL
-	pg    SourceProfileConnectionPostgreSQL
-	dydb  SourceProfileConnectionDynamoDB
+	Ty    SourceProfileConnectionType
+	Mysql SourceProfileConnectionMySQL
+	Pg    SourceProfileConnectionPostgreSQL
+	Dydb  SourceProfileConnectionDynamoDB
 }
 
 func NewSourceProfileConnection(source string, params map[string]string) (SourceProfileConnection, error) {
@@ -212,24 +212,24 @@ func NewSourceProfileConnection(source string, params map[string]string) (Source
 	switch strings.ToLower(source) {
 	case "mysql":
 		{
-			conn.ty = SourceProfileConnectionTypeMySQL
-			conn.mysql, err = NewSourceProfileConnectionMySQL(params)
+			conn.Ty = SourceProfileConnectionTypeMySQL
+			conn.Mysql, err = NewSourceProfileConnectionMySQL(params)
 			if err != nil {
 				return conn, err
 			}
 		}
 	case "postgresql", "postgres", "pg":
 		{
-			conn.ty = SourceProfileConnectionTypePostgreSQL
-			conn.pg, err = NewSourceProfileConnectionPostgreSQL(params)
+			conn.Ty = SourceProfileConnectionTypePostgreSQL
+			conn.Pg, err = NewSourceProfileConnectionPostgreSQL(params)
 			if err != nil {
 				return conn, err
 			}
 		}
 	case "dynamodb":
 		{
-			conn.ty = SourceProfileConnectionTypeDynamoDB
-			conn.dydb, err = NewSourceProfileConnectionDynamoDB(params)
+			conn.Ty = SourceProfileConnectionTypeDynamoDB
+			conn.Dydb, err = NewSourceProfileConnectionDynamoDB(params)
 			if err != nil {
 				return conn, err
 			}
@@ -249,10 +249,10 @@ func NewSourceProfileConfig(path string) SourceProfileConfig {
 }
 
 type SourceProfile struct {
-	ty     SourceProfileType
-	file   SourceProfileFile
-	conn   SourceProfileConnection
-	config SourceProfileConfig
+	Ty     SourceProfileType
+	File   SourceProfileFile
+	Conn   SourceProfileConnection
+	Config SourceProfileConfig
 }
 
 // ToLegacyDriver converts source-profile to equivalent legacy global flags
@@ -260,7 +260,7 @@ type SourceProfile struct {
 // same. TODO: Deprecate this function and pass around SourceProfile across the
 // codebase wherever information about source connection is required.
 func (src SourceProfile) ToLegacyDriver(source string) (string, error) {
-	switch src.ty {
+	switch src.Ty {
 	case SourceProfileTypeFile:
 		{
 			switch strings.ToLower(source) {
@@ -322,19 +322,19 @@ func NewSourceProfile(s string, source string) (SourceProfile, error) {
 
 	if _, ok := params["file"]; ok || filePipedToStdin() {
 		profile := NewSourceProfileFile(params)
-		return SourceProfile{ty: SourceProfileTypeFile, file: profile}, nil
+		return SourceProfile{Ty: SourceProfileTypeFile, File: profile}, nil
 	} else if format, ok := params["format"]; ok {
 		// File is not passed in from stdin or specified using "file" flag.
-		return SourceProfile{ty: SourceProfileTypeFile}, fmt.Errorf("file not specified, but format set to %v", format)
+		return SourceProfile{Ty: SourceProfileTypeFile}, fmt.Errorf("file not specified, but format set to %v", format)
 	} else if file, ok := params["config"]; ok {
 		config := NewSourceProfileConfig(file)
-		return SourceProfile{ty: SourceProfileTypeConfig, config: config}, fmt.Errorf("source-profile type config not yet implemented")
+		return SourceProfile{Ty: SourceProfileTypeConfig, Config: config}, fmt.Errorf("source-profile type config not yet implemented")
 	} else {
 		// Assume connection profile type connection by default, since
 		// connection parameters could be specified as part of environment
 		// variables.
 		conn, err := NewSourceProfileConnection(source, params)
-		return SourceProfile{ty: SourceProfileTypeConnection, conn: conn}, err
+		return SourceProfile{Ty: SourceProfileTypeConnection, Conn: conn}, err
 	}
 }
 
diff --git a/cmd/target_profile.go b/profiles/target_profile.go
similarity index 99%
rename from cmd/target_profile.go
rename to profiles/target_profile.go
index e5e5b09..e405d81 100644
--- a/cmd/target_profile.go
+++ b/profiles/target_profile.go
@@ -1,4 +1,4 @@
-package cmd
+package profiles
 
 import (
 	"fmt"
diff --git a/web/session.go b/web/session.go
index 6f66894..c30262d 100644
--- a/web/session.go
+++ b/web/session.go
@@ -22,6 +22,7 @@ import (
 	"os"
 	"time"
 
+	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/conversion"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 )
@@ -37,12 +38,12 @@ type session struct {
 }
 
 func createSession(w http.ResponseWriter, r *http.Request) {
-	ioHelper := &conversion.IOStreams{In: os.Stdin, Out: os.Stdout}
+	ioHelper := &utils.IOStreams{In: os.Stdin, Out: os.Stdout}
 	now := time.Now()
 	dbName := sessionState.dbName
 	var err error
 	if dbName == "" {
-		dbName, err = conversion.GetDatabaseName(sessionState.driver, now)
+		dbName, err = utils.GetDatabaseName(sessionState.driver, now)
 		if err != nil {
 			http.Error(w, fmt.Sprintf("Can not create database name : %v", err), http.StatusInternalServerError)
 		}
diff --git a/web/web.go b/web/web.go
index 8a2a882..f96288c 100644
--- a/web/web.go
+++ b/web/web.go
@@ -35,6 +35,7 @@ import (
 	"time"
 
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
+	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/conversion"
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
 	"github.com/cloudspannerecosystem/harbourbridge/sources/common"
@@ -167,7 +168,7 @@ func convertSchemaDump(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, fmt.Sprintf("failed to open dump file %v : %v", dc.FilePath, err), http.StatusNotFound)
 		return
 	}
-	conv, err := conversion.SchemaConv(dc.Driver, "", constants.TargetSpanner, &conversion.IOStreams{In: f, Out: os.Stdout}, 0)
+	conv, err := conversion.SchemaConv(dc.Driver, "", constants.TargetSpanner, &utils.IOStreams{In: f, Out: os.Stdout}, 0)
 	if err != nil {
 		http.Error(w, fmt.Sprintf("Schema Conversion Error : %v", err), http.StatusNotFound)
 		return
@@ -402,7 +403,7 @@ func getConversionRate(w http.ResponseWriter, r *http.Request) {
 
 // getSchemaFile generates schema file and returns file path.
 func getSchemaFile(w http.ResponseWriter, r *http.Request) {
-	ioHelper := &conversion.IOStreams{In: os.Stdin, Out: os.Stdout}
+	ioHelper := &utils.IOStreams{In: os.Stdin, Out: os.Stdout}
 	var err error
 	now := time.Now()
 	filePrefix, err := getFilePrefix(now)
@@ -421,7 +422,7 @@ func getSchemaFile(w http.ResponseWriter, r *http.Request) {
 
 // getReportFile generates report file and returns file path.
 func getReportFile(w http.ResponseWriter, r *http.Request) {
-	ioHelper := &conversion.IOStreams{In: os.Stdin, Out: os.Stdout}
+	ioHelper := &utils.IOStreams{In: os.Stdin, Out: os.Stdout}
 	var err error
 	now := time.Now()
 	filePrefix, err := getFilePrefix(now)
@@ -774,7 +775,7 @@ func dropSecondaryIndex(w http.ResponseWriter, r *http.Request) {
 // updateSessionFile updates the content of session file with
 // latest sessionState.conv while also dumping schemas and report.
 func updateSessionFile() error {
-	ioHelper := &conversion.IOStreams{In: os.Stdin, Out: os.Stdout}
+	ioHelper := &utils.IOStreams{In: os.Stdin, Out: os.Stdout}
 	_, err := conversion.WriteConvGeneratedFiles(sessionState.conv, sessionState.dbName, sessionState.driver, ioHelper.BytesRead, ioHelper.Out)
 	if err != nil {
 		return fmt.Errorf("encountered error %w. Cannot write files", err)
@@ -1093,7 +1094,7 @@ func getFilePrefix(now time.Time) (string, error) {
 	dbName := sessionState.dbName
 	var err error
 	if dbName == "" {
-		dbName, err = conversion.GetDatabaseName(sessionState.driver, now)
+		dbName, err = utils.GetDatabaseName(sessionState.driver, now)
 		if err != nil {
 			return "", fmt.Errorf("Can not create database name : %v", err)
 		}


Test patch:
diff --git a/cmd/source_profile_test.go b/profiles/source_profile_test.go
similarity index 91%
rename from cmd/source_profile_test.go
rename to profiles/source_profile_test.go
index 6562ee9..57fad6e 100644
--- a/cmd/source_profile_test.go
+++ b/profiles/source_profile_test.go
@@ -1,4 +1,4 @@
-package cmd
+package profiles
 
 import (
 	"testing"
@@ -17,31 +17,31 @@ func TestNewSourceProfileFile(t *testing.T) {
 			name:         "no params, file piped",
 			params:       map[string]string{},
 			pipedToStdin: true,
-			want:         SourceProfileFile{format: "dump"},
+			want:         SourceProfileFile{Format: "dump"},
 		},
 		{
 			name:         "format param, file piped",
-			params:       map[string]string{"format": "dump"},
+			params:       map[string]string{"Format": "dump"},
 			pipedToStdin: true,
-			want:         SourceProfileFile{format: "dump"},
+			want:         SourceProfileFile{Format: "dump"},
 		},
 		{
 			name:         "format and path param, file piped -- piped file takes precedence",
 			params:       map[string]string{"format": "dump", "file": "file1.mysqldump"},
 			pipedToStdin: true,
-			want:         SourceProfileFile{format: "dump"},
+			want:         SourceProfileFile{Format: "dump"},
 		},
 		{
 			name:         "format and path param, no file piped",
 			params:       map[string]string{"format": "dump", "file": "file1.mysqldump"},
 			pipedToStdin: false,
-			want:         SourceProfileFile{format: "dump", path: "file1.mysqldump"},
+			want:         SourceProfileFile{Format: "dump", Path: "file1.mysqldump"},
 		},
 		{
 			name:         "only path param, no file piped -- default dump format",
 			params:       map[string]string{"file": "file1.mysqldump"},
 			pipedToStdin: false,
-			want:         SourceProfileFile{format: "dump", path: "file1.mysqldump"},
+			want:         SourceProfileFile{Format: "dump", Path: "file1.mysqldump"},
 		},
 	}
 
diff --git a/testing/dynamodb/integration_test.go b/testing/dynamodb/integration_test.go
index 213c3bd..4711ba4 100644
--- a/testing/dynamodb/integration_test.go
+++ b/testing/dynamodb/integration_test.go
@@ -27,7 +27,7 @@ import (
 	"time"
 
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
-	"github.com/cloudspannerecosystem/harbourbridge/conversion"
+	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/testing/common"
 	"github.com/google/go-cmp/cmp"
 	"github.com/stretchr/testify/assert"
@@ -214,7 +214,7 @@ func TestIntegration_DYNAMODB_Command(t *testing.T) {
 	defer os.RemoveAll(tmpdir)
 
 	now := time.Now()
-	dbName, _ := conversion.GetDatabaseName(constants.DYNAMODB, now)
+	dbName, _ := utils.GetDatabaseName(constants.DYNAMODB, now)
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
 	filePrefix := filepath.Join(tmpdir, dbName+".")
 
diff --git a/testing/mysql/integration_test.go b/testing/mysql/integration_test.go
index 9ef60b2..c136718 100644
--- a/testing/mysql/integration_test.go
+++ b/testing/mysql/integration_test.go
@@ -26,7 +26,7 @@ import (
 	"time"
 
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
-	"github.com/cloudspannerecosystem/harbourbridge/conversion"
+	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/testing/common"
 
 	"cloud.google.com/go/spanner"
@@ -110,7 +110,7 @@ func TestIntegration_MYSQLDUMP_Command(t *testing.T) {
 	defer os.RemoveAll(tmpdir)
 
 	now := time.Now()
-	dbName, _ := conversion.GetDatabaseName(constants.MYSQLDUMP, now)
+	dbName, _ := utils.GetDatabaseName(constants.MYSQLDUMP, now)
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
 	dataFilepath := "../../test_data/mysqldump.test.out"
 	filePrefix := filepath.Join(tmpdir, dbName+".")
@@ -159,7 +159,7 @@ func TestIntegration_MYSQL_Command(t *testing.T) {
 	defer os.RemoveAll(tmpdir)
 
 	now := time.Now()
-	dbName, _ := conversion.GetDatabaseName(constants.MYSQL, now)
+	dbName, _ := utils.GetDatabaseName(constants.MYSQL, now)
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
 	filePrefix := filepath.Join(tmpdir, dbName+".")
 
diff --git a/testing/postgres/integration_test.go b/testing/postgres/integration_test.go
index 7673246..6696a15 100644
--- a/testing/postgres/integration_test.go
+++ b/testing/postgres/integration_test.go
@@ -30,7 +30,7 @@ import (
 	"cloud.google.com/go/spanner"
 	database "cloud.google.com/go/spanner/admin/database/apiv1"
 	"github.com/cloudspannerecosystem/harbourbridge/common/constants"
-	"github.com/cloudspannerecosystem/harbourbridge/conversion"
+	"github.com/cloudspannerecosystem/harbourbridge/common/utils"
 	"github.com/cloudspannerecosystem/harbourbridge/testing/common"
 	"google.golang.org/api/iterator"
 	databasepb "google.golang.org/genproto/googleapis/spanner/admin/database/v1"
@@ -112,7 +112,7 @@ func TestIntegration_PGDUMP_Command(t *testing.T) {
 	defer os.RemoveAll(tmpdir)
 
 	now := time.Now()
-	dbName, _ := conversion.GetDatabaseName(constants.PGDUMP, now)
+	dbName, _ := utils.GetDatabaseName(constants.PGDUMP, now)
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
 
 	dataFilepath := "../../test_data/pg_dump.test.out"
@@ -136,7 +136,7 @@ func TestIntegration_PGDUMP_SchemaAndDataSubcommand(t *testing.T) {
 	defer os.RemoveAll(tmpdir)
 
 	now := time.Now()
-	dbName, _ := conversion.GetDatabaseName(constants.PGDUMP, now)
+	dbName, _ := utils.GetDatabaseName(constants.PGDUMP, now)
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
 
 	dataFilepath := "../../test_data/pg_dump.test.out"
@@ -176,7 +176,7 @@ func TestIntegration_POSTGRES_Command(t *testing.T) {
 	defer os.RemoveAll(tmpdir)
 
 	now := time.Now()
-	dbName, _ := conversion.GetDatabaseName(constants.POSTGRES, now)
+	dbName, _ := utils.GetDatabaseName(constants.POSTGRES, now)
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
 	filePrefix := filepath.Join(tmpdir, dbName+".")
 
@@ -199,7 +199,7 @@ func TestIntegration_POSTGRES_SchemaAndDataSubcommand(t *testing.T) {
 	defer os.RemoveAll(tmpdir)
 
 	now := time.Now()
-	dbName, _ := conversion.GetDatabaseName(constants.POSTGRES, now)
+	dbName, _ := utils.GetDatabaseName(constants.POSTGRES, now)
 	dbURI := fmt.Sprintf("projects/%s/instances/%s/databases/%s", projectID, instanceID, dbName)
 	filePrefix := filepath.Join(tmpdir, dbName+".")