diff --git a/contrib/aws/secretsmanager/CHANGELOG.md b/contrib/aws/secretsmanager/CHANGELOG.md new file mode 100644 index 000000000..c89386dc8 --- /dev/null +++ b/contrib/aws/secretsmanager/CHANGELOG.md @@ -0,0 +1,12 @@ + + +# Changelog + +## [Unreleased] diff --git a/contrib/aws/secretsmanager/README.md b/contrib/aws/secretsmanager/README.md new file mode 100644 index 000000000..33fbe1195 --- /dev/null +++ b/contrib/aws/secretsmanager/README.md @@ -0,0 +1,160 @@ +# AWS Secrets Manager for Temporal Client Config + +> ⚠️ **This package is currently at an experimental release stage.** ⚠️ + +Package `go.temporal.io/sdk/contrib/aws/secretsmanager` is an +[`envconfig`](../../envconfig/README.md) extension that loads Temporal client +credentials (API key and/or mTLS material) from +[AWS Secrets Manager](https://aws.amazon.com/secrets-manager/). Only secret IDs +live in your config; secret material is fetched at runtime and kept in memory. + +## Add to your project + +From your application's Go module, run: + +```bash +go get go.temporal.io/sdk/contrib/aws/secretsmanager@latest +``` + +## Module versioning + +`secretsmanager` is released as a separate Go module from the core Temporal Go +SDK. See [CHANGELOG.md](CHANGELOG.md) for release notes. + +## Usage + +```go +import ( + "go.temporal.io/sdk/client" + "go.temporal.io/sdk/contrib/aws/secretsmanager" + "go.temporal.io/sdk/contrib/envconfig" +) + +opts, err := envconfig.LoadClientOptions(envconfig.LoadClientOptionsRequest{ + Extensions: []envconfig.ClientConfigExtension{secretsmanager.Extension{}}, +}) +if err != nil { + // handle error +} +c, err := client.Dial(opts) +``` + +## Configuration + +Each credential is configured with a Secrets Manager secret ID (friendly name +or full ARN), either in the TOML config file or with environment variables. +Environment variables take precedence over the file. + +| TOML (under `profile..ext.aws_secrets_manager`) | Environment variable | Replaces | +|---|---|---| +| `api_key_id` | `TEMPORAL_EXT_AWS_SECRETS_MANAGER_API_KEY_ID` | `api_key` / `TEMPORAL_API_KEY` | +| `tls.client_cert_id` | `TEMPORAL_EXT_AWS_SECRETS_MANAGER_TLS_CLIENT_CERT_ID` | `tls.client_cert_{path,data}` / `TEMPORAL_TLS_CLIENT_CERT_{PATH,DATA}` | +| `tls.client_key_id` | `TEMPORAL_EXT_AWS_SECRETS_MANAGER_TLS_CLIENT_KEY_ID` | `tls.client_key_{path,data}` / `TEMPORAL_TLS_CLIENT_KEY_{PATH,DATA}` | +| `tls.server_ca_cert_id` | `TEMPORAL_EXT_AWS_SECRETS_MANAGER_TLS_SERVER_CA_CERT_ID` | `tls.server_ca_cert_{path,data}` / `TEMPORAL_TLS_SERVER_CA_CERT_{PATH,DATA}` | + +For example: + +```toml +[profile.default] +address = "my-namespace.a1b2c.tmprl.cloud:7233" +namespace = "my-namespace.a1b2c" + +[profile.default.ext.aws_secrets_manager.tls] +client_cert_id = "temporal/client-cert-and-key" +client_key_id = "temporal/client-cert-and-key" +``` + +Rules: + +- A secret ID can't be combined with the credential it replaces, from any + source. For example, `api_key` in the file with + `TEMPORAL_EXT_AWS_SECRETS_MANAGER_API_KEY_ID` in the environment is an error. +- The client cert and key IDs must be set together. They may be the same ID, + in which case the secret holds both the PEM certificate and the PEM key. This + lets the pair be rotated atomically. +- TLS secret IDs can't be used when TLS is disabled. +- An API key secret enables TLS by default, as `api_key` does. +- Secret values are used as-is: the API key as a string, and TLS material as + PEM. Both `SecretString` and `SecretBinary` secrets are supported. +- Unknown keys under `ext.aws_secrets_manager` are rejected. + +## Fetching and caching + +When `Extension.SecretsClient` is nil, secrets are fetched through the +[AWS Parameters and Secrets Lambda Extension](https://docs.aws.amazon.com/secretsmanager/latest/userguide/retrieving-secrets_lambda.html) +if it is present. Otherwise they are fetched directly from Secrets Manager with +an AWS SDK v2 client built from the default AWS config. + +The Lambda extension is detected as follows: + +- If `PARAMETERS_SECRETS_EXTENSION_HTTP_PORT` is set to a port, the extension + is used on that port. +- If it is unset and the code is running in Lambda (`AWS_LAMBDA_FUNCTION_NAME` + is set), the extension's default port, 2773, is probed, and the extension is + used if something is listening there. Outside Lambda, the port is not probed + and secrets are fetched directly. +- Setting `PARAMETERS_SECRETS_EXTENSION_HTTP_PORT` to an empty value explicitly + disables detection so that secrets are always fetched directly. Use this if + another service in your function listens on port 2773. + +To use your own client, for example +with a specific region or credentials, set `Extension.SecretsClient`: + +```go +cfg, err := config.LoadDefaultConfig(ctx, config.WithRegion("us-west-2")) +// ... +ext := secretsmanager.Extension{SecretsClient: awssm.NewFromConfig(cfg)} +``` + +Secrets are first fetched when a client is created from the loaded options, +and client creation fails if a fetch fails. After that: + +- Secrets are cached by the plugin that the extension adds to the loaded + options, so clients created from the same options share the cache (for + example, across warm Lambda invocations). +- The API key and client cert/key are refreshed while a client is running once + the refresh interval has elapsed, so rotated secrets are picked up without + recreating the client. +- The refresh interval is `Extension.RefreshInterval` if set. Otherwise it + comes from the `SECRETS_MANAGER_TTL` environment variable, interpreted the + same way as by the Lambda extension: whole seconds, where 0 disables + caching. Unlike the Lambda extension, values over 300 are allowed. If neither is set, it defaults to 5 minutes. An invalid + `SECRETS_MANAGER_TTL` is an error when loading client options. +- If a refresh fails, the last fetched value keeps being used for up to + `Extension.MaxStaleness` (default 24 hours) after it was fetched. After a + failed refresh, the next fetch attempt waits 30 seconds. +- The server CA cert is fetched only when a client is created. + +## Observability + +The extension logs to the client's `Logger` and emits metrics to the client's +`MetricsHandler`. Clients created from the same loaded options share this +telemetry: fetches are reported to the logger and metrics handler of the most +recently created client. Load the options separately for each client that +needs its own. + +Metrics are all labeled with `secret_type` (`api_key`, `tls_client_cert`, +`tls_server_ca_cert`; client cert and client key metrics are both labeled +`tls_client_cert`) and `fetch_path` (`extension` or `sdk`). Currently, the following metrics are emitted: + +| Metric | Type | Extra labels | Description | +|---|---|---|---| +| `temporal_aws_secrets_manager_fetch_failed` | Counter | `failure_reason`, `fallback` (`stale`, `none`) | Failed fetches, and whether the previous value was used instead | +| `temporal_aws_secrets_manager_fetch_latency` | Timer | `outcome` (`success`, `failure`) | Latency of each fetch | +| `temporal_aws_secrets_manager_cache_lookup` | Counter | `result` (`hit`, `miss`, `miss_with_fallback`) | Lookups that returned a value | + +## Required IAM permissions + +The AWS credentials in use (for example, the Lambda execution role) need +`secretsmanager:GetSecretValue` on each configured secret: + +```json +{ + "Effect": "Allow", + "Action": "secretsmanager:GetSecretValue", + "Resource": "arn:aws:secretsmanager:us-west-2:123456789012:secret:temporal/*" +} +``` + +If a secret is encrypted with a customer-managed KMS key, `kms:Decrypt` on that +key is also required. diff --git a/contrib/aws/secretsmanager/cache.go b/contrib/aws/secretsmanager/cache.go new file mode 100644 index 000000000..b776ad613 --- /dev/null +++ b/contrib/aws/secretsmanager/cache.go @@ -0,0 +1,309 @@ +package secretsmanager + +import ( + "context" + "crypto/tls" + "crypto/x509" + "fmt" + "sync" + "time" + + "go.temporal.io/sdk/client" + "go.temporal.io/sdk/log" +) + +// After a failed refresh falls back to a stale value, wait this long before trying again so that an outage doesn't +// add a fetch attempt to every credential lookup. +const failedRefreshRetryInterval = 30 * time.Second + +// Bounds each secret fetch +const defaultFetchTimeout = 10 * time.Second + +// secretCache is a TTL cache of the configured credentials that falls back to the last fetched value, up to +// maxStaleness old, when a refresh fails. +type secretCache struct { + fetcher fetcher + refreshInterval time.Duration + maxStaleness time.Duration + fetchTimeout time.Duration + now func() time.Time // normally time.Now, but can be overridden in tests + + // nil if not configured + apiKey *cacheEntry[string] + // client cert and key, fetched and replaced together to prevent (cache-level) tearing on rotations + clientCert *cacheEntry[*tls.Certificate] + // the CA cert is only loaded once at client-config-time, but it's here for consistency + serverCACert *cacheEntry[*x509.CertPool] +} + +type secretRef struct { + id string + secretType secretType +} + +// cacheEntry is one or more secrets that are fetched and replaced together, cached as a single parsed value. +type cacheEntry[T any] struct { + // Distinct secret IDs. refs[0].secretType is used as the secret_type metric label. + refs []secretRef + // Called on newly fetched secrets (in refs order) to produce the cached value. An error is treated as a failed + // fetch. + parse func(secrets []secretValue) (T, error) + + // Guards the fields below + mu sync.Mutex + // The fetch in progress, or nil if none + refreshState *refreshState[T] + // In refs order, nil until the first successful fetch + versionIDs []string + value T + fetchedAt time.Time + // Last failed fetch since the last successful fetch, or zero if none + failedAt time.Time +} + +// A fetch in progress. value and err are set before done is closed. +type refreshState[T any] struct { + done chan struct{} + value T + err error +} + +func newSecretCache(f fetcher, ids secretIDs, refreshInterval time.Duration, maxStaleness time.Duration) *secretCache { + c := &secretCache{ + fetcher: f, + refreshInterval: refreshInterval, + maxStaleness: maxStaleness, + fetchTimeout: defaultFetchTimeout, + now: time.Now, + } + if ids.APIKey != "" { + c.apiKey = &cacheEntry[string]{ + refs: []secretRef{{id: ids.APIKey, secretType: secretTypeAPIKey}}, + parse: func(secrets []secretValue) (string, error) { + return string(secrets[0].value), nil + }, + } + } + if ids.ClientCert != "" { + c.clientCert = newClientCertEntry(ids.ClientCert, ids.ClientKey) + } + if ids.ServerCACert != "" { + c.serverCACert = newServerCACertEntry(ids.ServerCACert) + } + return c +} + +// A newly fetched pair that doesn't match (e.g. fetched partway through rotating separate cert and key secrets) is +// treated as a failed fetch, so the previous pair keeps being used. certID and keyID may be the same if bundled. +func newClientCertEntry(certID, keyID string) *cacheEntry[*tls.Certificate] { + refs := []secretRef{{id: certID, secretType: secretTypeClientCert}} + if keyID != certID { + refs = append(refs, secretRef{id: keyID, secretType: secretTypeClientKey}) + } + parse := func(secrets []secretValue) (*tls.Certificate, error) { + // for a bundled cert/key there's only one value, and X509KeyPair picks out the appropriate PEM blocks + certPEM := secrets[0].value + keyPEM := secrets[len(secrets)-1].value + cert, err := tls.X509KeyPair(certPEM, keyPEM) + if err != nil { + return nil, &invalidResponseError{fmt.Errorf( + "failed loading TLS client cert/key from AWS Secrets Manager secrets %q and %q: %w", certID, keyID, err)} + } + return &cert, nil + } + return &cacheEntry[*tls.Certificate]{refs: refs, parse: parse} +} + +func newServerCACertEntry(id string) *cacheEntry[*x509.CertPool] { + parse := func(secrets []secretValue) (*x509.CertPool, error) { + pool := x509.NewCertPool() + if !pool.AppendCertsFromPEM(secrets[0].value) { + return nil, &invalidResponseError{fmt.Errorf("no PEM certificates found in TLS server CA cert secret %q", id)} + } + return pool, nil + } + return &cacheEntry[*x509.CertPool]{ + refs: []secretRef{{id: id, secretType: secretTypeServerCACert}}, + parse: parse, + } +} + +// The get methods must only be called for configured credentials. Returned values are shared across callers and must +// not be mutated. + +func (c *secretCache) getAPIKey(ctx context.Context, tel telemetry) (string, error) { + return getEntry(ctx, c, c.apiKey, tel) +} + +func (c *secretCache) getClientCert(ctx context.Context, tel telemetry) (*tls.Certificate, error) { + return getEntry(ctx, c, c.clientCert, tel) +} + +func (c *secretCache) getServerCACert(ctx context.Context, tel telemetry) (*x509.CertPool, error) { + return getEntry(ctx, c, c.serverCACert, tel) +} + +// Get the parsed value associated with a given cache entry +func getEntry[T any](ctx context.Context, c *secretCache, e *cacheEntry[T], tel telemetry) (T, error) { + var zero T + metrics := tel.metrics + if metrics == nil { + metrics = client.MetricsNopHandler + } + metrics = metrics.WithTags(map[string]string{ + labelSecretType: string(e.refs[0].secretType), + labelFetchPath: string(c.fetcher.path()), + }) + logger := tel.logger + if logger == nil { + logger = nopLogger{} + } + recordLookup := func(result lookupResult) { + metrics.WithTags(map[string]string{labelResult: string(result)}).Counter(metricCacheLookup).Inc(1) + } + + // Check for a usable value in the cache + e.mu.Lock() + now := c.now() + hasValue := e.versionIDs != nil + canUseStale := hasValue && now.Sub(e.fetchedAt) < c.maxStaleness + stale := e.value + var result lookupResult + switch { + case hasValue && now.Sub(e.fetchedAt) < c.refreshInterval, + // Another caller is refreshing, so the current value is still the latest known one + canUseStale && e.refreshState != nil: + result = lookupHit + case canUseStale && now.Sub(e.failedAt) < failedRefreshRetryInterval: + result = lookupMissWithFallback + } + if result != "" { + e.mu.Unlock() + recordLookup(result) + return stale, nil + } + + // Wait for a fetch, starting one if none is in progress. + r := e.refreshState + if r == nil { + r = &refreshState[T]{done: make(chan struct{})} + e.refreshState = r + // separate context so the cache's failure state isn't affected by caller context deadlines + go refresh(context.WithoutCancel(ctx), c, e, r, now, canUseStale, metrics, logger) + } + e.mu.Unlock() + select { + case <-r.done: + case <-ctx.Done(): + if canUseStale { + recordLookup(lookupMissWithFallback) + return stale, nil + } + return zero, ctx.Err() + } + if r.err != nil { + if canUseStale { + recordLookup(lookupMissWithFallback) + return stale, nil + } + return zero, r.err + } + recordLookup(lookupMiss) + return r.value, nil +} + +// Fetch an entry's secrets and record the outcome in the cache. +func refresh[T any]( + ctx context.Context, + c *secretCache, + e *cacheEntry[T], + r *refreshState[T], + start time.Time, + canUseStale bool, + metrics client.MetricsHandler, + logger log.Logger, +) { + ctx, cancel := context.WithTimeout(ctx, c.fetchTimeout) + defer cancel() + value, versionIDs, err := fetch(ctx, c, e) + latency := c.now().Sub(start) + + // done is closed last so that the fetch's metrics and logs are recorded by the time waiters return + // (for the unit tests, and also for log ordering in production) + defer close(r.done) + e.mu.Lock() + r.value, r.err = value, err + e.refreshState = nil + prevVersionIDs, prevFetchedAt, prevFailed := e.versionIDs, e.fetchedAt, !e.failedAt.IsZero() + if err != nil { + e.failedAt = start + } else { + e.value = value + e.versionIDs = versionIDs + e.fetchedAt = start + e.failedAt = time.Time{} + } + e.mu.Unlock() + + // handle telemetry outside the lock + if err != nil { + metrics.WithTags(map[string]string{labelOutcome: string(outcomeFailure)}).Timer(metricFetchLatency).Record(latency) + fb := fallbackNone + if canUseStale { + fb = fallbackStale + } + metrics.WithTags(map[string]string{ + labelFailureReason: string(classifyFailure(err)), + labelFallback: string(fb), + }).Counter(metricFetchFailed).Inc(1) + + if canUseStale { + for _, ref := range e.refs { + logger.Warn("Failed refreshing AWS Secrets Manager secret, using previously fetched value", + "SecretID", ref.id, + "SecretType", ref.secretType, + "Error", err, + "Age", start.Sub(prevFetchedAt), + "MaxStaleness", c.maxStaleness) + } + } + return + } + + metrics.WithTags(map[string]string{labelOutcome: string(outcomeSuccess)}).Timer(metricFetchLatency).Record(latency) + for i, ref := range e.refs { + if prevVersionIDs == nil { + logger.Info("Fetched AWS Secrets Manager secret", + "SecretID", ref.id, + "SecretType", ref.secretType, + "FetchPath", c.fetcher.path(), + "VersionID", versionIDs[i]) + } else if prevFailed || versionIDs[i] != prevVersionIDs[i] { + logger.Info("Fetched updated AWS Secrets Manager secret", + "SecretID", ref.id, + "SecretType", ref.secretType, + "PreviousVersionID", prevVersionIDs[i], + "VersionID", versionIDs[i]) + } + } +} + +// Fetch all of an entry's secrets and parse them. Returns the parsed value and the version IDs in refs order. +func fetch[T any](ctx context.Context, c *secretCache, e *cacheEntry[T]) (T, []string, error) { + var zero T + secrets := make([]secretValue, 0, len(e.refs)) + versionIDs := make([]string, 0, len(e.refs)) + for _, ref := range e.refs { + s, err := c.fetcher.fetch(ctx, ref.id) + if err != nil { + return zero, nil, fmt.Errorf("failed fetching AWS Secrets Manager secret %q: %w", ref.id, err) + } + secrets = append(secrets, s) + versionIDs = append(versionIDs, s.versionID) + } + value, err := e.parse(secrets) + if err != nil { + return zero, nil, err + } + return value, versionIDs, nil +} diff --git a/contrib/aws/secretsmanager/cache_test.go b/contrib/aws/secretsmanager/cache_test.go new file mode 100644 index 000000000..9b3c353be --- /dev/null +++ b/contrib/aws/secretsmanager/cache_test.go @@ -0,0 +1,414 @@ +package secretsmanager + +import ( + "context" + "encoding/pem" + "errors" + "fmt" + "sync/atomic" + "testing" + "time" + + "github.com/aws/smithy-go" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +type funcFetcher func(ctx context.Context, secretID string) (secretValue, error) + +func (f funcFetcher) fetch(ctx context.Context, secretID string) (secretValue, error) { + return f(ctx, secretID) +} +func (funcFetcher) path() fetchPath { return fetchPathSDK } + +func TestSecretCache(t *testing.T) { + var ( + value = "v1" + version = "ver1" + fetchErr error + fetches int + now = time.Unix(1000, 0) + ) + c := newSecretCache(funcFetcher(func(context.Context, string) (secretValue, error) { + fetches++ + now = now.Add(time.Millisecond) // fetch latency + if fetchErr != nil { + return secretValue{}, fetchErr + } + return secretValue{value: []byte(value), versionID: version}, nil + }), secretIDs{APIKey: "id"}, time.Minute, time.Hour) + c.now = func() time.Time { return now } + logger := &recordingLogger{} + metrics := newRecordingMetrics() + tel := telemetry{logger: logger, metrics: metrics} + get := func() (string, error) { + return c.getAPIKey(context.Background(), tel) + } + lookups := func(result lookupResult) int64 { + return metrics.counter(metricCacheLookup, + labelSecretType, secretTypeAPIKey, labelFetchPath, fetchPathSDK, labelResult, result) + } + failures := func(fallback fallback) int64 { + return metrics.counter(metricFetchFailed, labelSecretType, secretTypeAPIKey, labelFetchPath, fetchPathSDK, + labelFailureReason, failureOther, labelFallback, fallback) + } + latencies := func(outcome fetchOutcome) []time.Duration { + return metrics.timer(metricFetchLatency, + labelSecretType, secretTypeAPIKey, labelFetchPath, fetchPathSDK, labelOutcome, outcome) + } + + // Initial fetch, then cached until refresh interval + v, err := get() + require.NoError(t, err) + require.Equal(t, "v1", v) + require.Equal(t, []string{ + "INFO Fetched AWS Secrets Manager secret SecretID=id SecretType=api_key FetchPath=sdk VersionID=ver1", + }, logger.take()) + require.Equal(t, int64(1), lookups(lookupMiss)) + require.Equal(t, []time.Duration{time.Millisecond}, latencies(outcomeSuccess)) + value = "v2" + now = now.Add(59 * time.Second) + v, err = get() + require.NoError(t, err) + require.Equal(t, "v1", v) + require.Equal(t, 1, fetches) + require.Equal(t, int64(1), lookups(lookupHit)) + + // Refreshed after interval. Same version isn't logged as a rotation. + now = now.Add(time.Second) + v, err = get() + require.NoError(t, err) + require.Equal(t, "v2", v) + require.Equal(t, 2, fetches) + require.Empty(t, logger.take()) + require.Equal(t, int64(2), lookups(lookupMiss)) + + // New version is logged as a rotation + version = "ver2" + now = now.Add(time.Minute) + _, err = get() + require.NoError(t, err) + require.Equal(t, []string{ + "INFO Fetched updated AWS Secrets Manager secret SecretID=id SecretType=api_key PreviousVersionID=ver1 " + + "VersionID=ver2", + }, logger.take()) + + // Failed refresh falls back to stale value, and doesn't retry until the retry interval passes + fetchErr = errors.New("outage") + now = now.Add(time.Minute) + v, err = get() + require.NoError(t, err) + require.Equal(t, "v2", v) + require.Equal(t, 4, fetches) + require.Equal(t, []string{ + "WARN Failed refreshing AWS Secrets Manager secret, using previously fetched value SecretID=id " + + "SecretType=api_key Error=failed fetching AWS Secrets Manager secret \"id\": outage Age=1m0.001s MaxStaleness=1h0m0s", + }, logger.take()) + require.Equal(t, int64(1), failures(fallbackStale)) + require.Equal(t, int64(1), lookups(lookupMissWithFallback)) + require.Equal(t, []time.Duration{time.Millisecond}, latencies(outcomeFailure)) + now = now.Add(failedRefreshRetryInterval - time.Second) + v, err = get() + require.NoError(t, err) + require.Equal(t, "v2", v) + require.Equal(t, 4, fetches) + require.Empty(t, logger.take()) + require.Equal(t, int64(2), lookups(lookupMissWithFallback)) + now = now.Add(time.Second) + _, err = get() + require.NoError(t, err) + require.Equal(t, 5, fetches) + require.Len(t, logger.take(), 1) + require.Equal(t, int64(2), failures(fallbackStale)) + + // Nil logger and metrics handler are allowed + now = now.Add(failedRefreshRetryInterval) + _, err = c.getAPIKey(context.Background(), telemetry{}) + require.NoError(t, err) + require.Equal(t, 6, fetches) + require.Empty(t, logger.take()) + + // Fails once past max staleness + now = now.Add(time.Hour) + _, err = get() + require.ErrorContains(t, err, `failed fetching AWS Secrets Manager secret "id": outage`) + require.Empty(t, logger.take()) + require.Equal(t, int64(1), failures(fallbackNone)) + require.Equal(t, int64(3), lookups(lookupMissWithFallback)) + + // Recovers. Logged even though the version is unchanged. + fetchErr = nil + value = "v3" + now = now.Add(time.Minute) + v, err = get() + require.NoError(t, err) + require.Equal(t, "v3", v) + require.Equal(t, []string{ + "INFO Fetched updated AWS Secrets Manager secret SecretID=id SecretType=api_key " + + "PreviousVersionID=ver2 VersionID=ver2", + }, logger.take()) + + // Later successes aren't logged as recoveries + now = now.Add(time.Minute) + _, err = get() + require.NoError(t, err) + require.Empty(t, logger.take()) +} + +func TestSecretCacheInitialFetchError(t *testing.T) { + fetchErr := error(&smithy.GenericAPIError{Code: "AccessDeniedException", Message: "denied"}) + caPEM, _ := generateCert(t, "ca") + c := newSecretCache(funcFetcher(func(context.Context, string) (secretValue, error) { + if fetchErr != nil { + return secretValue{}, fetchErr + } + return secretValue{value: []byte(caPEM), versionID: "ver"}, nil + }), secretIDs{ServerCACert: "id"}, time.Minute, time.Hour) + logger := &recordingLogger{} + metrics := newRecordingMetrics() + tel := telemetry{logger: logger, metrics: metrics} + _, err := c.getServerCACert(context.Background(), tel) + require.ErrorContains(t, err, `failed fetching AWS Secrets Manager secret "id"`) + require.ErrorContains(t, err, "denied") + require.Empty(t, logger.take()) + require.Equal(t, int64(1), metrics.counter(metricFetchFailed, labelSecretType, secretTypeServerCACert, + labelFetchPath, fetchPathSDK, labelFailureReason, failureAccessDenied, labelFallback, fallbackNone)) + require.Len(t, metrics.timer(metricFetchLatency, labelSecretType, secretTypeServerCACert, + labelFetchPath, fetchPathSDK, labelOutcome, outcomeFailure), 1) + require.Empty(t, metrics.counters(metricCacheLookup)) + + // A first success after failures is logged only as a first fetch + fetchErr = nil + _, err = c.getServerCACert(context.Background(), tel) + require.NoError(t, err) + require.Equal(t, []string{ + "INFO Fetched AWS Secrets Manager secret SecretID=id SecretType=tls_server_ca_cert FetchPath=sdk VersionID=ver", + }, logger.take()) +} + +func TestClientCertPair(t *testing.T) { + certA, keyA := generateCert(t, "a") + certB, keyB := generateCert(t, "b") + fake := newFakeClient(map[string]string{"cert": certA, "key": keyA}) + c := newSecretCache(&sdkFetcher{client: fake}, secretIDs{ClientCert: "cert", ClientKey: "key"}, time.Minute, + time.Hour) + now := time.Unix(1000, 0) + c.now = func() time.Time { return now } + logger := &recordingLogger{} + metrics := newRecordingMetrics() + tel := telemetry{logger: logger, metrics: metrics} + requireCert := func(expectedPEM string) { + t.Helper() + cert, err := c.getClientCert(context.Background(), tel) + require.NoError(t, err) + block, _ := pem.Decode([]byte(expectedPEM)) + require.Equal(t, block.Bytes, cert.Certificate[0]) + } + setSecret := func(id, value string) { + fake.mu.Lock() + defer fake.mu.Unlock() + fake.secrets[id] = value + } + + // Each secret is logged separately + requireCert(certA) + require.Equal(t, []string{ + "INFO Fetched AWS Secrets Manager secret SecretID=cert SecretType=tls_client_cert FetchPath=sdk VersionID=" + + fakeVersionID(certA), + "INFO Fetched AWS Secrets Manager secret SecretID=key SecretType=tls_client_key FetchPath=sdk VersionID=" + + fakeVersionID(keyA), + }, logger.take()) + + // Fetching partway through rotation gets a mismatched pair, which falls back to the previous pair + setSecret("cert", certB) + now = now.Add(time.Minute) + requireCert(certA) + logs := logger.take() + require.Len(t, logs, 2) + require.Contains(t, logs[0], "WARN Failed refreshing AWS Secrets Manager secret, using previously fetched value "+ + "SecretID=cert SecretType=tls_client_cert") + require.Contains(t, logs[1], "SecretID=key SecretType=tls_client_key") + require.Equal(t, int64(1), metrics.counter(metricFetchFailed, labelSecretType, secretTypeClientCert, + labelFetchPath, fetchPathSDK, labelFailureReason, failureInvalidResponse, labelFallback, fallbackStale)) + + // Once rotation completes, the new pair is used after the retry interval. Both secrets are always fetched + // together. + setSecret("key", keyB) + now = now.Add(failedRefreshRetryInterval - time.Second) + requireCert(certA) + require.Equal(t, 2, fake.callCount("cert")) + now = now.Add(time.Second) + requireCert(certB) + require.Equal(t, []string{ + "INFO Fetched updated AWS Secrets Manager secret SecretID=cert SecretType=tls_client_cert " + + "PreviousVersionID=" + fakeVersionID(certA) + " VersionID=" + fakeVersionID(certB), + "INFO Fetched updated AWS Secrets Manager secret SecretID=key SecretType=tls_client_key " + + "PreviousVersionID=" + fakeVersionID(keyA) + " VersionID=" + fakeVersionID(keyB), + }, logger.take()) + require.Equal(t, 3, fake.callCount("cert")) + require.Equal(t, 3, fake.callCount("key")) +} + +func TestSecretCacheCallerCanceled(t *testing.T) { + var fetches atomic.Int32 + var blocking atomic.Bool // blocks the fetch until release is closed + started := make(chan struct{}) // emits an event when a fetch blocks + release := make(chan struct{}) // unblocks a blocked fetch + c := newSecretCache(funcFetcher(func(ctx context.Context, _ string) (secretValue, error) { + n := fetches.Add(1) + if blocking.Load() { + started <- struct{}{} + <-release + } + if err := ctx.Err(); err != nil { + return secretValue{}, err + } + return secretValue{value: fmt.Appendf(nil, "v%d", n), versionID: fmt.Sprint(n)}, nil + }), secretIDs{APIKey: "id"}, time.Minute, time.Hour) + now := time.Unix(1000, 0) + c.now = func() time.Time { return now } + metrics := newRecordingMetrics() + tel := telemetry{metrics: metrics} + lookups := func(result lookupResult) int64 { + return metrics.counter(metricCacheLookup, + labelSecretType, secretTypeAPIKey, labelFetchPath, fetchPathSDK, labelResult, result) + } + _, err := c.getAPIKey(context.Background(), tel) + require.NoError(t, err) + + // A caller that gives up waiting for a refresh falls back to the stale value + now = now.Add(time.Minute) + blocking.Store(true) + ctx, cancel := context.WithCancel(context.Background()) + result := make(chan string, 1) + go func() { + v, err := c.getAPIKey(ctx, tel) + assert.NoError(t, err) + result <- v + }() + <-started + cancel() + require.Equal(t, "v1", <-result) + require.Equal(t, int64(1), lookups(lookupMissWithFallback)) + + // The fetch isn't canceled along with the caller, and its result is cached + c.apiKey.mu.Lock() + r := c.apiKey.refreshState + c.apiKey.mu.Unlock() + close(release) + <-r.done + v, err := c.getAPIKey(context.Background(), tel) + require.NoError(t, err) + require.Equal(t, "v2", v) + require.Equal(t, int32(2), fetches.Load()) + require.Equal(t, int64(1), lookups(lookupHit)) + require.Empty(t, metrics.counters(metricFetchFailed)) +} + +func TestSecretCacheFetchTimeout(t *testing.T) { + var hang atomic.Bool + c := newSecretCache(funcFetcher(func(ctx context.Context, _ string) (secretValue, error) { + if hang.Load() { + <-ctx.Done() + return secretValue{}, ctx.Err() + } + return secretValue{value: []byte("v"), versionID: "ver"}, nil + }), secretIDs{APIKey: "id"}, time.Minute, time.Hour) + c.fetchTimeout = time.Millisecond + now := time.Unix(1000, 0) + c.now = func() time.Time { return now } + metrics := newRecordingMetrics() + tel := telemetry{metrics: metrics} + failures := func(fallback fallback) int64 { + return metrics.counter(metricFetchFailed, labelSecretType, secretTypeAPIKey, labelFetchPath, fetchPathSDK, + labelFailureReason, failureTimeout, labelFallback, fallback) + } + _, err := c.getAPIKey(context.Background(), tel) + require.NoError(t, err) + + // A hung refresh times out even though the caller didn't, and is recorded as a failure with stale fallback + now = now.Add(time.Minute) + hang.Store(true) + v, err := c.getAPIKey(context.Background(), tel) + require.NoError(t, err) + require.Equal(t, "v", v) + require.Equal(t, int64(1), failures(fallbackStale)) + + // Without a usable value, the timeout is returned + now = now.Add(time.Hour) + _, err = c.getAPIKey(context.Background(), tel) + require.ErrorIs(t, err, context.DeadlineExceeded) + require.Equal(t, int64(1), failures(fallbackNone)) +} + +func TestSecretCacheConcurrentRefresh(t *testing.T) { + var fetches atomic.Int32 + var blocking atomic.Bool // blocks the fetch until release is closed + started := make(chan struct{}) // emits an event when a fetch first blocks + release := make(chan struct{}) // unblocks a blocked fetch + c := newSecretCache(funcFetcher(func(context.Context, string) (secretValue, error) { + n := fetches.Add(1) + if blocking.Load() { + started <- struct{}{} + <-release + } + // value/version increments with each fetch + return secretValue{value: fmt.Appendf(nil, "v%d", n), versionID: fmt.Sprint(n)}, nil + }), secretIDs{APIKey: "api-key"}, time.Minute, time.Hour) + now := time.Unix(1000, 0) + c.now = func() time.Time { return now } + metrics := newRecordingMetrics() + // api-key and other function as two independent entries, shouldn't block each other + other := &cacheEntry[string]{ + refs: []secretRef{{id: "other", secretType: secretTypeServerCACert}}, + parse: func(secrets []secretValue) (string, error) { return string(secrets[0].value), nil }, + } + get := func(ctx context.Context, id string) (string, error) { + tel := telemetry{metrics: metrics} + if id == "other" { + return getEntry(ctx, c, other, tel) + } + return c.getAPIKey(ctx, tel) + } + getAsync := func(id string) <-chan string { + ch := make(chan string, 1) + go func() { + v, err := get(context.Background(), id) + assert.NoError(t, err) + ch <- v + }() + <-started + return ch + } + + // Nothing blocks, get returns normally + v, err := get(context.Background(), "api-key") + require.NoError(t, err) + require.Equal(t, "v1", v) + + // Block an api-key refresh while in progress + now = now.Add(time.Minute) + blocking.Store(true) + refreshed := getAsync("api-key") + + // While a refresh is in progress, other callers get the current value without waiting + v, err = get(context.Background(), "api-key") + require.NoError(t, err) + require.Equal(t, "v1", v) // old value + require.Equal(t, int64(1), metrics.counter(metricCacheLookup, + labelSecretType, secretTypeAPIKey, labelFetchPath, fetchPathSDK, labelResult, lookupHit)) + + // Block a fetch of the other entry while in progress + fetched := getAsync("other") + // Without a usable value, callers wait for the fetch in progress, and can give up + ctx, cancel := context.WithCancel(context.Background()) + cancel() + _, err = get(ctx, "other") + require.ErrorIs(t, err, context.Canceled) + + // Unblock the refreshes and verify the final results + close(release) + require.Equal(t, "v2", <-refreshed) + require.Equal(t, "v3", <-fetched) + require.Equal(t, int32(3), fetches.Load()) +} diff --git a/contrib/aws/secretsmanager/doc.go b/contrib/aws/secretsmanager/doc.go new file mode 100644 index 000000000..4e84eff73 --- /dev/null +++ b/contrib/aws/secretsmanager/doc.go @@ -0,0 +1,72 @@ +// Package secretsmanager provides an [go.temporal.io/sdk/contrib/envconfig.ClientConfigExtension] that loads Temporal +// client credentials (API key and/or TLS material) from AWS Secrets Manager, so that only secret IDs, never secret +// material, need to live in static config. +// +// # Configuration +// +// Each credential is configured with a Secrets Manager secret ID (name or full ARN), either in the TOML config file: +// +// [profile.default.ext.aws_secrets_manager] +// api_key_id = "my-temporal-api-key" +// +// [profile.default.ext.aws_secrets_manager.tls] +// client_cert_id = "my-temporal-client-cert" +// client_key_id = "my-temporal-client-key" +// server_ca_cert_id = "my-temporal-server-ca-cert" +// +// or with environment variables, which take precedence over the file: +// +// TEMPORAL_EXT_AWS_SECRETS_MANAGER_API_KEY_ID +// TEMPORAL_EXT_AWS_SECRETS_MANAGER_TLS_CLIENT_CERT_ID +// TEMPORAL_EXT_AWS_SECRETS_MANAGER_TLS_CLIENT_KEY_ID +// TEMPORAL_EXT_AWS_SECRETS_MANAGER_TLS_SERVER_CA_CERT_ID +// +// A secret ID cannot be combined with its Temporal-native counterpart (e.g. api_key / TEMPORAL_API_KEY, or +// client_cert_path / client_cert_data), from any source. The client cert and key IDs must be set together. They may +// be the same ID, in which case the secret must contain both the PEM certificate and key, which allows the pair to be +// rotated atomically. +// +// Secret values are used as-is: the API key as a string, and TLS material as PEM. +// +// # Usage +// +// opts, err := envconfig.LoadClientOptions(envconfig.LoadClientOptionsRequest{ +// Extensions: []envconfig.ClientConfigExtension{secretsmanager.Extension{}}, +// }) +// +// # Fetching and caching +// +// By default, secrets are fetched through the AWS Parameters and Secrets Lambda Extension when it is detected, and +// otherwise directly from Secrets Manager using an AWS SDK client created from the default AWS config. Set +// [Extension.SecretsClient] to use a specific client instead. The Lambda extension is used on the port in the +// PARAMETERS_SECRETS_EXTENSION_HTTP_PORT environment variable if it is set, or, if it is unset and running in Lambda +// (AWS_LAMBDA_FUNCTION_NAME is set), on the extension's default port (2773) if something is listening there when +// options are loaded. If another service uses port 2773, set PARAMETERS_SECRETS_EXTENSION_HTTP_PORT to an empty value +// to disable detection and always fetch directly. +// +// Secrets are first fetched when a client is created from the loaded options, failing client creation if a fetch +// fails. Fetched secrets are cached by the plugin returned from the extension, so clients created from the same +// loaded options share them. The API key and client cert/key are refreshed during the lifetime of a client once +// [Extension.RefreshInterval] (which defaults from the SECRETS_MANAGER_TTL env var, or 5 minutes if unset, like the +// Lambda extension) elapses, so rotated secrets are picked up without recreating the client. If a refresh fails, +// the previous value continues to be used for up to [Extension.MaxStaleness]. The server CA cert is only fetched when +// a client is created. +// +// # Observability +// +// Fetches, rotations, fallbacks to stale values, and recoveries are logged to the client's [client.Options.Logger], +// and metrics are emitted to its [client.Options.MetricsHandler]. Clients created from the same loaded options share +// this telemetry: fetches are reported to the logger and metrics handler of the most recently created client. Load +// the options separately for each client that needs its own. +// Secret material is never logged or included in metrics. The metrics, all labeled with secret_type and fetch_path, +// are: +// +// temporal_aws_secrets_manager_fetch_failed (counter, also labeled with failure_reason and fallback) +// temporal_aws_secrets_manager_fetch_latency (timer, also labeled with outcome) +// temporal_aws_secrets_manager_cache_lookup (counter, also labeled with result) +// +// The AWS credentials in use must allow secretsmanager:GetSecretValue on the configured secrets. +// If a secret is encrypted with a customer-managed KMS key, `kms:Decrypt` on that key is also required. +// +// NOTE: Experimental +package secretsmanager diff --git a/contrib/aws/secretsmanager/extension.go b/contrib/aws/secretsmanager/extension.go new file mode 100644 index 000000000..1d95effdb --- /dev/null +++ b/contrib/aws/secretsmanager/extension.go @@ -0,0 +1,327 @@ +package secretsmanager + +import ( + "context" + "crypto/tls" + "fmt" + "math" + "os" + "sort" + "strconv" + "sync/atomic" + "time" + + awssm "github.com/aws/aws-sdk-go-v2/service/secretsmanager" + + "go.temporal.io/sdk/client" + "go.temporal.io/sdk/contrib/envconfig" +) + +const ( + // ExtensionName is the name of the extension, which is also its table key under profile..ext. + ExtensionName = "aws_secrets_manager" + + pluginName = "go.temporal.io/sdk/contrib/aws/secretsmanager" + + envAPIKeyID = "TEMPORAL_EXT_AWS_SECRETS_MANAGER_API_KEY_ID" + envClientCertID = "TEMPORAL_EXT_AWS_SECRETS_MANAGER_TLS_CLIENT_CERT_ID" + envClientKeyID = "TEMPORAL_EXT_AWS_SECRETS_MANAGER_TLS_CLIENT_KEY_ID" + envServerCACertID = "TEMPORAL_EXT_AWS_SECRETS_MANAGER_TLS_SERVER_CA_CERT_ID" + + // Same env var and default as the AWS Parameters and Secrets Lambda Extension's cache TTL + envSecretsManagerTTL = "SECRETS_MANAGER_TTL" + defaultRefreshInterval = 5 * time.Minute + defaultMaxStaleness = 24 * time.Hour +) + +// SecretsClient is the subset of the AWS SDK v2 Secrets Manager client used to fetch secrets. It is implemented by +// [awssm.Client]. +type SecretsClient interface { + GetSecretValue( + ctx context.Context, + params *awssm.GetSecretValueInput, + optFns ...func(*awssm.Options), + ) (*awssm.GetSecretValueOutput, error) +} + +// Extension is an [envconfig.ClientConfigExtension] that loads client credentials from AWS Secrets Manager. The zero +// value is ready to use. See the package documentation for configuration details. +// +// NOTE: Experimental +type Extension struct { + // Secrets Manager client used to fetch secrets. If nil, secrets are fetched through the AWS Parameters and Secrets + // Lambda Extension if it is detected, and otherwise with a client created from the default AWS config. See the + // package documentation for how the extension is detected. + SecretsClient SecretsClient + // How long a fetched secret is used before it is refreshed. If unset, defaults to the SECRETS_MANAGER_TTL env var + // (non-negative whole seconds, where 0 refreshes on every use) as used by the AWS Parameters and Secrets Lambda + // Extension, or 5 minutes if that is unset. + RefreshInterval time.Duration + // How long after its last successful fetch a secret may still be used if refreshing it fails. Defaults to 24 hours. + // Values less than RefreshInterval are treated as RefreshInterval. + MaxStaleness time.Duration +} + +var _ envconfig.ClientConfigExtension = Extension{} + +// Name implements [envconfig.ClientConfigExtension.Name]. +func (Extension) Name() string { return ExtensionName } + +// ClientPlugin implements [envconfig.ClientConfigExtension.ClientPlugin]. +func (e Extension) ClientPlugin(req envconfig.ClientConfigExtensionRequest) (client.Plugin, error) { + ids, err := loadSecretIDs(req) + if err != nil { + return nil, err + } + if ids == (secretIDs{}) { + return nil, nil + } + + refreshInterval, err := e.refreshInterval() + if err != nil { + return nil, err + } + maxStaleness := e.MaxStaleness + if maxStaleness <= 0 { + maxStaleness = defaultMaxStaleness + } + maxStaleness = max(maxStaleness, refreshInterval) + + var f fetcher + if e.SecretsClient != nil { + f = &sdkFetcher{client: e.SecretsClient} + } else { + f = newDefaultFetcher() + } + return &plugin{ids: ids, cache: newSecretCache(f, ids, refreshInterval, maxStaleness)}, nil +} + +func (e Extension) refreshInterval() (time.Duration, error) { + if e.RefreshInterval > 0 { + return e.RefreshInterval, nil + } + // Read from the process env rather than the config env, like PARAMETERS_SECRETS_EXTENSION_HTTP_PORT, since this + // is Lambda runtime config shared with the Lambda extension rather than Temporal client config + s, ok := os.LookupEnv(envSecretsManagerTTL) + if !ok || s == "" { + return defaultRefreshInterval, nil + } + secs, err := strconv.ParseInt(s, 10, 64) + if err != nil || secs < 0 || secs > int64(math.MaxInt64/time.Second) { + return 0, fmt.Errorf("invalid %v %q, must be a non-negative whole number of seconds", envSecretsManagerTTL, s) + } + return time.Duration(secs) * time.Second, nil +} + +type secretIDs struct { + APIKey string + ClientCert string + ClientKey string + ServerCACert string +} + +func (s secretIDs) hasTLS() bool { + return s.ClientCert != "" || s.ClientKey != "" || s.ServerCACert != "" +} + +func loadSecretIDs(req envconfig.ClientConfigExtensionRequest) (secretIDs, error) { + var ids secretIDs + if err := ids.applyTOML(req.Profile.Ext[ExtensionName]); err != nil { + return secretIDs{}, err + } + if req.Env != nil { + // Like core envconfig, set-but-empty env vars override (and so can unset) file values. + for env, field := range map[string]*string{ + envAPIKeyID: &ids.APIKey, + envClientCertID: &ids.ClientCert, + envClientKeyID: &ids.ClientKey, + envServerCACertID: &ids.ServerCACert, + } { + if s, ok := req.Env.LookupEnv(env); ok { + *field = s + } + } + } + if err := ids.validate(req.Profile); err != nil { + return secretIDs{}, err + } + return ids, nil +} + +func (s *secretIDs) applyTOML(table map[string]any) error { + for key, val := range table { + var err error + switch key { + case "api_key_id": + s.APIKey, err = tomlString(key, val) + case "tls": + tlsTable, ok := val.(map[string]any) + if !ok { + return fmt.Errorf("ext.%v.tls must be a table, got %T", ExtensionName, val) + } + err = s.applyTLSTOML(tlsTable) + default: + err = fmt.Errorf("unknown key %q in ext.%v", key, ExtensionName) + } + if err != nil { + return err + } + } + return nil +} + +func (s *secretIDs) applyTLSTOML(table map[string]any) error { + for key, val := range table { + var err error + switch key { + case "client_cert_id": + s.ClientCert, err = tomlString("tls."+key, val) + case "client_key_id": + s.ClientKey, err = tomlString("tls."+key, val) + case "server_ca_cert_id": + s.ServerCACert, err = tomlString("tls."+key, val) + default: + err = fmt.Errorf("unknown key %q in ext.%v.tls", key, ExtensionName) + } + if err != nil { + return err + } + } + return nil +} + +func tomlString(key string, val any) (string, error) { + s, ok := val.(string) + if !ok { + return "", fmt.Errorf("ext.%v.%v must be a string, got %T", ExtensionName, key, val) + } + return s, nil +} + +func (s secretIDs) validate(prof *envconfig.ClientConfigProfile) error { + var conflicts []string + if s.APIKey != "" && prof.APIKey != "" { + conflicts = append(conflicts, "API key secret ID cannot be set with an API key") + } + if prof.TLS != nil { + if s.ClientCert != "" && (prof.TLS.ClientCertPath != "" || len(prof.TLS.ClientCertData) > 0) { + conflicts = append(conflicts, "TLS client cert secret ID cannot be set with client cert path or data") + } + if s.ClientKey != "" && (prof.TLS.ClientKeyPath != "" || len(prof.TLS.ClientKeyData) > 0) { + conflicts = append(conflicts, "TLS client key secret ID cannot be set with client key path or data") + } + if s.ServerCACert != "" && (prof.TLS.ServerCACertPath != "" || len(prof.TLS.ServerCACertData) > 0) { + conflicts = append(conflicts, "TLS server CA cert secret ID cannot be set with server CA cert path or data") + } + if s.hasTLS() && prof.TLS.Disabled { + conflicts = append(conflicts, "TLS secret IDs cannot be set when TLS is disabled") + } + } + if (s.ClientCert == "") != (s.ClientKey == "") { + conflicts = append(conflicts, "TLS client cert and key secret IDs must be set together") + } + if len(conflicts) > 0 { + sort.Strings(conflicts) + return fmt.Errorf("invalid config: %v", conflicts) + } + return nil +} + +type plugin struct { + client.PluginBase + ids secretIDs + cache *secretCache + // Telemetry from the options of the most recently created client, or nil before the first + tel atomic.Pointer[telemetry] +} + +var _ client.Plugin = (*plugin)(nil) + +func (*plugin) Name() string { return pluginName } + +// Read at call time rather than captured, since the final options are only seen in NewClient, after ConfigureClient +// installs the callbacks. +func (p *plugin) telemetry() telemetry { + if tel := p.tel.Load(); tel != nil { + return *tel + } + return telemetry{} +} + +func (p *plugin) ConfigureClient(ctx context.Context, options client.PluginConfigureClientOptions) error { + opts := options.ClientOptions + if p.ids.APIKey != "" { + if opts.Credentials != nil { + return fmt.Errorf("cannot use API key from AWS Secrets Manager, client credentials already set") + } + opts.Credentials = client.NewAPIKeyDynamicCredentials(func(ctx context.Context) (string, error) { + return p.cache.getAPIKey(ctx, p.telemetry()) + }) + } + + if !p.ids.hasTLS() { + return nil + } + if opts.ConnectionOptions.TLSDisabled { + return fmt.Errorf("cannot use TLS config from AWS Secrets Manager, TLS is disabled") + } + var conf *tls.Config + if opts.ConnectionOptions.TLS != nil { + conf = opts.ConnectionOptions.TLS.Clone() + } else { + conf = &tls.Config{} + } + + if p.ids.ClientCert != "" { + if len(conf.Certificates) > 0 || conf.GetClientCertificate != nil { + return fmt.Errorf("cannot use TLS client cert from AWS Secrets Manager, client certificates already set") + } + conf.GetClientCertificate = func(info *tls.CertificateRequestInfo) (*tls.Certificate, error) { + ctx := info.Context() + if ctx == nil { + ctx = context.Background() + } + return p.cache.getClientCert(ctx, p.telemetry()) + } + } + + if p.ids.ServerCACert != "" { + if conf.RootCAs != nil { + return fmt.Errorf("cannot use TLS server CA cert from AWS Secrets Manager, root CAs already set") + } + // The pool must be set on the options, so this fetch can't wait for NewClient and only reports to telemetry + // that's already set + tel := telemetry{logger: opts.Logger, metrics: opts.MetricsHandler} + pool, err := p.cache.getServerCACert(ctx, tel) + if err != nil { + return err + } + conf.RootCAs = pool + } + + opts.ConnectionOptions.TLS = conf + return nil +} + +func (p *plugin) NewClient( + ctx context.Context, + options client.PluginNewClientOptions, + next func(context.Context, client.PluginNewClientOptions) error, +) error { + // Unlike in ConfigureClient, these are the final options, after all plugins have configured them and the SDK has + // applied its defaults + tel := telemetry{logger: options.ClientOptions.Logger, metrics: options.ClientOptions.MetricsHandler} + p.tel.Store(&tel) + // Fetch eagerly so that client creation fails on a bad secret instead of the first call + if p.ids.APIKey != "" { + if _, err := p.cache.getAPIKey(ctx, tel); err != nil { + return err + } + } + if p.ids.ClientCert != "" { + if _, err := p.cache.getClientCert(ctx, tel); err != nil { + return err + } + } + return next(ctx, options) +} diff --git a/contrib/aws/secretsmanager/extension_test.go b/contrib/aws/secretsmanager/extension_test.go new file mode 100644 index 000000000..7cfbd8a2b --- /dev/null +++ b/contrib/aws/secretsmanager/extension_test.go @@ -0,0 +1,533 @@ +package secretsmanager + +import ( + "context" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" + "crypto/sha256" + "crypto/tls" + "crypto/x509" + "crypto/x509/pkix" + "encoding/pem" + "errors" + "fmt" + "math/big" + "net" + "os" + "strings" + "sync" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + awssm "github.com/aws/aws-sdk-go-v2/service/secretsmanager" + "github.com/stretchr/testify/require" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" + + "go.temporal.io/sdk/client" + "go.temporal.io/sdk/contrib/envconfig" +) + +type fakeClient struct { + mu sync.Mutex + secrets map[string]string + err error + calls map[string]int +} + +func newFakeClient(secrets map[string]string) *fakeClient { + return &fakeClient{secrets: secrets, calls: map[string]int{}} +} + +func (f *fakeClient) GetSecretValue( + _ context.Context, + params *awssm.GetSecretValueInput, + _ ...func(*awssm.Options), +) (*awssm.GetSecretValueOutput, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.calls[*params.SecretId]++ + if f.err != nil { + return nil, f.err + } + s, ok := f.secrets[*params.SecretId] + if !ok { + return nil, errors.New("secret not found") + } + return &awssm.GetSecretValueOutput{SecretString: &s, VersionId: aws.String(fakeVersionID(s))}, nil +} + +// fakeVersionID derives a version ID from a secret value, so that changing a fake secret rotates it +func fakeVersionID(value string) string { + return fmt.Sprintf("%x", sha256.Sum256([]byte(value)))[:8] +} + +func (f *fakeClient) callCount(id string) int { + f.mu.Lock() + defer f.mu.Unlock() + return f.calls[id] +} + +type mapEnv map[string]string + +func (m mapEnv) Environ() []string { + ret := make([]string, 0, len(m)) + for k, v := range m { + ret = append(ret, k+"="+v) + } + return ret +} + +func (m mapEnv) LookupEnv(key string) (string, bool) { + v, ok := m[key] + return v, ok +} + +func loadOptions(t *testing.T, ext Extension, toml string, env mapEnv) (client.Options, error) { + t.Helper() + if env == nil { + env = mapEnv{} + } + return envconfig.LoadClientOptions(envconfig.LoadClientOptionsRequest{ + ConfigFileData: []byte(toml), + ConfigFileStrict: true, + EnvLookup: env, + Extensions: []envconfig.ClientConfigExtension{ext}, + }) +} + +// configure runs the plugins' ConfigureClient and then NewClient hooks like client creation does, without connecting. +func configure(t *testing.T, opts *client.Options) error { + t.Helper() + return configureWithDefaults(t, opts, nil) +} + +// configureWithDefaults is like configure, but applyDefaults (if non-nil) adjusts the options passed to NewClient, +// like the SDK does after ConfigureClient. +func configureWithDefaults(t *testing.T, opts *client.Options, applyDefaults func(*client.Options)) error { + t.Helper() + ctx := context.Background() + for _, p := range opts.Plugins { + if err := p.ConfigureClient(ctx, client.PluginConfigureClientOptions{ClientOptions: opts}); err != nil { + return err + } + } + finalOpts := *opts + if applyDefaults != nil { + applyDefaults(&finalOpts) + } + next := func(context.Context, client.PluginNewClientOptions) error { return nil } + for _, p := range opts.Plugins { + if err := p.NewClient(ctx, client.PluginNewClientOptions{ClientOptions: finalOpts}, next); err != nil { + return err + } + } + return nil +} + +func TestLoadSecretIDs(t *testing.T) { + toml := ` +[profile.default.ext.aws_secrets_manager] +api_key_id = "toml-api-key" +[profile.default.ext.aws_secrets_manager.tls] +client_cert_id = "toml-cert" +client_key_id = "toml-key" +server_ca_cert_id = "toml-ca" +` + load := func(toml string, env mapEnv) (secretIDs, error) { + var conf envconfig.ClientConfig + require.NoError(t, conf.FromTOML([]byte(toml), envconfig.ClientConfigFromTOMLOptions{})) + prof := conf.Profiles["default"] + if prof == nil { + prof = &envconfig.ClientConfigProfile{} + } + if env != nil { + require.NoError(t, prof.ApplyEnvVars(env)) + } + req := envconfig.ClientConfigExtensionRequest{Profile: prof} + if env != nil { + req.Env = env + } + return loadSecretIDs(req) + } + + ids, err := load(toml, nil) + require.NoError(t, err) + require.Equal(t, secretIDs{ + APIKey: "toml-api-key", ClientCert: "toml-cert", ClientKey: "toml-key", ServerCACert: "toml-ca", + }, ids) + + // Env overrides file, and set-but-empty env unsets + ids, err = load(toml, mapEnv{ + envAPIKeyID: "env-api-key", + envClientCertID: "env-bundle", + envClientKeyID: "env-bundle", + envServerCACertID: "", + }) + require.NoError(t, err) + require.Equal(t, secretIDs{APIKey: "env-api-key", ClientCert: "env-bundle", ClientKey: "env-bundle"}, ids) + + ids, err = load("", nil) + require.NoError(t, err) + require.Equal(t, secretIDs{}, ids) + + for name, tc := range map[string]struct { + toml string + env mapEnv + err string + }{ + "unknown key": { + toml: "[profile.default.ext.aws_secrets_manager]\napi_key = \"x\"", + err: `unknown key "api_key" in ext.aws_secrets_manager`, + }, + "unknown tls key": { + toml: "[profile.default.ext.aws_secrets_manager.tls]\nclient_cert = \"x\"", + err: `unknown key "client_cert" in ext.aws_secrets_manager.tls`, + }, + "non-string": { + toml: "[profile.default.ext.aws_secrets_manager]\napi_key_id = 5", + err: "ext.aws_secrets_manager.api_key_id must be a string", + }, + "non-table tls": { + toml: "[profile.default.ext.aws_secrets_manager]\ntls = \"x\"", + err: "ext.aws_secrets_manager.tls must be a table", + }, + "api key conflict in file": { + toml: "[profile.default]\napi_key = \"k\"\n[profile.default.ext.aws_secrets_manager]\napi_key_id = \"x\"", + err: "API key secret ID cannot be set with an API key", + }, + "api key conflict across file and env": { + toml: "[profile.default]\napi_key = \"k\"", + env: mapEnv{envAPIKeyID: "x"}, + err: "API key secret ID cannot be set with an API key", + }, + "api key conflict in env": { + env: mapEnv{"TEMPORAL_API_KEY": "k", envAPIKeyID: "x"}, + err: "API key secret ID cannot be set with an API key", + }, + "cert conflict": { + env: mapEnv{"TEMPORAL_TLS_CLIENT_CERT_DATA": "c", envClientCertID: "x", envClientKeyID: "x"}, + err: "TLS client cert secret ID cannot be set with client cert path or data", + }, + "key conflict": { + env: mapEnv{"TEMPORAL_TLS_CLIENT_KEY_PATH": "k", envClientCertID: "x", envClientKeyID: "x"}, + err: "TLS client key secret ID cannot be set with client key path or data", + }, + "ca conflict": { + env: mapEnv{"TEMPORAL_TLS_SERVER_CA_CERT_PATH": "ca", envServerCACertID: "x"}, + err: "TLS server CA cert secret ID cannot be set with server CA cert path or data", + }, + "cert without key": { + env: mapEnv{envClientCertID: "x"}, + err: "TLS client cert and key secret IDs must be set together", + }, + "tls disabled": { + env: mapEnv{"TEMPORAL_TLS": "false", envServerCACertID: "x"}, + err: "TLS secret IDs cannot be set when TLS is disabled", + }, + } { + t.Run(name, func(t *testing.T) { + _, err := load(tc.toml, tc.env) + require.ErrorContains(t, err, tc.err) + }) + } +} + +func TestNoConfig(t *testing.T) { + opts, err := loadOptions(t, Extension{SecretsClient: newFakeClient(nil)}, "", nil) + require.NoError(t, err) + require.Empty(t, opts.Plugins) +} + +func TestAPIKey(t *testing.T) { + // Capture headers on a fake server to observe the dynamic credentials + lis, err := net.Listen("tcp", "127.0.0.1:0") // random available ephemeral port on localhost + require.NoError(t, err) + authHeaders := make(chan string, 10) + srv := grpc.NewServer(grpc.UnknownServiceHandler(func(_ any, stream grpc.ServerStream) error { + md, _ := metadata.FromIncomingContext(stream.Context()) + authHeaders <- strings.Join(md.Get("authorization"), ",") + return status.Error(codes.Unimplemented, "unimplemented") + })) + go func() { _ = srv.Serve(lis) }() + t.Cleanup(srv.Stop) + + fake := newFakeClient(map[string]string{"api-key-secret": "my-api-key"}) + opts, err := loadOptions(t, Extension{SecretsClient: fake}, ` +[profile.default] +address = "`+lis.Addr().String()+`" +[profile.default.tls] +disabled = true +[profile.default.ext.aws_secrets_manager] +api_key_id = "api-key-secret" +`, nil) + require.NoError(t, err) + require.Len(t, opts.Plugins, 1) + + c, _ := client.Dial(opts) + if c != nil { + c.Close() + } + select { + case h := <-authHeaders: + require.Equal(t, "Bearer my-api-key", h) + case <-time.After(10 * time.Second): + t.Fatal("no request received") + } + require.Equal(t, 1, fake.callCount("api-key-secret")) +} + +func TestAPIKeyFetchFailure(t *testing.T) { + opts, err := loadOptions(t, Extension{SecretsClient: newFakeClient(nil)}, "", mapEnv{envAPIKeyID: "missing"}) + require.NoError(t, err) + require.ErrorContains(t, configure(t, &opts), `failed fetching AWS Secrets Manager secret "missing"`) +} + +func TestAPIKeyCredentialsAlreadySet(t *testing.T) { + fake := newFakeClient(map[string]string{"api-key-secret": "my-api-key"}) + opts, err := loadOptions(t, Extension{SecretsClient: fake}, "", mapEnv{envAPIKeyID: "api-key-secret"}) + require.NoError(t, err) + opts.Credentials = client.NewAPIKeyStaticCredentials("other") + require.ErrorContains(t, configure(t, &opts), "client credentials already set") +} + +func TestTLS(t *testing.T) { + certPEM, keyPEM := generateCert(t, "client") + caPEM, _ := generateCert(t, "ca") + + t.Run("separate cert and key", func(t *testing.T) { + fake := newFakeClient(map[string]string{"cert": certPEM, "key": keyPEM, "ca": caPEM}) + opts, err := loadOptions(t, Extension{SecretsClient: fake}, ` +[profile.default.tls] +server_name = "my-server" +[profile.default.ext.aws_secrets_manager.tls] +client_cert_id = "cert" +client_key_id = "key" +server_ca_cert_id = "ca" +`, nil) + require.NoError(t, err) + require.NoError(t, configure(t, &opts)) + conf := opts.ConnectionOptions.TLS + require.NotNil(t, conf) + require.Equal(t, "my-server", conf.ServerName) + require.NotNil(t, conf.RootCAs) + requireClientCert(t, conf, certPEM) + require.Equal(t, 1, fake.callCount("cert")) + require.Equal(t, 1, fake.callCount("key")) + require.Equal(t, 1, fake.callCount("ca")) + }) + + t.Run("bundled cert and key", func(t *testing.T) { + fake := newFakeClient(map[string]string{"bundle": certPEM + keyPEM}) + opts, err := loadOptions(t, Extension{SecretsClient: fake}, "", mapEnv{ + envClientCertID: "bundle", + envClientKeyID: "bundle", + }) + require.NoError(t, err) + require.NoError(t, configure(t, &opts)) + conf := opts.ConnectionOptions.TLS + require.NotNil(t, conf) + require.Nil(t, conf.RootCAs) + requireClientCert(t, conf, certPEM) + require.Equal(t, 1, fake.callCount("bundle")) + }) + + t.Run("refreshes cert", func(t *testing.T) { + fake := newFakeClient(map[string]string{"bundle": certPEM + keyPEM}) + opts, err := loadOptions(t, Extension{SecretsClient: fake, RefreshInterval: time.Nanosecond}, "", mapEnv{ + envClientCertID: "bundle", + envClientKeyID: "bundle", + }) + require.NoError(t, err) + require.NoError(t, configure(t, &opts)) + newCertPEM, newKeyPEM := generateCert(t, "client2") + fake.mu.Lock() + fake.secrets["bundle"] = newCertPEM + newKeyPEM + fake.mu.Unlock() + requireClientCert(t, opts.ConnectionOptions.TLS, newCertPEM) + // Every get refreshes at this interval, so this checks the bundle is fetched once per cert load, not once + // each for cert and key + require.Equal(t, 2, fake.callCount("bundle")) + }) + + t.Run("invalid key pair", func(t *testing.T) { + fake := newFakeClient(map[string]string{"cert": certPEM, "key": "not a key"}) + opts, err := loadOptions(t, Extension{SecretsClient: fake}, "", mapEnv{ + envClientCertID: "cert", + envClientKeyID: "key", + }) + require.NoError(t, err) + require.ErrorContains(t, configure(t, &opts), `failed loading TLS client cert/key from AWS Secrets Manager secrets "cert" and "key"`) + }) + + t.Run("invalid CA", func(t *testing.T) { + fake := newFakeClient(map[string]string{"ca": "not a cert"}) + opts, err := loadOptions(t, Extension{SecretsClient: fake}, "", mapEnv{envServerCACertID: "ca"}) + require.NoError(t, err) + require.ErrorContains(t, configure(t, &opts), `no PEM certificates found in TLS server CA cert secret "ca"`) + }) + + t.Run("TLS disabled after load", func(t *testing.T) { + fake := newFakeClient(map[string]string{"ca": caPEM}) + opts, err := loadOptions(t, Extension{SecretsClient: fake}, "", mapEnv{envServerCACertID: "ca"}) + require.NoError(t, err) + opts.ConnectionOptions.TLSDisabled = true + require.ErrorContains(t, configure(t, &opts), "TLS is disabled") + }) + + t.Run("does not mutate original TLS config", func(t *testing.T) { + fake := newFakeClient(map[string]string{"ca": caPEM}) + opts, err := loadOptions(t, Extension{SecretsClient: fake}, "", mapEnv{envServerCACertID: "ca"}) + require.NoError(t, err) + orig := &tls.Config{ServerName: "orig"} + opts.ConnectionOptions.TLS = orig + require.NoError(t, configure(t, &opts)) + require.Nil(t, orig.RootCAs) + require.NotNil(t, opts.ConnectionOptions.TLS.RootCAs) + require.Equal(t, "orig", opts.ConnectionOptions.TLS.ServerName) + }) +} + +func TestTelemetryFromClientOptions(t *testing.T) { + certPEM, keyPEM := generateCert(t, "client") + fake := newFakeClient(map[string]string{"bundle": certPEM + keyPEM}) + opts, err := loadOptions(t, Extension{SecretsClient: fake, RefreshInterval: time.Nanosecond}, "", mapEnv{ + envClientCertID: "bundle", + envClientKeyID: "bundle", + }) + require.NoError(t, err) + logger := &recordingLogger{} + metrics := newRecordingMetrics() + // Only set in the options NewClient sees, like SDK defaults + require.NoError(t, configureWithDefaults(t, &opts, func(o *client.Options) { + o.Logger = logger + o.MetricsHandler = metrics + })) + require.Equal(t, []string{ + "INFO Fetched AWS Secrets Manager secret SecretID=bundle SecretType=tls_client_cert FetchPath=sdk " + + "VersionID=" + fakeVersionID(certPEM+keyPEM), + }, logger.take()) + + // Rotation during the client's lifetime is logged + newCertPEM, newKeyPEM := generateCert(t, "client2") + fake.mu.Lock() + fake.secrets["bundle"] = newCertPEM + newKeyPEM + fake.mu.Unlock() + requireClientCert(t, opts.ConnectionOptions.TLS, newCertPEM) + require.Equal(t, []string{ + "INFO Fetched updated AWS Secrets Manager secret SecretID=bundle SecretType=tls_client_cert " + + "PreviousVersionID=" + fakeVersionID(certPEM+keyPEM) + " VersionID=" + fakeVersionID(newCertPEM+newKeyPEM), + }, logger.take()) + + // Bundled cert/key is fetched once and labeled as the client cert + require.Equal(t, map[string]int64{ + "temporal_aws_secrets_manager_cache_lookup fetch_path=sdk result=miss secret_type=tls_client_cert": 2, + }, metrics.counters(metricCacheLookup)) +} + +// Through real client creation, the plugin reports to the client's final telemetry rather than what was set when +// ConfigureClient ran. +func TestTelemetryFromFinalClientOptions(t *testing.T) { + fake := newFakeClient(map[string]string{"api-key-secret": "my-api-key"}) + opts, err := loadOptions(t, Extension{SecretsClient: fake}, "", mapEnv{envAPIKeyID: "api-key-secret"}) + require.NoError(t, err) + metrics := newRecordingMetrics() + opts.MetricsHandler = metrics + opts.Namespace = "my-ns" + c, err := client.NewLazyClient(opts) + require.NoError(t, err) + c.Close() + + lookups := metrics.counters(metricCacheLookup) + require.Len(t, lookups, 1) + for key := range lookups { + require.Contains(t, key, "secret_type=api_key") + // The SDK wraps the metrics handler with namespace tags after ConfigureClient, so these tags show the plugin + // used the final handler + require.Contains(t, key, "namespace=my-ns") + } +} + +func requireClientCert(t *testing.T, conf *tls.Config, expectedPEM string) { + t.Helper() + require.NotNil(t, conf.GetClientCertificate) + cert, err := conf.GetClientCertificate(&tls.CertificateRequestInfo{}) + require.NoError(t, err) + block, _ := pem.Decode([]byte(expectedPEM)) + require.Equal(t, block.Bytes, cert.Certificate[0]) +} + +func generateCert(t *testing.T, cn string) (certPEM, keyPEM string) { + t.Helper() + key, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + require.NoError(t, err) + tmpl := &x509.Certificate{ + SerialNumber: big.NewInt(1), + Subject: pkix.Name{CommonName: cn}, + NotBefore: time.Now().Add(-time.Hour), + NotAfter: time.Now().Add(time.Hour), + IsCA: true, + BasicConstraintsValid: true, + } + der, err := x509.CreateCertificate(rand.Reader, tmpl, tmpl, &key.PublicKey, key) + require.NoError(t, err) + keyDER, err := x509.MarshalECPrivateKey(key) + require.NoError(t, err) + return string(pem.EncodeToMemory(&pem.Block{Type: "CERTIFICATE", Bytes: der})), + string(pem.EncodeToMemory(&pem.Block{Type: "EC PRIVATE KEY", Bytes: keyDER})) +} + +func TestRefreshIntervalFromEnv(t *testing.T) { + for name, tc := range map[string]struct { + set bool + env string + field time.Duration + expected time.Duration + err bool + }{ + "unset": {expected: defaultRefreshInterval}, + "empty": {set: true, env: "", expected: defaultRefreshInterval}, + "seconds": {set: true, env: "60", expected: time.Minute}, + "zero": {set: true, env: "0", expected: 0}, + "over 300": {set: true, env: "3600", expected: time.Hour}, + "field wins": {set: true, env: "60", field: time.Hour, expected: time.Hour}, + "field wins bad": {set: true, env: "bad", field: time.Hour, expected: time.Hour}, + "negative": {set: true, env: "-1", err: true}, + "fractional": {set: true, env: "1.5", err: true}, + "not a number": {set: true, env: "5m", err: true}, + "overflow": {set: true, env: "9223372036854775807", err: true}, + } { + t.Run(name, func(t *testing.T) { + t.Setenv(envSecretsManagerTTL, tc.env) + if !tc.set { + require.NoError(t, os.Unsetenv(envSecretsManagerTTL)) + } + d, err := Extension{RefreshInterval: tc.field}.refreshInterval() + if tc.err { + require.ErrorContains(t, err, "invalid SECRETS_MANAGER_TTL") + return + } + require.NoError(t, err) + require.Equal(t, tc.expected, d) + }) + } +} + +func TestRefreshIntervalZeroDisablesCaching(t *testing.T) { + t.Setenv(envSecretsManagerTTL, "0") + certPEM, keyPEM := generateCert(t, "client") + fake := newFakeClient(map[string]string{"bundle": certPEM + keyPEM}) + opts, err := loadOptions(t, Extension{SecretsClient: fake}, "", mapEnv{ + envClientCertID: "bundle", + envClientKeyID: "bundle", + }) + require.NoError(t, err) + require.NoError(t, configure(t, &opts)) + requireClientCert(t, opts.ConnectionOptions.TLS, certPEM) + requireClientCert(t, opts.ConnectionOptions.TLS, certPEM) + require.Equal(t, 3, fake.callCount("bundle")) +} diff --git a/contrib/aws/secretsmanager/fetch.go b/contrib/aws/secretsmanager/fetch.go new file mode 100644 index 000000000..5ba9aecb7 --- /dev/null +++ b/contrib/aws/secretsmanager/fetch.go @@ -0,0 +1,179 @@ +package secretsmanager + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "io" + "net" + "net/http" + "net/url" + "os" + "sync" + "time" + + "github.com/aws/aws-sdk-go-v2/config" + awssm "github.com/aws/aws-sdk-go-v2/service/secretsmanager" +) + +const ( + envLambdaExtensionPort = "PARAMETERS_SECRETS_EXTENSION_HTTP_PORT" + envLambdaFunctionName = "AWS_LAMBDA_FUNCTION_NAME" + envAWSSessionToken = "AWS_SESSION_TOKEN" + lambdaExtensionTokenHdr = "X-Aws-Parameters-Secrets-Token" + lambdaExtensionTimeout = 10 * time.Second + lambdaExtensionHost = "localhost" + // Probing a local port is either accepted or refused almost immediately, so this only bounds pathological cases + lambdaExtensionProbeTimeout = 10 * time.Millisecond +) + +// Port the Lambda extension listens on when PARAMETERS_SECRETS_EXTENSION_HTTP_PORT is not set. A var for tests. +var defaultLambdaExtensionPort = "2773" + +// newDefaultFetcher returns a fetcher for the Lambda extension if it is detected, and otherwise an SDK fetcher with a +// client created from the default AWS config. The extension doesn't set PARAMETERS_SECRETS_EXTENSION_HTTP_PORT in the +// function's environment unless the user configured it, so if it is unset and we're running in Lambda, the extension's +// default port is probed. Setting it to empty disables detection, for when another service uses the default port. +func newDefaultFetcher() fetcher { + port, ok := os.LookupEnv(envLambdaExtensionPort) + if !ok { + // Only try to auto-detect the Lambda extension if we're in a Lambda function + if os.Getenv(envLambdaFunctionName) == "" { + return &sdkFetcher{} + } + conn, err := net.DialTimeout("tcp", net.JoinHostPort(lambdaExtensionHost, defaultLambdaExtensionPort), + lambdaExtensionProbeTimeout) + if err != nil { + return &sdkFetcher{} + } + _ = conn.Close() + port = defaultLambdaExtensionPort + } + if port == "" { + return &sdkFetcher{} + } + return newLambdaExtensionFetcher(port) +} + +type secretValue struct { + value []byte + versionID string +} + +type fetcher interface { + fetch(ctx context.Context, secretID string) (secretValue, error) + path() fetchPath +} + +// sdkFetcher fetches secrets directly from Secrets Manager. If client is nil, one is created from the default AWS +// config on first use. +type sdkFetcher struct { + mu sync.Mutex + client SecretsClient +} + +func (f *sdkFetcher) fetch(ctx context.Context, secretID string) (secretValue, error) { + c, err := f.getClient(ctx) + if err != nil { + return secretValue{}, err + } + out, err := c.GetSecretValue(ctx, &awssm.GetSecretValueInput{SecretId: &secretID}) + if err != nil { + return secretValue{}, err + } + return newSecretValue(out.SecretString, out.SecretBinary, out.VersionId) +} + +func (*sdkFetcher) path() fetchPath { return fetchPathSDK } + +func (f *sdkFetcher) getClient(ctx context.Context) (SecretsClient, error) { + f.mu.Lock() + defer f.mu.Unlock() + if f.client == nil { + cfg, err := config.LoadDefaultConfig(ctx) + if err != nil { + return nil, fmt.Errorf("failed loading default AWS config: %w", err) + } + f.client = awssm.NewFromConfig(cfg) + } + return f.client, nil +} + +// lambdaExtensionFetcher fetches secrets through the AWS Parameters and Secrets Lambda Extension. See +// https://docs.aws.amazon.com/secretsmanager/latest/userguide/retrieving-secrets_lambda.html. +type lambdaExtensionFetcher struct { + endpoint string + httpClient *http.Client +} + +func newLambdaExtensionFetcher(port string) *lambdaExtensionFetcher { + return &lambdaExtensionFetcher{ + endpoint: "http://" + net.JoinHostPort(lambdaExtensionHost, port) + "/secretsmanager/get", + httpClient: &http.Client{Timeout: lambdaExtensionTimeout}, + } +} + +func (f *lambdaExtensionFetcher) fetch(ctx context.Context, secretID string) (secretValue, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, + f.endpoint+"?secretId="+url.QueryEscape(secretID), nil) + if err != nil { + return secretValue{}, err + } + req.Header.Set(lambdaExtensionTokenHdr, os.Getenv(envAWSSessionToken)) + resp, err := f.httpClient.Do(req) + if err != nil { + return secretValue{}, fmt.Errorf("failed calling AWS Parameters and Secrets Lambda Extension: %w", err) + } + defer func() { _ = resp.Body.Close() }() + body, err := io.ReadAll(resp.Body) + if err != nil { + return secretValue{}, fmt.Errorf("failed reading AWS Parameters and Secrets Lambda Extension response: %w", err) + } + if resp.StatusCode != http.StatusOK { + return secretValue{}, &lambdaExtensionStatusError{statusCode: resp.StatusCode, body: string(body)} + } + // Same shape as the GetSecretValue response. SecretBinary is base64, which json decodes into []byte. + var out struct { + SecretString *string + SecretBinary []byte + VersionId *string + } + if err := json.Unmarshal(body, &out); err != nil { + return secretValue{}, &invalidResponseError{ + fmt.Errorf("failed decoding AWS Parameters and Secrets Lambda Extension response: %w", err)} + } + return newSecretValue(out.SecretString, out.SecretBinary, out.VersionId) +} + +func (*lambdaExtensionFetcher) path() fetchPath { return fetchPathExtension } + +func newSecretValue(secretString *string, secretBinary []byte, versionID *string) (secretValue, error) { + var ret secretValue + if versionID != nil { + ret.versionID = *versionID + } + if secretString != nil { + ret.value = []byte(*secretString) + } else if len(secretBinary) > 0 { + ret.value = secretBinary + } else { + return secretValue{}, &invalidResponseError{errors.New("secret has no value")} + } + return ret, nil +} + +type lambdaExtensionStatusError struct { + statusCode int + body string +} + +func (e *lambdaExtensionStatusError) Error() string { + return fmt.Sprintf("AWS Parameters and Secrets Lambda Extension returned status %v: %s", e.statusCode, e.body) +} + +// invalidResponseError is a response that was received but has no usable secret value. +type invalidResponseError struct{ err error } + +func (e *invalidResponseError) Error() string { return e.err.Error() } +func (e *invalidResponseError) Unwrap() error { return e.err } diff --git a/contrib/aws/secretsmanager/fetch_test.go b/contrib/aws/secretsmanager/fetch_test.go new file mode 100644 index 000000000..276fa08c5 --- /dev/null +++ b/contrib/aws/secretsmanager/fetch_test.go @@ -0,0 +1,144 @@ +package secretsmanager + +import ( + "context" + "encoding/json" + "net" + "net/http" + "net/http/httptest" + "net/url" + "os" + "testing" + + "github.com/stretchr/testify/require" + + "go.temporal.io/sdk/contrib/envconfig" +) + +func TestLambdaExtensionFetcher(t *testing.T) { + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/secretsmanager/get" || r.Header.Get(lambdaExtensionTokenHdr) != "session-token" { + w.WriteHeader(http.StatusBadRequest) + return + } + switch r.URL.Query().Get("secretId") { + case "arn:aws:secretsmanager:us-east-1:123:secret:string": + _ = json.NewEncoder(w).Encode(map[string]any{"SecretString": "string-value", "VersionId": "ver1"}) + case "binary": + _ = json.NewEncoder(w).Encode(map[string]any{"SecretBinary": []byte("binary-value")}) + case "empty": + _ = json.NewEncoder(w).Encode(map[string]any{"VersionId": "ver1"}) + case "bad-json": + _, _ = w.Write([]byte("{")) + case "server-error": + w.WriteHeader(http.StatusInternalServerError) + _, _ = w.Write([]byte("internal error")) + default: + w.WriteHeader(http.StatusBadRequest) + _, _ = w.Write([]byte("bad request")) + } + })) + t.Cleanup(srv.Close) + u, err := url.Parse(srv.URL) + require.NoError(t, err) + t.Setenv(envLambdaExtensionPort, u.Port()) + t.Setenv(envAWSSessionToken, "session-token") + + f := newLambdaExtensionFetcher(u.Port()) + require.Equal(t, fetchPathExtension, f.path()) + v, err := f.fetch(context.Background(), "arn:aws:secretsmanager:us-east-1:123:secret:string") + require.NoError(t, err) + require.Equal(t, secretValue{value: []byte("string-value"), versionID: "ver1"}, v) + v, err = f.fetch(context.Background(), "binary") + require.NoError(t, err) + require.Equal(t, secretValue{value: []byte("binary-value")}, v) + _, err = f.fetch(context.Background(), "missing") + require.ErrorContains(t, err, "returned status 400: bad request") + require.Equal(t, failureClientError, classifyFailure(err)) + _, err = f.fetch(context.Background(), "server-error") + require.ErrorContains(t, err, "returned status 500: internal error") + require.Equal(t, failureServerError, classifyFailure(err)) + _, err = f.fetch(context.Background(), "empty") + require.ErrorContains(t, err, "secret has no value") + require.Equal(t, failureInvalidResponse, classifyFailure(err)) + _, err = f.fetch(context.Background(), "bad-json") + require.ErrorContains(t, err, "failed decoding") + require.Equal(t, failureInvalidResponse, classifyFailure(err)) + + // Extension is used when no client is given and the port env var is set + // (configure triggers an eager load, which exercises the fetcher) + opts, err := envconfig.LoadClientOptions(envconfig.LoadClientOptionsRequest{ + DisableFile: true, + EnvLookup: mapEnv{envAPIKeyID: "arn:aws:secretsmanager:us-east-1:123:secret:string"}, + Extensions: []envconfig.ClientConfigExtension{Extension{}}, + }) + require.NoError(t, err) + metrics := newRecordingMetrics() + opts.MetricsHandler = metrics + require.NoError(t, configure(t, &opts)) + require.NotNil(t, opts.Credentials) + require.Equal(t, int64(1), metrics.counter(metricCacheLookup, + labelSecretType, secretTypeAPIKey, labelFetchPath, fetchPathExtension, labelResult, lookupMiss)) +} + +func TestDefaultFetcherDetection(t *testing.T) { + lis, err := net.Listen("tcp", "127.0.0.1:0") // random available ephemeral port + require.NoError(t, err) + t.Cleanup(func() { _ = lis.Close() }) + _, listeningPort, err := net.SplitHostPort(lis.Addr().String()) + require.NoError(t, err) + closed, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + _, closedPort, err := net.SplitHostPort(closed.Addr().String()) + require.NoError(t, err) + require.NoError(t, closed.Close()) + + for name, tc := range map[string]struct { + set bool + env string + notLambda bool + defaultPort string + expectedPort string + }{ + "unset, default port listening": {defaultPort: listeningPort, expectedPort: listeningPort}, + "unset, default port closed": {defaultPort: closedPort}, + "unset, not in Lambda, default port listening": {notLambda: true, defaultPort: listeningPort}, + "set, not probed": {set: true, env: "1234", defaultPort: closedPort, expectedPort: "1234"}, + "set, not in Lambda": {set: true, env: "1234", notLambda: true, expectedPort: "1234"}, + "empty, default port listening": {set: true, env: "", defaultPort: listeningPort}, + } { + t.Run(name, func(t *testing.T) { + t.Setenv(envLambdaExtensionPort, tc.env) + if !tc.set { + require.NoError(t, os.Unsetenv(envLambdaExtensionPort)) + } + t.Setenv(envLambdaFunctionName, "my-function") + if tc.notLambda { + require.NoError(t, os.Unsetenv(envLambdaFunctionName)) + } + + // We don't want to listen on a fixed well-known port in a test, so modify + // the default port to this test's ephemeral one + origDefault := defaultLambdaExtensionPort + defaultLambdaExtensionPort = tc.defaultPort + t.Cleanup(func() { defaultLambdaExtensionPort = origDefault }) + + f := newDefaultFetcher() + if tc.expectedPort == "" { + require.IsType(t, &sdkFetcher{}, f) + return + } + require.IsType(t, &lambdaExtensionFetcher{}, f) + require.Equal(t, "http://localhost:"+tc.expectedPort+"/secretsmanager/get", + f.(*lambdaExtensionFetcher).endpoint) + }) + } +} + +func TestSDKFetcher(t *testing.T) { + f := &sdkFetcher{client: newFakeClient(map[string]string{"id": "value"})} + require.Equal(t, fetchPathSDK, f.path()) + v, err := f.fetch(context.Background(), "id") + require.NoError(t, err) + require.Equal(t, secretValue{value: []byte("value"), versionID: fakeVersionID("value")}, v) +} diff --git a/contrib/aws/secretsmanager/go.mod b/contrib/aws/secretsmanager/go.mod new file mode 100644 index 000000000..a5026af8f --- /dev/null +++ b/contrib/aws/secretsmanager/go.mod @@ -0,0 +1,56 @@ +module go.temporal.io/sdk/contrib/aws/secretsmanager + +go 1.26.0 + +require ( + github.com/aws/aws-sdk-go-v2 v1.47.2 + github.com/aws/aws-sdk-go-v2/config v1.33.8 + github.com/aws/aws-sdk-go-v2/service/secretsmanager v1.50.3 + github.com/aws/smithy-go v1.28.4 + github.com/stretchr/testify v1.10.0 + go.temporal.io/sdk v1.39.0 + go.temporal.io/sdk/contrib/envconfig v1.0.2 + google.golang.org/grpc v1.83.2 +) + +require ( + github.com/BurntSushi/toml v1.4.0 // indirect + github.com/aws/aws-sdk-go-v2/credentials v1.20.8 // indirect + github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.20.2 // indirect + github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.5 // indirect + github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.5 // indirect + github.com/aws/aws-sdk-go-v2/internal/v4a v1.5.5 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.20 // indirect + github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.14.5 // indirect + github.com/aws/aws-sdk-go-v2/service/signin v1.10.3 // indirect + github.com/aws/aws-sdk-go-v2/service/sso v1.38.3 // indirect + github.com/aws/aws-sdk-go-v2/service/ssooidc v1.43.3 // indirect + github.com/aws/aws-sdk-go-v2/service/sts v1.51.3 // indirect + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/facebookgo/clock v0.0.0-20150410010913-600d898af40a // indirect + github.com/gogo/protobuf v1.3.2 // indirect + github.com/golang/mock v1.6.0 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.2 // indirect + github.com/grpc-ecosystem/grpc-gateway/v2 v2.22.0 // indirect + github.com/nexus-rpc/nexus-proto-annotations v0.1.0 // indirect + github.com/nexus-rpc/sdk-go v0.7.0 // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/robfig/cron v1.2.0 // indirect + github.com/stretchr/objx v0.5.2 // indirect + go.temporal.io/api v1.64.0 // indirect + golang.org/x/net v0.58.0 // indirect + golang.org/x/sync v0.22.0 // indirect + golang.org/x/sys v0.47.0 // indirect + golang.org/x/text v0.41.0 // indirect + golang.org/x/time v0.3.0 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa // indirect + google.golang.org/protobuf v1.36.11 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) + +replace ( + go.temporal.io/sdk => ../../../ + go.temporal.io/sdk/contrib/envconfig => ../../envconfig +) diff --git a/contrib/aws/secretsmanager/go.sum b/contrib/aws/secretsmanager/go.sum new file mode 100644 index 000000000..8ad8dacf8 --- /dev/null +++ b/contrib/aws/secretsmanager/go.sum @@ -0,0 +1,151 @@ +github.com/BurntSushi/toml v1.4.0 h1:kuoIxZQy2WRRk1pttg9asf+WVv6tWQuBNVmK8+nqPr0= +github.com/BurntSushi/toml v1.4.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho= +github.com/aws/aws-sdk-go-v2 v1.47.2 h1:4vrR4S7MUdvPiv/fxlYI/FaZCgwKuir2LpXxwCUtwWY= +github.com/aws/aws-sdk-go-v2 v1.47.2/go.mod h1:LYKdgB/k1/omdrfoQuyt6T10CvF1yw27ByBjd+ku1jk= +github.com/aws/aws-sdk-go-v2/config v1.33.8 h1:I9S23gMufs97T0CE0ZUzvdl/i2TDnxOJzgxw18d1RBo= +github.com/aws/aws-sdk-go-v2/config v1.33.8/go.mod h1:/WlDmOxSaE2dhWUmEcFF3QnckmOPH49GxsQI0CzRAtg= +github.com/aws/aws-sdk-go-v2/credentials v1.20.8 h1:ibO101be+3jJOMBCJphBXzXm9UzIq+zxr8plM2V3FHw= +github.com/aws/aws-sdk-go-v2/credentials v1.20.8/go.mod h1:gAFU+h+59h6frPO/KUG7sgcToVf1ti8i/VMImKiFpcM= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.20.2 h1:b4FhKL3xhR6BVxUTU+h1K8gK2XYhb+pU/sRiwJwLuCs= +github.com/aws/aws-sdk-go-v2/feature/ec2/imds v1.20.2/go.mod h1:FprVEOVKR2fKunYpdjOCpt1/n76MnL72PcTKylXlYRs= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.5 h1:JwchTNw8oqkFWMYrmFmdoS4sYgE6JJcnO+ElvME+uH4= +github.com/aws/aws-sdk-go-v2/internal/configsources v1.5.5/go.mod h1:8pc+vcibD0We5t3f/LqHxjBPkXXfgcjYQN3upxqz/Nw= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.5 h1:CNgkOhhgJ1anuboBDmjRLCpVlBKFV9cEvT16d4QNl/w= +github.com/aws/aws-sdk-go-v2/internal/endpoints/v2 v2.8.5/go.mod h1:SS3HIANtO8edxa4QkX+8SvbMxRm+k78QoE6EIRI81tU= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.5.5 h1:rkxirozlOuMs3QuaGsDV3ll5suBdeeH6iSLFPyA5zNk= +github.com/aws/aws-sdk-go-v2/internal/v4a v1.5.5/go.mod h1:OQjoYoCdNx6tLyZNEuD0HRfHQVStBIfQyigewExdEVU= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.20 h1:XBYeuBbMY5LRAKOdZgzUQO1BVj3B07uYAROSapq/IWg= +github.com/aws/aws-sdk-go-v2/service/internal/accept-encoding v1.13.20/go.mod h1:XL6ZfC3ePnooN2u2LT4RNhpr/xs5TiboydyUbgr8Dmg= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.14.5 h1:Q2MWL+9cTKHHiXWcsPobHiOD+wx8tTc8sWZaFvFSk9Q= +github.com/aws/aws-sdk-go-v2/service/internal/presigned-url v1.14.5/go.mod h1:U8tTityyPbgc90QwGOha8XsrQpLIjpGUS7+Q+4prM+Q= +github.com/aws/aws-sdk-go-v2/service/secretsmanager v1.50.3 h1:oPQLY2D2sJowKzZl1qHMH9Uo8C/rF3xCqRh5ORYg6fA= +github.com/aws/aws-sdk-go-v2/service/secretsmanager v1.50.3/go.mod h1:6ZMEilrwLy1zFaG2zndlY7cXtOL8Y6JsOGykkcSOALE= +github.com/aws/aws-sdk-go-v2/service/signin v1.10.3 h1:lRyGlN6mXianMWMQdvTFCnJE5DkmD54N9PC3m+U/GgA= +github.com/aws/aws-sdk-go-v2/service/signin v1.10.3/go.mod h1:9Y78vT13MZ36q2uxrSKqHJnk/6n9sbSgGp3ZOZS1s4w= +github.com/aws/aws-sdk-go-v2/service/sso v1.38.3 h1:NYlE1/vPhIhdPTLUnmvz/PSXS3klsToqQY6KVtsynhs= +github.com/aws/aws-sdk-go-v2/service/sso v1.38.3/go.mod h1:nVwtZaAwWFpiXGfNE0GPxbpjhWNVGJbJEZDOfrDCEm4= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.43.3 h1:Glo50xPemPwHaioR66Rl5BTEq62Ohet0dyxzt6L0H1M= +github.com/aws/aws-sdk-go-v2/service/ssooidc v1.43.3/go.mod h1:LaCSuxpt5Z2a8oThoMtWWB71u2MOmylEqHaQdp3J4aw= +github.com/aws/aws-sdk-go-v2/service/sts v1.51.3 h1:L5w+LXVzMaacz3iotdx2fWtU/O+TIrplsK2JVMNbXeA= +github.com/aws/aws-sdk-go-v2/service/sts v1.51.3/go.mod h1:8E586vK6OTj+qVzsTygGzw1GlZnOgNjul8toJEdtX5s= +github.com/aws/smithy-go v1.28.4 h1:vzBmFpcRUtotqGPfmnro2W2Pgg3mk103n9sZhNJY9YU= +github.com/aws/smithy-go v1.28.4/go.mod h1:YE2RhdIuDbA5E5bTdciG9KrW3+TiEONeUWCqxX9i1Fc= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/facebookgo/clock v0.0.0-20150410010913-600d898af40a h1:yDWHCSQ40h88yih2JAcL6Ls/kVkSE8GFACTGVnMPruw= +github.com/facebookgo/clock v0.0.0-20150410010913-600d898af40a/go.mod h1:7Ga40egUymuWXxAe151lTNnCv97MddSOVsjpPPkityA= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q= +github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q= +github.com/golang/mock v1.6.0 h1:ErTB+efbowRARo13NNdxyJji2egdxLGQhRaY+DUumQc= +github.com/golang/mock v1.6.0/go.mod h1:p6yTPP+5HYm5mzsMV8JkE6ZKdX+/wYM6Hr+LicevLPs= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.2 h1:sGm2vDRFUrQJO/Veii4h4zG2vvqG6uWNkBHSTqXOZk0= +github.com/grpc-ecosystem/go-grpc-middleware/v2 v2.3.2/go.mod h1:wd1YpapPLivG6nQgbf7ZkG1hhSOXDhhn4MLTknx2aAc= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.22.0 h1:asbCHRVmodnJTuQ3qamDwqVOIjwqUPTYmYuemVOx+Ys= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.22.0/go.mod h1:ggCgvZ2r7uOoQjOyu2Y1NhHmEPPzzuhWgcza5M1Ji1I= +github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= +github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/nexus-rpc/nexus-proto-annotations v0.1.0 h1:2fELd+9sqUtNu6Fg//pw8YFsxOvp8vZ8hfP0nHhNI80= +github.com/nexus-rpc/nexus-proto-annotations v0.1.0/go.mod h1:n3UjF1bPCW8llR8tHvbxJ+27yPWrhpo8w/Yg1IOuY0Y= +github.com/nexus-rpc/sdk-go v0.7.0 h1:38NrfY5rLnZAiMMs2ZfCKI/CSDzdfJG+27iAgfA8bUI= +github.com/nexus-rpc/sdk-go v0.7.0/go.mod h1:FHdPfVQwRuJFZFTF0Y2GOAxCrbIBNrcPna9slkGKPYk= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/robfig/cron v1.2.0 h1:ZjScXvvxeQ63Dbyxy76Fj3AT3Ut0aKsyd2/tl3DTMuQ= +github.com/robfig/cron v1.2.0/go.mod h1:JGuDeoQd7Z6yL4zQhZ3OPEVHB7fL6Ka6skscFHfmt2k= +github.com/rogpeppe/go-internal v1.11.0 h1:cWPaGQEPrBb5/AsnsZesgZZ9yb1OQ+GOISoDNXVBh4M= +github.com/rogpeppe/go-internal v1.11.0/go.mod h1:ddIwULY96R17DhadqLgMfk9H9tvdUzkipdSkR5nkCZA= +github.com/stretchr/objx v0.5.2 h1:xuMeJ0Sdp5ZMRXx/aWO6RZxdr3beISkG5/G/aIRr3pY= +github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA= +github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= +github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +github.com/yuin/goldmark v1.2.1/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= +github.com/yuin/goldmark v1.3.5/go.mod h1:mwnBkeHKe2W/ZEtQ+71ViKU8L12m81fl3OWwC1Zlc8k= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/otel v1.44.0 h1:JjwHmHpA4iZ3wBxluu2fbbE7j4kqlE8jXyAyPXH7HqU= +go.opentelemetry.io/otel v1.44.0/go.mod h1:BMgjTHL9WPRlRjL2oZCBTL4whCGtXch2H4BhOPIAyYc= +go.opentelemetry.io/otel/metric v1.44.0 h1:1w0gILTcHdr3YI+ixLyjemwrVnsMURbTZFrSYCdDdmc= +go.opentelemetry.io/otel/metric v1.44.0/go.mod h1:8O7hanEPBNgEMmybD3s2VBKcgWOCsA6tzHBPODAiquo= +go.opentelemetry.io/otel/sdk v1.44.0 h1:nHYwb9lK+fJPU/dnT6s7W7Z8itMWyqrnVfbheVYrZ58= +go.opentelemetry.io/otel/sdk v1.44.0/go.mod h1:Osuydd3Se74nqjAKxid74N5eC+jfEqfTegHRnq58oK0= +go.opentelemetry.io/otel/sdk/metric v1.44.0 h1:3LlKgI+VjbVsjNRFZJZAJ30WjXC5VkNRks6si09iEfI= +go.opentelemetry.io/otel/sdk/metric v1.44.0/go.mod h1:5B5pMARnXxKhltooO4xUuCBorl65a4EpnTalObqOigA= +go.opentelemetry.io/otel/trace v1.44.0 h1:jxF5CsGYCe74MCRx2X4g7WsY/VBKRqqpNvXlX/6gtIk= +go.opentelemetry.io/otel/trace v1.44.0/go.mod h1:oLl1jrMQAVo6v3GAggN+1VH9VIz9iUSvW53sW1Q8PIE= +go.temporal.io/api v1.64.0 h1:Wu7TMdVN/4KdU+v0As6SMm86uapnM1yI7PhC5Hvhqso= +go.temporal.io/api v1.64.0/go.mod h1:acM0I9WPuYg8W3Pd9jOZvEgi7mRUttUQ4+e7fowKVnM= +golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= +golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI= +golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto= +golang.org/x/mod v0.2.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA= +golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg= +golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= +golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= +golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM= +golang.org/x/net v0.58.0 h1:ynWG7rqYi4ccpTEuPZ2QGWHktVEM9DMCj9yzDE0Q7To= +golang.org/x/net v0.58.0/go.mod h1:YwCddHnFlT7eLQqVprV19OnhLGtc5xOKgE0RyqgfWAU= +golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.0.0-20210220032951-036812b2e83c/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= +golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek= +golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= +golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= +golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20201119102817-f84b799fce68/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= +golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= +golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs= +golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo= +golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= +golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= +golang.org/x/text v0.41.0 h1:vz/seA0lnX87Othu2f/0L24RcgrXD9/YFTSuGjj3rH8= +golang.org/x/text v0.41.0/go.mod h1:jvf1O8ajNzZqhSrQBPbutR/EB83Cc0CFrezNQIwbb5M= +golang.org/x/time v0.3.0 h1:rg5rLMjNzMS1RkNLzCG38eapWhnYLFYXDXj2gOlr8j4= +golang.org/x/time v0.3.0/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= +golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= +golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= +golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= +golang.org/x/tools v0.1.1/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk= +golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= +gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa h1:Kjn0N0tCrDgiAFW+lGO4JZ3ck44CehvJQMAwj9QF0G8= +google.golang.org/genproto/googleapis/api v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:q4lMZS6kskjT5HvCPrnnypcDPVJqT/f4nfxmkE7gryY= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa h1:mZHHdPZl0dbGHCflZgAq/Q468DWVFcU2whhB2KAo8fk= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260526163538-3dc84a4a5aaa/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.83.2 h1:EManeRomTObA0BU7I8vXgg/78uE5MJ9M8B39EX2WscU= +google.golang.org/grpc v1.83.2/go.mod h1:YPI1hK3kDked6iHvgX3tR0y+nX/qpMFKhPgFsokw1S8= +google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= +google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/contrib/aws/secretsmanager/telemetry.go b/contrib/aws/secretsmanager/telemetry.go new file mode 100644 index 000000000..d409216d2 --- /dev/null +++ b/contrib/aws/secretsmanager/telemetry.go @@ -0,0 +1,142 @@ +package secretsmanager + +import ( + "context" + "errors" + "net" + + "github.com/aws/smithy-go" + + "go.temporal.io/sdk/client" + "go.temporal.io/sdk/log" +) + +// telemetry is where the cache reports to, taken from a client's options. +type telemetry struct { + // May be nil, in which case nothing is logged + logger log.Logger + metrics client.MetricsHandler +} + +// Metric/label names, must be the same across SDKs +const ( + metricFetchFailed = "temporal_aws_secrets_manager_fetch_failed" + metricFetchLatency = "temporal_aws_secrets_manager_fetch_latency" + metricCacheLookup = "temporal_aws_secrets_manager_cache_lookup" + + labelSecretType = "secret_type" + labelFetchPath = "fetch_path" + labelFailureReason = "failure_reason" + labelFallback = "fallback" + labelOutcome = "outcome" + labelResult = "result" +) + +// secret_type label values +// Note: client cert/key secrets are bundled and reported as secretTypeClientCert in metrics +type secretType string + +const ( + secretTypeAPIKey secretType = "api_key" + secretTypeClientCert secretType = "tls_client_cert" + secretTypeClientKey secretType = "tls_client_key" + secretTypeServerCACert secretType = "tls_server_ca_cert" +) + +// fallback label values +type fallback string + +const ( + fallbackStale fallback = "stale" + fallbackNone fallback = "none" +) + +// outcome label values +type fetchOutcome string + +const ( + outcomeSuccess fetchOutcome = "success" + outcomeFailure fetchOutcome = "failure" +) + +// result label values +type lookupResult string + +const ( + lookupHit lookupResult = "hit" + lookupMiss lookupResult = "miss" + lookupMissWithFallback lookupResult = "miss_with_fallback" +) + +// fetch_path label values +type fetchPath string + +const ( + // AWS Parameters and Secrets Lambda Extension + fetchPathExtension fetchPath = "extension" + // Direct fetch from AWS Secrets Manager via AWS SDK + fetchPathSDK fetchPath = "sdk" +) + +// failure_reason label values +type failureReason string + +const ( + failureResourceNotFound failureReason = "resource_not_found" + failureAccessDenied failureReason = "access_denied" + failureDecryptionFailure failureReason = "decryption_failure" + failureInvalidRequest failureReason = "invalid_request" + failureThrottling failureReason = "throttling" + failureInternalServiceError failureReason = "internal_service_error" + failureTimeout failureReason = "timeout" + failureInvalidResponse failureReason = "invalid_response" + // The Lambda extension doesn't return the underlying AWS error, only an HTTP status with a generic message, so its + // errors are only classified by status class + failureClientError failureReason = "client_error" + failureServerError failureReason = "server_error" + failureOther failureReason = "other" +) + +// AWS error codes mapped to failure reasons. Other codes are failureOther. +var awsErrorCodeFailureReasons = map[string]failureReason{ + "ResourceNotFoundException": failureResourceNotFound, + "AccessDeniedException": failureAccessDenied, + "DecryptionFailure": failureDecryptionFailure, + "InvalidRequestException": failureInvalidRequest, + "InvalidParameterException": failureInvalidRequest, + "ThrottlingException": failureThrottling, + "InternalServiceError": failureInternalServiceError, +} + +// classifyFailure returns the failure_reason metric label for a fetch error. +func classifyFailure(err error) failureReason { + var apiErr smithy.APIError + var extErr *lambdaExtensionStatusError + var invalidErr *invalidResponseError + var netErr net.Error + switch { + case errors.As(err, &apiErr): + if reason, ok := awsErrorCodeFailureReasons[apiErr.ErrorCode()]; ok { + return reason + } + case errors.As(err, &extErr): + if extErr.statusCode >= 400 && extErr.statusCode < 500 { + return failureClientError + } else if extErr.statusCode >= 500 && extErr.statusCode < 600 { + return failureServerError + } + case errors.As(err, &invalidErr): + return failureInvalidResponse + case errors.Is(err, context.DeadlineExceeded), errors.Is(err, context.Canceled), + errors.As(err, &netErr) && netErr.Timeout(): + return failureTimeout + } + return failureOther +} + +type nopLogger struct{} + +func (nopLogger) Debug(string, ...any) {} +func (nopLogger) Info(string, ...any) {} +func (nopLogger) Warn(string, ...any) {} +func (nopLogger) Error(string, ...any) {} diff --git a/contrib/aws/secretsmanager/telemetry_test.go b/contrib/aws/secretsmanager/telemetry_test.go new file mode 100644 index 000000000..e04575117 --- /dev/null +++ b/contrib/aws/secretsmanager/telemetry_test.go @@ -0,0 +1,186 @@ +package secretsmanager + +import ( + "context" + "errors" + "fmt" + "maps" + "net/url" + "slices" + "strings" + "sync" + "testing" + "time" + + "github.com/aws/smithy-go" + "github.com/stretchr/testify/require" + + "go.temporal.io/sdk/client" +) + +type timeoutError struct{} + +func (timeoutError) Error() string { return "i/o timeout" } +func (timeoutError) Timeout() bool { return true } +func (timeoutError) Temporary() bool { return true } + +func TestFailureReason(t *testing.T) { + apiErr := func(code string) error { + return fmt.Errorf("operation error: %w", &smithy.GenericAPIError{Code: code, Message: "msg"}) + } + for name, tc := range map[string]struct { + err error + expected failureReason + }{ + "not found": {apiErr("ResourceNotFoundException"), failureResourceNotFound}, + "access denied": {apiErr("AccessDeniedException"), failureAccessDenied}, + "decryption": {apiErr("DecryptionFailure"), failureDecryptionFailure}, + "invalid request": {apiErr("InvalidRequestException"), failureInvalidRequest}, + "invalid parameter": {apiErr("InvalidParameterException"), failureInvalidRequest}, + "throttling": {apiErr("ThrottlingException"), failureThrottling}, + "internal": {apiErr("InternalServiceError"), failureInternalServiceError}, + "unknown code": {apiErr("SomethingElse"), failureOther}, + "extension 4xx": {&lambdaExtensionStatusError{statusCode: 400, body: "bad request"}, failureClientError}, + "extension 5xx": {&lambdaExtensionStatusError{statusCode: 500, body: "oops"}, failureServerError}, + "extension other": {&lambdaExtensionStatusError{statusCode: 302, body: "redirect"}, failureOther}, + "deadline": {fmt.Errorf("wrapped: %w", context.DeadlineExceeded), failureTimeout}, + "canceled": {&smithy.CanceledError{Err: context.Canceled}, failureTimeout}, + "net timeout": {&url.Error{Op: "Get", URL: "u", Err: timeoutError{}}, failureTimeout}, + "invalid response": {&invalidResponseError{errors.New("bad")}, failureInvalidResponse}, + "other": {errors.New("connection refused"), failureOther}, + } { + t.Run(name, func(t *testing.T) { + require.Equal(t, tc.expected, classifyFailure(tc.err)) + }) + } +} + +// Mock telemetry utilities for other tests // + +type recordingLogger struct { + mu sync.Mutex + logs []string +} + +func (*recordingLogger) Debug(string, ...any) {} +func (l *recordingLogger) Info(msg string, keyvals ...any) { l.log("INFO", msg, keyvals) } +func (l *recordingLogger) Warn(msg string, keyvals ...any) { l.log("WARN", msg, keyvals) } +func (l *recordingLogger) Error(msg string, keyvals ...any) { l.log("ERROR", msg, keyvals) } + +func (l *recordingLogger) log(level, msg string, keyvals []any) { + msg = level + " " + msg + for i := 0; i+1 < len(keyvals); i += 2 { + msg += fmt.Sprintf(" %v=%v", keyvals[i], keyvals[i+1]) + } + l.mu.Lock() + defer l.mu.Unlock() + l.logs = append(l.logs, msg) +} + +// take returns and clears the recorded logs +func (l *recordingLogger) take() []string { + l.mu.Lock() + defer l.mu.Unlock() + ret := l.logs + l.logs = nil + return ret +} + +// recordingMetrics is a metrics handler that records values by name and tags. +type recordingMetrics struct { + tags map[string]string + data *recordedMetrics +} + +type recordedMetrics struct { + mu sync.Mutex + counters map[string]int64 // keyed by metricKey + timers map[string][]time.Duration // keyed by metricKey +} + +func newRecordingMetrics() *recordingMetrics { + return &recordingMetrics{data: &recordedMetrics{ + counters: map[string]int64{}, + timers: map[string][]time.Duration{}, + }} +} + +func metricKey(name string, tags map[string]string) string { + key := name + for _, k := range slices.Sorted(maps.Keys(tags)) { + key += " " + k + "=" + tags[k] + } + return key +} + +func tagsFromPairs(kv []any) map[string]string { + tags := map[string]string{} + for i := 0; i+1 < len(kv); i += 2 { + tags[fmt.Sprint(kv[i])] = fmt.Sprint(kv[i+1]) + } + return tags +} + +func (m *recordingMetrics) WithTags(tags map[string]string) client.MetricsHandler { + newTags := maps.Clone(m.tags) + if newTags == nil { + newTags = map[string]string{} + } + maps.Copy(newTags, tags) + return &recordingMetrics{tags: newTags, data: m.data} +} + +func (m *recordingMetrics) Counter(name string) client.MetricsCounter { + key := metricKey(name, m.tags) + return counterFunc(func(d int64) { + m.data.mu.Lock() + defer m.data.mu.Unlock() + m.data.counters[key] += d + }) +} + +func (m *recordingMetrics) Gauge(string) client.MetricsGauge { + return client.MetricsNopHandler.Gauge("") +} + +func (m *recordingMetrics) Timer(name string) client.MetricsTimer { + key := metricKey(name, m.tags) + return timerFunc(func(d time.Duration) { + m.data.mu.Lock() + defer m.data.mu.Unlock() + m.data.timers[key] = append(m.data.timers[key], d) + }) +} + +type counterFunc func(int64) + +func (f counterFunc) Inc(d int64) { f(d) } + +type timerFunc func(time.Duration) + +func (f timerFunc) Record(d time.Duration) { f(d) } + +func (m *recordingMetrics) counter(name string, tagPairs ...any) int64 { + m.data.mu.Lock() + defer m.data.mu.Unlock() + return m.data.counters[metricKey(name, tagsFromPairs(tagPairs))] +} + +// counters returns all recorded counters with the given name, keyed by metricKey +func (m *recordingMetrics) counters(name string) map[string]int64 { + m.data.mu.Lock() + defer m.data.mu.Unlock() + ret := map[string]int64{} + for k, v := range m.data.counters { + if strings.HasPrefix(k, name+" ") { + ret[k] = v + } + } + return ret +} + +func (m *recordingMetrics) timer(name string, tagPairs ...any) []time.Duration { + m.data.mu.Lock() + defer m.data.mu.Unlock() + return m.data.timers[metricKey(name, tagsFromPairs(tagPairs))] +}