feat(remote/aws/secrets_manager): implement poller + update doc

Signed-off-by: hainenber <[email protected]>

docs/sources/reference/components/remote.aws.secrets_manager.md

@@ -7,8 +7,9 @@ title: remote.aws.secrets_manager
 # remote.aws.secret_manager
 
 `remote.aws.secrets_manager` securely exposes value of secrets located in [AWS Secrets Manager](https://aws.amazon.com/secrets-manager/) to other components.
-The secret would be fetched one time only at startup. Restart Alloy if you have updated to new value and would like 
-the component to fetch the latest version.
+By default, the secret would be fetched one time only at startup. If configured, the secret will be polled for changes so that the most recent value is always available. 
+
+Beware that this could incur cost due to frequent API calls.
 
 Multiple `remote.aws.secrets_manager` components can be specified using different name
 labels. By default, [AWS environment variables](https://docs.aws.amazon.com/cli/latest/userguide/cli-configure-envvars.html) are used to authenticate against AWS. The `key` and `secret` arguments inside `client` blocks can be used to provide custom authentication.
@@ -28,6 +29,7 @@ The following arguments are supported:
 Name             | Type       | Description                                                              | Default | Required
 -----------------|------------|--------------------------------------------------------------------------|---------|---------
 `id`             | `string`   | Secret ID.                                                               |         | yes
+`poll_frequency` | `duration` | How often to poll the API for changes.                                   |         | no
 
 ## Blocks
 

internal/component/remote/aws/secrets_manager/secrets_manager.go

@@ -2,11 +2,9 @@ package secrets_manager
 
 import (
 	"context"
-	"fmt"
 	"sync"
 	"time"
 
-	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/secretsmanager"
 	"github.com/grafana/alloy/internal/component"
 	aws_common_config "github.com/grafana/alloy/internal/component/common/config/aws"
@@ -28,11 +26,14 @@ func init() {
 }
 
 type Component struct {
-	opts         component.Options
-	mut          sync.Mutex
-	health       component.Health
-	errors       prometheus.Counter
-	lastAccessed prometheus.Gauge
+	opts          component.Options
+	mut           sync.Mutex
+	health        component.Health
+	pollFrequency time.Duration
+	watcher       *watcher
+	updateChan    chan result
+	errors        prometheus.Counter
+	lastAccessed  prometheus.Gauge
 }
 
 var (
@@ -45,6 +46,7 @@ type Arguments struct {
 	Options       aws_common_config.Client `alloy:"client,block,optional"`
 	SecretId      string                   `alloy:"id,attr"`
 	SecretVersion string                   `alloy:"version,attr,optional"`
+	PollFrequency time.Duration            `alloy:"poll_frequency,attr,optional"`
 }
 
 // DefaultArguments holds default settings for Arguments.
@@ -63,9 +65,18 @@ type Exports struct {
 
 // New initializes a new component
 func New(o component.Options, args Arguments) (*Component, error) {
+	// Create AWS and AWS's Secrets Manager client
+	awsCfg, err := aws_common_config.GenerateAWSConfig(args.Options)
+	if err != nil {
+		return nil, err
+	}
+	client := secretsmanager.NewFromConfig(*awsCfg)
+
 	s := &Component{
-		opts:   o,
-		health: component.Health{},
+		opts:          o,
+		pollFrequency: args.PollFrequency,
+		health:        component.Health{},
+		updateChan:    make(chan result),
 		errors: prometheus.NewCounter(prometheus.CounterOpts{
 			Name: "remote_aws_secrets_manager_errors_total",
 			Help: "Total number of errors when accessing AWS Secrets Manager",
@@ -76,66 +87,84 @@ func New(o component.Options, args Arguments) (*Component, error) {
 		}),
 	}
 
+	w := newWatcher(args.SecretId, args.SecretVersion, s.updateChan, args.PollFrequency, client)
+	s.watcher = w
+
 	if err := o.Registerer.Register(s.errors); err != nil {
 		return nil, err
 	}
 	if err := o.Registerer.Register(s.lastAccessed); err != nil {
 		return nil, err
 	}
 
-	if err := s.Update(args); err != nil {
-		return nil, err
+	res := w.getSecret(context.TODO())
+	if res.err != nil {
+		return nil, res.err
 	}
 
+	s.handlePolledSecret(res)
+
 	return s, nil
 }
 
-func (c *Component) Run(ctx context.Context) error {
+func (s *Component) Run(ctx context.Context) error {
+	if s.pollFrequency > 0 {
+		go s.handleSecretUpdate(ctx)
+		go s.watcher.run(ctx)
+	}
 	<-ctx.Done()
 	return nil
 }
 
 // Update is called whenever the arguments have changed.
-func (c *Component) Update(args component.Arguments) (err error) {
-	defer c.updateHealth(err)
+func (s *Component) Update(args component.Arguments) (err error) {
+	defer s.updateHealth(err)
 	newArgs := args.(Arguments)
 
+	// Create AWS and AWS's Secrets Manager client
 	awsCfg, err := aws_common_config.GenerateAWSConfig(newArgs.Options)
 	if err != nil {
 		return err
 	}
+	client := secretsmanager.NewFromConfig(*awsCfg)
 
-	// Create Secrets Manager client
-	svc := secretsmanager.NewFromConfig(*awsCfg)
+	s.mut.Lock()
+	defer s.mut.Unlock()
+	s.pollFrequency = newArgs.PollFrequency
+	s.watcher.updateValues(newArgs.SecretId, newArgs.SecretVersion, newArgs.PollFrequency, client)
 
-	result, err := svc.GetSecretValue(context.TODO(), &secretsmanager.GetSecretValueInput{
-		SecretId:     aws.String(newArgs.SecretId),
-		VersionStage: aws.String(newArgs.SecretVersion),
-	})
-	if err != nil {
-		return err
-	}
+	return nil
+}
 
-	if result == nil {
-		err = fmt.Errorf("unable to retrieve secret at path %s", newArgs.SecretId)
-		return err
+// handleSecretUpdate reads from update and error channels, setting as approriate
+func (s *Component) handleSecretUpdate(ctx context.Context) {
+	for {
+		select {
+		case r := <-s.updateChan:
+			s.handlePolledSecret(r)
+		case <-ctx.Done():
+			return
+		}
 	}
-
-	c.exportSecret(result)
-
-	return nil
 }
 
-// exportSecret converts the secret into exports and exports it to the
+// handledPolledSecret converts the secret into exports and exports it to the
 // controller.
-func (c *Component) exportSecret(secret *secretsmanager.GetSecretValueOutput) {
-	if secret != nil {
-		newExports := Exports{
-			Data: make(map[string]alloytypes.Secret),
-		}
-		newExports.Data[*secret.Name] = alloytypes.Secret(*secret.SecretString)
-		c.opts.OnStateChange(newExports)
+func (s *Component) handlePolledSecret(res result) {
+	var err error
+	if validated := res.Validate(); validated {
+		s.opts.OnStateChange(Exports{
+			Data: map[string]alloytypes.Secret{
+				res.secretId: alloytypes.Secret(res.secret),
+			},
+		})
+		s.lastAccessed.SetToCurrentTime()
+	} else {
+		s.errors.Inc()
+		err = res.err
 	}
+
+	s.updateHealth(err)
 }
 
 // CurrentHealth returns the health of the component.

internal/component/remote/aws/secrets_manager/secrets_manager_test.go

@@ -23,9 +23,10 @@ import (
 
 func Test_GetSecrets(t *testing.T) {
 	var (
-		ctx = componenttest.TestContext(t)
-		ep  = runTestLocalSecretManager(t)
-		l   = util.TestLogger(t)
+		secretId = "foo"
+		ctx      = componenttest.TestContext(t)
+		ep, _    = runTestLocalSecretManager(t, secretId)
+		l        = util.TestLogger(t)
 	)
 
 	cfg := fmt.Sprintf(`
@@ -34,8 +35,8 @@ func Test_GetSecrets(t *testing.T) {
 		key = "test"
 		secret = "test"
 	}
-	id = "foo"
-`, ep)
+	id = "%s"
+`, ep, secretId)
 
 	var args Arguments
 	require.NoError(t, syntax.Unmarshal([]byte(cfg), &args))
@@ -53,7 +54,7 @@ func Test_GetSecrets(t *testing.T) {
 	var (
 		expectExports = Exports{
 			Data: map[string]alloytypes.Secret{
-				"foo": alloytypes.Secret("bar"),
+				secretId: alloytypes.Secret("bar"),
 			},
 		}
 		actualExports = ctrl.Exports().(Exports)
@@ -64,7 +65,71 @@ func Test_GetSecrets(t *testing.T) {
 	require.Equal(t, innerComponent.CurrentHealth().Health, component.HealthTypeHealthy)
 }
 
-func runTestLocalSecretManager(t *testing.T) string {
+func Test_PollSecrets(t *testing.T) {
+	var (
+		secretId   = "foo_poll"
+		ctx        = componenttest.TestContext(t)
+		ep, client = runTestLocalSecretManager(t, secretId)
+		l          = util.TestLogger(t)
+	)
+
+	cfg := fmt.Sprintf(`
+	client {
+		endpoint = "%s"
+		key = "test"
+		secret = "test"
+	}
+
+	poll_frequency = "1s"
+
+	id = "%s"
+`, ep, secretId)
+
+	var args Arguments
+	require.NoError(t, syntax.Unmarshal([]byte(cfg), &args))
+
+	ctrl, err := componenttest.NewControllerFromID(l, "remote.aws.secrets_manager")
+	require.NoError(t, err)
+
+	go func() {
+		require.NoError(t, ctrl.Run(ctx, args))
+	}()
+
+	require.NoError(t, ctrl.WaitRunning(time.Minute))
+	require.NoError(t, ctrl.WaitExports(time.Minute))
+
+	var (
+		expectExports = Exports{
+			Data: map[string]alloytypes.Secret{
+				secretId: alloytypes.Secret("bar"),
+			},
+		}
+		actualExports = ctrl.Exports().(Exports)
+	)
+	require.Equal(t, expectExports, actualExports)
+
+	// Updated the secret to something else
+	updatedSecretString := "bar_poll"
+	result, err := client.UpdateSecret(context.TODO(), &secretsmanager.UpdateSecretInput{
+		SecretId:     &secretId,
+		SecretString: &updatedSecretString,
+	})
+	require.NoError(t, err)
+	require.NotNil(t, result)
+
+	require.NoError(t, ctrl.WaitExports(time.Minute))
+
+	expectExports = Exports{
+		Data: map[string]alloytypes.Secret{
+			secretId: alloytypes.Secret(updatedSecretString),
+		},
+	}
+	actualExports = ctrl.Exports().(Exports)
+	require.Equal(t, expectExports, actualExports)
+
+}
+
+func runTestLocalSecretManager(t *testing.T, secretId string) (string, *secretsmanager.Client) {
 	ctx := componenttest.TestContext(t)
 	container, err := testcontainers.GenericContainer(ctx, testcontainers.GenericContainerRequest{
 		ContainerRequest: testcontainers.ContainerRequest{
@@ -96,14 +161,13 @@ func runTestLocalSecretManager(t *testing.T) string {
 
 	svc := secretsmanager.NewFromConfig(*awsCfg)
 
-	secretName := "foo"
 	secretString := "bar"
 	result, err := svc.CreateSecret(context.TODO(), &secretsmanager.CreateSecretInput{
-		Name:         &secretName,
+		Name:         &secretId,
 		SecretString: &secretString,
 	})
 	require.NoError(t, err)
 	require.NotNil(t, result)
 
-	return ep
+	return ep, svc
 }