Repository: cloudspannerecosystem/harbourbridge
Pull request #327 (2022-06-29 05:47:21)

Added support for setting up DynamoDB Streams
**Things added**
- Support for setting up DynamoDB Streams used for Change Data Capture on the DynamoDB database. Unit tests are included.
- `enableStreaming` parameter in DynamoDB source-profile flag to start the streaming migration process. To use the streaming migration option follow all the bulk migration steps and specify a valid choice for `enableStreaming`.
```sh
harbourbridge schema-and-data -source=dynamodb -source-profile="aws-access-key-id=<>,...,enableStreaming=<>" -target-profile="instance=my-spanner-instance"
```
Valid choices for **enableStreaming**: yes, no, true, false

**Things not handled**
- This PR includes only DynamoDB Streams initialization. Reading data from DynamoDB Streams and writing to Cloud Spanner will be added in subsequent PRs.
- Changes to README will be added after complete streaming migration solution is ready.

- [x] Tests pass


Patch:
diff --git a/profiles/source_profile.go b/profiles/source_profile.go
index ef4d47a..ffb7d78 100644
--- a/profiles/source_profile.go
+++ b/profiles/source_profile.go
@@ -261,6 +261,7 @@ type SourceProfileConnectionDynamoDB struct {
 	AwsRegion          string // Same as AWS_REGION environment variable
 	DydbEndpoint       string // Same as DYNAMODB_ENDPOINT_OVERRIDE environment variable
 	SchemaSampleSize   int64  // Number of rows to use for inferring schema (default 100,000)
+	enableStreaming    string // Used for confirming streaming migration (valid options: `yes`,`no`,`true`,`false`)
 }
 
 func NewSourceProfileConnectionDynamoDB(params map[string]string) (SourceProfileConnectionDynamoDB, error) {
@@ -288,6 +289,16 @@ func NewSourceProfileConnectionDynamoDB(params map[string]string) (SourceProfile
 	if dydb.DydbEndpoint, ok = params["dydb-endpoint"]; ok {
 		os.Setenv("DYNAMODB_ENDPOINT_OVERRIDE", dydb.DydbEndpoint)
 	}
+	if dydb.enableStreaming, ok = params["enableStreaming"]; ok {
+		switch dydb.enableStreaming {
+		case "yes", "true":
+			dydb.enableStreaming = "yes"
+		case "no", "false":
+			dydb.enableStreaming = "no"
+		default:
+			return dydb, fmt.Errorf("please specify a valid choice for enableStreaming: available choices(yes, no, true, false)")
+		}
+	}
 	return dydb, nil
 }
 
@@ -377,6 +388,9 @@ func NewSourceProfileConnection(source string, params map[string]string) (Source
 			if err != nil {
 				return conn, err
 			}
+			if conn.Dydb.enableStreaming == "yes" {
+				conn.Streaming = true
+			}
 		}
 
 	case "sqlserver", "mssql":
diff --git a/sources/dynamodb/schema.go b/sources/dynamodb/schema.go
index 4e78870..178b502 100644
--- a/sources/dynamodb/schema.go
+++ b/sources/dynamodb/schema.go
@@ -192,11 +192,30 @@ func (isi InfoSchemaImpl) ProcessData(conv *internal.Conv, srcTable string, srcS
 	return nil
 }
 
+// StartChangeDataCapture initializes the DynamoDB Streams for the source database. It
+// returns the latestStreamArn for all tables in the source database.
 func (isi InfoSchemaImpl) StartChangeDataCapture(ctx context.Context, conv *internal.Conv) (map[string]interface{}, error) {
-	return nil, nil
+	fmt.Println("Starting DynamoDB Streams initialization...")
+
+	latestStreamArn := make(map[string]interface{})
+	orderTableNames := ddl.OrderTables(conv.SpSchema)
+
+	for _, spannerTable := range orderTableNames {
+		srcTable, _ := internal.GetSourceTable(conv, spannerTable)
+		streamArn, err := NewDynamoDBStream(isi.DynamoClient, srcTable)
+		if err != nil {
+			conv.Unexpected(fmt.Sprintf("Couldn't initialize DynamoDB Stream for table %s: %s", srcTable, err))
+			continue
+		}
+		latestStreamArn[srcTable] = streamArn
+	}
+
+	fmt.Println("DynamoDB Streams initialized successfully.")
+	return latestStreamArn, nil
 }
 
-func (isi InfoSchemaImpl) StartStreamingMigration(ctx context.Context, client *sp.Client, conv *internal.Conv, LatestStream map[string]interface{}) error {
+func (isi InfoSchemaImpl) StartStreamingMigration(ctx context.Context, client *sp.Client, conv *internal.Conv, latestStreamArn map[string]interface{}) error {
+	// TODO(nareshz): work in progress
 	return nil
 }
 
diff --git a/sources/dynamodb/streaming.go b/sources/dynamodb/streaming.go
new file mode 100644
index 0000000..e2bb5eb
--- /dev/null
+++ b/sources/dynamodb/streaming.go
@@ -0,0 +1,60 @@
+// Copyright 2022 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 dynamodb
+
+import (
+	"fmt"
+
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/dynamodb"
+	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbiface"
+)
+
+// NewDynamoDBStream initializes a new DynamoDB Stream for a table with NEW_AND_OLD_IMAGES
+// StreamViewType. If there exists a stream for a given table then it must be of type
+// NEW_IMAGE or NEW_AND_OLD_IMAGES otherwise streaming changes for this table won't be captured.
+// It returns latest Stream Arn for the table along with any error if encountered.
+func NewDynamoDBStream(client dynamodbiface.DynamoDBAPI, srcTable string) (string, error) {
+	describeTableInput := &dynamodb.DescribeTableInput{
+		TableName: aws.String(srcTable),
+	}
+	result, err := client.DescribeTable(describeTableInput)
+	if err != nil {
+		return "", fmt.Errorf("unexpected call to DescribeTable: %v", err)
+	}
+	if result.Table.StreamSpecification != nil {
+		switch *result.Table.StreamSpecification.StreamViewType {
+		case dynamodb.StreamViewTypeKeysOnly:
+			return "", fmt.Errorf("error! there exists a stream with KEYS_ONLY StreamViewType")
+		case dynamodb.StreamViewTypeOldImage:
+			return "", fmt.Errorf("error! there exists a stream with OLD_IMAGE StreamViewType")
+		default:
+			return *result.Table.LatestStreamArn, nil
+		}
+	} else {
+		streamSpecification := &dynamodb.StreamSpecification{
+			StreamEnabled:  aws.Bool(true),
+			StreamViewType: aws.String(dynamodb.StreamViewTypeNewAndOldImages),
+		}
+		updateTableInput := &dynamodb.UpdateTableInput{
+			StreamSpecification: streamSpecification,
+			TableName:           aws.String(srcTable),
+		}
+		res, err := client.UpdateTable(updateTableInput)
+		if err != nil {
+			return "", fmt.Errorf("unexpected call to UpdateTable: %v", err)
+		}
+		return *res.TableDescription.LatestStreamArn, nil
+	}
+}


Test patch:
diff --git a/sources/dynamodb/schema_test.go b/sources/dynamodb/schema_test.go
index 8c7d3f8..a41276b 100644
--- a/sources/dynamodb/schema_test.go
+++ b/sources/dynamodb/schema_test.go
@@ -15,12 +15,16 @@
 package dynamodb
 
 import (
+	"context"
 	"fmt"
 	"math/big"
+	"reflect"
 	"testing"
 
+	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/service/dynamodb"
 	"github.com/aws/aws-sdk-go/service/dynamodb/dynamodbiface"
+	"github.com/aws/aws-sdk-go/service/dynamodbstreams"
 	"github.com/stretchr/testify/assert"
 
 	"github.com/cloudspannerecosystem/harbourbridge/internal"
@@ -36,6 +40,8 @@ type mockDynamoClient struct {
 	describeTableOutputs   []dynamodb.DescribeTableOutput
 	scanCallCount          int
 	scanOutputs            []dynamodb.ScanOutput
+	updateTableCallCount   int
+	updateTableOutputs     []dynamodb.UpdateTableOutput
 	dynamodbiface.DynamoDBAPI
 }
 
@@ -63,6 +69,14 @@ func (m *mockDynamoClient) Scan(input *dynamodb.ScanInput) (*dynamodb.ScanOutput
 	return &m.scanOutputs[m.scanCallCount-1], nil
 }
 
+func (m *mockDynamoClient) UpdateTable(input *dynamodb.UpdateTableInput) (*dynamodb.UpdateTableOutput, error) {
+	if m.updateTableCallCount >= len(m.updateTableOutputs) {
+		return nil, fmt.Errorf("unexpected call to UpdateTable: %v", input)
+	}
+	m.updateTableCallCount++
+	return &m.updateTableOutputs[m.updateTableCallCount-1], nil
+}
+
 func TestProcessSchema(t *testing.T) {
 	tableNameA := "test_a"
 	tableNameB := "test_b"
@@ -912,3 +926,140 @@ func TestSetRowStats(t *testing.T) {
 	assert.Equal(t, tableItemCountA, conv.Stats.Rows[tableNameA])
 	assert.Equal(t, tableItemCountB, conv.Stats.Rows[tableNameB])
 }
+
+func TestInfoSchemaImpl_StartChangeDataCapture(t *testing.T) {
+	tableName := "testtable"
+	attrNameA := "a"
+	latestStreamArn := "arn:aws:dynamodb:dydb_endpoint:test_stream"
+
+	cols := []string{attrNameA}
+	spSchema := ddl.CreateTable{
+		Name:     tableName,
+		ColNames: cols,
+		ColDefs: map[string]ddl.ColumnDef{
+			attrNameA: {Name: attrNameA, T: ddl.Type{Name: ddl.String, Len: ddl.MaxLength}},
+		},
+		Pks: []ddl.IndexKey{{Col: attrNameA}},
+	}
+	srcTable := schema.Table{
+		Name:     tableName,
+		ColNames: cols,
+		ColDefs: map[string]schema.Column{
+			attrNameA: {Name: attrNameA, Type: schema.Type{Name: typeString}},
+		},
+		PrimaryKeys: []schema.Key{{Column: attrNameA}},
+	}
+	conv := buildConv(spSchema, srcTable)
+	type fields struct {
+		DynamoClient        dynamodbiface.DynamoDBAPI
+		DynamoStreamsClient *dynamodbstreams.DynamoDBStreams
+		SampleSize          int64
+	}
+	type args struct {
+		ctx  context.Context
+		conv *internal.Conv
+	}
+	arguments := args{
+		ctx:  context.Background(),
+		conv: conv,
+	}
+	tests := []struct {
+		name    string
+		fields  fields
+		args    args
+		want    map[string]interface{}
+		wantErr bool
+	}{
+		{
+			name: "test for checking correctness of output when stream exists already",
+			fields: fields{
+				DynamoClient: &mockDynamoClient{
+					describeTableOutputs: []dynamodb.DescribeTableOutput{
+						{
+							Table: &dynamodb.TableDescription{
+								TableName: &tableName,
+								StreamSpecification: &dynamodb.StreamSpecification{
+									StreamEnabled:  aws.Bool(true),
+									StreamViewType: aws.String(dynamodb.StreamViewTypeNewImage),
+								},
+								LatestStreamArn: &latestStreamArn,
+							},
+						},
+					},
+				},
+			},
+			args: arguments,
+			want: map[string]interface{}{
+				tableName: latestStreamArn,
+			},
+			wantErr: false,
+		},
+		{
+			name: "test for checking correctness of output when a new stream is created",
+			fields: fields{
+				DynamoClient: &mockDynamoClient{
+					describeTableOutputs: []dynamodb.DescribeTableOutput{
+						{
+							Table: &dynamodb.TableDescription{
+								TableName: &tableName,
+							},
+						},
+					},
+					updateTableOutputs: []dynamodb.UpdateTableOutput{
+						{
+							TableDescription: &dynamodb.TableDescription{
+								LatestStreamArn: &latestStreamArn,
+								StreamSpecification: &dynamodb.StreamSpecification{
+									StreamEnabled:  aws.Bool(true),
+									StreamViewType: aws.String(dynamodb.StreamViewTypeNewAndOldImages),
+								},
+							},
+						},
+					},
+				},
+			},
+			args: arguments,
+			want: map[string]interface{}{
+				tableName: latestStreamArn,
+			},
+			wantErr: false,
+		},
+		{
+			name: "test for handling api calls failure",
+			fields: fields{
+				DynamoClient: &mockDynamoClient{
+					describeTableOutputs: []dynamodb.DescribeTableOutput{
+						{
+							Table: &dynamodb.TableDescription{
+								TableName: &tableName,
+							},
+						},
+					},
+				},
+			},
+			args:    arguments,
+			want:    map[string]interface{}{},
+			wantErr: false,
+		},
+	}
+	totalUnexpecteds := int64(0)
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			isi := InfoSchemaImpl{
+				DynamoClient:        tt.fields.DynamoClient,
+				DynamoStreamsClient: tt.fields.DynamoStreamsClient,
+				SampleSize:          tt.fields.SampleSize,
+			}
+			got, err := isi.StartChangeDataCapture(tt.args.ctx, tt.args.conv)
+			if (err != nil) != tt.wantErr {
+				t.Errorf("InfoSchemaImpl.StartChangeDataCapture() error = %v, wantErr %v", err, tt.wantErr)
+				return
+			}
+			if !reflect.DeepEqual(got, tt.want) {
+				t.Errorf("InfoSchemaImpl.StartChangeDataCapture() = %v, want %v", got, tt.want)
+			}
+			totalUnexpecteds += conv.Unexpecteds()
+		})
+	}
+	assert.Equal(t, int64(1), totalUnexpecteds)
+}