feat: implement backoff for snowpipe streaming authorization errors #5399

shekhar-rudder · 2024-12-27T07:56:29Z

Description

This PR introduces a backoff mechanism when creating channels in the Snowpipe streaming destination. If an authorization error is encountered at the schema, table, or column level, the backoff ensures that we avoid repeatedly attempting to connect to Snowflake unnecessarily.

Added a new struct authzBackoff in manager to store backoff related state
Refactoring: Added managerCreator as a field in the manager for making the code testable.
Until now, any error encountered while creating the channel for the discards table resulted in the job being marked as aborted. In this PR, the behavior has been updated to mark the job as failed, allowing it to be retried.
Backoff is applied whenever an authorization error is encountered.
Backoff is being cleared whenver channels for all the tables are being created without any authz error

Security

The code changed/added as part of this pull request won't create any security issues with how the software is being used.

shekhar-rudder · 2024-12-27T07:59:19Z

warehouse/integrations/manager/manager.go

@@ -26,6 +26,9 @@ import (
 	warehouseutils "github.com/rudderlabs/rudder-server/warehouse/utils"
 )

+// Creating an alias since "model.TableSchema" is defined in an internal module
+type ModelTableSchema = model.TableSchema


Any alternatives ? I can't import model.TableSchema in the snowpipestreaming test since it is in an internal module.

We can use whutils.ModelTableSchema

codecov · 2024-12-27T08:26:05Z

Codecov Report

Attention: Patch coverage is 89.02439% with 9 lines in your changes missing coverage. Please review.

Project coverage is 74.80%. Comparing base (8a186e5) to head (e268f73).
Report is 1 commits behind head on master.

Files with missing lines	Patch %	Lines
...tionmanager/snowpipestreaming/snowpipestreaming.go	86.36%	4 Missing and 2 partials ⚠️
...yncdestinationmanager/snowpipestreaming/channel.go	80.00%	3 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##           master    #5399      +/-   ##
==========================================
- Coverage   74.82%   74.80%   -0.03%     
==========================================
  Files         438      438              
  Lines       61331    61404      +73     
==========================================
+ Hits        45893    45933      +40     
- Misses      12895    12932      +37     
+ Partials     2543     2539       -4

☔ View full report in Codecov by Sentry.
📢 Have feedback on the report? Share it here.

achettyiitr · 2024-12-30T06:53:04Z

warehouse/integrations/manager/manager.go

@@ -26,6 +26,9 @@ import (
 	warehouseutils "github.com/rudderlabs/rudder-server/warehouse/utils"
 )

+// Creating an alias since "model.TableSchema" is defined in an internal module
+type ModelTableSchema = model.TableSchema


We can use whutils.ModelTableSchema

achettyiitr · 2024-12-30T06:58:40Z

router/batchrouter/asyncdestinationmanager/snowpipestreaming/snowpipestreaming_test.go

@@ -43,6 +48,29 @@ func (m *mockAPI) GetStatus(_ context.Context, channelID string) (*model.StatusR
 	return m.getStatusOutputMap[channelID]()
 }

+type mockManager struct {
+	manager.Manager
+	throwSchemaErr bool


Instead of a boolean, we can set the following error like createSchemaErr, createTableErr.

achettyiitr · 2024-12-30T07:06:01Z

router/batchrouter/asyncdestinationmanager/snowpipestreaming/types.go

+		backoffDuration        time.Duration
+		initialBackoffDuration time.Duration


Can we use github.com/cenkalti/backoff/v4 instead of implementing this logic?

Replaced the custom implementation with the library

achettyiitr · 2024-12-30T07:07:53Z

router/batchrouter/asyncdestinationmanager/snowpipestreaming/snowpipestreaming.go

+			var authzErr *snowpipeAuthzError
+			if errors.As(err, &authzErr) {
+				if shouldResetBackoff {
+					shouldResetBackoff = false
+				}
+			}


Suggested change

var authzErr *snowpipeAuthzError

if errors.As(err, &authzErr) {

if shouldResetBackoff {

shouldResetBackoff = false

}

}

if errors.As(err, &authzErr) && !shouldResetBackoff {

shouldResetBackoff = false

}

Removed shouldResetBackoff since it was unnecessary

achettyiitr · 2024-12-30T07:14:22Z

router/batchrouter/asyncdestinationmanager/snowpipestreaming/snowpipestreaming.go

 	whutils "github.com/rudderlabs/rudder-server/warehouse/utils"
 )

 var json = jsoniter.ConfigCompatibleWithStandardLibrary

 func New(
 	conf *config.Config,
-	logger logger.Logger,
+	mLogger logger.Logger,


mLogger Do you mean module logger? Can we rename it as log?

I had renamed it to mLogger since managerCreator had an argument of type logger.Logger. Renaming it to log

achettyiitr · 2024-12-30T07:32:38Z

router/batchrouter/asyncdestinationmanager/snowpipestreaming/snowpipestreaming.go

+				logger.NewStringField("table", discardsTable()),
+				obskit.Error(err),
+			)
+			shouldResetBackoff = false


We can return from here itself. No point in going forward.

Suggested change

shouldResetBackoff = false

m.stats.jobs.failed.Count(len(asyncDest.ImportingJobIDs))

return common.AsyncUploadOutput{

FailedJobIDs: asyncDest.ImportingJobIDs,

FailedCount: len(asyncDest.ImportingJobIDs),

FailedReason: fmt.Errorf("failed to initialize channel with schema: %w", err).Error(),

DestinationID: asyncDest.Destination.ID,

}

Added an early return with failed job reporting

shekhar-rudder commented Dec 27, 2024

View reviewed changes

shekhar-rudder force-pushed the war-72-backoff branch from 9c22207 to 476cdb4 Compare December 27, 2024 08:37

RanjeetMishra requested a review from achettyiitr December 27, 2024 08:39

shekhar-rudder requested review from RanjeetMishra and vyeshwanth December 27, 2024 08:41

shekhar-rudder force-pushed the war-72-backoff branch from 476cdb4 to 5644298 Compare December 27, 2024 08:53

feat: implement backoff for snowpipe streaming authorization errors

141b31f

shekhar-rudder force-pushed the war-72-backoff branch from 5644298 to 141b31f Compare December 27, 2024 08:56

fix: use utc time

9df59d0

achettyiitr reviewed Dec 30, 2024

View reviewed changes

shekhar-rudder force-pushed the war-72-backoff branch from aa4f042 to 9779ca7 Compare December 30, 2024 08:08

refactor: review comments 1

e268f73

shekhar-rudder force-pushed the war-72-backoff branch from 9779ca7 to e268f73 Compare December 30, 2024 13:21

shekhar-rudder requested a review from achettyiitr December 30, 2024 13:22

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

feat: implement backoff for snowpipe streaming authorization errors #5399

feat: implement backoff for snowpipe streaming authorization errors #5399

shekhar-rudder commented Dec 27, 2024 •

edited

Loading

shekhar-rudder Dec 27, 2024 •

edited

Loading

achettyiitr Dec 30, 2024

codecov bot commented Dec 27, 2024 •

edited

Loading

achettyiitr Dec 30, 2024

achettyiitr Dec 30, 2024

shekhar-rudder Dec 30, 2024

achettyiitr Dec 30, 2024

shekhar-rudder Dec 30, 2024

achettyiitr Dec 30, 2024

shekhar-rudder Dec 30, 2024

achettyiitr Dec 30, 2024

shekhar-rudder Dec 30, 2024

achettyiitr Dec 30, 2024 •

edited

Loading

shekhar-rudder Dec 30, 2024

		backoffDuration time.Duration
		initialBackoffDuration time.Duration

-			shouldResetBackoff = false
+			m.stats.jobs.failed.Count(len(asyncDest.ImportingJobIDs))
+			return common.AsyncUploadOutput{
+				FailedJobIDs:  asyncDest.ImportingJobIDs,
+				FailedCount:   len(asyncDest.ImportingJobIDs),
+				FailedReason:  fmt.Errorf("failed to initialize channel with schema: %w", err).Error(),
+				DestinationID: asyncDest.Destination.ID,
+			}

feat: implement backoff for snowpipe streaming authorization errors #5399

Are you sure you want to change the base?

feat: implement backoff for snowpipe streaming authorization errors #5399

Conversation

shekhar-rudder commented Dec 27, 2024 • edited Loading

Description

Security

shekhar-rudder Dec 27, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov bot commented Dec 27, 2024 • edited Loading

Codecov Report

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

achettyiitr Dec 30, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

shekhar-rudder commented Dec 27, 2024 •

edited

Loading

shekhar-rudder Dec 27, 2024 •

edited

Loading

codecov bot commented Dec 27, 2024 •

edited

Loading

achettyiitr Dec 30, 2024 •

edited

Loading