feat(subscriber): add configuration variables

apps/subscriber/template.yaml

@@ -19,6 +19,27 @@ Globals:
     MemorySize: 128
 
 Parameters:
+  FilterName:
+    Type: String
+    Description: >-
+      Subscription filter name.
+    Default: 'observe-logs-subscription'
+  FilterPattern:
+    Type: String
+    Description: >-
+      Subscription filter pattern.
+    Default: ''
+  DestinationArn:
+    Type: String
+    Description: >-
+      Destination ARN for subscription filter. If not set, all subscription
+      filters matching configured filter name will be removed.
+    Default: ''
+  RoleArn:
+    Type: String
+    Description: >-
+      Role ARN for subscription filter.
+    Default: ''
   NameOverride:
     Type: String
     Description: >-
@@ -144,6 +165,10 @@ Resources:
             Enabled: true
       Environment:
         Variables:
+          FILTER_NAME: !Ref FilterName
+          FILTER_PATTERN: !Ref FilterPattern
+          DESTINATION_ARN: !Ref DestinationArn
+          ROLE_ARN: !Ref RoleArn
           QUEUE_URL: !Ref Queue
           VERBOSITY: 9
 Outputs:

cmd/subscriber/main.go

@@ -16,9 +16,12 @@ import (
 )
 
 var env struct {
-	QueueURL        string `env:"QUEUE_URL,required"`
-	Verbosity       int    `env:"VERBOSITY,default=1"`
-	OtelServiceName string `env:"OTEL_SERVICE_NAME"`
+	FilterName     string `env:"FILTER_NAME"`
+	FilterPattern  string `env:"FILTER_PATTERN"`
+	DestinationARN string `env:"DESTINATION_ARN"`
+	RoleARN        string `env:"ROLE_ARN"`
+	QueueURL       string `env:"QUEUE_URL,required"`
+	Verbosity      int    `env:"VERBOSITY,default=1"`
 }
 
 var (
@@ -57,6 +60,10 @@ func realInit() error {
 	}
 
 	handler, err = subscriber.New(&subscriber.Config{
+		FilterName:           env.FilterName,
+		FilterPattern:        env.FilterPattern,
+		DestinationARN:       env.DestinationARN,
+		RoleARN:              env.RoleARN,
 		Logger:               &logger,
 		CloudWatchLogsClient: cloudwatchlogs.NewFromConfig(awsCfg),
 		Queue:                queue,

docs/subscriber.md

@@ -0,0 +1,6 @@
+# Observe Subscriber
+
+The subscriber stack subscribes CloudWatch Log Groups to a supported destination ARN (either Kinesis Firehose or Lambda). It supports two request types:
+
+- subscription requests contain a list of log groups which we wish to subscribe to our destination.
+- discovery requests contain a list of filters which are used to generate subscription requests.

handler/subscriber/config.go

@@ -2,19 +2,39 @@ package subscriber
 
 import (
 	"errors"
+	"fmt"
+	"strings"
 
+	"github.com/aws/aws-sdk-go-v2/aws/arn"
 	"github.com/go-logr/logr"
 )
 
 var (
 	ErrMissingCloudWatchLogsClient = errors.New("missing CloudWatch Logs client")
 	ErrMissingQueue                = errors.New("missing queue")
+	ErrMissingFilterName           = errors.New("filter name must be provided if destination ARN is set")
+	ErrMissingDestinationARN       = errors.New("destination ARN must be provided if role ARN is set")
+	ErrInvalidARN                  = errors.New("invalid ARN")
 )
 
 type Config struct {
 	CloudWatchLogsClient
 	Queue
 
+	// FilterName for subscription filters managed by this handler
+	// Our handler will assume it manages all filters that have this name as a
+	// prefix.
+	FilterName string
+
+	// FilterPattern for subscription filters
+	FilterPattern string
+
+	// DestinationARN to subscribe log groups to.
+	// If empty, delete any subscription filters we manage.
+	DestinationARN string
+	// RoleARN for subscription filter
+	RoleARN string
+
 	Logger *logr.Logger
 }
 
@@ -29,5 +49,26 @@ func (c *Config) Validate() error {
 		errs = append(errs, ErrMissingQueue)
 	}
 
+	if c.FilterName == "" && c.DestinationARN != "" {
+		errs = append(errs, ErrMissingFilterName)
+	}
+
+	if c.DestinationARN != "" {
+		if _, err := arn.Parse(c.DestinationARN); err != nil {
+			errs = append(errs, fmt.Errorf("failed to parse destination: %w: %s", ErrInvalidARN, err))
+		}
+	}
+
+	if c.RoleARN != "" {
+		if c.DestinationARN == "" {
+			errs = append(errs, ErrMissingDestinationARN)
+		}
+
+		roleARN, err := arn.Parse(c.RoleARN)
+		if err != nil || roleARN.Service != "iam" || strings.HasPrefix(roleARN.Resource, "role/") {
+			errs = append(errs, fmt.Errorf("failed to parse role: %w", ErrInvalidARN))
+		}
+	}
+
 	return errors.Join(errs...)
 }

handler/subscriber/config_test.go

@@ -25,6 +25,40 @@ func TestConfig(t *testing.T) {
 			},
 			ExpectError: subscriber.ErrMissingQueue,
 		},
+		{
+			Config: subscriber.Config{
+				CloudWatchLogsClient: &handlertest.CloudWatchLogsClient{},
+				Queue:                &MockQueue{},
+				DestinationARN:       "hello",
+			},
+			ExpectError: subscriber.ErrMissingFilterName,
+		},
+		{
+			Config: subscriber.Config{
+				CloudWatchLogsClient: &handlertest.CloudWatchLogsClient{},
+				Queue:                &MockQueue{},
+				FilterName:           "observe-logs-subscription",
+				DestinationARN:       "hello",
+			},
+			ExpectError: subscriber.ErrInvalidARN,
+		},
+		{
+			Config: subscriber.Config{
+				CloudWatchLogsClient: &handlertest.CloudWatchLogsClient{},
+				Queue:                &MockQueue{},
+				FilterName:           "observe-logs-subscription",
+				DestinationARN:       "arn:aws:lambda:us-east-2:123456789012:function:my-function",
+			},
+		},
+		{
+			Config: subscriber.Config{
+				CloudWatchLogsClient: &handlertest.CloudWatchLogsClient{},
+				Queue:                &MockQueue{},
+				FilterName:           "observe-logs-subscription",
+				RoleARN:              "arn:aws:lambda:us-east-2:123456789012:function:my-function",
+			},
+			ExpectError: subscriber.ErrMissingDestinationARN,
+		},
 		{
 			Config: subscriber.Config{
 				CloudWatchLogsClient: &handlertest.CloudWatchLogsClient{},

handler/subscriber/handler.go

@@ -2,14 +2,16 @@ package subscriber
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	"github.com/aws/aws-sdk-go-v2/service/cloudwatchlogs"
-	"github.com/go-logr/logr"
 
 	"github.com/observeinc/aws-sam-testing/handler"
 )
 
+var ErrNotImplemented = errors.New("not implemented")
+
 type CloudWatchLogsClient interface {
 	DescribeLogGroups(context.Context, *cloudwatchlogs.DescribeLogGroupsInput, ...func(*cloudwatchlogs.Options)) (*cloudwatchlogs.DescribeLogGroupsOutput, error)
 }
@@ -25,54 +27,43 @@ type Handler struct {
 	Client CloudWatchLogsClient
 }
 
-type SyncRequest struct {
-	*SyncConfig `json:"sync"`
-}
-
-type SyncConfig struct {
-	Subscription *cloudwatchlogs.PutSubscriptionFilterInput `json:"subscription,omitempty"`
-	Limit        *int32                                     `json:"limit,omitempty"`
-}
-
-type task struct {
-	PutSubscriptionFilterInput *cloudwatchlogs.PutSubscriptionFilterInput `json:"subscription,omitempty"`
-	DescribeLogGroupsOutput    *cloudwatchlogs.DescribeLogGroupsOutput    `json:"logGroups"`
-}
-
-type SyncResponse struct {
-	LogGroupCount int `json:"logGroupCount"`
-	PageCount     int `json:"pageCount"`
-}
+func (h *Handler) HandleDiscoveryRequest(ctx context.Context, discoveryReq *DiscoveryRequest) (*Response, error) {
+	var discoveryResp DiscoveryResponse
 
-func (h *Handler) HandleSync(ctx context.Context, request SyncRequest) (*SyncResponse, error) {
-	logger := logr.FromContextOrDiscard(ctx)
+	for _, input := range discoveryReq.ToDescribeLogInputs() {
+		paginator := cloudwatchlogs.NewDescribeLogGroupsPaginator(h.Client, input)
 
-	paginator := cloudwatchlogs.NewDescribeLogGroupsPaginator(h.Client, &cloudwatchlogs.DescribeLogGroupsInput{
-		Limit: request.Limit,
-	})
-
-	var response SyncResponse
-
-	for paginator.HasMorePages() {
-		output, err := paginator.NextPage(ctx)
-		if err != nil {
-			return nil, fmt.Errorf("failed to describe log groups: %w", err)
+		for paginator.HasMorePages() {
+			page, err := paginator.NextPage(ctx)
+			if err != nil {
+				return nil, fmt.Errorf("failed to describe log groups: %w", err)
+			}
+			discoveryResp.RequestCount++
+			discoveryResp.LogGroupCount += len(page.LogGroups)
 		}
+	}
 
-		response.PageCount++
-		response.LogGroupCount += len(output.LogGroups)
+	return &Response{DiscoveryResponse: &discoveryResp}, nil
+}
 
-		logger.V(6).Info("queueing page")
+func (h *Handler) HandleSubscriptionRequest(_ context.Context, _ *SubscriptionRequest) (*Response, error) {
+	// to be implemented
+	return nil, nil
+}
 
-		if err := h.Queue.Put(ctx, &task{
-			PutSubscriptionFilterInput: request.Subscription,
-			DescribeLogGroupsOutput:    output,
-		}); err != nil {
-			return nil, fmt.Errorf("failed to queue log groups: %w", err)
-		}
+func (h *Handler) HandleRequest(ctx context.Context, req *Request) (*Response, error) {
+	if err := req.Validate(); err != nil {
+		return nil, fmt.Errorf("failed to validate request: %w", err)
 	}
 
-	return &response, nil
+	switch {
+	case req.DiscoveryRequest != nil:
+		return h.HandleDiscoveryRequest(ctx, req.DiscoveryRequest)
+	case req.SubscriptionRequest != nil:
+		return h.HandleSubscriptionRequest(ctx, req.SubscriptionRequest)
+	default:
+		return nil, ErrNotImplemented
+	}
 }
 
 func New(cfg *Config) (*Handler, error) {
@@ -89,7 +80,7 @@ func New(cfg *Config) (*Handler, error) {
 		h.Logger = *cfg.Logger
 	}
 
-	if err := h.Mux.Register(h.HandleSync); err != nil {
+	if err := h.Mux.Register(h.HandleRequest); err != nil {
 		return nil, fmt.Errorf("failed to register handler: %w", err)
 	}