From 58e68aaaa9fca13ce01d4e595398899f520b6791 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Thu, 6 Aug 2026 14:26:15 -0700 Subject: [PATCH 1/6] Guard PostgreSQL monitor configuration Keep MySQL-only heartbeat and plan defaults off PostgreSQL monitors, reject explicit unsupported settings, and sign IAM tokens with the database-specific default port. Co-authored-by: Codex Ai-assisted: true --- aws/iam_auth.go | 10 +++- config.go | 35 ++++++++++++-- config_postgres_test.go | 66 +++++++++++++++++++++++++ credentials/credentials.go | 6 ++- credentials/credentials_test.go | 82 ++++++++++++++++++++++++++++++++ docs/content/config/heartbeat.md | 5 ++ docs/content/plans/changing.md | 5 ++ docs/content/plans/table.md | 6 +++ 8 files changed, 207 insertions(+), 8 deletions(-) diff --git a/aws/iam_auth.go b/aws/iam_auth.go index 4e70e6f9..1cffefb3 100644 --- a/aws/iam_auth.go +++ b/aws/iam_auth.go @@ -19,9 +19,15 @@ type AuthToken struct { } func NewAuthToken(username, hostname string, cfg aws.Config) AuthToken { - // RDS auth tokens require the :3306 suffix + return NewAuthTokenWithDefaultPort(username, hostname, "3306", cfg) +} + +// NewAuthTokenWithDefaultPort constructs an RDS authentication token signer, +// adding defaultPort when hostname does not already include a port. +func NewAuthTokenWithDefaultPort(username, hostname, defaultPort string, cfg aws.Config) AuthToken { + // RDS auth tokens require the database port in the signed endpoint. if !portRe.MatchString(hostname) { - hostname += ":3306" + hostname += ":" + defaultPort } return AuthToken{ diff --git a/config.go b/config.go index 1ac52def..31c6f811 100644 --- a/config.go +++ b/config.go @@ -483,6 +483,15 @@ func (c ConfigMonitor) Validate() error { if c.MyCnf != "" { return fmt.Errorf("config.monitor.mycnf is only supported for database-type %q", DatabaseTypeMySQL) } + if c.Heartbeat.set() { + return fmt.Errorf("config.monitor.heartbeat is only supported for database-type %q", DatabaseTypeMySQL) + } + if c.Plans.Change.Enabled() { + return fmt.Errorf("config.monitor.plans.change is only supported for database-type %q", DatabaseTypeMySQL) + } + if c.Plans.Table != "" { + return fmt.Errorf("config.monitor.plans.table is only supported for database-type %q", DatabaseTypeMySQL) + } return c.Postgres.Validate() default: return fmt.Errorf("config.monitor.database-type: invalid database type %q", c.DatabaseType) @@ -503,7 +512,8 @@ func (c ConfigMonitor) redacted(seen map[*ConfigMonitor]*ConfigMonitor) ConfigMo } func (c *ConfigMonitor) ApplyDefaults(b Config) { - if c.EffectiveDatabaseType() == DatabaseTypeMySQL { + databaseType := c.EffectiveDatabaseType() + if databaseType == DatabaseTypeMySQL { if c.Socket == "" { c.Socket = b.MySQL.Socket } @@ -547,9 +557,14 @@ func (c *ConfigMonitor) ApplyDefaults(b Config) { c.AWS.ApplyDefaults(b) c.Exporter.ApplyDefaults(b) c.HA.ApplyDefaults(b) - c.Heartbeat.ApplyDefaults(b) - c.Plans.ApplyDefaults(b) - if c.EffectiveDatabaseType() == DatabaseTypePostgres { + // Heartbeat writes and plan state changes use MySQL-specific SQL. Do not + // inherit their global defaults into PostgreSQL monitors; explicit monitor + // values remain intact so Validate can report the unsupported configuration. + if databaseType == DatabaseTypeMySQL { + c.Heartbeat.ApplyDefaults(b) + } + c.Plans.applyDefaults(b, databaseType == DatabaseTypeMySQL) + if databaseType == DatabaseTypePostgres { postgresDefaults := DefaultConfigPostgres() postgresDefaults.ConnectTimeout = c.TimeoutConnect c.Postgres.ApplyDefaults(postgresDefaults) @@ -798,6 +813,10 @@ type ConfigHeartbeat struct { Table string `yaml:"table,omitempty"` } +func (c ConfigHeartbeat) set() bool { + return c.Freq != "" || c.SourceId != "" || c.Role != "" || c.Table != "" +} + const ( DEFAULT_HEARTBEAT_TABLE = "blip.heartbeat" ) @@ -986,11 +1005,17 @@ func (c ConfigPlans) redacted(seen map[*ConfigMonitor]*ConfigMonitor) ConfigPlan } func (c *ConfigPlans) ApplyDefaults(b Config) { + c.applyDefaults(b, true) +} + +func (c *ConfigPlans) applyDefaults(b Config, applyChange bool) { if len(c.Files) == 0 && len(b.Plans.Files) > 0 { c.Files = make([]string, len(b.Plans.Files)) copy(c.Files, b.Plans.Files) } - c.Change.ApplyDefaults(b) + if applyChange { + c.Change.ApplyDefaults(b) + } } func (c *ConfigPlans) InterpolateEnvVars() { diff --git a/config_postgres_test.go b/config_postgres_test.go index 8a38a17c..32dc0d44 100644 --- a/config_postgres_test.go +++ b/config_postgres_test.go @@ -124,6 +124,34 @@ func TestConfigMonitorDatabaseTypeValidation(t *testing.T) { }, wantError: "socket is only supported", }, + { + name: "heartbeat on PostgreSQL monitor", + monitor: blip.ConfigMonitor{ + DatabaseType: blip.DatabaseTypePostgres, + Heartbeat: blip.ConfigHeartbeat{Freq: "1s"}, + }, + wantError: "heartbeat is only supported", + }, + { + name: "plan changing on PostgreSQL monitor", + monitor: blip.ConfigMonitor{ + DatabaseType: blip.DatabaseTypePostgres, + Plans: blip.ConfigPlans{ + Change: blip.ConfigPlanChange{ + Active: blip.ConfigStatePlan{Plan: "active"}, + }, + }, + }, + wantError: "plans.change is only supported", + }, + { + name: "plan table on PostgreSQL monitor", + monitor: blip.ConfigMonitor{ + DatabaseType: blip.DatabaseTypePostgres, + Plans: blip.ConfigPlans{Table: "blip.plans"}, + }, + wantError: "plans.table is only supported", + }, } for _, tt := range tests { @@ -136,6 +164,44 @@ func TestConfigMonitorDatabaseTypeValidation(t *testing.T) { } } +func TestConfigMonitorDatabaseSpecificDefaults(t *testing.T) { + defaults := blip.DefaultConfig() + defaults.Heartbeat = blip.ConfigHeartbeat{ + Freq: "1s", + Table: "blip.heartbeat", + } + defaults.Plans = blip.ConfigPlans{ + Files: []string{"shared.yaml"}, + Change: blip.ConfigPlanChange{ + Active: blip.ConfigStatePlan{Plan: "active"}, + }, + } + + postgres := blip.ConfigMonitor{DatabaseType: blip.DatabaseTypePostgres} + postgres.ApplyDefaults(defaults) + if postgres.Heartbeat != (blip.ConfigHeartbeat{}) { + t.Fatalf("PostgreSQL monitor inherited heartbeat defaults: %+v", postgres.Heartbeat) + } + if postgres.Plans.Change.Enabled() { + t.Fatalf("PostgreSQL monitor inherited plan-changing defaults: %+v", postgres.Plans.Change) + } + if got := postgres.Plans.Files; len(got) != 1 || got[0] != "shared.yaml" { + t.Fatalf("PostgreSQL monitor plan files = %#v, expected shared plan defaults", got) + } + if err := postgres.Validate(); err != nil { + t.Fatalf("PostgreSQL monitor with shared defaults is invalid: %v", err) + } + + mysql := blip.ConfigMonitor{} + mysql.ApplyDefaults(defaults) + if mysql.Heartbeat == (blip.ConfigHeartbeat{}) { + t.Fatal("MySQL monitor did not inherit heartbeat defaults") + } + if !mysql.Plans.Change.Enabled() { + t.Fatal("MySQL monitor did not inherit plan-changing defaults") + } +} + func TestConfigPostgresAllowsExplicitUnlimitedPoolSettings(t *testing.T) { zero := 0 config := blip.ConfigPostgres{ diff --git a/credentials/credentials.go b/credentials/credentials.go index b1ddf897..880d7462 100644 --- a/credentials/credentials.go +++ b/credentials/credentials.go @@ -47,7 +47,11 @@ func (f Factory) Dynamic(cfg blip.ConfigMonitor) (Func, bool, error) { if err != nil { return nil, true, err } - token := blipaws.NewAuthToken(cfg.Username, cfg.Hostname, awscfg) + defaultPort := "3306" + if cfg.EffectiveDatabaseType() == blip.DatabaseTypePostgres { + defaultPort = "5432" + } + token := blipaws.NewAuthTokenWithDefaultPort(cfg.Username, cfg.Hostname, defaultPort, awscfg) return func(ctx context.Context) (blip.DbCredentials, error) { password, err := token.Password(ctx) if err != nil { diff --git a/credentials/credentials_test.go b/credentials/credentials_test.go index 0b54f089..8a193892 100644 --- a/credentials/credentials_test.go +++ b/credentials/credentials_test.go @@ -4,15 +4,26 @@ package credentials_test import ( "context" + "net/url" "os" "path/filepath" "strings" "testing" + awsv2 "github.com/aws/aws-sdk-go-v2/aws" + "github.com/cashapp/blip" "github.com/cashapp/blip/credentials" ) +type staticAWSConfigFactory struct { + config awsv2.Config +} + +func (f staticAWSConfigFactory) Make(blip.AWS, string) (awsv2.Config, error) { + return f.config, nil +} + func TestDynamicPasswordFileReloads(t *testing.T) { passwordFile := filepath.Join(t.TempDir(), "password") if err := os.WriteFile(passwordFile, []byte("first"), 0o600); err != nil { @@ -93,6 +104,77 @@ func TestDynamicPreservesSourcePrecedence(t *testing.T) { } } +func TestDynamicIAMUsesDatabaseDefaultPort(t *testing.T) { + iamAuth := true + factory := credentials.NewFactory(staticAWSConfigFactory{config: awsv2.Config{ + Region: "us-west-2", + Credentials: awsv2.CredentialsProviderFunc(func(context.Context) (awsv2.Credentials, error) { + return awsv2.Credentials{ + AccessKeyID: "test-access-key-do-not-use", + SecretAccessKey: "test-secret-key-do-not-use", + SessionToken: "test-session-token-do-not-use", + }, nil + }), + }}, nil) + + tests := []struct { + name string + databaseType blip.DatabaseType + hostname string + wantHost string + }{ + { + name: "implicit MySQL", + hostname: "mysql.example", + wantHost: "mysql.example:3306", + }, + { + name: "PostgreSQL", + databaseType: blip.DatabaseTypePostgres, + hostname: "postgres.example", + wantHost: "postgres.example:5432", + }, + { + name: "PostgreSQL custom port", + databaseType: blip.DatabaseTypePostgres, + hostname: "postgres.example:6432", + wantHost: "postgres.example:6432", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + credentialFunc, selected, err := factory.Dynamic(blip.ConfigMonitor{ + DatabaseType: tt.databaseType, + Hostname: tt.hostname, + Username: "metrics", + AWS: blip.ConfigAWS{ + IAMAuth: &iamAuth, + Region: "us-west-2", + }, + }) + if err != nil { + t.Fatal(err) + } + if !selected { + t.Fatal("IAM credentials were not selected") + } + + got, err := credentialFunc(context.Background()) + if err != nil { + t.Fatal(err) + } + tokenURL, err := url.Parse("https://" + got.Password) + if err != nil { + t.Fatal(err) + } + if tokenURL.Host != tt.wantHost { + t.Fatalf("IAM token host = %q, expected %q", tokenURL.Host, tt.wantHost) + } + }) + } +} + func TestDynamicReportsNoSharedSource(t *testing.T) { credentialFunc, selected, err := credentials.NewFactory(nil, nil).Dynamic(blip.ConfigMonitor{ Username: "metrics", diff --git a/docs/content/config/heartbeat.md b/docs/content/config/heartbeat.md index cb0c69da..9882bf96 100644 --- a/docs/content/config/heartbeat.md +++ b/docs/content/config/heartbeat.md @@ -6,6 +6,11 @@ Although MySQL has built-in replication heartbeats and lag metrics, they are not For example, `Seconds_Behind_Source` from `SHOW REPLICA STATUS` (or `Seconds_Behind_Master` from `SHOW SLAVE STATUS` before MySQL 8.022) is always on but infamously inaccurate: it reports zero when a network issue blocks replication. Consequently, external replication heartbeats are an industry norm because they are easy and accurate—and they work the same across all versions and distributions of MySQL, including the the cloud. +{{< hint type=note >}} +Blip heartbeat is MySQL-specific. PostgreSQL monitors do not inherit global +heartbeat defaults and reject monitor-level heartbeat configuration. +{{< /hint >}} + ## Quick Start Presuming one source MySQL instance and one read-only replica, the minimal configuration is: diff --git a/docs/content/plans/changing.md b/docs/content/plans/changing.md index 1394cdc7..60a484f8 100644 --- a/docs/content/plans/changing.md +++ b/docs/content/plans/changing.md @@ -4,6 +4,11 @@ title: "Changing" Plan changing makes Blip change plans while running (without restarting) based on the state of MySQL: +{{< hint type=note >}} +Plan changing is MySQL-specific. PostgreSQL monitors do not inherit global +plan-changing defaults and reject monitor-level plan-changing configuration. +{{< /hint >}} + |State|Connected to MySQL|Collecting Metrics|Description| |-----|------------------|------------------|-----------| |`offline`|no|no|Completely offline, no connection to MySQL| diff --git a/docs/content/plans/table.md b/docs/content/plans/table.md index 05dfca4c..1e217826 100644 --- a/docs/content/plans/table.md +++ b/docs/content/plans/table.md @@ -4,6 +4,12 @@ title: "Table" A plan table contains one plan per row: +{{< hint type=note >}} +Monitor-level plan tables are MySQL-specific. PostgreSQL monitors can use plan +files or compatible shared plans, but reject `plans.table` in their monitor +configuration. +{{< /hint >}} + ```sql CREATE TABLE IF NOT EXISTS plans ( name VARCHAR(100) NOT NULL PRIMARY KEY, From 1474cf8d22496d13c2de3ed4248faa14fba73bd8 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Thu, 6 Aug 2026 14:47:14 -0700 Subject: [PATCH 2/6] Close PostgreSQL integration gaps Cancel and join plan preparation before closing monitor-owned database resources, and reject the remaining MySQL-only plan configurations for PostgreSQL. Co-authored-by: Codex Ai-assisted: true --- config.go | 12 +++- config_postgres_test.go | 61 ++++++++++++++++ docs/content/plans/table.md | 7 +- monitor/level_collector.go | 49 ++++++++++--- monitor/provider_test.go | 136 ++++++++++++++++++++++++++++++++++++ 5 files changed, 250 insertions(+), 15 deletions(-) diff --git a/config.go b/config.go index 31c6f811..901d9ca9 100644 --- a/config.go +++ b/config.go @@ -486,7 +486,7 @@ func (c ConfigMonitor) Validate() error { if c.Heartbeat.set() { return fmt.Errorf("config.monitor.heartbeat is only supported for database-type %q", DatabaseTypeMySQL) } - if c.Plans.Change.Enabled() { + if c.Plans.Change.set() { return fmt.Errorf("config.monitor.plans.change is only supported for database-type %q", DatabaseTypeMySQL) } if c.Plans.Table != "" { @@ -985,6 +985,9 @@ func DefaultConfigPlans() ConfigPlans { } func (c ConfigPlans) Validate() error { + if c.Table != "" && c.Monitor != nil && c.Monitor.EffectiveDatabaseType() != DatabaseTypeMySQL { + return fmt.Errorf("config.plans.monitor.database-type: config.plans.table is only supported for database-type %q", DatabaseTypeMySQL) + } return nil } @@ -1104,6 +1107,13 @@ func (c ConfigPlanChange) Enabled() bool { c.Active.Plan != "" } +func (c ConfigPlanChange) set() bool { + return c.Offline.After != "" || c.Offline.Plan != "" || + c.Standby.After != "" || c.Standby.Plan != "" || + c.ReadOnly.After != "" || c.ReadOnly.Plan != "" || + c.Active.After != "" || c.Active.Plan != "" +} + // -------------------------------------------------------------------------- type ConfigSinks map[string]map[string]string diff --git a/config_postgres_test.go b/config_postgres_test.go index 32dc0d44..e35f8afb 100644 --- a/config_postgres_test.go +++ b/config_postgres_test.go @@ -144,6 +144,18 @@ func TestConfigMonitorDatabaseTypeValidation(t *testing.T) { }, wantError: "plans.change is only supported", }, + { + name: "plan change delay on PostgreSQL monitor", + monitor: blip.ConfigMonitor{ + DatabaseType: blip.DatabaseTypePostgres, + Plans: blip.ConfigPlans{ + Change: blip.ConfigPlanChange{ + Active: blip.ConfigStatePlan{After: "10s"}, + }, + }, + }, + wantError: "plans.change is only supported", + }, { name: "plan table on PostgreSQL monitor", monitor: blip.ConfigMonitor{ @@ -164,6 +176,55 @@ func TestConfigMonitorDatabaseTypeValidation(t *testing.T) { } } +func TestConfigPlansRejectsPostgresTableMonitor(t *testing.T) { + tests := []struct { + name string + plans blip.ConfigPlans + wantError string + }{ + { + name: "implicit MySQL", + plans: blip.ConfigPlans{ + Table: "blip.plans", + Monitor: &blip.ConfigMonitor{}, + }, + }, + { + name: "PostgreSQL", + plans: blip.ConfigPlans{ + Table: "blip.plans", + Monitor: &blip.ConfigMonitor{ + DatabaseType: blip.DatabaseTypePostgres, + }, + }, + wantError: "config.plans.table is only supported", + }, + { + name: "PostgreSQL without table", + plans: blip.ConfigPlans{ + Monitor: &blip.ConfigMonitor{ + DatabaseType: blip.DatabaseTypePostgres, + }, + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := tt.plans.Validate() + if tt.wantError == "" { + if err != nil { + t.Fatalf("unexpected error: %v", err) + } + return + } + if err == nil || !strings.Contains(err.Error(), tt.wantError) { + t.Fatalf("got error %v, expected it to contain %q", err, tt.wantError) + } + }) + } +} + func TestConfigMonitorDatabaseSpecificDefaults(t *testing.T) { defaults := blip.DefaultConfig() defaults.Heartbeat = blip.ConfigHeartbeat{ diff --git a/docs/content/plans/table.md b/docs/content/plans/table.md index 1e217826..ebf56bbf 100644 --- a/docs/content/plans/table.md +++ b/docs/content/plans/table.md @@ -5,9 +5,10 @@ title: "Table" A plan table contains one plan per row: {{< hint type=note >}} -Monitor-level plan tables are MySQL-specific. PostgreSQL monitors can use plan -files or compatible shared plans, but reject `plans.table` in their monitor -configuration. +Plan-table storage is MySQL-specific. The connection configured by a top-level +`plans.monitor` must be MySQL, and PostgreSQL monitors reject `plans.table` in +their monitor configuration. PostgreSQL monitors can use plan files or +compatible shared plans. {{< /hint >}} ```sql diff --git a/monitor/level_collector.go b/monitor/level_collector.go index 2b80a73b..4f932a22 100644 --- a/monitor/level_collector.go +++ b/monitor/level_collector.go @@ -219,15 +219,7 @@ func (c *lco) Run(stopChan, doneChan chan struct{}) error { break } blip.Debug("stopChan closed at %s s=%d interval=%d", startTime, s, interval) - c.changeMux.Lock() - defer c.changeMux.Unlock() - c.stopped = true // make ChangePlan do nothing - select { - case <-c.changePlanDoneChan: - c.changePlanCancelFunc() // stop --> changePlan goroutine - <-c.changePlanDoneChan // wait for changePlan goroutine - default: - } + c.stopChangePlan() c.engine.Stop() // stop all collectors and run their cleanup func return nil default: // no @@ -263,6 +255,23 @@ func (c *lco) Run(stopChan, doneChan chan struct{}) error { return nil } +// stopChangePlan prevents new plan changes, cancels the current plan change, +// and waits for its worker to stop using the engine and database provider. +func (c *lco) stopChangePlan() { + c.changeMux.Lock() + defer c.changeMux.Unlock() + + c.stopped = true + if c.changePlanCancelFunc == nil { + return + } + + c.changePlanCancelFunc() + <-c.changePlanDoneChan + c.changePlanCancelFunc = nil + c.changePlanDoneChan = nil +} + func (c *lco) collect(interval uint, levelName string, startTime time.Time) { status.Monitor(c.monitorId, status.LEVEL_COLLECT, "%s/%s: collecting", c.plan.Name, levelName) defer func() { @@ -393,7 +402,10 @@ func (c *lco) changePlan(ctx context.Context, doneChan chan struct{}, newState, errMsg := fmt.Sprintf("%s: error loading new plan %s: %s (retrying)", change, newPlanName, err) status.Monitor(c.monitorId, status.LEVEL_CHANGE_PLAN, "%s", errMsg) c.event.Sendf(event.CHANGE_PLAN_ERROR, "%s", errMsg) - time.Sleep(2 * time.Second) + if !waitForChangePlanRetry(ctx, 2*time.Second) { + blip.Debug("changePlan canceled while loading plan") + return + } } change = fmt.Sprintf("state:%s plan:%s -> state:%s plan:%s", oldState, oldPlanName, newState, newPlan.Name) @@ -468,13 +480,28 @@ func (c *lco) changePlan(ctx context.Context, doneChan chan struct{}, newState, return // changePlan goroutine has been cancelled } status.Monitor(c.monitorId, status.LEVEL_CHANGE_PLAN, "%s: error preparing new plan %s: %s (retrying)", change, newPlan.Name, err) - time.Sleep(retry.NextBackOff()) + if !waitForChangePlanRetry(ctx, retry.NextBackOff()) { + blip.Debug("changePlan canceled while waiting to retry plan preparation") + return + } } status.RemoveComponent(c.monitorId, status.LEVEL_CHANGE_PLAN) c.event.Sendf(event.CHANGE_PLAN_SUCCESS, "%s", change) } +func waitForChangePlanRetry(ctx context.Context, delay time.Duration) bool { + timer := time.NewTimer(delay) + defer timer.Stop() + + select { + case <-ctx.Done(): + return false + case <-timer.C: + return true + } +} + // Pause pauses metrics collection until ChangePlan is called. Run still runs, // but it doesn't collect when paused. The only way to resume after pausing is // to call ChangePlan again. diff --git a/monitor/provider_test.go b/monitor/provider_test.go index 0f6b49b7..4b140b45 100644 --- a/monitor/provider_test.go +++ b/monitor/provider_test.go @@ -9,9 +9,11 @@ import ( "fmt" "strings" "testing" + "time" "github.com/cashapp/blip" "github.com/cashapp/blip/metrics" + "github.com/cashapp/blip/plan" "github.com/cashapp/blip/test" "github.com/cashapp/blip/test/mock" ) @@ -131,6 +133,136 @@ func TestMonitorCloseDBClosesProviderOnce(t *testing.T) { } } +func TestMonitorStopWaitsForPlanPreparationBeforeClosingProvider(t *testing.T) { + TickerDuration(10*time.Millisecond, time.Second) + defer TickerDuration(time.Second, time.Second) + + const ( + domain = "test.provider-stop" + monitorID = "provider-stop" + planName = "provider-stop-plan" + levelName = "provider-stop-level" + stopCaller = "provider-stop-test" + ) + + prepareStarted := make(chan struct{}) + prepareStopped := make(chan struct{}) + releasePrepare := make(chan struct{}) + collector := mock.MetricsCollector{ + DomainFunc: func() string { return domain }, + PrepareFunc: func(ctx context.Context, _ blip.Plan) (func(), error) { + close(prepareStarted) + select { + case <-ctx.Done(): + close(prepareStopped) + return nil, ctx.Err() + case <-releasePrepare: + close(prepareStopped) + return nil, nil + } + }, + } + if err := metrics.Register(domain, mock.MetricFactory{ + MakeFunc: func(string, blip.CollectorFactoryArgs) (blip.Collector, error) { + return collector, nil + }, + }); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { metrics.Remove(domain) }) + + loader := plan.NewLoader(func(blip.ConfigPlans) ([]blip.Plan, error) { + return []blip.Plan{{ + Name: planName, + Levels: map[string]blip.Level{ + levelName: { + Name: levelName, + Freq: "1s", + Collect: map[string]blip.Domain{ + domain: {}, + }, + }, + }, + }}, nil + }) + if err := loader.LoadShared(blip.ConfigPlans{}, nil); err != nil { + t.Fatal(err) + } + config := blip.ConfigMonitor{MonitorId: monitorID} + if err := loader.LoadMonitor(config, nil); err != nil { + t.Fatal(err) + } + + _, primary, err := test.Connection(test.DefaultMySQLVersion) + if err != nil { + if test.Build { + t.Skip("MySQL test fixture not running") + } + t.Fatal(err) + } + defer primary.Close() + provider := &testDBProvider{ + primary: primary, + closeFunc: func() { + select { + case <-prepareStopped: + default: + close(releasePrepare) + t.Error("database provider closed before plan preparation stopped") + } + }, + } + lco := newLevelCollectorWithDBProvider(LevelCollectorArgs{ + Config: config, + DB: primary, + PlanLoader: loader, + }, provider) + monitor := NewMonitor(MonitorArgs{Config: config}) + monitor.runChan = make(chan struct{}) + monitor.db = primary + monitor.dbProvider = provider + + lcoDone := make(chan struct{}) + monitor.wg.Add(1) + go func() { + defer monitor.wg.Done() + _ = lco.Run(monitor.runChan, lcoDone) + }() + if err := lco.ChangePlan(blip.STATE_ACTIVE, planName); err != nil { + t.Fatal(err) + } + select { + case <-prepareStarted: + case <-time.After(time.Second): + t.Fatal("timeout waiting for plan preparation to start") + } + + stopDone := make(chan struct{}) + go func() { + monitor.stop(false, stopCaller) + close(stopDone) + }() + select { + case <-stopDone: + case <-time.After(2 * time.Second): + select { + case <-prepareStopped: + default: + close(releasePrepare) + } + t.Fatal("timeout waiting for monitor to stop") + } + + select { + case <-prepareStopped: + default: + t.Fatal("plan preparation was still running after monitor stop") + } + if provider.closeCalls != 1 { + t.Fatalf("provider Close calls = %d, expected 1", provider.closeCalls) + } +} + func TestNewEngineWithDBProviderRetainsProvider(t *testing.T) { provider := &testDBProvider{primary: &sql.DB{}} engine := newEngineWithDBProvider(blip.ConfigMonitor{MonitorId: "engine"}, provider.primary, provider) @@ -196,6 +328,7 @@ func TestEnginePassesProviderToCollectorFactory(t *testing.T) { type testDBProvider struct { primary *sql.DB closeCalls int + closeFunc func() } func (p *testDBProvider) Primary() *sql.DB { @@ -203,6 +336,9 @@ func (p *testDBProvider) Primary() *sql.DB { } func (p *testDBProvider) Close() error { + if p.closeFunc != nil { + p.closeFunc() + } p.closeCalls++ return nil } From 7f1ecca417735842459e6842e7be0f5c946e2aaa Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Thu, 6 Aug 2026 15:29:38 -0700 Subject: [PATCH 3/6] Require PostgreSQL exporter plans Avoid assigning the MySQL default exporter plan to PostgreSQL monitors. Require a named plan whenever exporter mode is enabled for PostgreSQL while preserving the public and MySQL defaulting behavior. Co-authored-by: Codex Ai-assisted: true --- config.go | 11 +++++++++-- config_postgres_test.go | 40 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 49 insertions(+), 2 deletions(-) diff --git a/config.go b/config.go index 901d9ca9..677251cc 100644 --- a/config.go +++ b/config.go @@ -492,6 +492,9 @@ func (c ConfigMonitor) Validate() error { if c.Plans.Table != "" { return fmt.Errorf("config.monitor.plans.table is only supported for database-type %q", DatabaseTypeMySQL) } + if c.Exporter.Mode != "" && c.Exporter.Plan == "" { + return fmt.Errorf("config.monitor.exporter.plan is required for database-type %q", DatabaseTypePostgres) + } return c.Postgres.Validate() default: return fmt.Errorf("config.monitor.database-type: invalid database type %q", c.DatabaseType) @@ -555,7 +558,7 @@ func (c *ConfigMonitor) ApplyDefaults(b Config) { c.Sinks = ConfigSinks{} } c.AWS.ApplyDefaults(b) - c.Exporter.ApplyDefaults(b) + c.Exporter.applyDefaults(b, databaseType == DatabaseTypeMySQL) c.HA.ApplyDefaults(b) // Heartbeat writes and plan state changes use MySQL-specific SQL. Do not // inherit their global defaults into PostgreSQL monitors; explicit monitor @@ -765,6 +768,10 @@ func (c ConfigExporter) Validate() error { } func (c *ConfigExporter) ApplyDefaults(b Config) { + c.applyDefaults(b, true) +} + +func (c *ConfigExporter) applyDefaults(b Config, useDefaultPlan bool) { if c.Mode == "" && b.Exporter.Mode != "" { c.Mode = b.Exporter.Mode } @@ -775,7 +782,7 @@ func (c *ConfigExporter) ApplyDefaults(b Config) { if c.Plan == "" && b.Exporter.Plan != "" { c.Plan = b.Exporter.Plan } - if c.Plan == "" { + if c.Plan == "" && useDefaultPlan { c.Plan = DEFAULT_EXPORTER_PLAN } if len(b.Exporter.Flags) > 0 { diff --git a/config_postgres_test.go b/config_postgres_test.go index e35f8afb..9126f52a 100644 --- a/config_postgres_test.go +++ b/config_postgres_test.go @@ -176,6 +176,46 @@ func TestConfigMonitorDatabaseTypeValidation(t *testing.T) { } } +func TestConfigMonitorPostgresExporterRequiresPlan(t *testing.T) { + defaults := blip.DefaultConfig() + defaults.Exporter.Mode = blip.EXPORTER_MODE_DUAL + monitor := blip.ConfigMonitor{DatabaseType: blip.DatabaseTypePostgres} + + monitor.ApplyDefaults(defaults) + if monitor.Exporter.Plan != "" { + t.Fatalf("PostgreSQL monitor inherited MySQL exporter plan %q", monitor.Exporter.Plan) + } + if err := monitor.Validate(); err == nil || !strings.Contains(err.Error(), "exporter.plan is required") { + t.Fatalf("validation error = %v, expected required exporter plan", err) + } +} + +func TestConfigMonitorPostgresExporterUsesConfiguredPlan(t *testing.T) { + defaults := blip.DefaultConfig() + defaults.Exporter.Mode = blip.EXPORTER_MODE_DUAL + defaults.Exporter.Plan = "postgres-exporter" + monitor := blip.ConfigMonitor{DatabaseType: blip.DatabaseTypePostgres} + + monitor.ApplyDefaults(defaults) + if monitor.Exporter.Plan != defaults.Exporter.Plan { + t.Fatalf("exporter plan = %q, expected %q", monitor.Exporter.Plan, defaults.Exporter.Plan) + } + if err := monitor.Validate(); err != nil { + t.Fatalf("PostgreSQL exporter config is invalid: %v", err) + } +} + +func TestConfigMonitorMySQLExporterRetainsDefaultPlan(t *testing.T) { + defaults := blip.DefaultConfig() + defaults.Exporter.Mode = blip.EXPORTER_MODE_DUAL + monitor := blip.ConfigMonitor{} + + monitor.ApplyDefaults(defaults) + if monitor.Exporter.Plan != blip.DEFAULT_EXPORTER_PLAN { + t.Fatalf("exporter plan = %q, expected %q", monitor.Exporter.Plan, blip.DEFAULT_EXPORTER_PLAN) + } +} + func TestConfigPlansRejectsPostgresTableMonitor(t *testing.T) { tests := []struct { name string From 768b27640d5bbc39a66cb963cd18712ca593ab1e Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Thu, 6 Aug 2026 15:42:57 -0700 Subject: [PATCH 4/6] Coordinate monitor subsystem teardown Roll back partial startup failures, bind subsystem stops to their startup generation, and retain exporter engines for cleanup before database providers close. Collector cleanup now also runs when a prepared collector is idle. Co-authored-by: Codex Ai-assisted: true --- monitor/engine.go | 10 +-- monitor/monitor.go | 68 ++++++++++++++------ monitor/mysqld_exporter.go | 10 +++ monitor/provider_test.go | 128 +++++++++++++++++++++++++++++++++++++ 4 files changed, 193 insertions(+), 23 deletions(-) diff --git a/monitor/engine.go b/monitor/engine.go index 9fe80973..644ec30f 100644 --- a/monitor/engine.go +++ b/monitor/engine.go @@ -440,16 +440,16 @@ func (e *Engine) stopCollectors() { /* -- CALLER MUST LOCK Engine -- */ for _, cl := range e.collectors { cl.Lock() - if !cl.running { - cl.Unlock() + if cl.running { + blip.Debug("%s: %s stopping", e.monitorId, cl.c.Domain()) + cl.cancel() + } else { blip.Debug("%s: %s not running", e.monitorId, cl.c.Domain()) - continue } - blip.Debug("%s: %s stopping", e.monitorId, cl.c.Domain()) - cl.cancel() if cl.cleanup != nil { blip.Debug("%s: %s cleanup", e.monitorId, cl.c.Domain()) cl.cleanup() + cl.cleanup = nil } cl.Unlock() } diff --git a/monitor/monitor.go b/monitor/monitor.go index 14b23d36..bfbf281e 100644 --- a/monitor/monitor.go +++ b/monitor/monitor.go @@ -63,6 +63,7 @@ type Monitor struct { dbProvider blip.DbProvider dsn string // redacted (no password) promAPI *prom.API + exporter *Exporter lco LevelCollector pch PlanChanger hbw *heartbeat.Writer @@ -231,7 +232,7 @@ func (m *Monitor) runLoop() { // then runLoop() calls startup again to restart monitoring. // // startup is called only by runLoop, which guards (serializes) and monitors it. -func (m *Monitor) startup() error { +func (m *Monitor) startup() (err error) { blip.Debug("%s: startup call", m.monitorId) defer blip.Debug("%s: startup return", m.monitorId) @@ -303,6 +304,11 @@ func (m *Monitor) startup() error { m.runMux.Lock() defer m.runMux.Unlock() + defer func() { + if err != nil { + m.stop(false, "startup error") + } + }() // ---------------------------------------------------------------------- // Heartbeat @@ -313,17 +319,19 @@ func (m *Monitor) startup() error { if m.cfg.Heartbeat.Freq != "" { status.Monitor(m.monitorId, status.MONITOR, "starting heartbeat") m.hbw = heartbeat.NewWriter(m.monitorId, m.db, m.cfg.Heartbeat) + hbw := m.hbw + runChan := m.runChan m.wg.Add(1) go func() { - defer m.stop(true, "heartbeat.Writer") // stop monitor subsystems - defer m.wg.Done() // notify stop() - defer func() { // catch panic in heartbeat.Writer + defer m.stopRun(runChan, "heartbeat.Writer") // stop monitor subsystems + defer m.wg.Done() // notify stop() + defer func() { // catch panic in heartbeat.Writer if r := recover(); r != nil { m.panic(r) } }() doneChan := make(chan struct{}) // Monitor uses wg - m.hbw.Write(m.runChan, doneChan) + hbw.Write(runChan, doneChan) }() } @@ -355,23 +363,26 @@ func (m *Monitor) startup() error { } // Run API to emulate an exporter, responding to GET /metrics + m.exporter = NewExporter(m.cfg.Exporter, promPlan, newEngineWithDBProvider(m.cfg, m.db, m.dbProvider)) m.promAPI = prom.NewAPI( m.cfg.Exporter, m.monitorId, - NewExporter(m.cfg.Exporter, promPlan, newEngineWithDBProvider(m.cfg, m.db, m.dbProvider)), + m.exporter, ) + promAPI := m.promAPI + runChan := m.runChan m.wg.Add(1) go func() { defer status.RemoveComponent(m.monitorId, "exporter") - defer m.stop(true, "prom.API") // stop monitor subsystems - defer m.wg.Done() // notify stop() - defer func() { // catch panic in exporter API + defer m.stopRun(runChan, "prom.API") // stop monitor subsystems + defer m.wg.Done() // notify stop() + defer func() { // catch panic in exporter API if r := recover(); r != nil { m.panic(r) } }() - err := m.promAPI.Run() + err := promAPI.Run() if err == nil { // shutdown blip.Debug("%s: prom api stopped", m.monitorId) return @@ -404,18 +415,20 @@ func (m *Monitor) startup() error { Sinks: m.sinks, TransformMetrics: m.transformMetric, }, m.dbProvider) + lco := m.lco + runChan := m.runChan m.wg.Add(1) go func() { - defer m.stop(true, "LCO") // stop monitor subsystems - defer m.wg.Done() // notify stop() - defer func() { // catch panic in LCO + defer m.stopRun(runChan, "LCO") // stop monitor subsystems + defer m.wg.Done() // notify stop() + defer func() { // catch panic in LCO if r := recover(); r != nil { m.panic(r) } }() doneChan := make(chan struct{}) // Monitor uses wg - m.lco.Run(m.runChan, doneChan) + lco.Run(runChan, doneChan) }() // ---------------------------------------------------------------------- @@ -433,18 +446,19 @@ func (m *Monitor) startup() error { LCO: m.lco, HA: m.ha, }) + pch := m.pch m.wg.Add(1) go func() { - defer m.stop(true, "PCH") // stop monitor subsystems - defer m.wg.Done() // notify stop() - defer func() { // catch panic in PCH + defer m.stopRun(runChan, "PCH") // stop monitor subsystems + defer m.wg.Done() // notify stop() + defer func() { // catch panic in PCH if r := recover(); r != nil { m.panic(r) } }() doneChan := make(chan struct{}) // Monitor uses wg - m.pch.Run(m.runChan, doneChan) // start LCO indirectly + pch.Run(runChan, doneChan) // start LCO indirectly }() } else { // When the PCH is not enabled, we must init the state and plan, @@ -528,11 +542,16 @@ func (m *Monitor) stop(lock bool, caller string) { // it's running an http.Server if m.promAPI != nil { m.promAPI.Stop() + m.promAPI = nil } // Wait for monitor subsystem goroutines to return status.Monitor(m.monitorId, status.MONITOR, "stopping goroutines") m.wg.Wait() + if m.exporter != nil { + m.exporter.Stop() + m.exporter = nil + } // A subsystem failure restarts the whole monitor, including its database // provider. Close the current provider only after every subsystem has @@ -540,6 +559,19 @@ func (m *Monitor) stop(lock bool, caller string) { m.closeDB() } +// stopRun stops a subsystem generation only if it is still current. A +// subsystem from an earlier startup can finish after runLoop has installed a +// new run channel; it must not stop that newer generation. +func (m *Monitor) stopRun(runChan chan struct{}, caller string) { + m.runMux.Lock() + defer m.runMux.Unlock() + if m.runChan != runChan { + blip.Debug("%s: stop called by %s for obsolete run (noop)", m.monitorId, caller) + return + } + m.stop(false, caller) +} + func (m *Monitor) setErr(err error, isPanic bool) { if err != nil { m.event.Error(event.MONITOR_ERROR, err.Error()) diff --git a/monitor/mysqld_exporter.go b/monitor/mysqld_exporter.go index 6827914a..e13df1be 100644 --- a/monitor/mysqld_exporter.go +++ b/monitor/mysqld_exporter.go @@ -107,3 +107,13 @@ func (e Exporter) Collect(ch chan<- prometheus.Metric) { tr.Translate(vals, ch) } } + +// Stop releases collector resources prepared by the exporter engine. The +// monitor calls this only after the HTTP API has stopped accepting and serving +// scrapes. +func (e *Exporter) Stop() { + e.Lock() + defer e.Unlock() + e.engine.Stop() + e.prepared = false +} diff --git a/monitor/provider_test.go b/monitor/provider_test.go index 4b140b45..84ff3965 100644 --- a/monitor/provider_test.go +++ b/monitor/provider_test.go @@ -8,10 +8,12 @@ import ( "errors" "fmt" "strings" + "sync" "testing" "time" "github.com/cashapp/blip" + "github.com/cashapp/blip/heartbeat" "github.com/cashapp/blip/metrics" "github.com/cashapp/blip/plan" "github.com/cashapp/blip/test" @@ -263,6 +265,132 @@ func TestMonitorStopWaitsForPlanPreparationBeforeClosingProvider(t *testing.T) { } } +func TestMonitorStartupErrorStopsPartialSubsystemsBeforeClosingProvider(t *testing.T) { + const ( + monitorID = "partial-startup" + exporterPlan = "invalid-exporter" + heartbeatDB = "blip_monitor_provider_test" + heartbeatTable = heartbeatDB + ".heartbeat" + ) + + _, primary, err := test.Connection(test.DefaultMySQLVersion) + if err != nil { + if test.Build { + t.Skip("MySQL test fixture not running") + } + t.Fatal(err) + } + defer primary.Close() + if _, err := primary.Exec("CREATE DATABASE IF NOT EXISTS " + heartbeatDB); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _, _ = primary.Exec("DROP DATABASE IF EXISTS " + heartbeatDB) }) + heartbeatDDL := strings.Replace(heartbeat.BLIP_TABLE_DDL, "heartbeat", heartbeatTable, 1) + if _, err := primary.Exec(heartbeatDDL); err != nil { + t.Fatal(err) + } + + loader := plan.NewLoader(func(blip.ConfigPlans) ([]blip.Plan, error) { + return []blip.Plan{{ + Name: exporterPlan, + Levels: map[string]blip.Level{ + "fast": {Name: "fast", Freq: "1s"}, + "slow": {Name: "slow", Freq: "5s"}, + }, + }}, nil + }) + if err := loader.LoadShared(blip.ConfigPlans{}, nil); err != nil { + t.Fatal(err) + } + + config := blip.ConfigMonitor{ + MonitorId: monitorID, + Heartbeat: blip.ConfigHeartbeat{ + Freq: "10ms", + Table: heartbeatTable, + }, + Exporter: blip.ConfigExporter{ + Mode: blip.EXPORTER_MODE_DUAL, + Plan: exporterPlan, + }, + } + provider := &testDBProvider{primary: primary} + monitor := NewMonitor(MonitorArgs{ + Config: config, + DbMaker: &testDBProviderFactory{provider: provider, dsn: "redacted"}, + PlanLoader: loader, + }) + monitor.runLoopChan = make(chan struct{}) + monitor.runChan = make(chan struct{}) + provider.closeFunc = func() { + select { + case <-monitor.runChan: + default: + t.Error("database provider closed before partial subsystems were stopped") + } + } + + err = monitor.startup() + if err == nil || !strings.Contains(err.Error(), "expected 1") { + t.Fatalf("startup error = %v, expected invalid exporter level count", err) + } + if provider.closeCalls != 1 { + t.Fatalf("provider Close calls = %d, expected 1", provider.closeCalls) + } + if monitor.db != nil || monitor.dbProvider != nil { + t.Fatalf("monitor retained database resources: db=%p provider=%T", monitor.db, monitor.dbProvider) + } +} + +func TestMonitorStopRunIgnoresObsoleteGeneration(t *testing.T) { + oldRun := make(chan struct{}) + currentRun := make(chan struct{}) + monitor := NewMonitor(MonitorArgs{Config: blip.ConfigMonitor{MonitorId: "run-generation"}}) + monitor.runChan = currentRun + + monitor.stopRun(oldRun, "obsolete") + + select { + case <-currentRun: + t.Fatal("obsolete subsystem stopped the current run") + default: + } +} + +func TestMonitorStopCleansExporterBeforeClosingProvider(t *testing.T) { + cleaned := false + provider := &testDBProvider{ + primary: &sql.DB{}, + closeFunc: func() { + if !cleaned { + t.Error("database provider closed before exporter cleanup") + } + }, + } + collector := mock.MetricsCollector{DomainFunc: func() string { return "test.exporter-cleanup" }} + engine := newEngineWithDBProvider(blip.ConfigMonitor{MonitorId: "exporter-cleanup"}, provider.primary, provider) + engine.collectors[collector.Domain()] = &clutch{ + c: collector, + cleanup: func() { cleaned = true }, + domain: collector.Domain(), + Mutex: &sync.Mutex{}, + } + monitor := NewMonitor(MonitorArgs{Config: blip.ConfigMonitor{MonitorId: "exporter-cleanup"}}) + monitor.runChan = make(chan struct{}) + monitor.db = provider.primary + monitor.dbProvider = provider + monitor.exporter = NewExporter(blip.ConfigExporter{}, blip.Plan{}, engine) + + monitor.stop(false, "exporter-cleanup-test") + + if !cleaned { + t.Fatal("exporter collector cleanup was not called") + } + if provider.closeCalls != 1 { + t.Fatalf("provider Close calls = %d, expected 1", provider.closeCalls) + } +} + func TestNewEngineWithDBProviderRetainsProvider(t *testing.T) { provider := &testDBProvider{primary: &sql.DB{}} engine := newEngineWithDBProvider(blip.ConfigMonitor{MonitorId: "engine"}, provider.primary, provider) From 8b6c341416733ce485d680f459d2931d0b50e5d6 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Thu, 6 Aug 2026 15:53:44 -0700 Subject: [PATCH 5/6] Join collectors before provider shutdown Track every engine collector goroutine, cancel active runs during teardown, and wait for foreground or ErrMore background work before invoking cleanup and releasing the database provider. Co-authored-by: Codex Ai-assisted: true --- monitor/engine.go | 29 +++++++++++++------ monitor/provider_test.go | 61 ++++++++++++++++++++++++++++++++++++++++ 2 files changed, 81 insertions(+), 9 deletions(-) diff --git a/monitor/engine.go b/monitor/engine.go index 644ec30f..a6ed281e 100644 --- a/monitor/engine.go +++ b/monitor/engine.go @@ -58,6 +58,7 @@ type Engine struct { collectAt map[string][]*clutch // keyed on level, sorted ascending by CMR checkAt map[string][]*clutch // keyed on level collectionChan chan collection + collectorWG sync.WaitGroup } func NewEngine(cfg blip.ConfigMonitor, db *sql.DB) *Engine { @@ -302,7 +303,11 @@ func (e *Engine) Collect(emrCtx context.Context, interval uint, levelName string for _, cl := range domains { select { case <-sem: - go cl.collect(*m, sem) + e.collectorWG.Add(1) + go func(cl *clutch, metrics blip.Metrics) { + defer e.collectorWG.Done() + cl.collect(metrics, sem) + }(cl, *m) running[cl.c.Domain()] = true case <-emrCtx.Done(): blip.Debug("EMR timeout starting collectors") @@ -418,13 +423,9 @@ SWEEP: return metrics, fmt.Errorf("%s: failed: zero metrics collected, %d errors", coId, errCount) } -// Stop the engine and cleanup any metrics associated with it. -// TODO: There is a possible race condition when this is called. Since -// Engine.Collect is called as a go-routine, we could have an invocation -// of the function block waiting for Engine.Stop to unlock -// after which Collect would run after cleanup has been called. -// This could result in a panic, though that should be caught and logged. -// Since the monitor is stopping anyway this isn't a huge issue. +// Stop the engine and cleanup any metrics associated with it. Collect calls +// are serialized by the engine lock; collectorWG also joins background +// collector runs that outlive the Collect call which started them. func (e *Engine) Stop() { blip.Debug("Engine.Stop called") e.Lock() @@ -442,10 +443,20 @@ func (e *Engine) stopCollectors() { cl.Lock() if cl.running { blip.Debug("%s: %s stopping", e.monitorId, cl.c.Domain()) - cl.cancel() + if cl.cancel != nil { + cl.cancel() + } } else { blip.Debug("%s: %s not running", e.monitorId, cl.c.Domain()) } + cl.Unlock() + } + + // Cleanup callbacks can release resources used by collector goroutines, so + // wait until every foreground or background run has observed cancellation. + e.collectorWG.Wait() + for _, cl := range e.collectors { + cl.Lock() if cl.cleanup != nil { blip.Debug("%s: %s cleanup", e.monitorId, cl.c.Domain()) cl.cleanup() diff --git a/monitor/provider_test.go b/monitor/provider_test.go index 84ff3965..1215bfd9 100644 --- a/monitor/provider_test.go +++ b/monitor/provider_test.go @@ -391,6 +391,67 @@ func TestMonitorStopCleansExporterBeforeClosingProvider(t *testing.T) { } } +func TestEngineStopWaitsForBackgroundCollector(t *testing.T) { + const ( + domain = "test.background-stop" + level = "fast" + ) + backgroundStarted := make(chan struct{}) + backgroundStopped := make(chan struct{}) + var collectCtx context.Context + collector := mock.MetricsCollector{ + DomainFunc: func() string { return domain }, + CollectFunc: func(ctx context.Context, _ string) ([]blip.MetricValue, error) { + if ctx != nil { + collectCtx = ctx + return nil, blip.ErrMore + } + close(backgroundStarted) + <-collectCtx.Done() + close(backgroundStopped) + return nil, nil + }, + } + engine := newEngineWithDBProvider(blip.ConfigMonitor{MonitorId: "background-stop"}, &sql.DB{}, nil) + cl := &clutch{ + c: collector, + cleanup: func() { + select { + case <-backgroundStopped: + default: + t.Error("collector cleanup ran before background collection stopped") + } + }, + domain: domain, + cmr: time.Second, + collectionChan: engine.collectionChan, + Mutex: &sync.Mutex{}, + } + engine.plan = blip.Plan{Name: "background", Levels: map[string]blip.Level{level: {Name: level, Freq: "1s"}}} + engine.collectors[domain] = cl + engine.collectAt[level] = []*clutch{cl} + engine.checkAt[level] = nil + + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if _, err := engine.Collect(ctx, 1, level, time.Now()); err != nil { + t.Fatal(err) + } + select { + case <-backgroundStarted: + case <-time.After(time.Second): + t.Fatal("timeout waiting for background collection to start") + } + + engine.Stop() + + select { + case <-backgroundStopped: + default: + t.Fatal("background collector was still running after Engine.Stop") + } +} + func TestNewEngineWithDBProviderRetainsProvider(t *testing.T) { provider := &testDBProvider{primary: &sql.DB{}} engine := newEngineWithDBProvider(blip.ConfigMonitor{MonitorId: "engine"}, provider.primary, provider) From 305591f00370ab0bec7ff4d3839b68286a8feb79 Mon Sep 17 00:00:00 2001 From: Ian Oberst Date: Fri, 7 Aug 2026 09:39:53 -0700 Subject: [PATCH 6/6] Generalize external database modules (#175) * Generalize external database modules Co-authored-by: Codex Ai-assisted: true * Interpolate typed database config values Preserve named and typed containers when Blip expands environment and monitor placeholders in opaque module configuration. This keeps programmatic config loaders consistent with YAML-loaded config. Co-authored-by: Codex Ai-assisted: true --------- Co-authored-by: Codex --- blip.go | 6 +- collector.go | 3 +- config.go | 144 +++--- config_database_test.go | 536 +++++++++++++++++++++++ config_postgres.go | 289 ------------ config_postgres_test.go | 410 ----------------- credentials/credentials.go | 10 +- credentials/credentials_test.go | 62 ++- database_module.go | 210 +++++++++ dbconn/factory.go | 2 +- docs/content/config/config-file.md | 23 +- docs/content/config/heartbeat.md | 4 +- docs/content/develop/collectors.md | 6 +- docs/content/develop/database-modules.md | 77 ++++ docs/content/plans/changing.md | 4 +- docs/content/plans/file.md | 6 +- docs/content/plans/table.md | 6 +- metrics/factory.go | 14 +- metrics/factory_test.go | 55 ++- monitor/engine.go | 6 +- monitor/level_collector.go | 10 +- monitor/monitor.go | 12 +- plan/loader.go | 13 +- plan/loader_test.go | 87 ++-- 24 files changed, 1098 insertions(+), 897 deletions(-) create mode 100644 config_database_test.go delete mode 100644 config_postgres.go delete mode 100644 config_postgres_test.go create mode 100644 database_module.go create mode 100644 docs/content/develop/database-modules.md diff --git a/blip.go b/blip.go index 62c0ef64..48d68866 100644 --- a/blip.go +++ b/blip.go @@ -34,11 +34,11 @@ const ( EVENT ) -// Metrics are metrics collected for one plan level, from one MySQL instance. +// Metrics are metrics collected for one plan level, from one monitor. type Metrics struct { Begin time.Time // when collection started End time.Time // when collection completed - MonitorId string // ID of monitor (MySQL) + MonitorId string // ID of monitor Plan string // plan name Level string // level name Interval uint // interval number @@ -99,7 +99,7 @@ type SinkFactoryArgs struct { Tags map[string]string // config.monitor.tags } -// DbCredentials are MySQL credentials parsed or loaded for a connection. +// DbCredentials are database credentials parsed or loaded for a connection. type DbCredentials struct { Username string Password string diff --git a/collector.go b/collector.go index ace30e19..55dd4080 100644 --- a/collector.go +++ b/collector.go @@ -151,7 +151,8 @@ type CollectorFactoryWithDBProvider interface { // // The domain argument allows one factory to serve domains with different // compatibility, such as MySQL collectors and database-neutral cloud metrics. -// Implementations must return at least one database type. +// Implementations must return at least one database type. Return +// DatabaseTypeAny by itself for a database-neutral domain. type CollectorFactoryDatabaseTypes interface { CollectorFactory DatabaseTypes(domain string) []DatabaseType diff --git a/config.go b/config.go index 677251cc..7df483ec 100644 --- a/config.go +++ b/config.go @@ -403,7 +403,8 @@ type ConfigMonitor struct { // Empty values retain Blip's historical MySQL behavior. DatabaseType DatabaseType `yaml:"database-type,omitempty"` - // ConfigMySQL: + // Shared connection identity and credentials. Socket and MyCnf are specific + // to Blip's built-in MySQL connection factory. Socket string `yaml:"socket,omitempty"` Hostname string `yaml:"hostname,omitempty"` MyCnf string `yaml:"mycnf,omitempty"` @@ -422,9 +423,12 @@ type ConfigMonitor struct { Heartbeat ConfigHeartbeat `yaml:"heartbeat,omitempty"` Plans ConfigPlans `yaml:"plans,omitempty"` Plan string `yaml:"plan,omitempty"` - Postgres ConfigPostgres `yaml:"postgres,omitempty"` - Sinks ConfigSinks `yaml:"sinks,omitempty"` - TLS ConfigTLS `yaml:"tls,omitempty"` + // DatabaseConfig is opaque configuration owned by the external module + // selected by DatabaseType. MySQL continues to use the historical monitor + // fields above and rejects this section. + DatabaseConfig ConfigDatabase `yaml:"database-config,omitempty"` + Sinks ConfigSinks `yaml:"sinks,omitempty"` + TLS ConfigTLS `yaml:"tls,omitempty"` Meta map[string]string `yaml:"meta,omitempty"` } @@ -434,13 +438,6 @@ const ( DEFAULT_MONITOR_TIMEOUT_CONNECT = "10s" ) -type DatabaseType string - -const ( - DatabaseTypeMySQL DatabaseType = "mysql" - DatabaseTypePostgres DatabaseType = "postgres" -) - func DefaultConfigMonitor() ConfigMonitor { return ConfigMonitor{ Username: DEFAULT_MONITOR_USERNAME, @@ -459,45 +456,55 @@ func DefaultConfigMonitor() ConfigMonitor { } // EffectiveDatabaseType returns the configured database type. An omitted -// value retains Blip's historical MySQL behavior. Environment interpolation is -// resolved here because monitor defaults are applied before the normal -// interpolation pass. +// value retains Blip's historical MySQL behavior. Direct environment-variable +// interpolation is resolved before defaults are selected; database type is a +// structural discriminator and does not support monitor-field interpolation. func (c ConfigMonitor) EffectiveDatabaseType() DatabaseType { - databaseType := DatabaseType(interpolateEnv(string(c.DatabaseType))) + databaseType := interpolateEnv(string(c.DatabaseType)) if databaseType == "" { return DatabaseTypeMySQL } - return databaseType + return DatabaseType(databaseType) } func (c ConfigMonitor) Validate() error { - switch c.EffectiveDatabaseType() { - case DatabaseTypeMySQL: - if c.Postgres.Set() { - return fmt.Errorf("config.monitor.postgres requires database-type %q", DatabaseTypePostgres) - } - case DatabaseTypePostgres: - if c.Socket != "" { - return fmt.Errorf("config.monitor.socket is only supported for database-type %q", DatabaseTypeMySQL) - } - if c.MyCnf != "" { - return fmt.Errorf("config.monitor.mycnf is only supported for database-type %q", DatabaseTypeMySQL) - } - if c.Heartbeat.set() { - return fmt.Errorf("config.monitor.heartbeat is only supported for database-type %q", DatabaseTypeMySQL) - } - if c.Plans.Change.set() { - return fmt.Errorf("config.monitor.plans.change is only supported for database-type %q", DatabaseTypeMySQL) - } - if c.Plans.Table != "" { - return fmt.Errorf("config.monitor.plans.table is only supported for database-type %q", DatabaseTypeMySQL) - } - if c.Exporter.Mode != "" && c.Exporter.Plan == "" { - return fmt.Errorf("config.monitor.exporter.plan is required for database-type %q", DatabaseTypePostgres) + databaseType := c.EffectiveDatabaseType() + if databaseType == DatabaseTypeMySQL { + if len(c.DatabaseConfig) > 0 { + return fmt.Errorf("config.monitor.database-config requires an external database type") } - return c.Postgres.Validate() - default: - return fmt.Errorf("config.monitor.database-type: invalid database type %q", c.DatabaseType) + return nil + } + if databaseType == DatabaseTypeAny { + return fmt.Errorf("config.monitor.database-type: %q is reserved for database-neutral collectors", databaseType) + } + if !ValidDatabaseType(databaseType) { + return fmt.Errorf("config.monitor.database-type: invalid database type %q", databaseType) + } + if c.Socket != "" { + return fmt.Errorf("config.monitor.socket is only supported for database-type %q", DatabaseTypeMySQL) + } + if c.MyCnf != "" { + return fmt.Errorf("config.monitor.mycnf is only supported for database-type %q", DatabaseTypeMySQL) + } + if c.Heartbeat.set() { + return fmt.Errorf("config.monitor.heartbeat is only supported for database-type %q", DatabaseTypeMySQL) + } + if c.Plans.Change.set() { + return fmt.Errorf("config.monitor.plans.change is only supported for database-type %q", DatabaseTypeMySQL) + } + if c.Plans.Table != "" { + return fmt.Errorf("config.monitor.plans.table is only supported for database-type %q", DatabaseTypeMySQL) + } + if c.Exporter.set() { + return fmt.Errorf("config.monitor.exporter is only supported for database-type %q", DatabaseTypeMySQL) + } + module, ok := registeredDatabaseModule(databaseType) + if !ok { + return fmt.Errorf("config.monitor.database-type: database module %q is not registered", databaseType) + } + if err := module.ValidateConfig(c); err != nil { + return fmt.Errorf("config.monitor.database-config for %q: %w", databaseType, err) } return nil } @@ -509,6 +516,9 @@ func (c ConfigMonitor) Redacted() ConfigMonitor { func (c ConfigMonitor) redacted(seen map[*ConfigMonitor]*ConfigMonitor) ConfigMonitor { c.Password = redactPassword(c.Password) + // External module configuration is opaque to Blip, so redact the entire + // section rather than guessing which module-owned fields might be secrets. + c.DatabaseConfig = nil c.Sinks = c.Sinks.redacted() c.Plans = c.Plans.redacted(seen) return c @@ -558,20 +568,16 @@ func (c *ConfigMonitor) ApplyDefaults(b Config) { c.Sinks = ConfigSinks{} } c.AWS.ApplyDefaults(b) - c.Exporter.applyDefaults(b, databaseType == DatabaseTypeMySQL) c.HA.ApplyDefaults(b) - // Heartbeat writes and plan state changes use MySQL-specific SQL. Do not - // inherit their global defaults into PostgreSQL monitors; explicit monitor - // values remain intact so Validate can report the unsupported configuration. + // Exporter emulation, heartbeat writes, and plan state changes are + // MySQL-specific. Do not inherit their global defaults into external database + // monitors; explicit monitor values remain intact so Validate can report the + // unsupported configuration. if databaseType == DatabaseTypeMySQL { + c.Exporter.ApplyDefaults(b) c.Heartbeat.ApplyDefaults(b) } c.Plans.applyDefaults(b, databaseType == DatabaseTypeMySQL) - if databaseType == DatabaseTypePostgres { - postgresDefaults := DefaultConfigPostgres() - postgresDefaults.ConnectTimeout = c.TimeoutConnect - c.Postgres.ApplyDefaults(postgresDefaults) - } c.Sinks.ApplyDefaults(b) c.TLS.ApplyDefaults(b) } @@ -598,7 +604,7 @@ func (c *ConfigMonitor) InterpolateEnvVars() { c.Heartbeat.InterpolateEnvVars() c.Plans.InterpolateEnvVars() c.Plan = interpolateEnv(c.Plan) - c.Postgres.InterpolateEnvVars() + c.DatabaseConfig.interpolate(interpolateEnv) c.Sinks.InterpolateEnvVars() c.TLS.InterpolateEnvVars() } @@ -624,7 +630,7 @@ func (c *ConfigMonitor) InterpolateMonitor() { c.Heartbeat.InterpolateMonitor(c) c.Plans.InterpolateMonitor(c) c.Plan = c.interpolateMon(c.Plan) - c.Postgres.InterpolateMonitor(c) + c.DatabaseConfig.interpolate(c.interpolateMon) c.Sinks.InterpolateMonitor(c) c.TLS.InterpolateMonitor(c) } @@ -676,20 +682,6 @@ func (c *ConfigMonitor) fieldValue(f string) string { return c.PasswordFile case "timeout-connect": return c.TimeoutConnect - case "postgres.database": - return c.Postgres.Database - case "postgres.application-name": - return c.Postgres.ApplicationName - case "postgres.ssl-mode": - return c.Postgres.SSLMode - case "postgres.connect-timeout": - return c.Postgres.ConnectTimeout - case "postgres.statement-timeout": - return c.Postgres.StatementTimeout - case "postgres.lock-timeout": - return c.Postgres.LockTimeout - case "postgres.dial-address": - return c.Postgres.DialAddress default: return "" } @@ -753,6 +745,10 @@ type ConfigExporter struct { Plan string `yaml:"plan,omitempty"` } +func (c ConfigExporter) set() bool { + return c.Mode != "" || c.Plan != "" || len(c.Flags) > 0 +} + func DefaultConfigExporter() ConfigExporter { return ConfigExporter{} } @@ -768,10 +764,6 @@ func (c ConfigExporter) Validate() error { } func (c *ConfigExporter) ApplyDefaults(b Config) { - c.applyDefaults(b, true) -} - -func (c *ConfigExporter) applyDefaults(b Config, useDefaultPlan bool) { if c.Mode == "" && b.Exporter.Mode != "" { c.Mode = b.Exporter.Mode } @@ -782,7 +774,7 @@ func (c *ConfigExporter) applyDefaults(b Config, useDefaultPlan bool) { if c.Plan == "" && b.Exporter.Plan != "" { c.Plan = b.Exporter.Plan } - if c.Plan == "" && useDefaultPlan { + if c.Plan == "" { c.Plan = DEFAULT_EXPORTER_PLAN } if len(b.Exporter.Flags) > 0 { @@ -992,9 +984,15 @@ func DefaultConfigPlans() ConfigPlans { } func (c ConfigPlans) Validate() error { - if c.Table != "" && c.Monitor != nil && c.Monitor.EffectiveDatabaseType() != DatabaseTypeMySQL { + if c.Table == "" || c.Monitor == nil { + return nil + } + if c.Monitor.EffectiveDatabaseType() != DatabaseTypeMySQL { return fmt.Errorf("config.plans.monitor.database-type: config.plans.table is only supported for database-type %q", DatabaseTypeMySQL) } + if err := c.Monitor.Validate(); err != nil { + return fmt.Errorf("config.plans.monitor: %w", err) + } return nil } diff --git a/config_database_test.go b/config_database_test.go new file mode 100644 index 00000000..947ec657 --- /dev/null +++ b/config_database_test.go @@ -0,0 +1,536 @@ +// Copyright 2026 Block, Inc. + +package blip_test + +import ( + "errors" + "os" + "path/filepath" + "strings" + "testing" + + "github.com/cashapp/blip" +) + +type testDatabaseModule struct { + databaseType blip.DatabaseType + validate func(blip.ConfigMonitor) error +} + +func (m testDatabaseModule) DatabaseType() blip.DatabaseType { + return m.databaseType +} + +func (m testDatabaseModule) ValidateConfig(cfg blip.ConfigMonitor) error { + if m.validate != nil { + return m.validate(cfg) + } + return nil +} + +func registerTestDatabaseModule(t *testing.T, databaseType blip.DatabaseType, validate func(blip.ConfigMonitor) error) { + t.Helper() + if err := blip.RegisterDatabaseModule(testDatabaseModule{databaseType: databaseType, validate: validate}); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { blip.RemoveDatabaseModule(databaseType) }) +} + +func TestConfigMonitorDatabaseTypeDefaultsToMySQLWithoutMutation(t *testing.T) { + monitor := blip.ConfigMonitor{} + monitor.ApplyDefaults(blip.DefaultConfig()) + + if monitor.DatabaseType != "" { + t.Fatalf("omitted database type mutated to %q", monitor.DatabaseType) + } + if monitor.EffectiveDatabaseType() != blip.DatabaseTypeMySQL { + t.Fatalf("effective database type = %q, expected mysql", monitor.EffectiveDatabaseType()) + } + if len(monitor.DatabaseConfig) != 0 { + t.Fatalf("external database config added to MySQL monitor: %#v", monitor.DatabaseConfig) + } +} + +func TestConfigMonitorExternalModuleValidationAndInterpolation(t *testing.T) { + const databaseType blip.DatabaseType = "test-interpolation" + t.Setenv("BLIP_TEST_DATABASE_TYPE", string(databaseType)) + t.Setenv("BLIP_TEST_DATABASE", "metrics_database") + t.Setenv("BLIP_TEST_INCLUDE", "app_*") + + type moduleConfig struct { + Database string `yaml:"database"` + Include []string `yaml:"include"` + ApplicationName string `yaml:"application-name"` + Nested struct { + Address string `yaml:"address"` + } `yaml:"nested"` + } + + var validated moduleConfig + registerTestDatabaseModule(t, databaseType, func(cfg blip.ConfigMonitor) error { + if cfg.TimeoutConnect != "7s" { + return errors.New("monitor defaults were not available to module validation") + } + return blip.DecodeDatabaseConfig(cfg.DatabaseConfig, &validated) + }) + + monitor := blip.ConfigMonitor{ + MonitorId: "external-monitor", + DatabaseType: "${BLIP_TEST_DATABASE_TYPE}", + TimeoutConnect: "7s", + DatabaseConfig: blip.ConfigDatabase{ + "database": "${BLIP_TEST_DATABASE}", + "include": []interface{}{"${BLIP_TEST_INCLUDE}"}, + "application-name": "%{monitor.id}", + "nested": map[interface{}]interface{}{ + "address": "%{monitor.hostname}", + }, + }, + Hostname: "database.example:1234", + } + monitor.ApplyDefaults(blip.DefaultConfig()) + monitor.InterpolateEnvVars() + monitor.InterpolateMonitor() + + if err := monitor.Validate(); err != nil { + t.Fatal(err) + } + if monitor.DatabaseType != databaseType { + t.Fatalf("database type = %q, expected %q", monitor.DatabaseType, databaseType) + } + if validated.Database != "metrics_database" { + t.Fatalf("database = %q", validated.Database) + } + if len(validated.Include) != 1 || validated.Include[0] != "app_*" { + t.Fatalf("include = %#v", validated.Include) + } + if validated.ApplicationName != monitor.MonitorId { + t.Fatalf("application name = %q", validated.ApplicationName) + } + if validated.Nested.Address != monitor.Hostname { + t.Fatalf("nested address = %q", validated.Nested.Address) + } +} + +func TestConfigMonitorExternalModuleInterpolationPreservesTypedContainers(t *testing.T) { + const databaseType blip.DatabaseType = "test-typed-interpolation" + t.Setenv("BLIP_TEST_TYPED_VALUE", "from-environment") + + type namedString string + type namedMap map[string]namedString + type namedSlice []namedString + type namedArray [2]namedString + type namedPointer *namedString + type namedConfig blip.ConfigDatabase + type nestedStruct struct { + Monitor namedString `yaml:"monitor"` + } + type namedStruct struct { + Environment namedString `yaml:"environment"` + Nested *nestedStruct `yaml:"nested"` + private namedString + } + type decodedStruct struct { + Environment string `yaml:"environment"` + Nested struct { + Monitor string `yaml:"monitor"` + } `yaml:"nested"` + } + + type moduleConfig struct { + NestedConfig map[string]string `yaml:"nested-config"` + TypedMap map[string]string `yaml:"typed-map"` + NamedMap map[string]string `yaml:"named-map"` + NamedSlice []string `yaml:"named-slice"` + NamedArray []string `yaml:"named-array"` + NamedStruct decodedStruct `yaml:"named-struct"` + NamedStructPointer decodedStruct `yaml:"named-struct-pointer"` + Pointer string `yaml:"pointer"` + NilMap map[string]string `yaml:"nil-map"` + NilSlice []string `yaml:"nil-slice"` + NilPointer *string `yaml:"nil-pointer"` + } + + var validated moduleConfig + registerTestDatabaseModule(t, databaseType, func(cfg blip.ConfigMonitor) error { + return blip.DecodeDatabaseConfig(cfg.DatabaseConfig, &validated) + }) + + pointer := namedString("%{monitor.id}") + structValue := namedStruct{ + Environment: "${BLIP_TEST_TYPED_VALUE}", + Nested: &nestedStruct{Monitor: "%{monitor.id}"}, + private: "${BLIP_TEST_TYPED_VALUE}", + } + monitor := blip.ConfigMonitor{ + MonitorId: "typed-monitor", + DatabaseType: databaseType, + DatabaseConfig: blip.ConfigDatabase{ + "nested-config": namedConfig{ + "environment": namedString("${BLIP_TEST_TYPED_VALUE}"), + }, + "typed-map": map[string]string{ + "monitor": "%{monitor.id}", + }, + "named-map": namedMap{ + "environment": "${BLIP_TEST_TYPED_VALUE}", + }, + "named-slice": namedSlice{"${BLIP_TEST_TYPED_VALUE}", "%{monitor.id}"}, + "named-array": namedArray{"${BLIP_TEST_TYPED_VALUE}", "%{monitor.id}"}, + "named-struct": structValue, + "named-struct-pointer": &structValue, + "pointer": namedPointer(&pointer), + "nil-map": map[string]string(nil), + "nil-slice": namedSlice(nil), + "nil-pointer": (*namedString)(nil), + }, + } + monitor.ApplyDefaults(blip.DefaultConfig()) + monitor.InterpolateEnvVars() + monitor.InterpolateMonitor() + + if err := monitor.Validate(); err != nil { + t.Fatal(err) + } + + nestedConfig, ok := monitor.DatabaseConfig["nested-config"].(namedConfig) + if !ok { + t.Fatalf("nested config type = %T, expected namedConfig", monitor.DatabaseConfig["nested-config"]) + } + if got := nestedConfig["environment"]; got != namedString("from-environment") { + t.Fatalf("nested environment = %q", got) + } + if _, ok := nestedConfig["environment"].(namedString); !ok { + t.Fatalf("nested environment type = %T, expected namedString", nestedConfig["environment"]) + } + + typedMap, ok := monitor.DatabaseConfig["typed-map"].(map[string]string) + if !ok || typedMap["monitor"] != monitor.MonitorId { + t.Fatalf("typed map = %#v (%T)", monitor.DatabaseConfig["typed-map"], monitor.DatabaseConfig["typed-map"]) + } + namedMapValue, ok := monitor.DatabaseConfig["named-map"].(namedMap) + if !ok || namedMapValue["environment"] != "from-environment" { + t.Fatalf("named map = %#v (%T)", monitor.DatabaseConfig["named-map"], monitor.DatabaseConfig["named-map"]) + } + namedSliceValue, ok := monitor.DatabaseConfig["named-slice"].(namedSlice) + if !ok || len(namedSliceValue) != 2 || namedSliceValue[0] != "from-environment" || namedSliceValue[1] != namedString(monitor.MonitorId) { + t.Fatalf("named slice = %#v (%T)", monitor.DatabaseConfig["named-slice"], monitor.DatabaseConfig["named-slice"]) + } + namedArrayValue, ok := monitor.DatabaseConfig["named-array"].(namedArray) + if !ok || namedArrayValue[0] != "from-environment" || namedArrayValue[1] != namedString(monitor.MonitorId) { + t.Fatalf("named array = %#v (%T)", monitor.DatabaseConfig["named-array"], monitor.DatabaseConfig["named-array"]) + } + namedStructValue, ok := monitor.DatabaseConfig["named-struct"].(namedStruct) + if !ok || namedStructValue.Environment != "from-environment" || namedStructValue.Nested == nil || namedStructValue.Nested.Monitor != namedString(monitor.MonitorId) { + t.Fatalf("named struct = %#v (%T)", monitor.DatabaseConfig["named-struct"], monitor.DatabaseConfig["named-struct"]) + } + if namedStructValue.private != "${BLIP_TEST_TYPED_VALUE}" { + t.Fatalf("unexported field was interpolated: %q", namedStructValue.private) + } + namedStructPointer, ok := monitor.DatabaseConfig["named-struct-pointer"].(*namedStruct) + if !ok || namedStructPointer == nil || namedStructPointer.Environment != "from-environment" || namedStructPointer.Nested == nil || namedStructPointer.Nested.Monitor != namedString(monitor.MonitorId) { + t.Fatalf("named struct pointer = %#v (%T)", monitor.DatabaseConfig["named-struct-pointer"], monitor.DatabaseConfig["named-struct-pointer"]) + } + if namedStructPointer.private != "${BLIP_TEST_TYPED_VALUE}" { + t.Fatalf("unexported pointer field was interpolated: %q", namedStructPointer.private) + } + pointerValue, ok := monitor.DatabaseConfig["pointer"].(namedPointer) + if !ok || pointerValue == nil || *pointerValue != namedString(monitor.MonitorId) { + t.Fatalf("pointer = %#v (%T)", monitor.DatabaseConfig["pointer"], monitor.DatabaseConfig["pointer"]) + } + if value, ok := monitor.DatabaseConfig["nil-map"].(map[string]string); !ok || value != nil { + t.Fatalf("nil map = %#v (%T)", monitor.DatabaseConfig["nil-map"], monitor.DatabaseConfig["nil-map"]) + } + if value, ok := monitor.DatabaseConfig["nil-slice"].(namedSlice); !ok || value != nil { + t.Fatalf("nil slice = %#v (%T)", monitor.DatabaseConfig["nil-slice"], monitor.DatabaseConfig["nil-slice"]) + } + if value, ok := monitor.DatabaseConfig["nil-pointer"].(*namedString); !ok || value != nil { + t.Fatalf("nil pointer = %#v (%T)", monitor.DatabaseConfig["nil-pointer"], monitor.DatabaseConfig["nil-pointer"]) + } + + if validated.NestedConfig["environment"] != "from-environment" || + validated.TypedMap["monitor"] != monitor.MonitorId || + validated.NamedMap["environment"] != "from-environment" || + len(validated.NamedSlice) != 2 || validated.NamedSlice[1] != monitor.MonitorId || + len(validated.NamedArray) != 2 || validated.NamedArray[1] != monitor.MonitorId || + validated.NamedStruct.Environment != "from-environment" || validated.NamedStruct.Nested.Monitor != monitor.MonitorId || + validated.NamedStructPointer.Environment != "from-environment" || validated.NamedStructPointer.Nested.Monitor != monitor.MonitorId || + validated.Pointer != monitor.MonitorId || + len(validated.NilMap) != 0 || len(validated.NilSlice) != 0 || validated.NilPointer != nil { + t.Fatalf("decoded module config = %#v", validated) + } +} + +func TestRegisterDatabaseModuleValidation(t *testing.T) { + tests := []struct { + name string + databaseType blip.DatabaseType + wantError string + }{ + {name: "empty", wantError: "invalid database type"}, + {name: "whitespace", databaseType: " test ", wantError: "invalid database type"}, + {name: "uppercase", databaseType: "Test", wantError: "invalid database type"}, + {name: "MySQL", databaseType: blip.DatabaseTypeMySQL, wantError: "built into Blip"}, + {name: "neutral marker", databaseType: blip.DatabaseTypeAny, wantError: "reserved"}, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := blip.RegisterDatabaseModule(testDatabaseModule{databaseType: tt.databaseType}) + if err == nil || !strings.Contains(err.Error(), tt.wantError) { + t.Fatalf("RegisterDatabaseModule error = %v", err) + } + }) + } + + const duplicate blip.DatabaseType = "test-duplicate" + registerTestDatabaseModule(t, duplicate, nil) + if err := blip.RegisterDatabaseModule(testDatabaseModule{databaseType: duplicate}); err == nil || !strings.Contains(err.Error(), "already registered") { + t.Fatalf("duplicate registration error = %v", err) + } + if err := blip.RegisterDatabaseModule(nil); err == nil || !strings.Contains(err.Error(), "nil") { + t.Fatalf("nil registration error = %v", err) + } + var nilModule *testDatabaseModule + if err := blip.RegisterDatabaseModule(nilModule); err == nil || !strings.Contains(err.Error(), "nil") { + t.Fatalf("typed nil registration error = %v", err) + } +} + +func TestConfigMonitorExternalDatabaseValidation(t *testing.T) { + const databaseType blip.DatabaseType = "test-guards" + registerTestDatabaseModule(t, databaseType, func(blip.ConfigMonitor) error { + return errors.New("module validation failed") + }) + + tests := []struct { + name string + monitor blip.ConfigMonitor + wantError string + }{ + { + name: "unregistered database type", + monitor: blip.ConfigMonitor{DatabaseType: "unregistered"}, + wantError: "is not registered", + }, + { + name: "database config on implicit MySQL monitor", + monitor: blip.ConfigMonitor{ + DatabaseConfig: blip.ConfigDatabase{"database": "example"}, + }, + wantError: "requires an external database type", + }, + { + name: "neutral marker as monitor type", + monitor: blip.ConfigMonitor{DatabaseType: blip.DatabaseTypeAny}, + wantError: "reserved", + }, + { + name: "monitor interpolation in database type", + monitor: blip.ConfigMonitor{DatabaseType: "%{monitor.meta.engine}", Meta: map[string]string{"engine": string(databaseType)}}, + wantError: "invalid database type", + }, + { + name: "my.cnf on external monitor", + monitor: blip.ConfigMonitor{DatabaseType: databaseType, MyCnf: "/etc/blip/my.cnf"}, + wantError: "mycnf is only supported", + }, + { + name: "socket on external monitor", + monitor: blip.ConfigMonitor{DatabaseType: databaseType, Socket: "/tmp/database.sock"}, + wantError: "socket is only supported", + }, + { + name: "heartbeat on external monitor", + monitor: blip.ConfigMonitor{DatabaseType: databaseType, Heartbeat: blip.ConfigHeartbeat{Freq: "1s"}}, + wantError: "heartbeat is only supported", + }, + { + name: "plan changing on external monitor", + monitor: blip.ConfigMonitor{DatabaseType: databaseType, Plans: blip.ConfigPlans{Change: blip.ConfigPlanChange{ + Active: blip.ConfigStatePlan{Plan: "active"}, + }}}, + wantError: "plans.change is only supported", + }, + { + name: "plan table on external monitor", + monitor: blip.ConfigMonitor{DatabaseType: databaseType, Plans: blip.ConfigPlans{Table: "blip.plans"}}, + wantError: "plans.table is only supported", + }, + { + name: "exporter on external monitor", + monitor: blip.ConfigMonitor{DatabaseType: databaseType, Exporter: blip.ConfigExporter{Mode: blip.EXPORTER_MODE_DUAL, Plan: "external"}}, + wantError: "exporter is only supported", + }, + { + name: "module validation", + monitor: blip.ConfigMonitor{DatabaseType: databaseType}, + wantError: "module validation failed", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + monitor := tt.monitor + monitor.InterpolateEnvVars() + monitor.InterpolateMonitor() + err := monitor.Validate() + if err == nil || !strings.Contains(err.Error(), tt.wantError) { + t.Fatalf("got error %v, expected it to contain %q", err, tt.wantError) + } + }) + } +} + +func TestConfigMonitorDatabaseSpecificDefaults(t *testing.T) { + const databaseType blip.DatabaseType = "test-defaults" + registerTestDatabaseModule(t, databaseType, nil) + + defaults := blip.DefaultConfig() + defaults.MySQL.Hostname = "mysql.example:3306" + defaults.Exporter = blip.ConfigExporter{Mode: blip.EXPORTER_MODE_DUAL} + defaults.Heartbeat = blip.ConfigHeartbeat{Freq: "1s", Table: "blip.heartbeat"} + defaults.Plans = blip.ConfigPlans{ + Files: []string{"shared.yaml"}, + Change: blip.ConfigPlanChange{Active: blip.ConfigStatePlan{Plan: "active"}}, + } + + external := blip.ConfigMonitor{DatabaseType: databaseType} + external.ApplyDefaults(defaults) + if external.Hostname != "" { + t.Fatalf("external monitor inherited MySQL hostname %q", external.Hostname) + } + if external.Exporter.Mode != "" || external.Exporter.Plan != "" || len(external.Exporter.Flags) != 0 { + t.Fatalf("external monitor inherited exporter defaults: %+v", external.Exporter) + } + if external.Heartbeat != (blip.ConfigHeartbeat{}) { + t.Fatalf("external monitor inherited heartbeat defaults: %+v", external.Heartbeat) + } + if external.Plans.Change.Enabled() { + t.Fatalf("external monitor inherited plan-changing defaults: %+v", external.Plans.Change) + } + if got := external.Plans.Files; len(got) != 1 || got[0] != "shared.yaml" { + t.Fatalf("external monitor plan files = %#v", got) + } + if err := external.Validate(); err != nil { + t.Fatalf("external monitor with shared defaults is invalid: %v", err) + } + + mysql := blip.ConfigMonitor{} + mysql.ApplyDefaults(defaults) + if mysql.Hostname != defaults.MySQL.Hostname { + t.Fatalf("MySQL hostname = %q", mysql.Hostname) + } + if mysql.Exporter.Plan != blip.DEFAULT_EXPORTER_PLAN { + t.Fatalf("MySQL exporter plan = %q", mysql.Exporter.Plan) + } + if mysql.Heartbeat == (blip.ConfigHeartbeat{}) || !mysql.Plans.Change.Enabled() { + t.Fatal("MySQL monitor did not inherit MySQL-specific defaults") + } +} + +func TestConfigMonitorEnvironmentTypeSelectsDefaultsBeforeInterpolation(t *testing.T) { + const databaseType blip.DatabaseType = "test-env-defaults" + registerTestDatabaseModule(t, databaseType, nil) + t.Setenv("BLIP_TEST_ENV_DATABASE_TYPE", string(databaseType)) + + defaults := blip.DefaultConfig() + defaults.MySQL.Hostname = "mysql.example:3306" + monitor := blip.ConfigMonitor{DatabaseType: "${BLIP_TEST_ENV_DATABASE_TYPE}"} + monitor.ApplyDefaults(defaults) + if monitor.Hostname != "" { + t.Fatalf("external monitor inherited MySQL hostname %q", monitor.Hostname) + } + monitor.InterpolateEnvVars() + monitor.InterpolateMonitor() + if monitor.DatabaseType != databaseType { + t.Fatalf("database type = %q", monitor.DatabaseType) + } + if err := monitor.Validate(); err != nil { + t.Fatal(err) + } +} + +func TestConfigPlansRejectsExternalTableMonitor(t *testing.T) { + plans := blip.ConfigPlans{ + Table: "blip.plans", + Monitor: &blip.ConfigMonitor{DatabaseType: "external-without-registration"}, + } + err := plans.Validate() + if err == nil || !strings.Contains(err.Error(), "config.plans.table is only supported") { + t.Fatalf("ConfigPlans.Validate error = %v", err) + } +} + +func TestConfigMonitorRedactsOpaqueDatabaseConfig(t *testing.T) { + monitor := blip.ConfigMonitor{ + DatabaseType: "external", + DatabaseConfig: blip.ConfigDatabase{ + "password": "do-not-log", + "nested": map[string]interface{}{"token": "also-do-not-log"}, + }, + } + redacted := monitor.Redacted() + if redacted.DatabaseConfig != nil { + t.Fatalf("redacted database config = %#v", redacted.DatabaseConfig) + } + if monitor.DatabaseConfig["password"] != "do-not-log" { + t.Fatal("redaction modified live database config") + } +} + +func TestDecodeDatabaseConfigIsStrict(t *testing.T) { + type moduleConfig struct { + Database string `yaml:"database"` + } + + var decoded moduleConfig + if err := blip.DecodeDatabaseConfig(blip.ConfigDatabase{"database": "metrics"}, &decoded); err != nil { + t.Fatal(err) + } + if decoded.Database != "metrics" { + t.Fatalf("decoded database = %q", decoded.Database) + } + if err := blip.DecodeDatabaseConfig(blip.ConfigDatabase{"unknown": true}, &decoded); err == nil || !strings.Contains(err.Error(), "field unknown not found") { + t.Fatalf("strict decode error = %v", err) + } + if err := blip.DecodeDatabaseConfig(nil, nil); err == nil { + t.Fatal("nil decode output succeeded") + } + var nilOutput *moduleConfig + if err := blip.DecodeDatabaseConfig(nil, nilOutput); err == nil { + t.Fatal("typed nil decode output succeeded") + } +} + +func TestLoadConfigAcceptsOpaqueModuleShapeForStrictModuleDecode(t *testing.T) { + const databaseType blip.DatabaseType = "test-yaml" + registerTestDatabaseModule(t, databaseType, func(cfg blip.ConfigMonitor) error { + var decoded struct { + Database string `yaml:"database"` + } + return blip.DecodeDatabaseConfig(cfg.DatabaseConfig, &decoded) + }) + + configFile := filepath.Join(t.TempDir(), "blip.yaml") + contents := `monitors: + - id: external + database-type: test-yaml + database-config: + database: metrics + unknown: rejected-by-module +` + if err := os.WriteFile(configFile, []byte(contents), 0o600); err != nil { + t.Fatal(err) + } + cfg, err := blip.LoadConfig(configFile, blip.DefaultConfig(), true) + if err != nil { + t.Fatalf("Blip strict YAML decode rejected opaque module config: %v", err) + } + monitor := cfg.Monitors[0] + monitor.ApplyDefaults(cfg) + monitor.InterpolateEnvVars() + monitor.InterpolateMonitor() + if err := monitor.Validate(); err == nil || !strings.Contains(err.Error(), "field unknown not found") { + t.Fatalf("module strict validation error = %v", err) + } +} diff --git a/config_postgres.go b/config_postgres.go deleted file mode 100644 index 7c3d9f4f..00000000 --- a/config_postgres.go +++ /dev/null @@ -1,289 +0,0 @@ -// Copyright 2026 Block, Inc. - -package blip - -import ( - "fmt" - "strings" - "time" -) - -const ( - DEFAULT_POSTGRES_DATABASE = "postgres" - DEFAULT_POSTGRES_APPLICATION_NAME = "pgblip" - DEFAULT_POSTGRES_MAX_OPEN_CONNECTIONS = 4 - DEFAULT_POSTGRES_MAX_IDLE_CONNECTIONS = 2 - DEFAULT_POSTGRES_MAX_CONNECTION_IDLE_TIME = "30s" - DEFAULT_POSTGRES_MAX_CONNECTION_LIFETIME = "0" - DEFAULT_POSTGRES_DATABASE_REFRESH = "5m" - DEFAULT_POSTGRES_DATABASE_MAX_CONCURRENCY = 4 -) - -// ConfigPostgres configures the PostgreSQL database/sql pool owned by one -// monitor. Credentials and TLS certificate files remain in the existing -// monitor-level fields so all Blip credential sources can be shared by -// database-specific connection factories. -type ConfigPostgres struct { - Database string `yaml:"database,omitempty"` - Databases ConfigPostgresDatabases `yaml:"databases,omitempty"` - ApplicationName string `yaml:"application-name,omitempty"` - SSLMode string `yaml:"ssl-mode,omitempty"` - MaxOpenConnections *int `yaml:"max-open-connections,omitempty"` - MaxIdleConnections *int `yaml:"max-idle-connections,omitempty"` - MaxConnectionIdleTime string `yaml:"max-connection-idle-time,omitempty"` - MaxConnectionLifetime string `yaml:"max-connection-lifetime,omitempty"` - ConnectTimeout string `yaml:"connect-timeout,omitempty"` - StatementTimeout string `yaml:"statement-timeout,omitempty"` - LockTimeout string `yaml:"lock-timeout,omitempty"` - DialAddress string `yaml:"dial-address,omitempty"` -} - -// ConfigPostgresDatabases selects the databases that database-local -// PostgreSQL collectors monitor. An empty Include selects every eligible -// database, and Exclude patterns always take precedence. Patterns are -// case-sensitive and support * and ? wildcards. -type ConfigPostgresDatabases struct { - Enabled *bool `yaml:"enabled,omitempty"` - Include []string `yaml:"include,omitempty"` - Exclude []string `yaml:"exclude,omitempty"` - Refresh string `yaml:"refresh,omitempty"` - MaxConcurrency *int `yaml:"max-concurrency,omitempty"` -} - -func DefaultConfigPostgres() ConfigPostgres { - return ConfigPostgres{ - Database: DEFAULT_POSTGRES_DATABASE, - Databases: DefaultConfigPostgresDatabases(), - ApplicationName: DEFAULT_POSTGRES_APPLICATION_NAME, - MaxOpenConnections: postgresInt(DEFAULT_POSTGRES_MAX_OPEN_CONNECTIONS), - MaxIdleConnections: postgresInt(DEFAULT_POSTGRES_MAX_IDLE_CONNECTIONS), - MaxConnectionIdleTime: DEFAULT_POSTGRES_MAX_CONNECTION_IDLE_TIME, - MaxConnectionLifetime: DEFAULT_POSTGRES_MAX_CONNECTION_LIFETIME, - ConnectTimeout: DEFAULT_MONITOR_TIMEOUT_CONNECT, - } -} - -func DefaultConfigPostgresDatabases() ConfigPostgresDatabases { - return ConfigPostgresDatabases{ - Enabled: postgresBool(true), - Refresh: DEFAULT_POSTGRES_DATABASE_REFRESH, - MaxConcurrency: postgresInt(DEFAULT_POSTGRES_DATABASE_MAX_CONCURRENCY), - } -} - -// Set reports whether a monitor explicitly contains PostgreSQL configuration. -func (c ConfigPostgres) Set() bool { - return c.Database != "" || - c.Databases.Set() || - c.ApplicationName != "" || - c.SSLMode != "" || - c.MaxOpenConnections != nil || - c.MaxIdleConnections != nil || - c.MaxConnectionIdleTime != "" || - c.MaxConnectionLifetime != "" || - c.ConnectTimeout != "" || - c.StatementTimeout != "" || - c.LockTimeout != "" || - c.DialAddress != "" -} - -func (c *ConfigPostgres) ApplyDefaults(defaults ConfigPostgres) { - if c.Database == "" { - c.Database = defaults.Database - } - c.Databases.ApplyDefaults(defaults.Databases) - if c.ApplicationName == "" { - c.ApplicationName = defaults.ApplicationName - } - if c.SSLMode == "" { - c.SSLMode = defaults.SSLMode - } - c.MaxOpenConnections = setPostgresInt(c.MaxOpenConnections, defaults.MaxOpenConnections) - c.MaxIdleConnections = setPostgresInt(c.MaxIdleConnections, defaults.MaxIdleConnections) - if c.MaxConnectionIdleTime == "" { - c.MaxConnectionIdleTime = defaults.MaxConnectionIdleTime - } - if c.MaxConnectionLifetime == "" { - c.MaxConnectionLifetime = defaults.MaxConnectionLifetime - } - if c.ConnectTimeout == "" { - c.ConnectTimeout = defaults.ConnectTimeout - } - if c.StatementTimeout == "" { - c.StatementTimeout = defaults.StatementTimeout - } - if c.LockTimeout == "" { - c.LockTimeout = defaults.LockTimeout - } - if c.DialAddress == "" { - c.DialAddress = defaults.DialAddress - } -} - -func (c ConfigPostgresDatabases) Set() bool { - return c.Enabled != nil || - c.Include != nil || - c.Exclude != nil || - c.Refresh != "" || - c.MaxConcurrency != nil -} - -func (c *ConfigPostgresDatabases) ApplyDefaults(defaults ConfigPostgresDatabases) { - if c.Enabled == nil && defaults.Enabled != nil { - c.Enabled = postgresBool(*defaults.Enabled) - } - if c.Include == nil && defaults.Include != nil { - c.Include = append([]string(nil), defaults.Include...) - } - if c.Exclude == nil && defaults.Exclude != nil { - c.Exclude = append([]string(nil), defaults.Exclude...) - } - if c.Refresh == "" { - c.Refresh = defaults.Refresh - } - c.MaxConcurrency = setPostgresInt(c.MaxConcurrency, defaults.MaxConcurrency) -} - -func (c ConfigPostgres) Validate() error { - validSSLModes := map[string]bool{ - "": true, - "disable": true, - "allow": true, - "prefer": true, - "require": true, - "verify-ca": true, - "verify-full": true, - } - if !validSSLModes[strings.ToLower(c.SSLMode)] { - return fmt.Errorf("config.postgres.ssl-mode: invalid PostgreSQL SSL mode %q", c.SSLMode) - } - if c.MaxOpenConnections != nil && *c.MaxOpenConnections < 0 { - return fmt.Errorf("config.postgres.max-open-connections: must be greater than or equal to zero") - } - if c.MaxIdleConnections != nil && *c.MaxIdleConnections < 0 { - return fmt.Errorf("config.postgres.max-idle-connections: must be greater than or equal to zero") - } - if c.MaxOpenConnections != nil && c.MaxIdleConnections != nil && - *c.MaxOpenConnections > 0 && *c.MaxIdleConnections > *c.MaxOpenConnections { - return fmt.Errorf("config.postgres.max-idle-connections: cannot exceed max-open-connections") - } - if err := validatePostgresDuration("connect-timeout", c.ConnectTimeout, false); err != nil { - return err - } - if err := validatePostgresDuration("max-connection-idle-time", c.MaxConnectionIdleTime, true); err != nil { - return err - } - if err := validatePostgresDuration("max-connection-lifetime", c.MaxConnectionLifetime, true); err != nil { - return err - } - if err := validatePostgresDuration("statement-timeout", c.StatementTimeout, true); err != nil { - return err - } - if err := validatePostgresDuration("lock-timeout", c.LockTimeout, true); err != nil { - return err - } - return c.Databases.Validate() -} - -func (c ConfigPostgresDatabases) Validate() error { - if err := validatePostgresDuration("databases.refresh", c.Refresh, false); err != nil { - return err - } - if c.MaxConcurrency != nil && *c.MaxConcurrency <= 0 { - return fmt.Errorf("config.postgres.databases.max-concurrency: must be greater than zero") - } - for _, patterns := range []struct { - name string - values []string - }{ - {name: "include", values: c.Include}, - {name: "exclude", values: c.Exclude}, - } { - for _, pattern := range patterns.values { - if pattern == "" { - return fmt.Errorf("config.postgres.databases.%s: patterns cannot be empty", patterns.name) - } - } - } - return nil -} - -func postgresBool(value bool) *bool { - return &value -} - -func postgresInt(value int) *int { - return &value -} - -func setPostgresInt(value, defaultValue *int) *int { - if value != nil || defaultValue == nil { - return value - } - copy := *defaultValue - return © -} - -func validatePostgresDuration(name, value string, allowZero bool) error { - if value == "" { - return nil - } - duration, err := time.ParseDuration(value) - if err != nil { - return fmt.Errorf("config.postgres.%s: invalid duration %q: %w", name, value, err) - } - if duration < 0 || (!allowZero && duration == 0) { - constraint := "greater than zero" - if allowZero { - constraint = "greater than or equal to zero" - } - return fmt.Errorf("config.postgres.%s: must be %s", name, constraint) - } - return nil -} - -func (c *ConfigPostgres) InterpolateEnvVars() { - c.Database = interpolateEnv(c.Database) - c.Databases.InterpolateEnvVars() - c.ApplicationName = interpolateEnv(c.ApplicationName) - c.SSLMode = interpolateEnv(c.SSLMode) - c.MaxConnectionIdleTime = interpolateEnv(c.MaxConnectionIdleTime) - c.MaxConnectionLifetime = interpolateEnv(c.MaxConnectionLifetime) - c.ConnectTimeout = interpolateEnv(c.ConnectTimeout) - c.StatementTimeout = interpolateEnv(c.StatementTimeout) - c.LockTimeout = interpolateEnv(c.LockTimeout) - c.DialAddress = interpolateEnv(c.DialAddress) -} - -func (c *ConfigPostgresDatabases) InterpolateEnvVars() { - for i := range c.Include { - c.Include[i] = interpolateEnv(c.Include[i]) - } - for i := range c.Exclude { - c.Exclude[i] = interpolateEnv(c.Exclude[i]) - } - c.Refresh = interpolateEnv(c.Refresh) -} - -func (c *ConfigPostgres) InterpolateMonitor(m *ConfigMonitor) { - c.Database = m.interpolateMon(c.Database) - c.Databases.InterpolateMonitor(m) - c.ApplicationName = m.interpolateMon(c.ApplicationName) - c.SSLMode = m.interpolateMon(c.SSLMode) - c.MaxConnectionIdleTime = m.interpolateMon(c.MaxConnectionIdleTime) - c.MaxConnectionLifetime = m.interpolateMon(c.MaxConnectionLifetime) - c.ConnectTimeout = m.interpolateMon(c.ConnectTimeout) - c.StatementTimeout = m.interpolateMon(c.StatementTimeout) - c.LockTimeout = m.interpolateMon(c.LockTimeout) - c.DialAddress = m.interpolateMon(c.DialAddress) -} - -func (c *ConfigPostgresDatabases) InterpolateMonitor(m *ConfigMonitor) { - for i := range c.Include { - c.Include[i] = m.interpolateMon(c.Include[i]) - } - for i := range c.Exclude { - c.Exclude[i] = m.interpolateMon(c.Exclude[i]) - } - c.Refresh = m.interpolateMon(c.Refresh) -} diff --git a/config_postgres_test.go b/config_postgres_test.go deleted file mode 100644 index 9126f52a..00000000 --- a/config_postgres_test.go +++ /dev/null @@ -1,410 +0,0 @@ -// Copyright 2026 Block, Inc. - -package blip_test - -import ( - "strings" - "testing" - - "github.com/cashapp/blip" -) - -func TestConfigMonitorDatabaseTypeDefaultsToMySQLWithoutMutation(t *testing.T) { - monitor := blip.ConfigMonitor{} - monitor.ApplyDefaults(blip.DefaultConfig()) - - if monitor.DatabaseType != "" { - t.Fatalf("omitted database type mutated to %q", monitor.DatabaseType) - } - if monitor.EffectiveDatabaseType() != blip.DatabaseTypeMySQL { - t.Fatalf("effective database type = %q, expected mysql", monitor.EffectiveDatabaseType()) - } - if monitor.Postgres.Set() { - t.Fatalf("PostgreSQL defaults added to MySQL monitor: %+v", monitor.Postgres) - } -} - -func TestConfigMonitorPostgresDefaultsAndInterpolation(t *testing.T) { - t.Setenv("BLIP_TEST_DATABASE_TYPE", "postgres") - t.Setenv("BLIP_TEST_POSTGRES_DATABASE", "metrics_database") - t.Setenv("BLIP_TEST_POSTGRES_INCLUDE", "app_*") - t.Setenv("BLIP_TEST_POSTGRES_DIAL_ADDRESS", "127.0.0.1:35432") - - monitor := blip.ConfigMonitor{ - MonitorId: "postgres-monitor", - DatabaseType: "${BLIP_TEST_DATABASE_TYPE}", - TimeoutConnect: "7s", - Postgres: blip.ConfigPostgres{ - Database: "${BLIP_TEST_POSTGRES_DATABASE}", - Databases: blip.ConfigPostgresDatabases{ - Include: []string{"${BLIP_TEST_POSTGRES_INCLUDE}"}, - Exclude: []string{"%{monitor.id}_scratch"}, - }, - ApplicationName: "%{monitor.id}", - DialAddress: "${BLIP_TEST_POSTGRES_DIAL_ADDRESS}", - }, - } - monitor.ApplyDefaults(blip.DefaultConfig()) - monitor.InterpolateEnvVars() - monitor.InterpolateMonitor() - - if err := monitor.Validate(); err != nil { - t.Fatal(err) - } - if monitor.DatabaseType != blip.DatabaseTypePostgres { - t.Fatalf("database type = %q, expected postgres", monitor.DatabaseType) - } - if monitor.Postgres.Database != "metrics_database" { - t.Fatalf("database = %q, expected metrics_database", monitor.Postgres.Database) - } - if monitor.Postgres.ApplicationName != monitor.MonitorId { - t.Fatalf("application name = %q, expected monitor ID %q", monitor.Postgres.ApplicationName, monitor.MonitorId) - } - if monitor.Postgres.DialAddress != "127.0.0.1:35432" { - t.Fatalf("dial address = %q, expected interpolated address", monitor.Postgres.DialAddress) - } - if monitor.Postgres.ConnectTimeout != "7s" { - t.Fatalf("connect timeout = %q, expected inherited monitor timeout", monitor.Postgres.ConnectTimeout) - } - if monitor.Postgres.Databases.Enabled == nil || !*monitor.Postgres.Databases.Enabled { - t.Fatalf("database discovery not enabled by default: %+v", monitor.Postgres.Databases.Enabled) - } - if got := monitor.Postgres.Databases.Include; len(got) != 1 || got[0] != "app_*" { - t.Fatalf("database includes not interpolated: %#v", got) - } - if got := monitor.Postgres.Databases.Exclude; len(got) != 1 || got[0] != "postgres-monitor_scratch" { - t.Fatalf("database excludes not monitor-interpolated: %#v", got) - } - if monitor.Postgres.Databases.Refresh != blip.DEFAULT_POSTGRES_DATABASE_REFRESH { - t.Fatalf("database refresh = %q, expected %q", monitor.Postgres.Databases.Refresh, blip.DEFAULT_POSTGRES_DATABASE_REFRESH) - } - if monitor.Postgres.Databases.MaxConcurrency == nil || - *monitor.Postgres.Databases.MaxConcurrency != blip.DEFAULT_POSTGRES_DATABASE_MAX_CONCURRENCY { - t.Fatalf("database max concurrency not defaulted: %+v", monitor.Postgres.Databases.MaxConcurrency) - } - if monitor.Postgres.MaxOpenConnections == nil || *monitor.Postgres.MaxOpenConnections != blip.DEFAULT_POSTGRES_MAX_OPEN_CONNECTIONS { - t.Fatalf("max open connections not defaulted: %+v", monitor.Postgres.MaxOpenConnections) - } - if monitor.Postgres.MaxIdleConnections == nil || *monitor.Postgres.MaxIdleConnections != blip.DEFAULT_POSTGRES_MAX_IDLE_CONNECTIONS { - t.Fatalf("max idle connections not defaulted: %+v", monitor.Postgres.MaxIdleConnections) - } -} - -func TestConfigMonitorDatabaseTypeValidation(t *testing.T) { - tests := []struct { - name string - monitor blip.ConfigMonitor - wantError string - }{ - { - name: "unknown database type", - monitor: blip.ConfigMonitor{DatabaseType: "oracle"}, - wantError: "invalid database type", - }, - { - name: "PostgreSQL config on implicit MySQL monitor", - monitor: blip.ConfigMonitor{ - Postgres: blip.ConfigPostgres{Database: "postgres"}, - }, - wantError: "requires database-type", - }, - { - name: "my.cnf on PostgreSQL monitor", - monitor: blip.ConfigMonitor{ - DatabaseType: blip.DatabaseTypePostgres, - MyCnf: "/etc/blip/my.cnf", - }, - wantError: "mycnf is only supported", - }, - { - name: "socket on PostgreSQL monitor", - monitor: blip.ConfigMonitor{ - DatabaseType: blip.DatabaseTypePostgres, - Socket: "/tmp/.s.PGSQL.5432", - }, - wantError: "socket is only supported", - }, - { - name: "heartbeat on PostgreSQL monitor", - monitor: blip.ConfigMonitor{ - DatabaseType: blip.DatabaseTypePostgres, - Heartbeat: blip.ConfigHeartbeat{Freq: "1s"}, - }, - wantError: "heartbeat is only supported", - }, - { - name: "plan changing on PostgreSQL monitor", - monitor: blip.ConfigMonitor{ - DatabaseType: blip.DatabaseTypePostgres, - Plans: blip.ConfigPlans{ - Change: blip.ConfigPlanChange{ - Active: blip.ConfigStatePlan{Plan: "active"}, - }, - }, - }, - wantError: "plans.change is only supported", - }, - { - name: "plan change delay on PostgreSQL monitor", - monitor: blip.ConfigMonitor{ - DatabaseType: blip.DatabaseTypePostgres, - Plans: blip.ConfigPlans{ - Change: blip.ConfigPlanChange{ - Active: blip.ConfigStatePlan{After: "10s"}, - }, - }, - }, - wantError: "plans.change is only supported", - }, - { - name: "plan table on PostgreSQL monitor", - monitor: blip.ConfigMonitor{ - DatabaseType: blip.DatabaseTypePostgres, - Plans: blip.ConfigPlans{Table: "blip.plans"}, - }, - wantError: "plans.table is only supported", - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - err := tt.monitor.Validate() - if err == nil || !strings.Contains(err.Error(), tt.wantError) { - t.Fatalf("got error %v, expected it to contain %q", err, tt.wantError) - } - }) - } -} - -func TestConfigMonitorPostgresExporterRequiresPlan(t *testing.T) { - defaults := blip.DefaultConfig() - defaults.Exporter.Mode = blip.EXPORTER_MODE_DUAL - monitor := blip.ConfigMonitor{DatabaseType: blip.DatabaseTypePostgres} - - monitor.ApplyDefaults(defaults) - if monitor.Exporter.Plan != "" { - t.Fatalf("PostgreSQL monitor inherited MySQL exporter plan %q", monitor.Exporter.Plan) - } - if err := monitor.Validate(); err == nil || !strings.Contains(err.Error(), "exporter.plan is required") { - t.Fatalf("validation error = %v, expected required exporter plan", err) - } -} - -func TestConfigMonitorPostgresExporterUsesConfiguredPlan(t *testing.T) { - defaults := blip.DefaultConfig() - defaults.Exporter.Mode = blip.EXPORTER_MODE_DUAL - defaults.Exporter.Plan = "postgres-exporter" - monitor := blip.ConfigMonitor{DatabaseType: blip.DatabaseTypePostgres} - - monitor.ApplyDefaults(defaults) - if monitor.Exporter.Plan != defaults.Exporter.Plan { - t.Fatalf("exporter plan = %q, expected %q", monitor.Exporter.Plan, defaults.Exporter.Plan) - } - if err := monitor.Validate(); err != nil { - t.Fatalf("PostgreSQL exporter config is invalid: %v", err) - } -} - -func TestConfigMonitorMySQLExporterRetainsDefaultPlan(t *testing.T) { - defaults := blip.DefaultConfig() - defaults.Exporter.Mode = blip.EXPORTER_MODE_DUAL - monitor := blip.ConfigMonitor{} - - monitor.ApplyDefaults(defaults) - if monitor.Exporter.Plan != blip.DEFAULT_EXPORTER_PLAN { - t.Fatalf("exporter plan = %q, expected %q", monitor.Exporter.Plan, blip.DEFAULT_EXPORTER_PLAN) - } -} - -func TestConfigPlansRejectsPostgresTableMonitor(t *testing.T) { - tests := []struct { - name string - plans blip.ConfigPlans - wantError string - }{ - { - name: "implicit MySQL", - plans: blip.ConfigPlans{ - Table: "blip.plans", - Monitor: &blip.ConfigMonitor{}, - }, - }, - { - name: "PostgreSQL", - plans: blip.ConfigPlans{ - Table: "blip.plans", - Monitor: &blip.ConfigMonitor{ - DatabaseType: blip.DatabaseTypePostgres, - }, - }, - wantError: "config.plans.table is only supported", - }, - { - name: "PostgreSQL without table", - plans: blip.ConfigPlans{ - Monitor: &blip.ConfigMonitor{ - DatabaseType: blip.DatabaseTypePostgres, - }, - }, - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - err := tt.plans.Validate() - if tt.wantError == "" { - if err != nil { - t.Fatalf("unexpected error: %v", err) - } - return - } - if err == nil || !strings.Contains(err.Error(), tt.wantError) { - t.Fatalf("got error %v, expected it to contain %q", err, tt.wantError) - } - }) - } -} - -func TestConfigMonitorDatabaseSpecificDefaults(t *testing.T) { - defaults := blip.DefaultConfig() - defaults.Heartbeat = blip.ConfigHeartbeat{ - Freq: "1s", - Table: "blip.heartbeat", - } - defaults.Plans = blip.ConfigPlans{ - Files: []string{"shared.yaml"}, - Change: blip.ConfigPlanChange{ - Active: blip.ConfigStatePlan{Plan: "active"}, - }, - } - - postgres := blip.ConfigMonitor{DatabaseType: blip.DatabaseTypePostgres} - postgres.ApplyDefaults(defaults) - if postgres.Heartbeat != (blip.ConfigHeartbeat{}) { - t.Fatalf("PostgreSQL monitor inherited heartbeat defaults: %+v", postgres.Heartbeat) - } - if postgres.Plans.Change.Enabled() { - t.Fatalf("PostgreSQL monitor inherited plan-changing defaults: %+v", postgres.Plans.Change) - } - if got := postgres.Plans.Files; len(got) != 1 || got[0] != "shared.yaml" { - t.Fatalf("PostgreSQL monitor plan files = %#v, expected shared plan defaults", got) - } - if err := postgres.Validate(); err != nil { - t.Fatalf("PostgreSQL monitor with shared defaults is invalid: %v", err) - } - - mysql := blip.ConfigMonitor{} - mysql.ApplyDefaults(defaults) - if mysql.Heartbeat == (blip.ConfigHeartbeat{}) { - t.Fatal("MySQL monitor did not inherit heartbeat defaults") - } - if !mysql.Plans.Change.Enabled() { - t.Fatal("MySQL monitor did not inherit plan-changing defaults") - } -} - -func TestConfigPostgresAllowsExplicitUnlimitedPoolSettings(t *testing.T) { - zero := 0 - config := blip.ConfigPostgres{ - Database: "postgres", - MaxOpenConnections: &zero, - MaxIdleConnections: &zero, - } - config.ApplyDefaults(blip.DefaultConfigPostgres()) - - if *config.MaxOpenConnections != 0 || *config.MaxIdleConnections != 0 { - t.Fatalf("explicit zero pool settings were overwritten: %+v", config) - } -} - -func TestConfigPostgresValidation(t *testing.T) { - minusOne := -1 - zero := 0 - one := 1 - two := 2 - tests := []struct { - name string - config blip.ConfigPostgres - wantError string - }{ - { - name: "invalid SSL mode", - config: blip.ConfigPostgres{SSLMode: "invalid"}, - wantError: "ssl-mode", - }, - { - name: "negative max open", - config: blip.ConfigPostgres{MaxOpenConnections: &minusOne}, - wantError: "max-open-connections", - }, - { - name: "idle exceeds open", - config: blip.ConfigPostgres{ - MaxOpenConnections: &one, - MaxIdleConnections: &two, - }, - wantError: "cannot exceed", - }, - { - name: "zero connect timeout", - config: blip.ConfigPostgres{ConnectTimeout: "0"}, - wantError: "connect-timeout", - }, - { - name: "invalid lifetime", - config: blip.ConfigPostgres{MaxConnectionLifetime: "tomorrow"}, - wantError: "max-connection-lifetime", - }, - { - name: "zero database concurrency", - config: blip.ConfigPostgres{ - Databases: blip.ConfigPostgresDatabases{MaxConcurrency: &zero}, - }, - wantError: "databases.max-concurrency", - }, - { - name: "invalid database refresh", - config: blip.ConfigPostgres{ - Databases: blip.ConfigPostgresDatabases{Refresh: "tomorrow"}, - }, - wantError: "databases.refresh", - }, - { - name: "empty database include", - config: blip.ConfigPostgres{ - Databases: blip.ConfigPostgresDatabases{Include: []string{""}}, - }, - wantError: "databases.include", - }, - } - - for _, tt := range tests { - t.Run(tt.name, func(t *testing.T) { - err := tt.config.Validate() - if err == nil || !strings.Contains(err.Error(), tt.wantError) { - t.Fatalf("got error %v, expected it to contain %q", err, tt.wantError) - } - }) - } -} - -func TestConfigPostgresDatabasesPreservesExplicitDisabled(t *testing.T) { - disabled := false - one := 1 - config := blip.ConfigPostgresDatabases{ - Enabled: &disabled, - Include: []string{"app_*"}, - MaxConcurrency: &one, - } - config.ApplyDefaults(blip.DefaultConfigPostgresDatabases()) - - if config.Enabled == nil || *config.Enabled { - t.Fatalf("explicit disabled discovery was overwritten: %+v", config.Enabled) - } - if config.MaxConcurrency == nil || *config.MaxConcurrency != 1 { - t.Fatalf("explicit concurrency was overwritten: %+v", config.MaxConcurrency) - } - if got := config.Include; len(got) != 1 || got[0] != "app_*" { - t.Fatalf("explicit includes were overwritten: %#v", got) - } -} diff --git a/credentials/credentials.go b/credentials/credentials.go index 880d7462..e6d32028 100644 --- a/credentials/credentials.go +++ b/credentials/credentials.go @@ -36,8 +36,9 @@ func NewFactory(awsConfig blip.AWSConfigFactory, passwordSecretParser blip.Passw // Dynamic returns the first configured shared reloadable source in Blip's // established precedence order: IAM, Secrets Manager, then password file. The // boolean reports whether a source was selected. Engine-specific factories can -// insert their own sources before falling back to Static. -func (f Factory) Dynamic(cfg blip.ConfigMonitor) (Func, bool, error) { +// insert their own sources before falling back to Static. defaultPort is used +// only when signing an IAM token for a hostname without an explicit port. +func (f Factory) Dynamic(cfg blip.ConfigMonitor, defaultPort string) (Func, bool, error) { if blip.True(cfg.AWS.IAMAuth) { blip.Debug("%s: AWS IAM auth token password", cfg.MonitorId) if f.awsConfig == nil { @@ -47,9 +48,8 @@ func (f Factory) Dynamic(cfg blip.ConfigMonitor) (Func, bool, error) { if err != nil { return nil, true, err } - defaultPort := "3306" - if cfg.EffectiveDatabaseType() == blip.DatabaseTypePostgres { - defaultPort = "5432" + if defaultPort == "" { + return nil, true, fmt.Errorf("AWS IAM authentication requires a database default port") } token := blipaws.NewAuthTokenWithDefaultPort(cfg.Username, cfg.Hostname, defaultPort, awscfg) return func(ctx context.Context) (blip.DbCredentials, error) { diff --git a/credentials/credentials_test.go b/credentials/credentials_test.go index 8a193892..f388d48c 100644 --- a/credentials/credentials_test.go +++ b/credentials/credentials_test.go @@ -33,7 +33,7 @@ func TestDynamicPasswordFileReloads(t *testing.T) { credentialFunc, selected, err := credentials.NewFactory(nil, nil).Dynamic(blip.ConfigMonitor{ Username: "metrics", PasswordFile: passwordFile, - }) + }, "3306") if err != nil { t.Fatal(err) } @@ -90,7 +90,7 @@ func TestDynamicPreservesSourcePrecedence(t *testing.T) { for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { - credentialFunc, selected, err := credentials.NewFactory(nil, nil).Dynamic(tt.config) + credentialFunc, selected, err := credentials.NewFactory(nil, nil).Dynamic(tt.config, "3306") if !selected { t.Fatal("configured source was not selected") } @@ -104,7 +104,7 @@ func TestDynamicPreservesSourcePrecedence(t *testing.T) { } } -func TestDynamicIAMUsesDatabaseDefaultPort(t *testing.T) { +func TestDynamicIAMUsesCallerDefaultPort(t *testing.T) { iamAuth := true factory := credentials.NewFactory(staticAWSConfigFactory{config: awsv2.Config{ Region: "us-west-2", @@ -118,41 +118,41 @@ func TestDynamicIAMUsesDatabaseDefaultPort(t *testing.T) { }}, nil) tests := []struct { - name string - databaseType blip.DatabaseType - hostname string - wantHost string + name string + defaultPort string + hostname string + wantHost string }{ { - name: "implicit MySQL", - hostname: "mysql.example", - wantHost: "mysql.example:3306", + name: "MySQL default", + defaultPort: "3306", + hostname: "mysql.example", + wantHost: "mysql.example:3306", }, { - name: "PostgreSQL", - databaseType: blip.DatabaseTypePostgres, - hostname: "postgres.example", - wantHost: "postgres.example:5432", + name: "external engine default", + defaultPort: "5432", + hostname: "postgres.example", + wantHost: "postgres.example:5432", }, { - name: "PostgreSQL custom port", - databaseType: blip.DatabaseTypePostgres, - hostname: "postgres.example:6432", - wantHost: "postgres.example:6432", + name: "explicit port", + defaultPort: "5432", + hostname: "postgres.example:6432", + wantHost: "postgres.example:6432", }, } for _, tt := range tests { t.Run(tt.name, func(t *testing.T) { credentialFunc, selected, err := factory.Dynamic(blip.ConfigMonitor{ - DatabaseType: tt.databaseType, - Hostname: tt.hostname, - Username: "metrics", + Hostname: tt.hostname, + Username: "metrics", AWS: blip.ConfigAWS{ IAMAuth: &iamAuth, Region: "us-west-2", }, - }) + }, tt.defaultPort) if err != nil { t.Fatal(err) } @@ -175,11 +175,27 @@ func TestDynamicIAMUsesDatabaseDefaultPort(t *testing.T) { } } +func TestDynamicIAMRequiresCallerDefaultPort(t *testing.T) { + iamAuth := true + factory := credentials.NewFactory(staticAWSConfigFactory{config: awsv2.Config{Region: "us-west-2"}}, nil) + credentialFunc, selected, err := factory.Dynamic(blip.ConfigMonitor{ + Hostname: "database.example", + Username: "metrics", + AWS: blip.ConfigAWS{ + IAMAuth: &iamAuth, + Region: "us-west-2", + }, + }, "") + if !selected || credentialFunc != nil || err == nil || !strings.Contains(err.Error(), "default port") { + t.Fatalf("Dynamic returned func=%v selected=%t err=%v", credentialFunc != nil, selected, err) + } +} + func TestDynamicReportsNoSharedSource(t *testing.T) { credentialFunc, selected, err := credentials.NewFactory(nil, nil).Dynamic(blip.ConfigMonitor{ Username: "metrics", Password: "static", - }) + }, "3306") if err != nil || selected || credentialFunc != nil { t.Fatalf("Dynamic returned func=%v selected=%t err=%v", credentialFunc != nil, selected, err) } diff --git a/database_module.go b/database_module.go new file mode 100644 index 00000000..c7dbaa64 --- /dev/null +++ b/database_module.go @@ -0,0 +1,210 @@ +// Copyright 2026 Block, Inc. + +package blip + +import ( + "fmt" + "reflect" + "regexp" + "sync" + + "gopkg.in/yaml.v2" +) + +// DatabaseType identifies the database engine used by one monitor. +type DatabaseType string + +const ( + // DatabaseTypeMySQL is Blip's built-in database type. An omitted monitor + // database type retains this historical behavior. + DatabaseTypeMySQL DatabaseType = "mysql" + + // DatabaseTypeAny declares that a collector is database-neutral. It is a + // collector compatibility marker, not a valid monitor database type. + DatabaseTypeAny DatabaseType = "*" +) + +// ConfigDatabase contains configuration owned by an external database module. +// Blip interpolates string values but otherwise treats this map as opaque. A +// module should use DecodeDatabaseConfig to strictly decode it into a typed +// configuration before applying its defaults and validation. +type ConfigDatabase map[string]interface{} + +// DatabaseModule validates configuration for one external database type. +// Modules register before server boot. MySQL is built into Blip and does not +// use this interface. +type DatabaseModule interface { + DatabaseType() DatabaseType + ValidateConfig(ConfigMonitor) error +} + +var databaseModuleRegistry = struct { + sync.RWMutex + modules map[DatabaseType]DatabaseModule +}{ + modules: map[DatabaseType]DatabaseModule{}, +} + +var validDatabaseType = regexp.MustCompile(`^[a-z][a-z0-9._-]*$`) + +// ValidDatabaseType reports whether a value is a valid concrete monitor and +// collector database type. DatabaseTypeAny is not a concrete type. +func ValidDatabaseType(databaseType DatabaseType) bool { + return validDatabaseType.MatchString(string(databaseType)) +} + +// RegisterDatabaseModule registers one external database module. Registering a +// duplicate type or one of Blip's reserved database types returns an error. +func RegisterDatabaseModule(module DatabaseModule) error { + if nilInterface(module) { + return fmt.Errorf("database module is nil") + } + databaseType := module.DatabaseType() + if databaseType == DatabaseTypeMySQL { + return fmt.Errorf("database type %q is built into Blip", databaseType) + } + if databaseType == DatabaseTypeAny { + return fmt.Errorf("database type %q is reserved for database-neutral collectors", databaseType) + } + if !ValidDatabaseType(databaseType) { + return fmt.Errorf("database module has invalid database type %q", databaseType) + } + + databaseModuleRegistry.Lock() + defer databaseModuleRegistry.Unlock() + if _, ok := databaseModuleRegistry.modules[databaseType]; ok { + return fmt.Errorf("database module %q already registered", databaseType) + } + databaseModuleRegistry.modules[databaseType] = module + return nil +} + +// RemoveDatabaseModule removes an external database module. It supports test +// isolation and registration rollback when a larger module enable operation +// fails after registering its database type. +func RemoveDatabaseModule(databaseType DatabaseType) { + databaseModuleRegistry.Lock() + defer databaseModuleRegistry.Unlock() + delete(databaseModuleRegistry.modules, databaseType) +} + +func registeredDatabaseModule(databaseType DatabaseType) (DatabaseModule, bool) { + databaseModuleRegistry.RLock() + defer databaseModuleRegistry.RUnlock() + module, ok := databaseModuleRegistry.modules[databaseType] + return module, ok +} + +// DecodeDatabaseConfig strictly decodes opaque monitor database configuration +// into a module-owned typed value. +func DecodeDatabaseConfig(config ConfigDatabase, out interface{}) error { + if nilInterface(out) { + return fmt.Errorf("database config output is nil") + } + encoded, err := yaml.Marshal(config) + if err != nil { + return fmt.Errorf("encode database config: %w", err) + } + if err := yaml.UnmarshalStrict(encoded, out); err != nil { + return fmt.Errorf("decode database config: %w", err) + } + return nil +} + +func nilInterface(value interface{}) bool { + if value == nil { + return true + } + reflected := reflect.ValueOf(value) + switch reflected.Kind() { + case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Ptr, reflect.Slice: + return reflected.IsNil() + default: + return false + } +} + +func (c ConfigDatabase) interpolate(interpolate func(string) string) { + for key, value := range c { + c[key] = interpolateDatabaseConfigValue(value, interpolate) + } +} + +func interpolateDatabaseConfigValue(value interface{}, interpolate func(string) string) interface{} { + interpolated := interpolateDatabaseConfigReflect(reflect.ValueOf(value), interpolate) + if !interpolated.IsValid() { + return nil + } + return interpolated.Interface() +} + +func interpolateDatabaseConfigReflect(value reflect.Value, interpolate func(string) string) reflect.Value { + if !value.IsValid() { + return value + } + + switch value.Kind() { + case reflect.Interface: + if value.IsNil() { + return reflect.Zero(value.Type()) + } + interpolated := interpolateDatabaseConfigReflect(value.Elem(), interpolate) + result := reflect.New(value.Type()).Elem() + result.Set(interpolated) + return result + case reflect.String: + result := reflect.New(value.Type()).Elem() + result.SetString(interpolate(value.String())) + return result + case reflect.Map: + if value.IsNil() { + return reflect.Zero(value.Type()) + } + result := reflect.MakeMapWithSize(value.Type(), value.Len()) + iterator := value.MapRange() + for iterator.Next() { + result.SetMapIndex( + iterator.Key(), + interpolateDatabaseConfigReflect(iterator.Value(), interpolate), + ) + } + return result + case reflect.Slice: + if value.IsNil() { + return reflect.Zero(value.Type()) + } + result := reflect.MakeSlice(value.Type(), value.Len(), value.Len()) + for i := 0; i < value.Len(); i++ { + result.Index(i).Set(interpolateDatabaseConfigReflect(value.Index(i), interpolate)) + } + return result + case reflect.Array: + result := reflect.New(value.Type()).Elem() + for i := 0; i < value.Len(); i++ { + result.Index(i).Set(interpolateDatabaseConfigReflect(value.Index(i), interpolate)) + } + return result + case reflect.Struct: + result := reflect.New(value.Type()).Elem() + result.Set(value) + for i := 0; i < value.NumField(); i++ { + if !value.Type().Field(i).IsExported() { + continue + } + result.Field(i).Set(interpolateDatabaseConfigReflect(value.Field(i), interpolate)) + } + return result + case reflect.Ptr: + if value.IsNil() { + return reflect.Zero(value.Type()) + } + result := reflect.New(value.Type().Elem()) + result.Elem().Set(interpolateDatabaseConfigReflect(value.Elem(), interpolate)) + if result.Type() != value.Type() { + result = result.Convert(value.Type()) + } + return result + default: + return value + } +} diff --git a/dbconn/factory.go b/dbconn/factory.go index dd148a38..0feb2f0d 100644 --- a/dbconn/factory.go +++ b/dbconn/factory.go @@ -275,7 +275,7 @@ func (f factory) Make(cfg blip.ConfigMonitor) (*sql.DB, string, error) { // credentials are fetched via a reload func, even a static credential specified // in the Blip config file. func (f factory) Credentials(cfg blip.ConfigMonitor) (CredentialFunc, error) { - credentialFunc, selected, err := credentials.NewFactory(f.awsConfig, f.passwordSecretParser).Dynamic(cfg) + credentialFunc, selected, err := credentials.NewFactory(f.awsConfig, f.passwordSecretParser).Dynamic(cfg, "3306") if err != nil || selected { return credentialFunc, err } diff --git a/docs/content/config/config-file.md b/docs/content/config/config-file.md index 3848a63d..1bb5d7dc 100644 --- a/docs/content/config/config-file.md +++ b/docs/content/config/config-file.md @@ -761,7 +761,28 @@ Section [`exporter`](#exporter) is exactly the same in a monitor. Refer to [Monitor Defaults](#monitor-defaults) for configuring MySQL instances, and remember: [`mysql`](#mysql) variables are top-level in a monitor (omit `mysql:` and include the variables directly). -Monitors have three variables that only appear in monitors: `id`, `meta`, and `plan`. +All monitors have three variables that only appear in monitor entries: `id`, +`meta`, and `plan`. A binary that enables an external database module can also +use `database-type` and `database-config`: + +```yaml +monitors: + - id: external + database-type: my-database + hostname: database.example:1234 + database-config: + module-option: value +``` + +An omitted `database-type` retains Blip's built-in MySQL behavior. The type is +a literal module identifier, although direct environment-variable interpolation +such as `${DATABASE_TYPE}` is supported. Monitor-field interpolation is not +supported for this structural value. + +The contents of `database-config` belong to the selected module. Blip +interpolates string values, redacts the complete section from logged config, +and asks the registered module to validate it. Refer to the external module for +its supported fields. Blip itself does not register a non-MySQL module. ### `id` diff --git a/docs/content/config/heartbeat.md b/docs/content/config/heartbeat.md index 9882bf96..aa7755b3 100644 --- a/docs/content/config/heartbeat.md +++ b/docs/content/config/heartbeat.md @@ -7,8 +7,8 @@ For example, `Seconds_Behind_Source` from `SHOW REPLICA STATUS` (or `Seconds_Beh Consequently, external replication heartbeats are an industry norm because they are easy and accurate—and they work the same across all versions and distributions of MySQL, including the the cloud. {{< hint type=note >}} -Blip heartbeat is MySQL-specific. PostgreSQL monitors do not inherit global -heartbeat defaults and reject monitor-level heartbeat configuration. +Blip heartbeat is MySQL-specific. External database monitors do not inherit +global heartbeat defaults and reject monitor-level heartbeat configuration. {{< /hint >}} ## Quick Start diff --git a/docs/content/develop/collectors.md b/docs/content/develop/collectors.md index 807b2b0d..7bc2999c 100644 --- a/docs/content/develop/collectors.md +++ b/docs/content/develop/collectors.md @@ -133,10 +133,14 @@ preparation: ```go func (myFactory) DatabaseTypes(string) []blip.DatabaseType { - return []blip.DatabaseType{blip.DatabaseTypePostgres} + return []blip.DatabaseType{"my-database"} } ``` +A collector that does not depend on the monitor's database engine returns +`blip.DatabaseTypeAny`. The database-neutral type cannot be combined with +specific database types. + ## Long-running As of Blip v1.2.0, long-running collectors are possible using one of two approaches: diff --git a/docs/content/develop/database-modules.md b/docs/content/develop/database-modules.md new file mode 100644 index 00000000..9024bb49 --- /dev/null +++ b/docs/content/develop/database-modules.md @@ -0,0 +1,77 @@ +--- +--- + +Blip is purpose-built for MySQL. An external database module can reuse its +monitor, plan, collection, transformation, and sink runtime without adding that +database implementation to Blip itself. + +{{< toc >}} + +## Enable Before Boot + +An external module is enabled by the integrating binary before `server.Boot`. +The module must: + +1. Register its database type with `blip.RegisterDatabaseModule`. +2. Register its collectors through the existing `metrics.Register` API. +3. Decorate `Factories.DbConn` with its connection factory. + +The module's connection factory should handle only its own database type and +delegate every other type to the previous factory. It must also preserve +`DbProviderFactory` delegation when the previous factory implements that +optional capability. This convention allows multiple external modules to +compose without changing Blip's built-in MySQL factory. + +## Configuration + +Every external monitor sets a literal `database-type` and can provide an opaque +`database-config` map: + +```yaml +monitors: + - database-type: my-database + hostname: database.example:1234 + username: metrics + database-config: + module-option: value +``` + +An omitted database type remains MySQL. Direct `${ENV_VAR}` interpolation is +supported in `database-type`; monitor-field interpolation is intentionally not +supported because the type selects defaults before the rest of monitor +initialization. + +Blip recursively interpolates string values inside `database-config` and +redacts the entire opaque map when logging monitor configuration. A module uses +`blip.DecodeDatabaseConfig` to strictly decode the map into its own typed +configuration, then applies and validates its defaults in module code. Its +`DatabaseModule.ValidateConfig` implementation provides early validation during +monitor loading. + +MySQL socket, `my.cnf`, heartbeat, plan changing, plan-table storage, and +`mysqld_exporter` emulation are rejected for external database monitors. + +## Connections and Credentials + +`DbProviderFactory` is optional. A module that needs multiple connection pools +returns a `DbProvider`; Blip uses `Primary` for ordinary collectors and closes +the provider only after monitor subsystems and collectors stop. A specialized +collector factory implements `CollectorFactoryWithDBProvider` and type-asserts +the generic provider to a module-owned extension interface. + +The shared `credentials.Factory` supports IAM, Secrets Manager, password files, +static passwords, and passwordless authentication. Its `Dynamic` method takes +the engine's default port explicitly. The module remains responsible for +endpoint normalization, credential caching and refresh, authentication-error +classification, and connection retry behavior. + +## Collector Compatibility + +A module collector implements `CollectorFactoryDatabaseTypes` and returns its +database type. A collector that is independent of the monitor database returns +`DatabaseTypeAny`. A factory that does not implement the optional interface +retains Blip's historical MySQL compatibility. + +Blip validates that database-specific collectors in one plan have a common +database type, and it validates the selected plan against each monitor before +collector preparation. Database-neutral collectors do not constrain the plan. diff --git a/docs/content/plans/changing.md b/docs/content/plans/changing.md index 60a484f8..b5dbaf95 100644 --- a/docs/content/plans/changing.md +++ b/docs/content/plans/changing.md @@ -5,8 +5,8 @@ title: "Changing" Plan changing makes Blip change plans while running (without restarting) based on the state of MySQL: {{< hint type=note >}} -Plan changing is MySQL-specific. PostgreSQL monitors do not inherit global -plan-changing defaults and reject monitor-level plan-changing configuration. +Plan changing is MySQL-specific. External database monitors do not inherit +global plan-changing defaults and reject monitor-level plan-changing configuration. {{< /hint >}} |State|Connected to MySQL|Collecting Metrics|Description| diff --git a/docs/content/plans/file.md b/docs/content/plans/file.md index 7e4eb4a7..7fe90c20 100644 --- a/docs/content/plans/file.md +++ b/docs/content/plans/file.md @@ -50,9 +50,9 @@ And each domain has a domain-specific configuration that includes: These values are documented for each [domain]({{< ref "/metrics/domains" >}}) and printed on the command line by [`--print-domains`]({{< ref "/config/blip#--print-domains" >}}). All domains in a plan, across every level, must support at least one common -database type. A MySQL-only domain and a domain that supports both MySQL and -PostgreSQL can share a plan, but a MySQL-only domain and a PostgreSQL-only -domain cannot. A shared plan configuration can still contain separate plans +database type. A database-neutral domain can share a plan with any +database-specific domain, but domains for two different database types cannot +share one plan. A shared plan configuration can still contain separate plans for different database types. Since Blip automatically levels up overlapping frequencies (described in [Intro / Plans]({{< ref "intro/plans" >}})), it's conventional to define levels from most to least frequent, as in this example: diff --git a/docs/content/plans/table.md b/docs/content/plans/table.md index ebf56bbf..5d6c3ddb 100644 --- a/docs/content/plans/table.md +++ b/docs/content/plans/table.md @@ -6,9 +6,9 @@ A plan table contains one plan per row: {{< hint type=note >}} Plan-table storage is MySQL-specific. The connection configured by a top-level -`plans.monitor` must be MySQL, and PostgreSQL monitors reject `plans.table` in -their monitor configuration. PostgreSQL monitors can use plan files or -compatible shared plans. +`plans.monitor` must be MySQL, and external database monitors reject +`plans.table` in their monitor configuration. External database monitors can +use plan files or compatible shared plans. {{< /hint >}} ```sql diff --git a/metrics/factory.go b/metrics/factory.go index 81b890d5..c76d1eea 100644 --- a/metrics/factory.go +++ b/metrics/factory.go @@ -80,9 +80,7 @@ func normalizeDatabaseTypes(domain string, databaseTypes []blip.DatabaseType) ([ seen := map[blip.DatabaseType]bool{} normalized := make([]blip.DatabaseType, 0, len(databaseTypes)) for _, databaseType := range databaseTypes { - switch databaseType { - case blip.DatabaseTypeMySQL, blip.DatabaseTypePostgres: - default: + if databaseType != blip.DatabaseTypeAny && !blip.ValidDatabaseType(databaseType) { return nil, fmt.Errorf("collector %s declares invalid database type %q", domain, databaseType) } if seen[databaseType] { @@ -91,6 +89,9 @@ func normalizeDatabaseTypes(domain string, databaseTypes []blip.DatabaseType) ([ seen[databaseType] = true normalized = append(normalized, databaseType) } + if seen[blip.DatabaseTypeAny] && len(normalized) != 1 { + return nil, fmt.Errorf("collector %s declares database-neutral compatibility with specific database types", domain) + } sort.Slice(normalized, func(i, j int) bool { return normalized[i] < normalized[j] }) @@ -155,7 +156,7 @@ func validateDatabase(domain string, databaseType blip.DatabaseType) error { return fmt.Errorf("invalid domain: %s (no factory registered)", domain) } for _, supportedType := range registered.databaseTypes { - if supportedType == databaseType { + if supportedType == blip.DatabaseTypeAny || supportedType == databaseType { return nil } } @@ -361,10 +362,7 @@ func InitFactory(factories blip.Factories) { func (f *factory) DatabaseTypes(domain string) []blip.DatabaseType { if domain == awsrds.DOMAIN { - return []blip.DatabaseType{ - blip.DatabaseTypeMySQL, - blip.DatabaseTypePostgres, - } + return []blip.DatabaseType{blip.DatabaseTypeAny} } return []blip.DatabaseType{blip.DatabaseTypeMySQL} } diff --git a/metrics/factory_test.go b/metrics/factory_test.go index a443c631..825e6c74 100644 --- a/metrics/factory_test.go +++ b/metrics/factory_test.go @@ -12,6 +12,8 @@ import ( "github.com/cashapp/blip/test/mock" ) +const externalType blip.DatabaseType = "test-database" + type databaseTypesFactory struct { mock.MetricFactory databaseTypes func(string) []blip.DatabaseType @@ -65,20 +67,20 @@ func TestRegisterDefaultsToMySQL(t *testing.T) { t.Fatalf("Make(default mysql): %v", err) } - err := metrics.ValidateDatabase(domain, blip.DatabaseTypePostgres) - if err == nil || !strings.Contains(err.Error(), `does not support database type "postgres" (supported: [mysql])`) { - t.Fatalf("ValidateDatabase(postgres) error = %v", err) + err := metrics.ValidateDatabase(domain, externalType) + if err == nil || !strings.Contains(err.Error(), `does not support database type "test-database" (supported: [mysql])`) { + t.Fatalf("ValidateDatabase(external) error = %v", err) } } func TestRegisterUsesFactoryDatabaseTypes(t *testing.T) { - const domain = "test.postgres-only" + const domain = "test.external-only" factory := databaseTypesFactory{ databaseTypes: func(gotDomain string) []blip.DatabaseType { if gotDomain != domain { t.Fatalf("DatabaseTypes domain = %q, expected %q", gotDomain, domain) } - return []blip.DatabaseType{blip.DatabaseTypePostgres} + return []blip.DatabaseType{externalType} }, } @@ -87,17 +89,17 @@ func TestRegisterUsesFactoryDatabaseTypes(t *testing.T) { } t.Cleanup(func() { metrics.Remove(domain) }) - if err := metrics.ValidateDatabase(domain, blip.DatabaseTypePostgres); err != nil { - t.Fatalf("ValidateDatabase(postgres): %v", err) + if err := metrics.ValidateDatabase(domain, externalType); err != nil { + t.Fatalf("ValidateDatabase(external): %v", err) } if _, err := metrics.Make(domain, blip.CollectorFactoryArgs{ - Config: blip.ConfigMonitor{DatabaseType: blip.DatabaseTypePostgres}, + Config: blip.ConfigMonitor{DatabaseType: externalType}, }); err != nil { - t.Fatalf("Make(postgres): %v", err) + t.Fatalf("Make(external): %v", err) } err := metrics.ValidateDatabase(domain, blip.DatabaseTypeMySQL) - if err == nil || !strings.Contains(err.Error(), `does not support database type "mysql" (supported: [postgres])`) { + if err == nil || !strings.Contains(err.Error(), `does not support database type "mysql" (supported: [test-database])`) { t.Fatalf("ValidateDatabase(mysql) error = %v", err) } if _, err := metrics.Make(domain, blip.CollectorFactoryArgs{}); err == nil { @@ -116,9 +118,9 @@ func TestRegisterSupportsMultipleDatabaseTypes(t *testing.T) { factory := databaseTypesFactory{ databaseTypes: func(string) []blip.DatabaseType { return []blip.DatabaseType{ - blip.DatabaseTypePostgres, + externalType, blip.DatabaseTypeMySQL, - blip.DatabaseTypePostgres, + externalType, } }, } @@ -130,7 +132,7 @@ func TestRegisterSupportsMultipleDatabaseTypes(t *testing.T) { for _, databaseType := range []blip.DatabaseType{ blip.DatabaseTypeMySQL, - blip.DatabaseTypePostgres, + externalType, } { if err := metrics.ValidateDatabase(domain, databaseType); err != nil { t.Fatalf("ValidateDatabase(%s): %v", databaseType, err) @@ -143,7 +145,7 @@ func TestRegisterSupportsMultipleDatabaseTypes(t *testing.T) { } if len(databaseTypes) != 2 || databaseTypes[0] != blip.DatabaseTypeMySQL || - databaseTypes[1] != blip.DatabaseTypePostgres { + databaseTypes[1] != externalType { t.Fatalf("SupportedDatabaseTypes = %v", databaseTypes) } @@ -163,9 +165,21 @@ func TestRegisterValidatesFactoryDatabaseTypes(t *testing.T) { databaseTypes: nil, errorContains: "supports no database types", }, - "invalid": { - databaseTypes: []blip.DatabaseType{"oracle"}, - errorContains: `declares invalid database type "oracle"`, + "empty value": { + databaseTypes: []blip.DatabaseType{""}, + errorContains: `declares invalid database type ""`, + }, + "whitespace": { + databaseTypes: []blip.DatabaseType{" oracle "}, + errorContains: `declares invalid database type " oracle "`, + }, + "uppercase": { + databaseTypes: []blip.DatabaseType{"Oracle"}, + errorContains: `declares invalid database type "Oracle"`, + }, + "neutral with specific": { + databaseTypes: []blip.DatabaseType{blip.DatabaseTypeAny, "oracle"}, + errorContains: "database-neutral compatibility with specific database types", }, } for name, tt := range tests { @@ -266,13 +280,14 @@ func TestBuiltInCollectorDatabaseCompatibility(t *testing.T) { if err := metrics.ValidateDatabase("status.global", blip.DatabaseTypeMySQL); err != nil { t.Fatalf("status.global with MySQL: %v", err) } - if err := metrics.ValidateDatabase("status.global", blip.DatabaseTypePostgres); err == nil { - t.Fatal("status.global supports PostgreSQL") + if err := metrics.ValidateDatabase("status.global", externalType); err == nil { + t.Fatal("status.global supports an external database") } for _, databaseType := range []blip.DatabaseType{ blip.DatabaseTypeMySQL, - blip.DatabaseTypePostgres, + externalType, + blip.DatabaseType("future-rds-engine"), } { if err := metrics.ValidateDatabase("aws.rds", databaseType); err != nil { t.Fatalf("aws.rds with %s: %v", databaseType, err) diff --git a/monitor/engine.go b/monitor/engine.go index a6ed281e..250c2e61 100644 --- a/monitor/engine.go +++ b/monitor/engine.go @@ -116,14 +116,14 @@ func (e *Engine) Prepare(ctx context.Context, plan blip.Plan, before, after func } }() - // Connect to MySQL. DO NOT loop and retry; try once and return on error + // Connect to the database. DO NOT loop and retry; try once and return on error // to let the caller (a LevelCollector.changePlan goroutine) retry with backoff. - status.Monitor(e.monitorId, status.ENGINE_PREPARE, "%s: connect to MySQL", plan.Name) + status.Monitor(e.monitorId, status.ENGINE_PREPARE, "%s: connect to database", plan.Name) dbctx, cancel := context.WithTimeout(ctx, 5*time.Second) err := e.db.PingContext(dbctx) cancel() if err != nil { - lerr = fmt.Errorf("while connecting to MySQL: %s", err) + lerr = fmt.Errorf("while connecting to database: %s", err) return lerr } diff --git a/monitor/level_collector.go b/monitor/level_collector.go index 4f932a22..fd88a61d 100644 --- a/monitor/level_collector.go +++ b/monitor/level_collector.go @@ -23,12 +23,12 @@ import ( // // The term "collector" is a little misleading because the LCO doesn't collect // metrics, but it is the first step in the metrics collection process, which -// looks roughly like: LCO -> Engine -> metric collectors -> MySQL. +// looks roughly like: LCO -> Engine -> metric collectors -> database. // In Run, the LCO checks every 1s for the highest level in the plan to collect. // For example, after 5s it'll collect levels with a frequency divisible by 5s. // See https://block.github.io/blip/plans/file/. // -// Metrics from MySQL flow back to the LCO as blip.Metrics, which the LCO +// Metrics from the database flow back to the LCO as blip.Metrics, which the LCO // passes to blip.Plugin.TransformMetrics if specified, then to all sinks // specified for the monitor. type LevelCollector interface { @@ -370,7 +370,7 @@ func (c *lco) ChangePlan(newState, newPlanName string) error { // changePlan is a gorountine run by ChangePlan It's potentially long-running // because it waits for Engine.Prepare. If that function returns an error -// (e.g. MySQL is offline), then this function retires forever, or until canceled +// (e.g. the database is offline), then this function retries forever, or until canceled // by either another call to ChangePlan or Run is stopped (LCO is terminated). // // Never all this function directly; it's only called via ChangePlan, which @@ -454,9 +454,9 @@ func (c *lco) changePlan(ctx context.Context, doneChan chan struct{}, newState, c.stateMux.Unlock() // -- X unlock -- } - // Try forever, or until context is cancelled, because it could be that MySQL is + // Try forever, or until context is cancelled, because it could be that the database is // temporarily offline. In the real world, this is not uncommon: Blip might be - // started before MySQL, for example. We're running in a goroutine from ChangePlan + // started before the database, for example. We're running in a goroutine from ChangePlan // that already returned to its caller, so we're not blocking anything here. // More importantly, as documented in several place: this is _the code_ that // all other code relies on to try "forever" because a plan must be prepared diff --git a/monitor/monitor.go b/monitor/monitor.go index bfbf281e..4f4db952 100644 --- a/monitor/monitor.go +++ b/monitor/monitor.go @@ -1,7 +1,7 @@ // Copyright 2024 Block, Inc. // Package monitor provides core Blip components that, together, monitor one -// MySQL instance. Most monitoring logic happens in the package, but package +// database target. Most monitoring logic happens in the package, but package // metrics is closely related: this latter actually collect metrics, but it // is driven by this package. Other Blip packages are mostly set up and support // of monitors. @@ -25,7 +25,7 @@ import ( "github.com/cashapp/blip/status" ) -// Monitor monitors one MySQL instance. The monitor is a high-level component +// Monitor monitors one database target. The monitor is a high-level component // that runs (and keeps running) four monitor subsystems: // - Plan changer (PCH) // - Level collector (LCO) @@ -37,7 +37,7 @@ import ( // If any subsystem crashes (returns for any reason or panics), the monitor // stops and restarts all subsystems. The monitor doesn't stop until Stop is // called. Consequently, if a monitor is not configured correctly (for example, -// it can't connect to MySQL), it tries and reports every forever. +// it can't connect to the database), it tries and reports every forever. // // Monitors are loaded, created, and initially started only by the MonitorLoader. // A monitor can be stopped and started (again) via the server API. @@ -90,7 +90,7 @@ type MonitorArgs struct { // NewMonitor creates a new Monitor with the given arguments. The caller must // call Boot then, if that does not return an error, Run to start monitoring -// the MySQL instance. +// the database target. func NewMonitor(args MonitorArgs) *Monitor { retry := backoff.NewExponentialBackOff() retry.MaxElapsedTime = 0 @@ -256,7 +256,7 @@ func (m *Monitor) startup() (err error) { m.runMux.Unlock() // ---------------------------------------------------------------------- - // Make DSN and *sql.DB. This does NOT connect to MySQL. + // Make DSN and *sql.DB. This does NOT connect to the database. for { status.Monitor(m.monitorId, status.MONITOR, "making DB/DSN (not connecting)") dbProvider, db, dsnRedacted, err := m.makeDB() @@ -280,7 +280,7 @@ func (m *Monitor) startup() (err error) { } // ---------------------------------------------------------------------- - // Load monitor plans, if any. This MIGHT connect to MySQL if the plan + // Load monitor plans, if any. This MIGHT connect to the database if the plan // is stored in a table. for { status.Monitor(m.monitorId, status.MONITOR, "loading plans") diff --git a/plan/loader.go b/plan/loader.go index af7dce16..10b4c690 100644 --- a/plan/loader.go +++ b/plan/loader.go @@ -609,21 +609,26 @@ func validatePlanDatabaseCompatibility(plan blip.Plan) error { commonTypes := map[blip.DatabaseType]bool{} domainTypes := make([]string, 0, len(domains)) - for i, domain := range domains { + hasDatabaseConstraint := false + for _, domain := range domains { supportedTypes, err := metrics.SupportedDatabaseTypes(domain) if err != nil { return err } domainTypes = append(domainTypes, fmt.Sprintf("%s=%v", domain, supportedTypes)) + if len(supportedTypes) == 1 && supportedTypes[0] == blip.DatabaseTypeAny { + continue + } supported := map[blip.DatabaseType]bool{} for _, databaseType := range supportedTypes { supported[databaseType] = true - if i == 0 { + if !hasDatabaseConstraint { commonTypes[databaseType] = true } } - if i == 0 { + if !hasDatabaseConstraint { + hasDatabaseConstraint = true continue } for databaseType := range commonTypes { @@ -633,7 +638,7 @@ func validatePlanDatabaseCompatibility(plan blip.Plan) error { } } - if len(commonTypes) == 0 { + if hasDatabaseConstraint && len(commonTypes) == 0 { return fmt.Errorf("collectors have no common database type: %s", strings.Join(domainTypes, ", ")) } return nil diff --git a/plan/loader_test.go b/plan/loader_test.go index a80a0365..51c145f2 100644 --- a/plan/loader_test.go +++ b/plan/loader_test.go @@ -19,6 +19,8 @@ import ( "github.com/cashapp/blip/test/mock" ) +const externalType blip.DatabaseType = "test-database" + // -------------------------------------------------------------------------- func TestLoadDefault(t *testing.T) { @@ -178,7 +180,7 @@ func (f planDatabaseTypesFactory) DatabaseTypes(string) []blip.DatabaseType { func TestSharedPlansValidateDatabaseCompatibilityPerMonitor(t *testing.T) { const ( mysqlDomain = "test.mysql-plan" - postgresDomain = "test.postgres-plan" + externalDomain = "test.external-plan" sharedDomain = "test.shared-plan" ) factory := mock.MetricFactory{} @@ -186,17 +188,14 @@ func TestSharedPlansValidateDatabaseCompatibilityPerMonitor(t *testing.T) { t.Fatal(err) } t.Cleanup(func() { metrics.Remove(mysqlDomain) }) - if err := metrics.Register(postgresDomain, planDatabaseTypesFactory{ - databaseTypes: []blip.DatabaseType{blip.DatabaseTypePostgres}, + if err := metrics.Register(externalDomain, planDatabaseTypesFactory{ + databaseTypes: []blip.DatabaseType{externalType}, }); err != nil { t.Fatal(err) } - t.Cleanup(func() { metrics.Remove(postgresDomain) }) + t.Cleanup(func() { metrics.Remove(externalDomain) }) if err := metrics.Register(sharedDomain, planDatabaseTypesFactory{ - databaseTypes: []blip.DatabaseType{ - blip.DatabaseTypeMySQL, - blip.DatabaseTypePostgres, - }, + databaseTypes: []blip.DatabaseType{blip.DatabaseTypeAny}, }); err != nil { t.Fatal(err) } @@ -218,10 +217,10 @@ func TestSharedPlansValidateDatabaseCompatibilityPerMonitor(t *testing.T) { } } mysqlPlan := newPlan("mysql-plan", mysqlDomain) - postgresPlan := newPlan("postgres-plan", postgresDomain) + externalPlan := newPlan("external-plan", externalDomain) pl := plan.NewLoader(func(blip.ConfigPlans) ([]blip.Plan, error) { - return []blip.Plan{mysqlPlan, postgresPlan}, nil + return []blip.Plan{mysqlPlan, externalPlan}, nil }) if err := pl.LoadShared(blip.ConfigPlans{}, nil); err != nil { t.Fatalf("LoadShared: %v", err) @@ -232,50 +231,47 @@ func TestSharedPlansValidateDatabaseCompatibilityPerMonitor(t *testing.T) { t.Fatalf("LoadMonitor(mysql): %v", err) } if err := pl.LoadMonitor(blip.ConfigMonitor{ - MonitorId: "postgres", - DatabaseType: blip.DatabaseTypePostgres, + MonitorId: "external", + DatabaseType: externalType, }, nil); err != nil { - t.Fatalf("LoadMonitor(postgres): %v", err) + t.Fatalf("LoadMonitor(external): %v", err) } if _, err := pl.Plan("mysql", mysqlPlan.Name, nil); err != nil { t.Fatalf("MySQL plan for MySQL monitor: %v", err) } - if _, err := pl.Plan("postgres", postgresPlan.Name, nil); err != nil { - t.Fatalf("PostgreSQL plan for PostgreSQL monitor: %v", err) + if _, err := pl.Plan("external", externalPlan.Name, nil); err != nil { + t.Fatalf("external plan for external monitor: %v", err) } - if _, err := pl.Plan("mysql", postgresPlan.Name, nil); err == nil || - !strings.Contains(err.Error(), `collector test.postgres-plan does not support database type "mysql" (supported: [postgres])`) { - t.Fatalf("PostgreSQL plan for MySQL monitor error = %v", err) + if _, err := pl.Plan("mysql", externalPlan.Name, nil); err == nil || + !strings.Contains(err.Error(), `collector test.external-plan does not support database type "mysql" (supported: [test-database])`) { + t.Fatalf("external plan for MySQL monitor error = %v", err) } - if _, err := pl.Plan("postgres", mysqlPlan.Name, nil); err == nil || - !strings.Contains(err.Error(), `collector test.mysql-plan does not support database type "postgres" (supported: [mysql])`) { - t.Fatalf("MySQL plan for PostgreSQL monitor error = %v", err) + if _, err := pl.Plan("external", mysqlPlan.Name, nil); err == nil || + !strings.Contains(err.Error(), `collector test.mysql-plan does not support database type "test-database" (supported: [mysql])`) { + t.Fatalf("MySQL plan for external monitor error = %v", err) } } func TestValidatePlansRejectsCollectorsWithoutCommonDatabaseType(t *testing.T) { const ( mysqlDomain = "test.no-common-mysql" - postgresDomain = "test.no-common-postgres" + externalDomain = "test.no-common-external" sharedDomain = "test.no-common-shared" ) if err := metrics.Register(mysqlDomain, mock.MetricFactory{}); err != nil { t.Fatal(err) } t.Cleanup(func() { metrics.Remove(mysqlDomain) }) - if err := metrics.Register(postgresDomain, planDatabaseTypesFactory{ - databaseTypes: []blip.DatabaseType{blip.DatabaseTypePostgres}, + if err := metrics.Register(externalDomain, planDatabaseTypesFactory{ + databaseTypes: []blip.DatabaseType{externalType}, }); err != nil { t.Fatal(err) } - t.Cleanup(func() { metrics.Remove(postgresDomain) }) + t.Cleanup(func() { metrics.Remove(externalDomain) }) if err := metrics.Register(sharedDomain, planDatabaseTypesFactory{ - databaseTypes: []blip.DatabaseType{ - blip.DatabaseTypeMySQL, - blip.DatabaseTypePostgres, - }, + databaseTypes: []blip.DatabaseType{blip.DatabaseTypeAny}, }); err != nil { t.Fatal(err) } @@ -291,10 +287,10 @@ func TestValidatePlansRejectsCollectorsWithoutCommonDatabaseType(t *testing.T) { sharedDomain: {}, }, }, - "postgres": { + "external": { Freq: "5s", Collect: map[string]blip.Domain{ - postgresDomain: {}, + externalDomain: {}, }, }, }, @@ -302,13 +298,36 @@ func TestValidatePlansRejectsCollectorsWithoutCommonDatabaseType(t *testing.T) { err := plan.ValidatePlans([]blip.Plan{mixedPlan}) if err == nil { - t.Fatal("mixed MySQL and PostgreSQL plan is valid") + t.Fatal("mixed MySQL and external plan is valid") } expected := "collectors have no common database type: " + + "test.no-common-external=[test-database], " + "test.no-common-mysql=[mysql], " + - "test.no-common-postgres=[postgres], " + - "test.no-common-shared=[mysql postgres]" + "test.no-common-shared=[*]" if !strings.Contains(err.Error(), expected) { t.Fatalf("ValidatePlans error = %v", err) } } + +func TestValidatePlansAllowsDatabaseNeutralCollectors(t *testing.T) { + const domain = "test.database-neutral-plan" + if err := metrics.Register(domain, planDatabaseTypesFactory{ + databaseTypes: []blip.DatabaseType{blip.DatabaseTypeAny}, + }); err != nil { + t.Fatal(err) + } + t.Cleanup(func() { metrics.Remove(domain) }) + + neutralPlan := blip.Plan{ + Name: "database-neutral-plan", + Levels: map[string]blip.Level{ + "level": { + Freq: "1s", + Collect: map[string]blip.Domain{domain: {}}, + }, + }, + } + if err := plan.ValidatePlans([]blip.Plan{neutralPlan}); err != nil { + t.Fatalf("database-neutral plan is invalid: %v", err) + } +}