diff --git a/aws/iam_auth.go b/aws/iam_auth.go index 4e70e6f..1cffefb 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/blip.go b/blip.go index 62c0ef6..48d6886 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 ace30e1..55dd408 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 1ac52de..7df483e 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,33 +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) + databaseType := c.EffectiveDatabaseType() + if databaseType == DatabaseTypeMySQL { + if len(c.DatabaseConfig) > 0 { + return fmt.Errorf("config.monitor.database-config requires an external database type") } - 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) - } - 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 } @@ -497,13 +516,17 @@ 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 } func (c *ConfigMonitor) ApplyDefaults(b Config) { - if c.EffectiveDatabaseType() == DatabaseTypeMySQL { + databaseType := c.EffectiveDatabaseType() + if databaseType == DatabaseTypeMySQL { if c.Socket == "" { c.Socket = b.MySQL.Socket } @@ -545,15 +568,16 @@ func (c *ConfigMonitor) ApplyDefaults(b Config) { c.Sinks = ConfigSinks{} } c.AWS.ApplyDefaults(b) - c.Exporter.ApplyDefaults(b) c.HA.ApplyDefaults(b) - c.Heartbeat.ApplyDefaults(b) - c.Plans.ApplyDefaults(b) - if c.EffectiveDatabaseType() == DatabaseTypePostgres { - postgresDefaults := DefaultConfigPostgres() - postgresDefaults.ConnectTimeout = c.TimeoutConnect - c.Postgres.ApplyDefaults(postgresDefaults) - } + // 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) c.Sinks.ApplyDefaults(b) c.TLS.ApplyDefaults(b) } @@ -580,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() } @@ -606,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) } @@ -658,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 "" } @@ -735,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{} } @@ -798,6 +812,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" ) @@ -966,6 +984,15 @@ func DefaultConfigPlans() ConfigPlans { } func (c ConfigPlans) Validate() error { + 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 } @@ -986,11 +1013,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() { @@ -1079,6 +1112,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_database_test.go b/config_database_test.go new file mode 100644 index 0000000..947ec65 --- /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 7c3d9f4..0000000 --- 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 8a38a17..0000000 --- a/config_postgres_test.go +++ /dev/null @@ -1,243 +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", - }, - } - - 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 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 b1ddf89..e6d3202 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,7 +48,10 @@ 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) + 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) { password, err := token.Password(ctx) if err != nil { diff --git a/credentials/credentials_test.go b/credentials/credentials_test.go index 0b54f08..f388d48 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 { @@ -22,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) } @@ -79,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") } @@ -93,11 +104,98 @@ func TestDynamicPreservesSourcePrecedence(t *testing.T) { } } +func TestDynamicIAMUsesCallerDefaultPort(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 + defaultPort string + hostname string + wantHost string + }{ + { + name: "MySQL default", + defaultPort: "3306", + hostname: "mysql.example", + wantHost: "mysql.example:3306", + }, + { + name: "external engine default", + defaultPort: "5432", + hostname: "postgres.example", + wantHost: "postgres.example:5432", + }, + { + 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{ + Hostname: tt.hostname, + Username: "metrics", + AWS: blip.ConfigAWS{ + IAMAuth: &iamAuth, + Region: "us-west-2", + }, + }, tt.defaultPort) + 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 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 0000000..c7dbaa6 --- /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 dd148a3..0feb2f0 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 3848a63..1bb5d7d 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 cb0c69d..aa7755b 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. External database 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/develop/collectors.md b/docs/content/develop/collectors.md index 807b2b0..7bc2999 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 0000000..9024bb4 --- /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 1394cdc..b5dbaf9 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. 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| |-----|------------------|------------------|-----------| |`offline`|no|no|Completely offline, no connection to MySQL| diff --git a/docs/content/plans/file.md b/docs/content/plans/file.md index 7e4eb4a..7fe90c2 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 05dfca4..5d6c3dd 100644 --- a/docs/content/plans/table.md +++ b/docs/content/plans/table.md @@ -4,6 +4,13 @@ title: "Table" 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 external database monitors reject +`plans.table` in their monitor configuration. External database monitors can +use plan files or compatible shared plans. +{{< /hint >}} + ```sql CREATE TABLE IF NOT EXISTS plans ( name VARCHAR(100) NOT NULL PRIMARY KEY, diff --git a/metrics/factory.go b/metrics/factory.go index 81b890d..c76d1ee 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 a443c63..825e6c7 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 9fe8097..250c2e6 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 { @@ -115,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 } @@ -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() @@ -440,16 +441,26 @@ 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()) + if cl.cancel != nil { + 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() + 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() + cl.cleanup = nil } cl.Unlock() } diff --git a/monitor/level_collector.go b/monitor/level_collector.go index 2b80a73..fd88a61 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 { @@ -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() { @@ -361,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 @@ -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) @@ -442,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 @@ -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/monitor.go b/monitor/monitor.go index 14b23d3..4f4db95 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. @@ -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 @@ -89,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 @@ -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) @@ -255,7 +256,7 @@ func (m *Monitor) startup() 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() @@ -279,7 +280,7 @@ func (m *Monitor) startup() 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") @@ -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 6827914..e13df1b 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 0f6b49b..1215bfd 100644 --- a/monitor/provider_test.go +++ b/monitor/provider_test.go @@ -8,10 +8,14 @@ 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" "github.com/cashapp/blip/test/mock" ) @@ -131,6 +135,323 @@ 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 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 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) @@ -196,6 +517,7 @@ func TestEnginePassesProviderToCollectorFactory(t *testing.T) { type testDBProvider struct { primary *sql.DB closeCalls int + closeFunc func() } func (p *testDBProvider) Primary() *sql.DB { @@ -203,6 +525,9 @@ func (p *testDBProvider) Primary() *sql.DB { } func (p *testDBProvider) Close() error { + if p.closeFunc != nil { + p.closeFunc() + } p.closeCalls++ return nil } diff --git a/plan/loader.go b/plan/loader.go index af7dce1..10b4c69 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 a80a036..51c145f 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) + } +}