fix(subscriber): do not update on match (#303)

This commit fixes a bug whereby the subscriber would run `PutSubscriptionFilter` for every previously subscribed log group on discovery.
observeinc · Jun 12, 2024 · 9068f5f · 9068f5f
1 parent e950207
commit 9068f5f
Show file tree

Hide file tree

Showing 4 changed files with 38 additions and 15 deletions.
diff --git a/apps/logwriter/template.yaml b/apps/logwriter/template.yaml
@@ -477,7 +477,8 @@ Resources:
     Condition: HasDiscoveryRate
     DependsOn: Subscriber
     Properties:
-      Description: "Subscribe new log groups"
+      Description: >-
+        Subscribe new log groups. Requires CloudTrail in target region.
       State: ENABLED
       EventPattern:
         source:

diff --git a/pkg/handler/subscriber/subscription.go b/pkg/handler/subscriber/subscription.go
@@ -103,8 +103,8 @@ func subscriptionFilterEquals(a, b types.SubscriptionFilter) bool {
 // cloudwatch API in order to converge to our intended configuration state.
 func (h *Handler) SubscriptionFilterDiff(subscriptionFilters []types.SubscriptionFilter) (actions []any) {
 	var (
-		deleted, updated int
-		deleteOnly       = aws.ToString(h.subscriptionFilter.DestinationArn) == ""
+		deleted, found int
+		deleteOnly     = aws.ToString(h.subscriptionFilter.DestinationArn) == ""
 	)
 
 	for _, f := range subscriptionFilters {
@@ -117,18 +117,20 @@ func (h *Handler) SubscriptionFilterDiff(subscriptionFilters []types.Subscriptio
 			actions = append(actions, &cloudwatchlogs.DeleteSubscriptionFilterInput{
 				FilterName: f.FilterName,
 			})
-		} else if !subscriptionFilterEquals(h.subscriptionFilter, f) {
-			updated++
-			actions = append(actions, &cloudwatchlogs.PutSubscriptionFilterInput{
-				FilterName:     h.subscriptionFilter.FilterName,
-				FilterPattern:  h.subscriptionFilter.FilterPattern,
-				DestinationArn: h.subscriptionFilter.DestinationArn,
-				RoleArn:        h.subscriptionFilter.RoleArn,
-			})
+		} else {
+			found++
+			if !subscriptionFilterEquals(h.subscriptionFilter, f) {
+				actions = append(actions, &cloudwatchlogs.PutSubscriptionFilterInput{
+					FilterName:     h.subscriptionFilter.FilterName,
+					FilterPattern:  h.subscriptionFilter.FilterPattern,
+					DestinationArn: h.subscriptionFilter.DestinationArn,
+					RoleArn:        h.subscriptionFilter.RoleArn,
+				})
+			}
 		}
 	}
 
-	if !deleteOnly && updated == 0 && len(subscriptionFilters)-deleted < MaxSubscriptionFilterCount {
+	if !deleteOnly && found == 0 && len(subscriptionFilters)-deleted < MaxSubscriptionFilterCount {
 		actions = append(actions, &cloudwatchlogs.PutSubscriptionFilterInput{
 			FilterName:     h.subscriptionFilter.FilterName,
 			FilterPattern:  h.subscriptionFilter.FilterPattern,

diff --git a/pkg/handler/subscriber/subscription_test.go b/pkg/handler/subscriber/subscription_test.go
@@ -185,9 +185,6 @@ func TestSubscriptionFilterDiff(t *testing.T) {
 			},
 		},
 		{
-			/*
-				Do nothing if we exceed the two subscription filter limit
-			*/
 			Configure: types.SubscriptionFilter{
 				FilterName:     aws.String("observe"),
 				DestinationArn: aws.String("arn:aws:lambda:us-west-2:123456789012:function:example"),
@@ -202,6 +199,9 @@ func TestSubscriptionFilterDiff(t *testing.T) {
 			},
 		},
 		{
+			/*
+				Do nothing if we exceed the two subscription filter limit
+			*/
 			Configure: types.SubscriptionFilter{
 				FilterName:     aws.String("observe"),
 				DestinationArn: aws.String("arn:aws:lambda:us-west-2:123456789012:function:example"),
@@ -216,6 +216,24 @@ func TestSubscriptionFilterDiff(t *testing.T) {
 			},
 			// no expected actions
 		},
+		{
+			// Do nothing if filter already matches.
+			Configure: types.SubscriptionFilter{
+				FilterName:     aws.String("observe"),
+				DestinationArn: aws.String("arn:aws:lambda:us-west-2:123456789012:function:example"),
+			},
+			Existing: []types.SubscriptionFilter{
+				{
+					FilterName:     aws.String("observe"),
+					DestinationArn: aws.String("arn:aws:lambda:us-west-2:123456789012:function:example"),
+				},
+				{
+					FilterName:     aws.String("something-else"),
+					DestinationArn: aws.String("arn:aws:lambda:us-west-2:123456789012:function:another"),
+				},
+			},
+			// no expected actions
+		},
 	}
 
 	for i, tt := range testcases {

diff --git a/pkg/lambda/subscriber/lambda.go b/pkg/lambda/subscriber/lambda.go
@@ -53,6 +53,8 @@ func New(ctx context.Context, cfg *Config) (*Lambda, error) {
 	logger := logging.New(cfg.Logging)
 	logger.V(4).Info("initialized", "config", cfg)
 
+	tracing.SetLogger(logger)
+
 	tracerProvider, err := tracing.NewTracerProvider(ctx)
 	if err != nil {
 		return nil, fmt.Errorf("failed to initialize tracing: %w", err)