-
Notifications
You must be signed in to change notification settings - Fork 61
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Display equivalent gcloud cli command for Dataflow for forward migrat…
…ions (#641)
- Loading branch information
Showing
14 changed files
with
337 additions
and
13 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,111 @@ | ||
| // Copyright 2023 Google LLC | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| // Package utils contains common helper functions used across multiple other packages. | ||
| // Utils should not import any Spanner migration tool packages. | ||
| package utils | ||
|
|
||
| import ( | ||
| "encoding/json" | ||
| "fmt" | ||
| "sort" | ||
| "strings" | ||
|
|
||
| "cloud.google.com/go/dataflow/apiv1beta3/dataflowpb" | ||
| "golang.org/x/exp/maps" | ||
| ) | ||
|
|
||
| // Generate the equivalent gCloud CLI command to launch a dataflow job with the same parameters and environment flags | ||
| // as the input body. | ||
| func GetGcloudDataflowCommand(req *dataflowpb.LaunchFlexTemplateRequest) string { | ||
| lp := req.LaunchParameter | ||
| templatePath := lp.Template.(*dataflowpb.LaunchFlexTemplateParameter_ContainerSpecGcsPath).ContainerSpecGcsPath | ||
| cmd := fmt.Sprintf("gcloud dataflow flex-template run %s --project=%s --region=%s --template-file-gcs-location=%s %s %s", | ||
| lp.JobName, req.ProjectId, req.Location, templatePath, getEnvironmentFlags(lp.Environment), getParametersFlag(lp.Parameters)) | ||
| return strings.Trim(cmd, " ") | ||
| } | ||
|
|
||
| // Generate the equivalent parameter flag string, returning empty string if none are specified. | ||
| func getParametersFlag(parameters map[string]string) string { | ||
| if len(parameters) == 0 { | ||
| return "" | ||
| } | ||
| params := "" | ||
| keys := maps.Keys(parameters) | ||
| sort.Strings(keys) | ||
| for _, k := range keys { | ||
| params = params + k + "=" + parameters[k] + "," | ||
| } | ||
| params = strings.TrimSuffix(params, ",") | ||
| return fmt.Sprintf("--parameters %s", params) | ||
| } | ||
|
|
||
| // We don't populate all flags in the API because certain flags (like AutoscalingAlgorithm, DumpHeapOnOom etc.) | ||
| // are not supported in gCloud. | ||
| func getEnvironmentFlags(environment *dataflowpb.FlexTemplateRuntimeEnvironment) string { | ||
| flag := "" | ||
| if environment.NumWorkers != 0 { | ||
| flag += fmt.Sprintf("--num-workers %d ", environment.NumWorkers) | ||
| } | ||
| if environment.MaxWorkers != 0 { | ||
| flag += fmt.Sprintf("--max-workers %d ", environment.MaxWorkers) | ||
| } | ||
| if environment.ServiceAccountEmail != "" { | ||
| flag += fmt.Sprintf("--service-account-email %s ", environment.ServiceAccountEmail) | ||
| } | ||
| if environment.TempLocation != "" { | ||
| flag += fmt.Sprintf("--temp-location %s ", environment.TempLocation) | ||
| } | ||
| if environment.MachineType != "" { | ||
| flag += fmt.Sprintf("--worker-machine-type %s ", environment.MachineType) | ||
| } | ||
| if environment.AdditionalExperiments != nil && len(environment.AdditionalExperiments) > 0 { | ||
| flag += fmt.Sprintf("--additional-experiments %s ", strings.Join(environment.AdditionalExperiments, ",")) | ||
| } | ||
| if environment.Network != "" { | ||
| flag += fmt.Sprintf("--network %s ", environment.Network) | ||
| } | ||
| if environment.Subnetwork != "" { | ||
| flag += fmt.Sprintf("--subnetwork %s ", environment.Subnetwork) | ||
| } | ||
| if environment.AdditionalUserLabels != nil && len(environment.AdditionalUserLabels) > 0 { | ||
| jsonByteStr, err := json.Marshal(environment.AdditionalUserLabels) | ||
| // If error is not nil, omit this flag and move on. We don't need error handling here. | ||
| if err == nil { | ||
| flag += fmt.Sprintf("--additional-user-labels %s ", string(jsonByteStr)) | ||
| } | ||
| } | ||
| if environment.KmsKeyName != "" { | ||
| flag += fmt.Sprintf("--dataflow-kms-key %s ", environment.KmsKeyName) | ||
| } | ||
| if environment.IpConfiguration == dataflowpb.WorkerIPAddressConfiguration_WORKER_IP_PRIVATE { | ||
| flag += "--disable-public-ips " | ||
| } | ||
| if environment.WorkerRegion != "" { | ||
| flag += fmt.Sprintf("--worker-region %s ", environment.WorkerRegion) | ||
| } | ||
| if environment.WorkerZone != "" { | ||
| flag += fmt.Sprintf("--worker-zone %s ", environment.WorkerZone) | ||
| } | ||
| if environment.EnableStreamingEngine { | ||
| flag += "--enable-streaming-engine " | ||
| } | ||
| if environment.FlexrsGoal != dataflowpb.FlexResourceSchedulingGoal_FLEXRS_UNSPECIFIED { | ||
| flag += fmt.Sprintf("--flexrs-goal %s ", environment.FlexrsGoal) | ||
| } | ||
| if environment.StagingLocation != "" { | ||
| flag += fmt.Sprintf("--staging-location %s ", environment.StagingLocation) | ||
| } | ||
| return strings.Trim(flag, " ") | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,117 @@ | ||
| // Copyright 2023 Google LLC | ||
| // | ||
| // Licensed under the Apache License, Version 2.0 (the "License"); | ||
| // you may not use this file except in compliance with the License. | ||
| // You may obtain a copy of the License at | ||
| // | ||
| // http://www.apache.org/licenses/LICENSE-2.0 | ||
| // | ||
| // Unless required by applicable law or agreed to in writing, software | ||
| // distributed under the License is distributed on an "AS IS" BASIS, | ||
| // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| // See the License for the specific language governing permissions and | ||
| // limitations under the License. | ||
|
|
||
| // TODO: Refactor this file and other integration tests by moving all common code | ||
| // to remove redundancy. | ||
|
|
||
| package utils_test | ||
|
|
||
| import ( | ||
| "os" | ||
| "testing" | ||
|
|
||
| "cloud.google.com/go/dataflow/apiv1beta3/dataflowpb" | ||
| "github.com/GoogleCloudPlatform/spanner-migration-tool/common/utils" | ||
| "github.com/stretchr/testify/assert" | ||
| ) | ||
|
|
||
| func TestMain(m *testing.M) { | ||
| res := m.Run() | ||
| os.Exit(res) | ||
| } | ||
|
|
||
| func getTemplateDfRequest() *dataflowpb.LaunchFlexTemplateRequest { | ||
| launchParameters := &dataflowpb.LaunchFlexTemplateParameter{ | ||
| JobName: "test-job", | ||
| Template: &dataflowpb.LaunchFlexTemplateParameter_ContainerSpecGcsPath{ContainerSpecGcsPath: "gs://template/Cloud_Datastream_to_Spanner"}, | ||
| Parameters: map[string]string{ | ||
| "inputFilePattern": "gs://inputFilePattern", | ||
| "streamName": "my-stream", | ||
| "instanceId": "my-instance", | ||
| "databaseId": "my-dbName", | ||
| "sessionFilePath": "gs://session.json", | ||
| "deadLetterQueueDirectory": "gs://dlq", | ||
| "transformationContextFilePath": "gs://transformationContext.json", | ||
| "directoryWatchDurationInMinutes": "480", // Setting directory watch timeout to 8 hours | ||
| }, | ||
| Environment: &dataflowpb.FlexTemplateRuntimeEnvironment{ | ||
| MaxWorkers: 50, | ||
| NumWorkers: 10, | ||
| ServiceAccountEmail: "[email protected]", | ||
| TempLocation: "gs://temp-location", | ||
| MachineType: "n2-standard-16", | ||
| AdditionalExperiments: []string{"use_runner_V2", "test-experiment"}, | ||
| Network: "my-network", | ||
| Subnetwork: "my-subnetwork", | ||
| AdditionalUserLabels: map[string]string{"name": "wrench", "count": "3"}, | ||
| KmsKeyName: "sample-kms-key", | ||
| IpConfiguration: dataflowpb.WorkerIPAddressConfiguration_WORKER_IP_PRIVATE, | ||
| WorkerRegion: "test-worker-region", | ||
| WorkerZone: "test-worker-zone", | ||
| EnableStreamingEngine: true, | ||
| FlexrsGoal: 1, | ||
| StagingLocation: "gs://staging-location", | ||
| }, | ||
| } | ||
| req := &dataflowpb.LaunchFlexTemplateRequest{ | ||
| ProjectId: "test-project", | ||
| LaunchParameter: launchParameters, | ||
| Location: "us-central1", | ||
| } | ||
| return req | ||
| } | ||
|
|
||
| func TestGcloudCmdWithAllParams(t *testing.T) { | ||
|
|
||
| req := getTemplateDfRequest() | ||
| expectedCmd := "gcloud dataflow flex-template run test-job " + | ||
| "--project=test-project --region=us-central1 " + | ||
| "--template-file-gcs-location=gs://template/Cloud_Datastream_to_Spanner " + | ||
| "--num-workers 10 --max-workers 50 --service-account-email [email protected] " + | ||
| "--temp-location gs://temp-location --worker-machine-type n2-standard-16 " + | ||
| "--additional-experiments use_runner_V2,test-experiment --network my-network " + | ||
| "--subnetwork my-subnetwork --additional-user-labels {\"count\":\"3\",\"name\":\"wrench\"} " + | ||
| "--dataflow-kms-key sample-kms-key --disable-public-ips --worker-region test-worker-region " + | ||
| "--worker-zone test-worker-zone --enable-streaming-engine " + | ||
| "--flexrs-goal FLEXRS_SPEED_OPTIMIZED --staging-location gs://staging-location " + | ||
| "--parameters databaseId=my-dbName,deadLetterQueueDirectory=gs://dlq," + | ||
| "directoryWatchDurationInMinutes=480,inputFilePattern=gs://inputFilePattern," + | ||
| "instanceId=my-instance,sessionFilePath=gs://session.json,streamName=my-stream," + | ||
| "transformationContextFilePath=gs://transformationContext.json" | ||
| assert.Equal(t, expectedCmd, utils.GetGcloudDataflowCommand(req)) | ||
| } | ||
|
|
||
| func TestGcloudCmdWithPartialParams(t *testing.T) { | ||
|
|
||
| req := getTemplateDfRequest() | ||
| req.LaunchParameter.Parameters = make(map[string]string) | ||
| req.LaunchParameter.Environment.FlexrsGoal = 0 | ||
| req.LaunchParameter.Environment.IpConfiguration = 0 | ||
| req.LaunchParameter.Environment.EnableStreamingEngine = false | ||
| req.LaunchParameter.Environment.AdditionalExperiments = []string{} | ||
| req.LaunchParameter.Environment.AdditionalUserLabels = make(map[string]string) | ||
| req.LaunchParameter.Environment.WorkerRegion = "" | ||
| req.LaunchParameter.Environment.NumWorkers = 0 | ||
|
|
||
| expectedCmd := "gcloud dataflow flex-template run test-job " + | ||
| "--project=test-project --region=us-central1 " + | ||
| "--template-file-gcs-location=gs://template/Cloud_Datastream_to_Spanner " + | ||
| "--max-workers 50 --service-account-email [email protected] " + | ||
| "--temp-location gs://temp-location --worker-machine-type n2-standard-16 " + | ||
| "--network my-network --subnetwork my-subnetwork " + | ||
| "--dataflow-kms-key sample-kms-key " + | ||
| "--worker-zone test-worker-zone " + | ||
| "--staging-location gs://staging-location" | ||
| assert.Equal(t, expectedCmd, utils.GetGcloudDataflowCommand(req)) | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
12 changes: 12 additions & 0 deletions
12
ui/src/app/components/equivalent-gcloud-command/equivalent-gcloud-command.component.html
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,12 @@ | ||
| <div mat-dialog-content> | ||
| <h2>Equivalent gcloud command line<mat-icon class="configure" | ||
| matTooltip='This is the gcloud command to launch the dataflow job with the same parameters. Can be used to re-run a dataflow job manually in case of failure.'> | ||
| info</mat-icon></h2> | ||
|
|
||
| <span class="left-text">{{ gcloudCmd }}</span> | ||
|
|
||
| <div mat-dialog-actions class="buttons-container"> | ||
| <button mat-button color="primary" mat-dialog-close>Close</button> | ||
| <button mat-button color="primary" [cdkCopyToClipboard]="gcloudCmd">Copy</button> | ||
| </div> | ||
| </div> |
Empty file.
25 changes: 25 additions & 0 deletions
25
ui/src/app/components/equivalent-gcloud-command/equivalent-gcloud-command.component.spec.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,25 @@ | ||
| import { ComponentFixture, TestBed } from '@angular/core/testing'; | ||
|
|
||
| import { EquivalentGcloudCommandComponent } from './equivalent-gcloud-command.component'; | ||
|
|
||
| describe('EquivalentGcloudCommandComponent', () => { | ||
| let component: EquivalentGcloudCommandComponent; | ||
| let fixture: ComponentFixture<EquivalentGcloudCommandComponent>; | ||
|
|
||
| beforeEach(async () => { | ||
| await TestBed.configureTestingModule({ | ||
| declarations: [ EquivalentGcloudCommandComponent ] | ||
| }) | ||
| .compileComponents(); | ||
| }); | ||
|
|
||
| beforeEach(() => { | ||
| fixture = TestBed.createComponent(EquivalentGcloudCommandComponent); | ||
| component = fixture.componentInstance; | ||
| fixture.detectChanges(); | ||
| }); | ||
|
|
||
| it('should create', () => { | ||
| expect(component).toBeTruthy(); | ||
| }); | ||
| }); |
21 changes: 21 additions & 0 deletions
21
ui/src/app/components/equivalent-gcloud-command/equivalent-gcloud-command.component.ts
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,21 @@ | ||
| import { Component, Inject, OnInit } from '@angular/core'; | ||
| import { MatDialogRef, MAT_DIALOG_DATA } from '@angular/material/dialog'; | ||
|
|
||
| @Component({ | ||
| selector: 'app-equivalent-gcloud-command', | ||
| templateUrl: './equivalent-gcloud-command.component.html', | ||
| styleUrls: ['./equivalent-gcloud-command.component.scss'] | ||
| }) | ||
| export class EquivalentGcloudCommandComponent implements OnInit { | ||
| gcloudCmd: string | ||
|
|
||
| constructor( | ||
| @Inject(MAT_DIALOG_DATA) public data: string, | ||
| private dialofRef: MatDialogRef<EquivalentGcloudCommandComponent> | ||
| ) { | ||
| this.gcloudCmd = data | ||
| } | ||
|
|
||
| ngOnInit(): void { | ||
| } | ||
| } |
Oops, something went wrong.