Fail Stream Replicator on startup if source or target isn't reachable #205

TiganeteaRobert · 2022-08-22T12:37:55Z

Added connection checks to:

PubSub source (also added check for existing subscription in project)
Kinesis source (added DescribeStream to check for connection and test)
SQS source (check existing, added unit test)

PubSub target (also added check for existing topic in project)
Eventhub target (added check for runtime info)
HTTP target (added HEAD request on start-up to check for a connection to the host)
Kinesis target (check existing, added unit test)
Kafka target (check existing, added unit test)
SQS target (check existing, added unit test)

#194)

…85)

TiganeteaRobert · 2022-08-22T12:38:10Z

Duplicate of #183

TiganeteaRobert · 2022-08-22T13:11:54Z

Responded to comment about a removed unit test here https://github.com/snowplow-devops/stream-replicator/pull/183/files/adb452ba5f917aa0fc01e8f02a2cc25661c33ef5..5e9cf7c2dbc230c4449093f48efb17ab1cf305e4#r944267068

TiganeteaRobert · 2022-08-22T13:53:09Z

pkg/target/http_test.go

@@ -220,6 +227,37 @@ func TestHttpWrite_Concurrent(t *testing.T) {
 	assert.Equal(int64(10), ackOps)
 }

+func TestHttpWrite_RequestFailure(t *testing.T) {


Added test back after discussion here: #183 (comment)

TiganeteaRobert · 2022-08-22T13:53:16Z

pkg/target/http_test.go

@@ -32,6 +32,13 @@ func createTestServer(results *[][]byte, waitgroup *sync.WaitGroup) *httptest.Se
 			panic(err)
 		}
 		mutex.Lock()
+		if req.URL.Path == `/error` {


Added after discussion here: #183 (comment)

Nice, this is clean and gives us better coverage of the cases we want.

TiganeteaRobert · 2022-08-22T13:53:36Z

pkg/source/sqs/sqs_source_test.go

@@ -65,7 +64,9 @@ func TestNewSQSSourceWithInterfaces_Failure(t *testing.T) {
 	assert.Nil(err)

 	err = source.Read(nil)
-	assert.True(strings.HasPrefix(err.Error(), `Failed to get SQS queue URL`))
+	if err != nil {


Added after discussion here: #183 (comment)

TiganeteaRobert · 2022-08-22T13:53:59Z

Besides the fixed I've done I have responded to some comments on #183

pkg/target/eventhub_test.go

colmsnowplow

Looks like we're missing the following:

kinesis source
kafka target
kinesis target
sqs target

Some of them have tests already written but commented out, some need tests to be written.

colmsnowplow · 2022-08-23T09:36:03Z

Well that's good- less work involved.

I think every source and target needs a test for this - so as well as uncommenting the ones that are there already, if there's any that don't have tests for any, then let's add them.

TiganeteaRobert · 2022-08-23T11:44:45Z

Added checks (or unit tests if check exists) for all sources and targets. Also updated PR desc.

TiganeteaRobert · 2022-08-23T11:47:14Z

pkg/source/kinesis/kinesis_source.go

@@ -243,7 +249,7 @@ func (ks *kinesisSource) Read(sf *sourceiface.SourceFunctions) error {
 	case <-time.After(10 * time.Second):
 		// Append errors and crash
 		multierror.Append(kinesisPullErr, errors.Errorf("wg.Wait() took too long, forcing app close."))
-		ks.log.WithFields(log.Fields{"error": err}).Fatal(err)
+		ks.log.WithFields(log.Fields{"error": kinesisPullErr}).Fatal(kinesisPullErr)


Noticed randomly that the error we logged here was the wrong one, fixed it.

Good spot - Can we pull this into its own issue for auditability - it's a distinct bug which might be important to some debugging investigation in future - so I'd prefer it to have its own audit trail.

(Doesn't need a separate PR though, just its own commit and issue within this PR is fine)

Created issue here: #209 and a separate commit here for it.

colmsnowplow · 2022-08-23T16:52:28Z

pkg/source/kinesis/kinesis_source_test.go


 	assert.Nil(source)
 	assert.EqualError(err, `Failed to create Kinsumer client: leaderActionFrequency config value is mandatory and must be at least as long as ShardCheckFrequency`)
+
+	if testing.Short() {


copy paste error here I think?

Yup, weird how that got there

colmsnowplow · 2022-08-23T17:00:14Z

pkg/source/sqs/sqs_source_test.go

@@ -61,10 +74,14 @@ func TestNewSQSSourceWithInterfaces_Failure(t *testing.T) {

 	source, err := newSQSSourceWithInterfaces(client, "00000000000", 10, testutil.AWSLocalstackRegion, "nonexistent-queue")

-	assert.Nil(source)
-	assert.NotNil(err)
+	assert.NotNil(source)


This is still the opposite behaviour to what we want: #183 (comment)

If the source can't be reached then newSQSSourceWithInterfaces should throw an error.

I added a call to GetQueueUrl inside newSQSSourceWithInterfaces which should fail if there is no connection to SQS.

colmsnowplow · 2022-08-23T17:05:36Z

pkg/source/sqs/sqs_source_test.go

@@ -30,6 +30,20 @@ func TestMain(m *testing.M) {
 	os.Exit(exitVal)
 }

+// TestNewSqsSource_ConnectionCheck tests that the SQS source fails on start-up if the connection to SQS fails
+func TestNewSqsSource_ConnectionCheck(t *testing.T) {


This doesn't test a connection check with the SQS queue. It tests if an AWS client session can be created.

We can have a valid AWS session and without having a valid connection to the SQS queue, but not vice-versa.

Indeed, this tests the connection to AWS. I am now testing the case where a connection to SQS cannot be done in TestNewSQSSourceWithInterfaces_Failure

colmsnowplow · 2022-08-23T17:06:41Z

pkg/target/kinesis_test.go

@@ -15,6 +15,19 @@ import (
 	"github.com/snowplow-devops/stream-replicator/pkg/testutil"
 )

+// TestNewKinesisTarget_ConnectionCheck tests that the Kinesis target fails on start-up if the connection to Kinesis fails
+func TestNewKinesisTarget_ConnectionCheck(t *testing.T) {


Same as my coment for the sqs source - there's no check for a connection to a stream, this failure only looks at the AWS session.

I don't see anywhere in the kinesis target codebase where we're doing a connection check at all

I just added a DescribeStream call to the Kinesis target too. The case where the connection to Kinesis could not be done is now tested in TestKinesisTarget_WriteFailure

Are you sure you've pushed all your changes? I can't see any TestKinesisTarget_WriteFailure test. I can see you've renamed it and replaced it with TestNewKinesisTarget_ConnectionCheck and TestKinesisTarget_KinesisConnectionFailure, neither of which appear to test this feature.

colmsnowplow · 2022-08-23T17:08:37Z

pkg/target/sqs_test.go

@@ -15,6 +15,15 @@ import (
 	"github.com/snowplow-devops/stream-replicator/pkg/testutil"
 )

+// TestNewSqsTarget_ConnectionCheck tests that the SQS target fails on start-up if the connection to SQS fails
+func TestNewSqsTarget_ConnectionCheck(t *testing.T) {


Same comment as for the sqs source and the kinesis target.

Same as in the SQS source, I added a quick call to GetQueueUrl in newSQSTargetWithInterfaces so we test the connection to SQS instead of AWS.

…closes #182)

colmsnowplow · 2022-08-31T09:40:05Z

pkg/source/kinesis/kinesis_source_test.go

-	assert.Equal("arn:aws:kinesis:us-east-1:00000000000:stream/not-exists", source.GetID())
-
-	err = source.Read(nil)
-	assert.NotNil(err)
 	if err != nil {


This test will pass when we don't throw any error. It needs to assert against the error existing first.

colmsnowplow · 2022-08-31T09:51:17Z

pkg/target/kinesis_test.go

 	if testing.Short() {
 		t.Skip("skipping integration test")
 	}

 	assert := assert.New(t)

-	client := testutil.GetAWSLocalstackKinesisClient()
+	client := kinesis.New(session.Must(session.NewSession(&aws.Config{


I don't understand why you're manually creating another new kinesis client here?

colmsnowplow

Please stop force pushing before requested changes have been reviewed. It just creates needless work and makes reviews of your PRs take longer than they need to.

It's easier for everyone to squash once at the end, when the entire review is done.

colmsnowplow · 2022-11-11T10:39:22Z

I'm gonna reopen this and move it to draft - there's been a lot of messing around in between but I still plan on picking this up from where we left off and pulling it into an upcoming release!

TiganeteaRobert added 2 commits August 22, 2022 13:46

Fix panic on failure target WriteOversized() and WriteInvalid() (closes

b1063e9

#194)

Add transformLatency and sum metrics to statsd implementation (closes #…

9b219a0

…85)

TiganeteaRobert marked this as a duplicate of #183 Aug 22, 2022

TiganeteaRobert commented Aug 22, 2022

View reviewed changes

colmsnowplow reviewed Aug 22, 2022

View reviewed changes

pkg/target/eventhub_test.go Outdated Show resolved Hide resolved

colmsnowplow reviewed Aug 22, 2022

View reviewed changes

TiganeteaRobert commented Aug 23, 2022

View reviewed changes

TiganeteaRobert force-pushed the startup_fail_rebased branch from 5b99e66 to b5f0154 Compare August 23, 2022 11:52

TiganeteaRobert marked this pull request as ready for review August 23, 2022 11:52

TiganeteaRobert requested a review from colmsnowplow August 23, 2022 11:52

TiganeteaRobert added 2 commits August 23, 2022 15:26

Add clientRecordMaxAge() to kinesis source (closes #117)

54c45ba

Allow setting properties metadata for EventHubs target (closes #99)

d832b17

TiganeteaRobert force-pushed the startup_fail_rebased branch from b5f0154 to bfaa707 Compare August 23, 2022 12:31

colmsnowplow reviewed Aug 23, 2022

View reviewed changes

TiganeteaRobert requested a review from colmsnowplow August 26, 2022 08:43

TiganeteaRobert force-pushed the startup_fail_rebased branch from e12d862 to a1f30ad Compare August 26, 2022 08:49

TiganeteaRobert added 2 commits August 29, 2022 15:49

Fail Stream Replicator on startup if source or target isn't reachable (…

35cc351

…closes #182)

Kinesis source logging wrong error message (closes #209)

d1aa932

TiganeteaRobert force-pushed the startup_fail_rebased branch from 420d2ca to d1aa932 Compare August 29, 2022 12:49

colmsnowplow reviewed Aug 31, 2022

View reviewed changes

TiganeteaRobert closed this Nov 4, 2022

colmsnowplow reopened this Nov 11, 2022

colmsnowplow marked this pull request as draft November 11, 2022 10:39

adatzer force-pushed the develop branch from d832b17 to 642c6cd Compare March 27, 2024 10:51

colmsnowplow force-pushed the develop branch 3 times, most recently from dd7c31c to e4c848a Compare April 15, 2024 10:49

Base automatically changed from develop to master April 15, 2024 11:28

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fail Stream Replicator on startup if source or target isn't reachable #205

Fail Stream Replicator on startup if source or target isn't reachable #205

TiganeteaRobert commented Aug 22, 2022 •

edited

Loading

TiganeteaRobert commented Aug 22, 2022

TiganeteaRobert commented Aug 22, 2022

TiganeteaRobert Aug 22, 2022

TiganeteaRobert Aug 22, 2022

colmsnowplow Aug 22, 2022

TiganeteaRobert Aug 22, 2022

TiganeteaRobert commented Aug 22, 2022

colmsnowplow left a comment

colmsnowplow commented Aug 23, 2022 •

edited

Loading

TiganeteaRobert commented Aug 23, 2022

TiganeteaRobert Aug 23, 2022

colmsnowplow Aug 23, 2022 •

edited

Loading

TiganeteaRobert Aug 26, 2022

TiganeteaRobert Aug 26, 2022

colmsnowplow Aug 23, 2022

TiganeteaRobert Aug 26, 2022

colmsnowplow Aug 23, 2022

TiganeteaRobert Aug 26, 2022

colmsnowplow Aug 23, 2022

TiganeteaRobert Aug 26, 2022

colmsnowplow Aug 23, 2022

TiganeteaRobert Aug 26, 2022

colmsnowplow Aug 31, 2022

colmsnowplow Aug 23, 2022

TiganeteaRobert Aug 26, 2022

colmsnowplow Aug 31, 2022

colmsnowplow Aug 31, 2022

colmsnowplow left a comment

colmsnowplow commented Nov 11, 2022

Fail Stream Replicator on startup if source or target isn't reachable #205

Are you sure you want to change the base?

Fail Stream Replicator on startup if source or target isn't reachable #205

Conversation

TiganeteaRobert commented Aug 22, 2022 • edited Loading

TiganeteaRobert commented Aug 22, 2022

TiganeteaRobert commented Aug 22, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

TiganeteaRobert commented Aug 22, 2022

colmsnowplow left a comment

Choose a reason for hiding this comment

colmsnowplow commented Aug 23, 2022 • edited Loading

TiganeteaRobert commented Aug 23, 2022

Choose a reason for hiding this comment

colmsnowplow Aug 23, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

colmsnowplow left a comment

Choose a reason for hiding this comment

colmsnowplow commented Nov 11, 2022

TiganeteaRobert commented Aug 22, 2022 •

edited

Loading

colmsnowplow commented Aug 23, 2022 •

edited

Loading

colmsnowplow Aug 23, 2022 •

edited

Loading