diff --git a/profiles/source_profile.go b/profiles/source_profile.go index 8176117c1c..5966b3abb2 100644 --- a/profiles/source_profile.go +++ b/profiles/source_profile.go @@ -28,7 +28,6 @@ import ( type SourceProfileType int - const ( SourceProfileTypeUnset = iota SourceProfileTypeFile @@ -54,7 +53,7 @@ type SourceProfileDialectInterface interface { NewSourceProfileConnectionOracle(params map[string]string, g utils.GetUtilInfoInterface) (SourceProfileConnectionOracle, error) } -type SourceProfileDialectImpl struct {} +type SourceProfileDialectImpl struct{} // Interface to create new source profiles for different input types type NewSourceProfileInterface interface { @@ -101,11 +100,11 @@ const ( ) type SourceProfileConnectionCloudSQLMySQL struct { - User string - Db string + User string + Db string InstanceName string - Project string - Region string + Project string + Region string } func (spd *SourceProfileDialectImpl) NewSourceProfileConnectionCloudSQLMySQL(params map[string]string, g utils.GetUtilInfoInterface) (SourceProfileConnectionCloudSQLMySQL, error) { @@ -202,11 +201,11 @@ func (spd *SourceProfileDialectImpl) NewSourceProfileConnectionMySQL(params map[ } type SourceProfileConnectionCloudSQLPostgreSQL struct { - User string - Db string + User string + Db string InstanceName string - Project string - Region string + Project string + Region string } func (spd *SourceProfileDialectImpl) NewSourceProfileConnectionCloudSQLPostgreSQL(params map[string]string, g utils.GetUtilInfoInterface) (SourceProfileConnectionCloudSQLPostgreSQL, error) { @@ -468,9 +467,9 @@ type SourceProfileConnection struct { } type SourceProfileConnectionCloudSQL struct { - Ty SourceProfileConnectionTypeCloudSQL - Mysql SourceProfileConnectionCloudSQLMySQL - Pg SourceProfileConnectionCloudSQLPostgreSQL + Ty SourceProfileConnectionTypeCloudSQL + Mysql SourceProfileConnectionCloudSQLMySQL + Pg SourceProfileConnectionCloudSQLPostgreSQL } func (nsp *NewSourceProfileImpl) NewSourceProfileConnection(source string, params map[string]string, s SourceProfileDialectInterface) (SourceProfileConnection, error) { @@ -571,16 +570,16 @@ type DirectConnectionConfig struct { type DatastreamConnProfileSource struct { Name string `json:"name"` - Host string `json:"host"` - User string `json:"user"` - Port string `json:"port"` - Password string `json:"password"` - Location string `json:"location"` + Host string `json:"host"` + User string `json:"user"` + Port string `json:"port"` + Password string `json:"password"` + Location string `json:"location"` } type DatastreamConnProfileTarget struct { Name string `json:"name"` - Location string `json:"location"` + Location string `json:"location"` } type DatastreamConfig struct { @@ -607,18 +606,21 @@ type DataflowConfig struct { AdditionalUserLabels string `json:"additionalUserLabels"` KmsKeyName string `json:"kmsKeyName"` GcsTemplatePath string `json:"gcsTemplatePath"` + CustomJarPath string `json:"customJarPath"` + CustomClassName string `json:"customClassName"` + CustomParameter string `json:"customParameter"` } type DataShard struct { - DataShardId string `json:"dataShardId"` + DataShardId string `json:"dataShardId"` SrcConnectionProfile DatastreamConnProfileSource `json:"srcConnectionProfile"` DstConnectionProfile DatastreamConnProfileTarget `json:"dstConnectionProfile"` - DatastreamConfig DatastreamConfig `json:"datastreamConfig"` - GcsConfig GcsConfig `json:"gcsConfig"` - DataflowConfig DataflowConfig `json:"dataflowConfig"` - TmpDir string `json:"tmpDir"` - StreamLocation string `json:"streamLocation"` - LogicalShards []LogicalShard `json:"databases"` + DatastreamConfig DatastreamConfig `json:"datastreamConfig"` + GcsConfig GcsConfig `json:"gcsConfig"` + DataflowConfig DataflowConfig `json:"dataflowConfig"` + TmpDir string `json:"tmpDir"` + StreamLocation string `json:"streamLocation"` + LogicalShards []LogicalShard `json:"databases"` } type LogicalShard struct { @@ -690,13 +692,13 @@ func NewSourceProfileCsv(params map[string]string) SourceProfileCsv { } type SourceProfile struct { - Driver string - Ty SourceProfileType - File SourceProfileFile - Conn SourceProfileConnection + Driver string + Ty SourceProfileType + File SourceProfileFile + Conn SourceProfileConnection ConnCloudSQL SourceProfileConnectionCloudSQL - Config SourceProfileConfig - Csv SourceProfileCsv + Config SourceProfileConfig + Csv SourceProfileCsv } // UseTargetSchema returns true if the driver expects an existing schema diff --git a/streaming/streaming.go b/streaming/streaming.go index f93e193387..f3f28a7c71 100644 --- a/streaming/streaming.go +++ b/streaming/streaming.go @@ -153,6 +153,9 @@ type DataflowCfg struct { KmsKeyName string `json:"kmsKeyName"` GcsTemplatePath string `json:"gcsTemplatePath"` DbNameToShardIdMap map[string]string `json:"dbNameToShardIdMap"` + CustomJarPath string `json:"customJarPath"` + CustomClassName string `json:"customClassName"` + CustomParameter string `json:"customParameter"` } type StreamingCfg struct { @@ -763,6 +766,15 @@ func LaunchDataflowJob(ctx context.Context, migrationProjectId string, targetPro KmsKeyName: dataflowCfg.KmsKeyName, }, } + + if dataflowCfg.CustomClassName != "" && dataflowCfg.CustomJarPath != "" { + launchParameters.Parameters["transformationJarPath"] = dataflowCfg.CustomJarPath + launchParameters.Parameters["transformationClassName"] = dataflowCfg.CustomClassName + launchParameters.Parameters["transformationCustomParameters"] = dataflowCfg.CustomParameter + } else if (dataflowCfg.CustomClassName != "" && dataflowCfg.CustomJarPath == "") || (dataflowCfg.CustomClassName == "" && dataflowCfg.CustomJarPath != "") { + return internal.DataflowOutput{}, fmt.Errorf("specify both the custom class name and custom jar GCS path, or specify neither") + } + req := &dataflowpb.LaunchFlexTemplateRequest{ ProjectId: dataflowProjectId, LaunchParameter: launchParameters, @@ -836,6 +848,9 @@ func CreateStreamingConfig(pl profiles.DataShard) StreamingCfg { AdditionalUserLabels: inputDataflowConfig.AdditionalUserLabels, KmsKeyName: inputDataflowConfig.KmsKeyName, GcsTemplatePath: inputDataflowConfig.GcsTemplatePath, + CustomJarPath: inputDataflowConfig.CustomJarPath, + CustomClassName: inputDataflowConfig.CustomClassName, + CustomParameter: inputDataflowConfig.CustomParameter, } //create src and dst datastream from pl receiver object datastreamCfg := DatastreamCfg{ diff --git a/streaming/streaming_test.go b/streaming/streaming_test.go index 0dc8a5ea9f..46eeb22fe4 100644 --- a/streaming/streaming_test.go +++ b/streaming/streaming_test.go @@ -431,3 +431,95 @@ func TestGetUpdateDataStreamLRORetryBackoff(t *testing.T) { assert.Equal(t, backoff.MaxInterval, DEFAULT_DATASTREAM_LRO_POLL_MAX_DELAY) assert.Equal(t, backoff.MaxElapsedTime, DEFAULT_DATASTREAM_LRO_POLL_MAX_ELAPSED_TIME) } + +func TestCreateStreamingConfig(t *testing.T) { + // Mock data + inputDataflowConfig := profiles.DataflowConfig{ + ProjectId: "project-id", + Location: "us-central1", + Network: "network", + VpcHostProjectId: "vpc-host-project-id", + Subnetwork: "subnetwork", + MaxWorkers: "10", + NumWorkers: "5", + ServiceAccountEmail: "service-account-email", + MachineType: "machine-type", + AdditionalUserLabels: "", + KmsKeyName: "kms-key-name", + GcsTemplatePath: "gcs-template-path", + CustomJarPath: "custom-jar-path", + CustomClassName: "custom-class-name", + CustomParameter: "custom-parameter", + } + + inputDatastreamConfig := profiles.DatastreamConfig{ + MaxConcurrentBackfillTasks: "3", + MaxConcurrentCdcTasks: "5", + } + + inputSrcConnProfile := profiles.DatastreamConnProfileSource{ + Location: "us-central1", + Name: "src-conn-profile", + } + + inputDstConnProfile := profiles.DatastreamConnProfileTarget{ + Name: "dst-conn-profile", + Location: "us-central1", + } + + inputGcsConfig := profiles.GcsConfig{ + TtlInDays: 7, + TtlInDaysSet: true, + } + + inputDataShard := profiles.DataShard{ + DataflowConfig: inputDataflowConfig, + StreamLocation: "us-central1", + DatastreamConfig: inputDatastreamConfig, + SrcConnectionProfile: inputSrcConnProfile, + DstConnectionProfile: inputDstConnProfile, + GcsConfig: inputGcsConfig, + TmpDir: "gs://my-bucket/tmp/", + DataShardId: "data-shard-id", + } + + // Expected output + expectedStreamingCfg := StreamingCfg{ + DatastreamCfg: DatastreamCfg{ + StreamLocation: "us-central1", + MaxConcurrentBackfillTasks: "3", + MaxConcurrentCdcTasks: "5", + SourceConnectionConfig: SrcConnCfg{Location: "us-central1", Name: "src-conn-profile"}, + DestinationConnectionConfig: DstConnCfg{Name: "dst-conn-profile", Location: "us-central1"}, + }, + GcsCfg: GcsCfg{ + TtlInDays: 7, + TtlInDaysSet: true, + }, + DataflowCfg: DataflowCfg{ + ProjectId: "project-id", + Location: "us-central1", + Network: "network", + VpcHostProjectId: "vpc-host-project-id", + Subnetwork: "subnetwork", + MaxWorkers: "10", + NumWorkers: "5", + ServiceAccountEmail: "service-account-email", + MachineType: "machine-type", + AdditionalUserLabels: "", + KmsKeyName: "kms-key-name", + GcsTemplatePath: "gcs-template-path", + CustomJarPath: "custom-jar-path", + CustomClassName: "custom-class-name", + CustomParameter: "custom-parameter", + }, + TmpDir: "gs://my-bucket/tmp/", + DataShardId: "data-shard-id", + } + + // Call function + actualStreamingCfg := CreateStreamingConfig(inputDataShard) + + // Compare expected and actual output + assert.Equal(t, expectedStreamingCfg, actualStreamingCfg, "The streaming configuration should match the expected configuration") +} diff --git a/ui/dist/ui/index.html b/ui/dist/ui/index.html index 56f2228dce..b84774b8a9 100644 --- a/ui/dist/ui/index.html +++ b/ui/dist/ui/index.html @@ -12,5 +12,5 @@