diff --git a/aws/aws.go b/aws/aws.go index 934da62..c2f5646 100644 --- a/aws/aws.go +++ b/aws/aws.go @@ -4,7 +4,6 @@ package aws import ( "context" - "fmt" "strings" "time" @@ -15,6 +14,24 @@ import ( "github.com/cashapp/blip/v2" ) +// regionError is returned when AWS region auto-detection fails. It keeps the +// IMDS error so callers can inspect it with errors.Is and errors.As. +type regionError struct { + err error +} + +func (e regionError) Error() string { + return "cannot auto-detect AWS region (EC2 IMDS query failed)" +} + +func (e regionError) Unwrap() error { + return e.err +} + +// imdsOptions configures the IMDS client that Region uses. Tests set it to use +// a local IMDS server. +var imdsOptions = imds.Options{} + type ConfigFactory struct { region string } @@ -27,7 +44,7 @@ func (f *ConfigFactory) Make(ba blip.AWS, endpoint string) (aws.Config, error) { ba.Region, err = Region(ctx) if err != nil { blip.Debug("cannot auto-detect AWS region: %s", err) - return aws.Config{}, fmt.Errorf("cannot auto-detect AWS region (EC2 IMDS query failed)") + return aws.Config{}, regionError{err: err} } if f.region == "" { blip.Debug("set factory region to %s", ba.Region) @@ -60,7 +77,7 @@ func (f *ConfigFactory) Make(ba blip.AWS, endpoint string) (aws.Config, error) { // If the region cannot be detect, it returns an empty string. func Region(ctx context.Context) (string, error) { blip.Debug("auto-detect AWS region from IMDS") - client := imds.New(imds.Options{}) + client := imds.New(imdsOptions) ec2, err := client.GetInstanceIdentityDocument(ctx, &imds.GetInstanceIdentityDocumentInput{}) if err != nil { return "", err diff --git a/aws/aws_test.go b/aws/aws_test.go new file mode 100644 index 0000000..2147eef --- /dev/null +++ b/aws/aws_test.go @@ -0,0 +1,41 @@ +// Copyright 2026 Block, Inc. + +package aws + +import ( + "errors" + "net/http" + "net/http/httptest" + "testing" + + "github.com/aws/aws-sdk-go-v2/feature/ec2/imds" + + "github.com/cashapp/blip/v2" +) + +func TestConfigFactoryWrapsRegionError(t *testing.T) { + // IMDS that fails every request, so region auto-detection fails + server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + http.Error(w, "not found", http.StatusNotFound) + })) + defer server.Close() + imdsOptions = imds.Options{Endpoint: server.URL} + defer func() { imdsOptions = imds.Options{} }() + + f := &ConfigFactory{} + _, err := f.Make(blip.AWS{Region: "auto"}, "") + if err == nil { + t.Fatal("got nil error, expected region auto-detection error") + } + if got, expect := err.Error(), "cannot auto-detect AWS region (EC2 IMDS query failed)"; got != expect { + t.Errorf("got error %q, expected %q", got, expect) + } + // The IMDS HTTP response error, which has the status code + var respErr interface{ HTTPStatusCode() int } + if !errors.As(err, &respErr) { + t.Fatalf("errors.As(%v, HTTPStatusCode() int) = false, expected the IMDS error", err) + } + if respErr.HTTPStatusCode() != http.StatusNotFound { + t.Errorf("got IMDS status %d, expected %d", respErr.HTTPStatusCode(), http.StatusNotFound) + } +} diff --git a/aws/secret.go b/aws/secret.go index 63ecb79..a94ba80 100644 --- a/aws/secret.go +++ b/aws/secret.go @@ -33,7 +33,7 @@ func (s Secret) GetSecret(ctx context.Context) (map[string]interface{}, error) { var v map[string]interface{} if err := json.Unmarshal(payload, &v); err != nil { - return nil, fmt.Errorf("cannot decode secret string as map[string]interface{}: %s", err) + return nil, fmt.Errorf("cannot decode secret string as map[string]interface{}: %w", err) } if v == nil { return nil, fmt.Errorf("secret value is 'null' literal") @@ -49,7 +49,7 @@ func (s Secret) GetSecretPayload(ctx context.Context) ([]byte, error) { sv, err := s.client.GetSecretValue(ctx, input) if err != nil { - return nil, fmt.Errorf("Secrets Manager API error: %s", err) + return nil, fmt.Errorf("Secrets Manager API error: %w", err) } name := "" if sv.Name != nil { diff --git a/aws/secret_test.go b/aws/secret_test.go new file mode 100644 index 0000000..f218420 --- /dev/null +++ b/aws/secret_test.go @@ -0,0 +1,104 @@ +// Copyright 2026 Block, Inc. + +package aws_test + +import ( + "context" + "encoding/json" + "errors" + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + "github.com/aws/aws-sdk-go-v2/credentials" + "github.com/aws/aws-sdk-go-v2/service/secretsmanager/types" + + blipAWS "github.com/cashapp/blip/v2/aws" +) + +func testSecretsManager(t *testing.T, handler http.HandlerFunc) aws.Config { + t.Helper() + server := httptest.NewServer(handler) + t.Cleanup(server.Close) + return aws.Config{ + Credentials: credentials.NewStaticCredentialsProvider("access-key", "secret-key", ""), + EndpointResolver: aws.EndpointResolverFunc(func(service, region string) (aws.Endpoint, error) { + return aws.Endpoint{URL: server.URL, SigningRegion: "us-east-1"}, nil + }), + HTTPClient: server.Client(), + Region: "us-east-1", + Retryer: func() aws.Retryer { + return aws.NopRetryer{} + }, + } +} + +func TestGetSecretPayloadWrapsAPIError(t *testing.T) { + cfg := testSecretsManager(t, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/x-amz-json-1.1") + w.WriteHeader(http.StatusBadRequest) + w.Write([]byte(`{"__type":"ResourceNotFoundException","message":"no such secret"}`)) + }) + + _, err := blipAWS.NewSecret("test-secret", cfg).GetSecretPayload(context.Background()) + if err == nil { + t.Fatal("got nil error, expected API error") + } + if !strings.HasPrefix(err.Error(), "Secrets Manager API error: ") { + t.Errorf("got error %q, expected prefix 'Secrets Manager API error: '", err) + } + var notFound *types.ResourceNotFoundException + if !errors.As(err, ¬Found) { + t.Errorf("errors.As(%v, *types.ResourceNotFoundException) = false, expected true", err) + } +} + +func TestGetSecretPayloadWrapsCanceled(t *testing.T) { + // The first call might have to fetch AWS credentials (e.g. IRSA). If the + // caller's context expires first, the SDK credentials cache returns + // *aws.RequestCanceledError, which callers must be able to detect. + cfg := testSecretsManager(t, func(w http.ResponseWriter, r *http.Request) { + t.Error("Secrets Manager called without credentials") + }) + cfg.Credentials = aws.NewCredentialsCache(aws.CredentialsProviderFunc(func(ctx context.Context) (aws.Credentials, error) { + select { + case <-time.After(2 * time.Second): + case <-ctx.Done(): + } + return aws.Credentials{}, errors.New("slow credentials") + })) + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer cancel() + _, err := blipAWS.NewSecret("test-secret", cfg).GetSecretPayload(ctx) + if err == nil { + t.Fatal("got nil error, expected request canceled error") + } + var canceled *aws.RequestCanceledError + if !errors.As(err, &canceled) { + t.Errorf("errors.As(%v, *aws.RequestCanceledError) = false, expected true", err) + } + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("errors.Is(%v, context.DeadlineExceeded) = false, expected true", err) + } +} + +func TestGetSecretWrapsDecodeError(t *testing.T) { + cfg := testSecretsManager(t, func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/x-amz-json-1.1") + json.NewEncoder(w).Encode(map[string]string{ + "Name": "test-secret", + "SecretString": "not-json", + "VersionId": "test-version", + }) + }) + + _, err := blipAWS.NewSecret("test-secret", cfg).GetSecret(context.Background()) + var syntaxErr *json.SyntaxError + if !errors.As(err, &syntaxErr) { + t.Errorf("errors.As(%v, *json.SyntaxError) = false, expected true", err) + } +} diff --git a/blip.go b/blip.go index 6a5ccb8..4278bee 100644 --- a/blip.go +++ b/blip.go @@ -121,7 +121,7 @@ func DefaultPasswordSecretParser(_ context.Context, cfg ConfigMonitor, payload [ var secretPayload map[string]interface{} if err := json.Unmarshal(payload, &secretPayload); err != nil { - return fmt.Errorf("cannot decode secret as JSON object: %s", err) + return fmt.Errorf("cannot decode secret as JSON object: %w", err) } if secretPayload == nil { return fmt.Errorf("secret value is 'null' literal") diff --git a/dbconn/credential_timeout_test.go b/dbconn/credential_timeout_test.go new file mode 100644 index 0000000..8b3bd3e --- /dev/null +++ b/dbconn/credential_timeout_test.go @@ -0,0 +1,125 @@ +// Copyright 2026 Block, Inc. + +package dbconn + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/aws/aws-sdk-go-v2/aws" + + "github.com/cashapp/blip/v2" +) + +// slowCredentialsConfig returns an AWS config for a Secrets Manager stub +// with AWS credentials that take delay to fetch, like the first IRSA +// AssumeRoleWithWebIdentity call. +func slowCredentialsConfig(t *testing.T, delay time.Duration) aws.Config { + t.Helper() + awscfg, cleanup := testPasswordSecretConfig(t, `{"password":"secret-pass"}`) + t.Cleanup(cleanup) + awscfg.Credentials = aws.NewCredentialsCache(aws.CredentialsProviderFunc(func(ctx context.Context) (aws.Credentials, error) { + select { + case <-time.After(delay): + return aws.Credentials{AccessKeyID: "access-key", SecretAccessKey: "secret-key", Source: "test"}, nil + case <-ctx.Done(): + return aws.Credentials{}, ctx.Err() + } + })) + return awscfg +} + +func passwordSecretMonitor() blip.ConfigMonitor { + return blip.ConfigMonitor{ + MonitorId: "m1", + Hostname: "127.0.0.1:33997", + Username: "config-user", + AWS: blip.ConfigAWS{ + PasswordSecret: "test-secret", + Region: "us-east-1", + }, + } +} + +func TestMakeCredentialTimeoutDefault(t *testing.T) { + // The first credentials from AWS can take seconds (fetching AWS credentials + // via IRSA, AssumeRole, etc.), so Make must wait longer than the old 500ms. + awscfg := slowCredentialsConfig(t, 700*time.Millisecond) + f := NewConnFactory(testAWSConfigFactory{cfg: awscfg}, nil) + + db, _, err := f.Make(passwordSecretMonitor()) + if err != nil { + t.Fatalf("got error %s, expected nil", err) + } + db.Close() +} + +func TestMakeCredentialTimeoutOption(t *testing.T) { + awscfg := slowCredentialsConfig(t, 2*time.Second) + f := NewConnFactory(testAWSConfigFactory{cfg: awscfg}, nil, WithCredentialTimeout(50*time.Millisecond)) + + t0 := time.Now() + db, _, err := f.Make(passwordSecretMonitor()) + elapsed := time.Since(t0) + if err == nil { + db.Close() + t.Fatal("got nil error, expected credential timeout") + } + if elapsed > 5*time.Second { + t.Errorf("Make took %s, expected about 50ms", elapsed) + } + + // The SDK error must survive wrapping so callers can tell a timeout or + // cancellation from, e.g., a bad secret + var canceled *aws.RequestCanceledError + if !errors.As(err, &canceled) { + t.Errorf("errors.As(%v, *aws.RequestCanceledError) = false, expected true", err) + } + if !errors.Is(err, context.DeadlineExceeded) { + t.Errorf("errors.Is(%v, context.DeadlineExceeded) = false, expected true", err) + } +} + +func TestCredentialContext(t *testing.T) { + tests := []struct { + name string + f factory + expectDeadline time.Duration // 0: no deadline + }{ + {"default", NewConnFactory(nil, nil), DefaultCredentialTimeout}, + {"zero value", factory{}, DefaultCredentialTimeout}, + {"option", NewConnFactory(nil, nil, WithCredentialTimeout(5*time.Second)), 5 * time.Second}, + {"disabled", NewConnFactory(nil, nil, WithCredentialTimeout(0)), 0}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + t0 := time.Now() + ctx, cancel := tt.f.credentialContext() + defer cancel() + deadline, ok := ctx.Deadline() + if tt.expectDeadline == 0 { + if ok { + t.Errorf("got deadline in %s, expected no deadline", deadline.Sub(t0)) + } + return + } + if !ok { + t.Fatalf("got no deadline, expected deadline in %s", tt.expectDeadline) + } + if d := deadline.Sub(t0); d < tt.expectDeadline-time.Second || d > tt.expectDeadline+time.Second { + t.Errorf("got deadline in %s, expected about %s", d, tt.expectDeadline) + } + }) + } +} + +func TestWithCredentialTimeoutNegativePanics(t *testing.T) { + defer func() { + if r := recover(); r == nil { + t.Error("WithCredentialTimeout(-1s) didn't panic") + } + }() + WithCredentialTimeout(-time.Second) +} diff --git a/dbconn/factory.go b/dbconn/factory.go index 278c268..4705e6b 100644 --- a/dbconn/factory.go +++ b/dbconn/factory.go @@ -31,11 +31,23 @@ var rdsAddr = regexp.MustCompile(`rds\.amazonaws\.com(:\d+)?$`) // strip the port suffix before passing the hostname to LoadTLS. var portSuffix = regexp.MustCompile(`:\d+$`) +// DefaultCredentialTimeout bounds how long Make waits for the first +// credentials. It's long because the first call to an AWS source (Secrets +// Manager or IAM auth) might have to fetch AWS credentials, for example EKS +// IRSA AssumeRoleWithWebIdentity plus a cross-account AssumeRole, which can +// take several seconds. Use WithCredentialTimeout to change it. +const DefaultCredentialTimeout = 30 * time.Second + // factory is the internal implementation of blip.DbFactory. type factory struct { awsConfig blip.AWSConfigFactory modifyDB func(*sql.DB, string) passwordSecretParser blip.PasswordSecretParser + + // credentialTimeout is set by WithCredentialTimeout. Zero means + // DefaultCredentialTimeout, so the zero value factory has the default. + // Negative means no timeout. + credentialTimeout time.Duration } // ConnFactoryOption configures the default MySQL connection factory. @@ -49,6 +61,25 @@ func WithPasswordSecretParser(parser blip.PasswordSecretParser) ConnFactoryOptio } } +// WithCredentialTimeout sets how long Make waits for the credentials reload +// func to return the first credentials. The default is DefaultCredentialTimeout. +// Zero disables the timeout, so a hung credentials source blocks Make, and +// therefore a monitor stop or Blip startup, until it returns. It panics if d is +// negative, so validate a value from user config before passing it. It does +// not affect credential reloads by the mysql-hotswap-dsn driver, which use the +// connect context. +func WithCredentialTimeout(d time.Duration) ConnFactoryOption { + if d < 0 { + panic(fmt.Sprintf("dbconn.WithCredentialTimeout: negative timeout %s", d)) + } + if d == 0 { + d = -1 // no timeout; see factory.credentialTimeout + } + return func(f *factory) { + f.credentialTimeout = d + } +} + // NewConnFactory returns a blip.NewConnFactory that connects to MySQL. // This is the only blip.NewConnFactor. It is created in Server.Defaults. func NewConnFactory(awsConfig blip.AWSConfigFactory, modifyDB func(*sql.DB, string), opts ...ConnFactoryOption) factory { @@ -128,7 +159,7 @@ func (f factory) Make(cfg blip.ConfigMonitor) (*sql.DB, string, error) { // Test the credentials reload func, i.e. get the current credentials, which // might just be a static credentials in the Blip config file or another file, // but it could be something dynamic like an Amazon IAM auth token. - ctx, cancel := context.WithTimeout(context.Background(), 500*time.Millisecond) + ctx, cancel := f.credentialContext() defer cancel() credentials, err := credentialFunc(ctx) if err != nil { @@ -284,6 +315,20 @@ func (f factory) Make(cfg blip.ConfigMonitor) (*sql.DB, string, error) { return db, RedactedDSN(dsn), nil } +// credentialContext returns the context for the first call to the credentials +// reload func in Make. It has a deadline unless WithCredentialTimeout(0) +// disabled it. +func (f factory) credentialContext() (context.Context, context.CancelFunc) { + switch { + case f.credentialTimeout < 0: + return context.WithCancel(context.Background()) + case f.credentialTimeout == 0: + return context.WithTimeout(context.Background(), DefaultCredentialTimeout) + default: + return context.WithTimeout(context.Background(), f.credentialTimeout) + } +} + // Credentials creates a credentials reload function (callback) based on the // configured credential method. This function is used by the mysql-hotswap-dsn // driver (see reload_password.go). For a consistent abstraction, all diff --git a/docs/content/about/release-notes.md b/docs/content/about/release-notes.md index 4367805..d8c147d 100644 --- a/docs/content/about/release-notes.md +++ b/docs/content/about/release-notes.md @@ -66,6 +66,8 @@ Default sink HTTP clients now have a 10-second whole-request timeout plus bounde * **Behavior change:** Plans that set the `repl.lag` `writer` to `pfs` at one level and `blip` at another now fail to prepare, and the monitor keeps retrying the plan. Previously each level used its own writer. Set the same `writer` at every level, or set it at one level only. * Changed `heartbeat.BlipReader.Stop` to wait for the reader to return, which takes up to the heartbeat read timeout (2s) if a heartbeat query is running. Previously the reader could report status after its monitor was stopped and removed, so the removed monitor reappeared in [`GET /status/monitors`]({{< ref "/api/status#get-statusmonitors" >}}). * **Behavior change:** Fixed `monitor-loader.stop-loss` to count the net drop in the number of monitors. Previously it compared the number of removed monitors with the number loaded, so a small removal could trigger stop-loss and removing every monitor could not. A percent stop-loss (for example, `50%`) never triggered; it now triggers when the drop is greater than that percent of the monitors loaded before the reload. +* **Behavior change:** Changed the default MySQL connection factory to wait up to 30s (`dbconn.DefaultCredentialTimeout`) for a monitor's first credentials. Previously it waited 500ms, so AWS Secrets Manager or IAM auth credentials often failed on the first try with `request canceled, context deadline exceeded` when the AWS SDK first had to fetch AWS credentials (for example, EKS IRSA plus a cross-account `AssumeRole`). The monitor retried, but loading shared plans from `plans.table` failed, so Blip exited at startup. Added the `dbconn.WithCredentialTimeout` option to change it; a negative value panics. A monitor stopped while it waits for credentials can take up to the credential timeout to stop. If its plans are in a table, it can take twice that, because loading them makes another connection. Blip startup can wait up to the credential timeout when it loads shared plans from `plans.table`. When no monitors are configured and the top-level `aws` config sets `iam-auth` or `password-secret`, local auto-detect can also wait up to the credential timeout for each user and socket it tries. Zero disables the timeout, so a hung credentials source can block a monitor stop or Blip startup indefinitely. +* Fixed `aws.Secret.GetSecretPayload`, `aws.Secret.GetSecret`, and `blip.DefaultPasswordSecretParser`, which formatted errors with `%s`, and `aws.ConfigFactory.Make`, which dropped the IMDS error when AWS region auto-detection failed. Callers couldn't use `errors.As` or `errors.Is` to detect the underlying error, such as `*aws.RequestCanceledError` or a Secrets Manager API error type. They now wrap the error; error messages are unchanged. ### v2.1.0 (10 Aug 2026) diff --git a/docs/content/config/config-file.md b/docs/content/config/config-file.md index 3acf17e..90ffde6 100644 --- a/docs/content/config/config-file.md +++ b/docs/content/config/config-file.md @@ -206,7 +206,7 @@ A stop can take longer in these cases: * A collector, or its cleanup function, doesn't return promptly when its collection is canceled. A stop waits for every collector to return, then runs their cleanup functions. * A plan change is in progress, and a collector's `Prepare` doesn't return promptly when it's canceled. A stop cancels the plan change and waits for it. -* The monitor is starting. A stop waits for it to make its connections, which calls its credentials function (for example, for an IAM auth token, up to 500 milliseconds each time) or a custom `DbProviderFactory`, and to load its plans, which queries MySQL only if they're in a table ([`plans.table`](#table-1)): up to 5 seconds, connecting included, when MySQL is unreachable or hung. +* The monitor is starting. A stop waits for it to make its connections, which calls its credentials function (for example, for an IAM auth token or AWS Secrets Manager password, up to 30 seconds each time by default, while the AWS SDK fetches AWS credentials) or a custom `DbProviderFactory`, and to load its plans, which queries MySQL only if they're in a table ([`plans.table`](#table-1)): one more call to its credentials function (up to the credential timeout), then up to 5 seconds, connecting included, when MySQL is unreachable or hung. --- diff --git a/secret_test.go b/secret_test.go index d621485..157a0a6 100644 --- a/secret_test.go +++ b/secret_test.go @@ -4,6 +4,8 @@ package blip_test import ( "context" + "encoding/json" + "errors" "testing" "github.com/cashapp/blip/v2" @@ -87,3 +89,11 @@ func TestDefaultPasswordSecretParserNilCredentials(t *testing.T) { t.Fatal("got nil error, expected non-nil error") } } + +func TestDefaultPasswordSecretParserWrapsDecodeError(t *testing.T) { + err := blip.DefaultPasswordSecretParser(context.Background(), blip.ConfigMonitor{}, []byte(`not-json`), &blip.DbCredentials{}) + var syntaxErr *json.SyntaxError + if !errors.As(err, &syntaxErr) { + t.Errorf("errors.As(%v, *json.SyntaxError) = false, expected true", err) + } +}