diff --git a/CHANGELOG.md b/CHANGELOG.md index 9b14789..96aa5b8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,29 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Added + +- **One metrics endpoint for the whole container ([#135]).** The main + `/metrics` response can now tell the complete story in a single scrape: + - The embedded runtime PHP-FPM autotuner's `fpm_tune_*` series always + appear on the main endpoint while the tuner runs — no second listener + required. `fpm_tune.metrics_addr` stays as an optional extra listener + for standalone-tool parity. + - `global.metrics_federate` declares local exporters whose exposition is + appended to every scrape (per-source timeout and TTL cache, 8 MiB body + cap). Each source contributes `cbox_init_federate_up{name}`; one that is + down degrades to 0 there instead of failing the scrape. URLs are + validated to loopback only — federation merges exporters inside the + container, it is not a proxy. + - With federation enabled the endpoint serves the plain-text exposition + format unconditionally, since federated bodies are appended verbatim. + See `configs/examples/metrics-federate.yaml` and + [Prometheus Metrics](docs/observability/metrics.md). + +[#135]: https://github.com/cboxdk/init/issues/135 + ## [3.1.2] - 2026-09-04 ### Fixed diff --git a/cmd/cbox-init/fpmtune.go b/cmd/cbox-init/fpmtune.go index 92b987c..742c3e8 100644 --- a/cmd/cbox-init/fpmtune.go +++ b/cmd/cbox-init/fpmtune.go @@ -10,6 +10,7 @@ import ( "github.com/cboxdk/fpm-tune/state" "github.com/cboxdk/init/internal/config" + "github.com/prometheus/client_golang/prometheus" ) // The p95 hybrid is the intended default sizing basis: size on the 95th @@ -34,10 +35,13 @@ const ( // are up, is fine even if php-fpm is still coming up. Its config is loaded per // round, so a pool's boot-time pm.max_children (set by the calculator before // php-fpm started) is the seed it refines, not something it fights. -func startFPMTune(ctx context.Context, cfg *config.Config, log *slog.Logger) (func(), error) { +// It also returns the tuner's Prometheus registry so the caller can merge the +// fpm_tune_* series onto the main metrics endpoint — one scrape, one story — +// regardless of whether the optional metrics_addr listener is configured. +func startFPMTune(ctx context.Context, cfg *config.Config, log *slog.Logger) (func(), *prometheus.Registry, error) { ft := cfg.Global.FPMTune if ft == nil || !ft.Enabled { - return nil, nil + return nil, nil, nil } sc := serve.Config{ @@ -62,7 +66,7 @@ func startFPMTune(ctx context.Context, cfg *config.Config, log *slog.Logger) (fu loop, err := serve.New(sc, log) if err != nil { - return nil, fmt.Errorf("fpm-tune: %w", err) + return nil, nil, fmt.Errorf("fpm-tune: %w", err) } // The loop gets its own context so shutdown can stop it independently: the @@ -90,7 +94,7 @@ func startFPMTune(ctx context.Context, cfg *config.Config, log *slog.Logger) (fu <-done // Run's deferred Close() releases the state lock and saves baselines. } - return stop, nil + return stop, loop.Metrics().Registry, nil } // resolveFPMWorkload maps the configured workload name to a class, warning on an diff --git a/cmd/cbox-init/fpmtune_test.go b/cmd/cbox-init/fpmtune_test.go index 3f8b8e7..c0f5b83 100644 --- a/cmd/cbox-init/fpmtune_test.go +++ b/cmd/cbox-init/fpmtune_test.go @@ -20,7 +20,7 @@ func discardLogger() *slog.Logger { func TestStartFPMTuneDisabled(t *testing.T) { for _, ft := range []*config.FPMTuneConfig{nil, {Enabled: false}} { cfg := &config.Config{Global: config.GlobalConfig{FPMTune: ft}} - stop, err := startFPMTune(context.Background(), cfg, discardLogger()) + stop, _, err := startFPMTune(context.Background(), cfg, discardLogger()) if err != nil { t.Fatalf("disabled autotuner returned an error: %v", err) } @@ -49,7 +49,7 @@ func TestStartFPMTuneStartsAndStops(t *testing.T) { }, } - stop, err := startFPMTune(context.Background(), cfg, discardLogger()) + stop, _, err := startFPMTune(context.Background(), cfg, discardLogger()) if err != nil { t.Fatalf("startFPMTune: %v", err) } @@ -70,7 +70,7 @@ func TestStartFPMTuneStartsAndStops(t *testing.T) { } // The state lock is released, so a second loop on the same state path starts. - stop2, err := startFPMTune(context.Background(), cfg, discardLogger()) + stop2, _, err := startFPMTune(context.Background(), cfg, discardLogger()) if err != nil { t.Fatalf("second start after stop failed (lock not released?): %v", err) } diff --git a/cmd/cbox-init/serve.go b/cmd/cbox-init/serve.go index 5163a0b..43b9486 100644 --- a/cmd/cbox-init/serve.go +++ b/cmd/cbox-init/serve.go @@ -392,10 +392,17 @@ func runServe(cmd *cobra.Command, args []string) { // startup step that can os.Exit, so its stop is never skipped. Non-critical: // if it cannot start (for example a second copy already holds the state lock), // php-fpm keeps its boot-time size and the container runs on. - stopFPMTune, err := startFPMTune(ctx, cfg, log) + stopFPMTune, fpmTuneRegistry, err := startFPMTune(ctx, cfg, log) if err != nil { slog.Warn("Runtime PHP-FPM autotuner not started", "error", err) } + if fpmTuneRegistry != nil && metricsServer != nil { + // One scrape, one story: the tuner's fpm_tune_* series ride on the + // main metrics endpoint; the separate metrics_addr listener stays + // optional for standalone-tool parity. + metricsServer.AddGatherer(fpmTuneRegistry) + slog.Info("fpm-tune metrics merged onto the main metrics endpoint") + } // Main event loop - handles shutdown signals and config reloads var shutdownReason string @@ -684,6 +691,14 @@ func startMetricsServer(ctx context.Context, cfg *config.Config, log *slog.Logge server := metrics.NewServer(metricsPort, metricsPath, cfg.Global.MetricsACL, cfg.Global.MetricsTLS, log) server.SetBindHost(cfg.Global.MetricsHost) + if len(cfg.Global.MetricsFederate) > 0 { + sources := make([]metrics.FederateSource, len(cfg.Global.MetricsFederate)) + for i, fs := range cfg.Global.MetricsFederate { + sources[i] = metrics.FederateSource{Name: fs.Name, URL: fs.URL, Timeout: fs.Timeout, CacheTTL: fs.CacheTTL} + } + server.SetFederator(metrics.NewFederator(sources, log)) + slog.Info("Metrics federation enabled", "sources", len(sources)) + } if err := server.Start(ctx); err != nil { slog.Warn("Failed to start metrics server (continuing without metrics)", "error", err) return nil diff --git a/configs/examples/metrics-federate.yaml b/configs/examples/metrics-federate.yaml new file mode 100644 index 0000000..ad09a6a --- /dev/null +++ b/configs/examples/metrics-federate.yaml @@ -0,0 +1,54 @@ +# One scrape, one story. +# +# A PHP container easily grows three metrics endpoints: cbox-init's own +# (supervision), the embedded fpm-tune's (capacity), and an application +# exporter's (php-fpm status, queues). Federation folds the local exporters +# into cbox-init's main /metrics response, and since 3.2.0 the embedded +# fpm-tune series are always there natively - so Prometheus scrapes ONE port +# and sees the whole story: +# +# cbox_init_* - is everything running? +# fpm_tune_* - does the workload fit this container? (vertical) +# phpfpm_* - is this container saturated? (horizontal - listen_queue) +# +# A source that is down contributes cbox_init_federate_up{name} 0 instead of +# failing the scrape. URLs must be loopback: federation is for exporters +# inside the container, not a proxy. + +version: "1.0" + +global: + shutdown_timeout: 30 + log_format: json + + metrics_enabled: true + metrics_port: 9090 + + metrics_federate: + - name: fpm-exporter + url: http://127.0.0.1:9114/metrics + # timeout: 2s # per-fetch timeout (default) + # cache_ttl: 5s # serve the cached body this long between fetches (default) + + # The runtime autotuner's fpm_tune_* series ride on the main endpoint + # automatically; metrics_addr remains optional for standalone-tool parity. + fpm_tune: + enabled: true + mode: apply + interval: 30s + +processes: + php-fpm: + enabled: true + command: ["php-fpm", "-F", "-R"] + restart: always + + fpm-exporter: + enabled: true + command: ["fpm-exporter", "serve"] + restart: always + depends_on: + - php-fpm + health_check: + type: tcp + address: "127.0.0.1:9114" diff --git a/docs/observability/metrics.md b/docs/observability/metrics.md index d8a0a3d..c3abe86 100644 --- a/docs/observability/metrics.md +++ b/docs/observability/metrics.md @@ -19,6 +19,49 @@ global: metrics_path: /metrics ``` +## One Endpoint: Federation and Embedded Engines + +Since 3.2.0 the main `/metrics` response can carry the whole container's +story, so Prometheus scrapes one port: + +- **Embedded engines merge natively.** The runtime PHP-FPM autotuner's + `fpm_tune_*` series always appear on the main endpoint when `fpm_tune` is + enabled. `fpm_tune.metrics_addr` remains optional, for parity with the + standalone tool. +- **Local exporters federate.** Declare them under + `global.metrics_federate` and their exposition is appended to every + scrape, with a short cache so heavy scraping does not multiply load: + +```yaml +global: + metrics_federate: + - name: fpm-exporter + url: http://127.0.0.1:9114/metrics + timeout: 2s # per-fetch (default) + cache_ttl: 5s # cached between fetches (default) +``` + +Each source contributes a health gauge; a source that is down degrades +instead of failing the scrape: + +```text +cbox_init_federate_up{name="fpm-exporter"} 1 +``` + +Rules and caveats: + +- URLs must point at **loopback** (`127.0.0.1`, `::1`, `localhost`) — + federation merges exporters inside the container; it is not a proxy, and + config validation rejects anything else. +- Bodies over 8 MiB and non-200 responses count as down. +- Metric names must not collide across sources — federation appends + expositions verbatim and does not rewrite names. +- With federation enabled the endpoint always serves the plain-text + exposition format (no content negotiation), because federated bodies are + appended as-is. + +A complete example lives in `configs/examples/metrics-federate.yaml`. + ## Available Metrics ### Process Lifecycle Metrics diff --git a/go.mod b/go.mod index 9d60a88..f776466 100644 --- a/go.mod +++ b/go.mod @@ -16,6 +16,8 @@ require ( github.com/charmbracelet/lipgloss v1.1.0 github.com/fsnotify/fsnotify v1.10.1 github.com/prometheus/client_golang v1.24.1 + github.com/prometheus/client_model v0.6.2 + github.com/prometheus/common v0.70.1 github.com/robfig/cron/v3 v3.0.1 github.com/shirou/gopsutil/v4 v4.26.7 github.com/spf13/cobra v1.10.2 @@ -62,8 +64,6 @@ require ( github.com/muesli/termenv v0.16.0 // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/power-devops/perfstat v0.0.0-20240221224432-82ca36839d55 // indirect - github.com/prometheus/client_model v0.6.2 // indirect - github.com/prometheus/common v0.70.1 // indirect github.com/prometheus/procfs v0.21.1 // indirect github.com/rivo/uniseg v0.4.7 // indirect github.com/shirou/gopsutil/v3 v3.24.5 // indirect diff --git a/internal/config/federate_test.go b/internal/config/federate_test.go new file mode 100644 index 0000000..28b5f6e --- /dev/null +++ b/internal/config/federate_test.go @@ -0,0 +1,65 @@ +package config + +import ( + "strings" + "testing" +) + +func federateConfig(sources ...FederateSourceConfig) *Config { + cfg := &Config{ + Version: "1.0", + Processes: map[string]*Process{ + "dummy": {Command: []string{"sleep", "1"}}, + }, + } + cfg.Global.MetricsFederate = sources + cfg.SetDefaults() + return cfg +} + +func validationErrors(cfg *Config) []string { + res, _ := cfg.ValidateComprehensive() + var msgs []string + for _, iss := range res.Errors { + msgs = append(msgs, iss.Field+": "+iss.Message) + } + return msgs +} + +func TestFederateValidationAcceptsLoopback(t *testing.T) { + for _, url := range []string{ + "http://127.0.0.1:9114/metrics", + "http://localhost:9114/metrics", + "http://[::1]:9114/metrics", + } { + cfg := federateConfig(FederateSourceConfig{Name: "ok", URL: url}) + if errs := validationErrors(cfg); len(errs) != 0 { + t.Fatalf("loopback URL %q rejected: %v", url, errs) + } + } +} + +func TestFederateValidationRejectsNonLoopback(t *testing.T) { + cfg := federateConfig(FederateSourceConfig{Name: "bad", URL: "http://10.0.0.5:9114/metrics"}) + errs := validationErrors(cfg) + if len(errs) == 0 { + t.Fatal("non-loopback URL must be rejected - federation is not a proxy") + } + if !strings.Contains(strings.Join(errs, " "), "Non-loopback") { + t.Fatalf("unexpected errors: %v", errs) + } +} + +func TestFederateValidationRejectsDuplicateNamesAndMissingFields(t *testing.T) { + cfg := federateConfig( + FederateSourceConfig{Name: "dup", URL: "http://127.0.0.1:1/metrics"}, + FederateSourceConfig{Name: "dup", URL: "http://127.0.0.1:2/metrics"}, + FederateSourceConfig{URL: "ftp://127.0.0.1/metrics"}, + ) + errs := strings.Join(validationErrors(cfg), " | ") + for _, want := range []string{"Duplicate", "needs a name", "Unsupported scheme"} { + if !strings.Contains(errs, want) { + t.Fatalf("missing %q in: %s", want, errs) + } + } +} diff --git a/internal/config/types.go b/internal/config/types.go index 0b11726..a147494 100644 --- a/internal/config/types.go +++ b/internal/config/types.go @@ -18,39 +18,40 @@ type Config struct { // GlobalConfig contains global settings for the process manager type GlobalConfig struct { - ShutdownTimeout int `yaml:"shutdown_timeout" json:"shutdown_timeout"` // seconds - HealthCheckInterval int `yaml:"health_check_interval" json:"health_check_interval"` // seconds - RestartPolicy string `yaml:"restart_policy" json:"restart_policy"` // always | on-failure | never - MaxRestartAttempts int `yaml:"max_restart_attempts" json:"max_restart_attempts"` // - RestartBackoff int `yaml:"restart_backoff" json:"restart_backoff"` // seconds (legacy, prefer restart_backoff_initial/max) - RestartBackoffInitial time.Duration `yaml:"restart_backoff_initial" json:"restart_backoff_initial"` // initial duration (supports "5s" style) - RestartBackoffMax time.Duration `yaml:"restart_backoff_max" json:"restart_backoff_max"` // max duration - RestartStabilityWindow time.Duration `yaml:"restart_stability_window" json:"restart_stability_window"` // uptime after which the restart budget resets (default 60s; negative disables) - AutotuneMemoryThreshold float64 `yaml:"autotune_memory_threshold" json:"autotune_memory_threshold"` // 0.0-2.0, overrides profile MaxMemoryUsage - AutotuneStrict bool `yaml:"autotune_strict" json:"autotune_strict"` // Fail boot (exit PID 1) when the profile does not fit; default false clamps and boots - LogFormat string `yaml:"log_format" json:"log_format"` // json | text - LogLevel string `yaml:"log_level" json:"log_level"` // debug | info | warn | error - LogTimestamps bool `yaml:"log_timestamps" json:"log_timestamps"` // - MetricsEnabled *bool `yaml:"metrics_enabled" json:"metrics_enabled"` // - MetricsPort int `yaml:"metrics_port" json:"metrics_port"` // - MetricsPath string `yaml:"metrics_path" json:"metrics_path"` // - MetricsHost string `yaml:"metrics_host" json:"metrics_host"` // Bind host for metrics (default: all interfaces) - APIEnabled *bool `yaml:"api_enabled" json:"api_enabled"` // - APIPort int `yaml:"api_port" json:"api_port"` // - APIHost string `yaml:"api_host" json:"api_host"` // Bind host for the management API (default: 127.0.0.1, loopback-only; set 0.0.0.0 to expose — requires api_auth or api_acl) - APISocket string `yaml:"api_socket" json:"api_socket"` // Unix socket path (e.g. /var/run/cbox-init.sock) - APIAuth string `yaml:"api_auth" json:"api_auth"` // Bearer token - APITLS *TLSConfig `yaml:"api_tls" json:"api_tls"` // TLS configuration for API - APIACL *ACLConfig `yaml:"api_acl" json:"api_acl"` // IP ACL for API - MetricsTLS *TLSConfig `yaml:"metrics_tls" json:"metrics_tls"` // TLS configuration for metrics - MetricsACL *ACLConfig `yaml:"metrics_acl" json:"metrics_acl"` // IP ACL for metrics - ResourceMetricsEnabled *bool `yaml:"resource_metrics_enabled" json:"resource_metrics_enabled"` // Enable CPU/RAM collection - ResourceMetricsInterval int `yaml:"resource_metrics_interval" json:"resource_metrics_interval"` // seconds (default: 5) - ResourceMetricsMaxSamples int `yaml:"resource_metrics_max_samples" json:"resource_metrics_max_samples"` // Per-instance buffer size (default: 720 = 1h at 5s) - AuditEnabled bool `yaml:"audit_enabled" json:"audit_enabled"` // Enable audit logging - TracingEnabled bool `yaml:"tracing_enabled" json:"tracing_enabled"` // Enable distributed tracing - TracingExporter string `yaml:"tracing_exporter" json:"tracing_exporter"` // otlp-grpc | stdout - TracingEndpoint string `yaml:"tracing_endpoint" json:"tracing_endpoint"` // Exporter endpoint (e.g., localhost:4317) + ShutdownTimeout int `yaml:"shutdown_timeout" json:"shutdown_timeout"` // seconds + HealthCheckInterval int `yaml:"health_check_interval" json:"health_check_interval"` // seconds + RestartPolicy string `yaml:"restart_policy" json:"restart_policy"` // always | on-failure | never + MaxRestartAttempts int `yaml:"max_restart_attempts" json:"max_restart_attempts"` // + RestartBackoff int `yaml:"restart_backoff" json:"restart_backoff"` // seconds (legacy, prefer restart_backoff_initial/max) + RestartBackoffInitial time.Duration `yaml:"restart_backoff_initial" json:"restart_backoff_initial"` // initial duration (supports "5s" style) + RestartBackoffMax time.Duration `yaml:"restart_backoff_max" json:"restart_backoff_max"` // max duration + RestartStabilityWindow time.Duration `yaml:"restart_stability_window" json:"restart_stability_window"` // uptime after which the restart budget resets (default 60s; negative disables) + AutotuneMemoryThreshold float64 `yaml:"autotune_memory_threshold" json:"autotune_memory_threshold"` // 0.0-2.0, overrides profile MaxMemoryUsage + AutotuneStrict bool `yaml:"autotune_strict" json:"autotune_strict"` // Fail boot (exit PID 1) when the profile does not fit; default false clamps and boots + LogFormat string `yaml:"log_format" json:"log_format"` // json | text + LogLevel string `yaml:"log_level" json:"log_level"` // debug | info | warn | error + LogTimestamps bool `yaml:"log_timestamps" json:"log_timestamps"` // + MetricsEnabled *bool `yaml:"metrics_enabled" json:"metrics_enabled"` // + MetricsPort int `yaml:"metrics_port" json:"metrics_port"` // + MetricsPath string `yaml:"metrics_path" json:"metrics_path"` // + MetricsHost string `yaml:"metrics_host" json:"metrics_host"` // Bind host for metrics (default: all interfaces) + APIEnabled *bool `yaml:"api_enabled" json:"api_enabled"` // + APIPort int `yaml:"api_port" json:"api_port"` // + APIHost string `yaml:"api_host" json:"api_host"` // Bind host for the management API (default: 127.0.0.1, loopback-only; set 0.0.0.0 to expose — requires api_auth or api_acl) + APISocket string `yaml:"api_socket" json:"api_socket"` // Unix socket path (e.g. /var/run/cbox-init.sock) + APIAuth string `yaml:"api_auth" json:"api_auth"` // Bearer token + APITLS *TLSConfig `yaml:"api_tls" json:"api_tls"` // TLS configuration for API + APIACL *ACLConfig `yaml:"api_acl" json:"api_acl"` // IP ACL for API + MetricsTLS *TLSConfig `yaml:"metrics_tls" json:"metrics_tls"` // TLS configuration for metrics + MetricsACL *ACLConfig `yaml:"metrics_acl" json:"metrics_acl"` // IP ACL for metrics + MetricsFederate []FederateSourceConfig `yaml:"metrics_federate" json:"metrics_federate"` // Local exporters merged into the main /metrics response (see FederateSourceConfig) + ResourceMetricsEnabled *bool `yaml:"resource_metrics_enabled" json:"resource_metrics_enabled"` // Enable CPU/RAM collection + ResourceMetricsInterval int `yaml:"resource_metrics_interval" json:"resource_metrics_interval"` // seconds (default: 5) + ResourceMetricsMaxSamples int `yaml:"resource_metrics_max_samples" json:"resource_metrics_max_samples"` // Per-instance buffer size (default: 720 = 1h at 5s) + AuditEnabled bool `yaml:"audit_enabled" json:"audit_enabled"` // Enable audit logging + TracingEnabled bool `yaml:"tracing_enabled" json:"tracing_enabled"` // Enable distributed tracing + TracingExporter string `yaml:"tracing_exporter" json:"tracing_exporter"` // otlp-grpc | stdout + TracingEndpoint string `yaml:"tracing_endpoint" json:"tracing_endpoint"` // Exporter endpoint (e.g., localhost:4317) // TracingSampleRate is a pointer so an explicit 0.0 — the documented way to // sample nothing — can be told from an absent key. Treating 0 as "unset" // turned it into 100% sampling, the exact opposite of what was asked for. @@ -328,10 +329,24 @@ type FPMTuneConfig struct { DropInDir string `yaml:"drop_in_dir" json:"drop_in_dir"` // Where pool drop-ins are written; empty = the directory the master includes StatePath string `yaml:"state_path" json:"state_path"` // Where learned baselines persist (empty = fpm-tune's default) BackupDir string `yaml:"backup_dir" json:"backup_dir"` // Rollback / self-repair directory (empty = fpm-tune's default) - MetricsAddr string `yaml:"metrics_addr" json:"metrics_addr"` // Address for fpm-tune's own /metrics, e.g. ":9110" (empty disables it) + MetricsAddr string `yaml:"metrics_addr" json:"metrics_addr"` // OPTIONAL separate listener for fpm-tune's own /metrics, e.g. ":9110". Since 3.2.0 the fpm_tune_* series are always on the main metrics endpoint too; this exists for standalone-tool parity RecommendPath string `yaml:"recommend_path" json:"recommend_path"` // Advisory mode: write the plan here for copying by hand (empty disables it) } +// FederateSourceConfig declares one local metrics endpoint whose exposition is +// merged into the main /metrics response, so one scrape of cbox-init tells the +// whole story (supervision + capacity + application metrics). Each source +// contributes a cbox_init_federate_up{name} gauge; a source that is down +// degrades to 0 there instead of failing the scrape. URLs must point at +// loopback: federation is for exporters INSIDE the container, and anything +// else would turn the metrics port into a proxy. +type FederateSourceConfig struct { + Name string `yaml:"name" json:"name"` // Label value for cbox_init_federate_up; required, unique + URL string `yaml:"url" json:"url"` // Loopback http(s) URL, e.g. http://127.0.0.1:9114/metrics + Timeout time.Duration `yaml:"timeout" json:"timeout"` // Per-fetch timeout (default 2s) + CacheTTL time.Duration `yaml:"cache_ttl" json:"cache_ttl"` // Serve a cached body this long between fetches (default 5s) +} + // setGlobalDefaults sets default values for global configuration func (c *Config) setGlobalDefaults() { c.setGlobalBasicDefaults() diff --git a/internal/config/validation.go b/internal/config/validation.go index f4c9950..0faf5b0 100644 --- a/internal/config/validation.go +++ b/internal/config/validation.go @@ -2,6 +2,8 @@ package config import ( "fmt" + "net" + "net/url" "os" "path/filepath" "runtime" @@ -186,6 +188,7 @@ func (c *Config) validateGlobalSettings(result *ValidationResult) { c.validateGlobalLimits(result) c.validateGlobalAPISettings(result) c.validateGlobalMetricsSettings(result) + c.validateGlobalMetricsFederate(result) c.validateGlobalReadinessSettings(result) c.validateGlobalFPMTuneSettings(result) } @@ -370,6 +373,47 @@ func (c *Config) validateGlobalMetricsSettings(result *ValidationResult) { } } +// validateGlobalMetricsFederate validates the federated-source declarations. +// URLs are restricted to loopback: federation exists to merge exporters that +// run INSIDE the container onto one endpoint — a non-loopback URL would turn +// the metrics port into an open proxy for whatever the operator points it at. +func (c *Config) validateGlobalMetricsFederate(result *ValidationResult) { + seen := make(map[string]bool, len(c.Global.MetricsFederate)) + for i, src := range c.Global.MetricsFederate { + field := fmt.Sprintf("global.metrics_federate[%d]", i) + if src.Name == "" { + result.AddError(field+".name", "Federated source needs a name", "It labels cbox_init_federate_up for this source") + } else if seen[src.Name] { + result.AddError(field+".name", fmt.Sprintf("Duplicate federated source name %q", src.Name), "Names must be unique") + } + seen[src.Name] = true + + u, err := url.Parse(src.URL) + if err != nil || u.Scheme == "" || u.Host == "" { + result.AddError(field+".url", fmt.Sprintf("Invalid URL %q", src.URL), "Use e.g. http://127.0.0.1:9114/metrics") + continue + } + if u.Scheme != "http" && u.Scheme != "https" { + result.AddError(field+".url", fmt.Sprintf("Unsupported scheme %q", u.Scheme), "Only http and https are federated") + continue + } + host := u.Hostname() + if ip := net.ParseIP(host); ip != nil { + if !ip.IsLoopback() { + result.AddError(field+".url", fmt.Sprintf("Non-loopback address %q", host), "Federation is for exporters inside the container; use 127.0.0.1 or ::1") + } + } else if host != "localhost" { + result.AddError(field+".url", fmt.Sprintf("Non-loopback host %q", host), "Federation is for exporters inside the container; use 127.0.0.1, ::1 or localhost") + } + if src.Timeout < 0 { + result.AddError(field+".timeout", "Timeout cannot be negative", "Omit it for the 2s default") + } + if src.CacheTTL < 0 { + result.AddError(field+".cache_ttl", "cache_ttl cannot be negative", "Omit it for the 5s default") + } + } +} + // validateGlobalReadinessSettings validates readiness file configuration for Kubernetes func (c *Config) validateGlobalReadinessSettings(result *ValidationResult) { if c.Global.Readiness == nil || !c.Global.Readiness.Enabled { diff --git a/internal/metrics/federate.go b/internal/metrics/federate.go new file mode 100644 index 0000000..ff16dd4 --- /dev/null +++ b/internal/metrics/federate.go @@ -0,0 +1,186 @@ +package metrics + +import ( + "bytes" + "context" + "fmt" + "io" + "log/slog" + "net/http" + "strings" + "sync" + "time" +) + +// Federation limits. A local exporter that suddenly returns hundreds of +// megabytes should degrade to "down", not balloon every scrape of cbox-init. +const ( + federateMaxBodyBytes = 8 << 20 // 8 MiB per source + federateDefaultTimeout = 2 * time.Second + federateDefaultCacheTTL = 5 * time.Second +) + +// FederateSource is one local exporter merged into the main /metrics response. +type FederateSource struct { + Name string + URL string + Timeout time.Duration + CacheTTL time.Duration +} + +type federateEntry struct { + mu sync.Mutex + body []byte + fetchedAt time.Time + up bool +} + +// Federator fetches declared local exporters and appends their exposition to +// the main metrics response. Each source is cached for its TTL so a busy +// Prometheus (or several) does not multiply load onto small exporters, and a +// source that is down contributes cbox_init_federate_up{name} 0 instead of +// failing the scrape. +type Federator struct { + sources []FederateSource + entries []*federateEntry + client *http.Client + logger *slog.Logger +} + +// NewFederator builds a federator over the declared sources. Defaults are +// applied here (not in config) so the zero-value config stays honest. +func NewFederator(sources []FederateSource, logger *slog.Logger) *Federator { + entries := make([]*federateEntry, len(sources)) + maxTimeout := time.Duration(0) + for i := range sources { + if sources[i].Timeout <= 0 { + sources[i].Timeout = federateDefaultTimeout + } + if sources[i].CacheTTL <= 0 { + sources[i].CacheTTL = federateDefaultCacheTTL + } + if sources[i].Timeout > maxTimeout { + maxTimeout = sources[i].Timeout + } + entries[i] = &federateEntry{} + } + + return &Federator{ + sources: sources, + entries: entries, + // The client timeout is a backstop; the per-request context carries + // each source's own timeout. + client: &http.Client{Timeout: maxTimeout + time.Second}, + logger: logger, + } +} + +// Append writes every source's exposition (cached or freshly fetched) followed +// by the cbox_init_federate_up block. It never returns an error for a source +// being down — that is what the gauge is for. +func (f *Federator) Append(ctx context.Context, w io.Writer) { + up := make([]bool, len(f.sources)) + for i := range f.sources { + body, ok := f.fetch(ctx, i) + up[i] = ok + if !ok || len(body) == 0 { + continue + } + fmt.Fprintf(w, "\n# Federated from %s (%s)\n", f.sources[i].Name, f.sources[i].URL) + _, _ = w.Write(body) + if body[len(body)-1] != '\n' { + _, _ = io.WriteString(w, "\n") + } + } + + _, _ = io.WriteString(w, "\n# HELP cbox_init_federate_up Whether the federated metrics source responded on the last fetch (1) or is being skipped as down (0).\n") + _, _ = io.WriteString(w, "# TYPE cbox_init_federate_up gauge\n") + for i, src := range f.sources { + v := 0 + if up[i] { + v = 1 + } + fmt.Fprintf(w, "cbox_init_federate_up{name=%q} %d\n", src.Name, v) + } +} + +// fetch returns the cached body when fresh, otherwise fetches. A failed fetch +// marks the source down and drops the cached body: stale metrics presented as +// current are worse than an honest gap plus federate_up 0. +func (f *Federator) fetch(ctx context.Context, i int) ([]byte, bool) { + src := f.sources[i] + e := f.entries[i] + + e.mu.Lock() + defer e.mu.Unlock() + + if time.Since(e.fetchedAt) < src.CacheTTL { + return e.body, e.up + } + + reqCtx, cancel := context.WithTimeout(ctx, src.Timeout) + defer cancel() + + body, err := f.fetchOnce(reqCtx, src.URL) + e.fetchedAt = time.Now() + if err != nil { + if e.up { // log the transition, not every scrape + f.logger.Warn("Federated metrics source is down", "name", src.Name, "error", err) + } + e.up = false + e.body = nil + return nil, false + } + if !e.up && e.fetchedAt != (time.Time{}) { + f.logger.Info("Federated metrics source is up", "name", src.Name) + } + e.up = true + e.body = body + return body, true +} + +func (f *Federator) fetchOnce(ctx context.Context, url string) ([]byte, error) { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil) + if err != nil { + return nil, err + } + // Ask for the plain text exposition; the response is appended verbatim. + req.Header.Set("Accept", "text/plain;version=0.0.4") + + resp, err := f.client.Do(req) + if err != nil { + return nil, err + } + defer func() { _ = resp.Body.Close() }() + + if resp.StatusCode != http.StatusOK { + return nil, fmt.Errorf("status %d", resp.StatusCode) + } + + raw, err := io.ReadAll(io.LimitReader(resp.Body, federateMaxBodyBytes+1)) + if err != nil { + return nil, err + } + if len(raw) > federateMaxBodyBytes { + return nil, fmt.Errorf("body exceeds %d bytes", federateMaxBodyBytes) + } + + return sanitizeExposition(raw), nil +} + +// sanitizeExposition drops OpenMetrics terminators — "# EOF" in the middle of +// a concatenated response would truncate parsing for some scrapers. +func sanitizeExposition(raw []byte) []byte { + if !bytes.Contains(raw, []byte("# EOF")) { + return raw + } + var b bytes.Buffer + b.Grow(len(raw)) + for _, line := range strings.SplitAfter(string(raw), "\n") { + if strings.TrimSpace(line) == "# EOF" { + continue + } + b.WriteString(line) + } + return b.Bytes() +} diff --git a/internal/metrics/federate_test.go b/internal/metrics/federate_test.go new file mode 100644 index 0000000..884f7c0 --- /dev/null +++ b/internal/metrics/federate_test.go @@ -0,0 +1,156 @@ +package metrics + +import ( + "context" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus" +) + +func testLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +func federateOutput(t *testing.T, f *Federator) string { + t.Helper() + var b strings.Builder + f.Append(context.Background(), &b) + return b.String() +} + +func TestFederatorAppendsUpSource(t *testing.T) { + src := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + io.WriteString(w, "phpfpm_listen_queue{pool=\"www\"} 3\n") + })) + defer src.Close() + + f := NewFederator([]FederateSource{{Name: "fpm-exporter", URL: src.URL}}, testLogger()) + out := federateOutput(t, f) + + if !strings.Contains(out, `phpfpm_listen_queue{pool="www"} 3`) { + t.Fatalf("federated body missing from output:\n%s", out) + } + if !strings.Contains(out, `cbox_init_federate_up{name="fpm-exporter"} 1`) { + t.Fatalf("federate_up 1 missing:\n%s", out) + } +} + +func TestFederatorDownSourceDegrades(t *testing.T) { + f := NewFederator([]FederateSource{{ + Name: "dead", + URL: "http://127.0.0.1:1/metrics", // nothing listens on port 1 + Timeout: 200 * time.Millisecond, + }}, testLogger()) + out := federateOutput(t, f) + + if !strings.Contains(out, `cbox_init_federate_up{name="dead"} 0`) { + t.Fatalf("down source must contribute federate_up 0:\n%s", out) + } +} + +func TestFederatorCachesWithinTTL(t *testing.T) { + var hits atomic.Int64 + src := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + hits.Add(1) + io.WriteString(w, "x_total 1\n") + })) + defer src.Close() + + f := NewFederator([]FederateSource{{Name: "cached", URL: src.URL, CacheTTL: time.Hour}}, testLogger()) + for range 5 { + federateOutput(t, f) + } + if got := hits.Load(); got != 1 { + t.Fatalf("expected exactly 1 upstream fetch within the TTL, got %d", got) + } +} + +func TestFederatorStripsOpenMetricsEOF(t *testing.T) { + src := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + io.WriteString(w, "y_total 2\n# EOF\n") + })) + defer src.Close() + + f := NewFederator([]FederateSource{{Name: "om", URL: src.URL}}, testLogger()) + out := federateOutput(t, f) + + if strings.Contains(out, "# EOF") { + t.Fatalf("OpenMetrics terminator must be stripped:\n%s", out) + } + if !strings.Contains(out, "y_total 2") { + t.Fatalf("body lost while stripping EOF:\n%s", out) + } +} + +func TestFederatorRejectsOversizedBody(t *testing.T) { + big := strings.Repeat("a_metric 1\n", (federateMaxBodyBytes/11)+2) + src := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + io.WriteString(w, big) + })) + defer src.Close() + + f := NewFederator([]FederateSource{{Name: "huge", URL: src.URL}}, testLogger()) + out := federateOutput(t, f) + + if !strings.Contains(out, `cbox_init_federate_up{name="huge"} 0`) { + t.Fatalf("oversized body must degrade to down:\n%s", out) + } +} + +func TestMetricsHandlerMergesExtraGatherer(t *testing.T) { + reg := prometheus.NewRegistry() + g := prometheus.NewGauge(prometheus.GaugeOpts{Name: "fpm_tune_test_gauge", Help: "test"}) + g.Set(42) + reg.MustRegister(g) + + s := NewServer(0, "/metrics", nil, nil, testLogger()) + s.AddGatherer(reg) // after construction, as serve.go does after Start + + rr := httptest.NewRecorder() + s.metricsHandler().ServeHTTP(rr, httptest.NewRequest(http.MethodGet, "/metrics", nil)) + + if !strings.Contains(rr.Body.String(), "fpm_tune_test_gauge 42") { + t.Fatalf("extra gatherer not merged into scrape:\n%s", rr.Body.String()) + } +} + +func TestMetricsHandlerWithFederation(t *testing.T) { + src := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { + io.WriteString(w, "laravel_queue_size{queue=\"default\"} 7\n") + })) + defer src.Close() + + reg := prometheus.NewRegistry() + g := prometheus.NewGauge(prometheus.GaugeOpts{Name: "fpm_tune_other_gauge", Help: "test"}) + g.Set(1) + reg.MustRegister(g) + + s := NewServer(0, "/metrics", nil, nil, testLogger()) + s.AddGatherer(reg) + s.SetFederator(NewFederator([]FederateSource{{Name: "app", URL: src.URL}}, testLogger())) + + rr := httptest.NewRecorder() + s.metricsHandler().ServeHTTP(rr, httptest.NewRequest(http.MethodGet, "/metrics", nil)) + body := rr.Body.String() + + for _, want := range []string{ + "fpm_tune_other_gauge 1", // merged gatherer + `laravel_queue_size{queue="default"} 7`, // federated body + `cbox_init_federate_up{name="app"} 1`, // health of the source + "go_goroutines", // default registry still present + } { + if !strings.Contains(body, want) { + t.Fatalf("missing %q in federated scrape:\n%s", want, body) + } + } + if ct := rr.Header().Get("Content-Type"); !strings.HasPrefix(ct, "text/plain") { + t.Fatalf("federated response must be plain text, got %q", ct) + } +} diff --git a/internal/metrics/server.go b/internal/metrics/server.go index 7c3d2c0..ab387c3 100644 --- a/internal/metrics/server.go +++ b/internal/metrics/server.go @@ -13,7 +13,10 @@ import ( "github.com/cboxdk/init/internal/acl" "github.com/cboxdk/init/internal/config" tlsmgr "github.com/cboxdk/init/internal/tls" + "github.com/prometheus/client_golang/prometheus" "github.com/prometheus/client_golang/prometheus/promhttp" + dto "github.com/prometheus/client_model/go" + "github.com/prometheus/common/expfmt" ) // Server serves Prometheus metrics @@ -29,6 +32,73 @@ type Server struct { aclInitErr error // ACL was enabled but its checker failed to build; fail closed at Start tlsConfig *config.TLSConfig tlsManager *tlsmgr.Manager + + // extraGatherers are merged into every scrape alongside the default + // registry — embedded engines (fpm-tune) register here so their series + // appear on the main endpoint without a second listener. Guarded by a + // mutex because the tuner starts after the metrics server does. + extraMu sync.RWMutex + extraGatherers []prometheus.Gatherer + federator *Federator +} + +// AddGatherer merges an additional registry into the main /metrics response. +// Safe to call after Start; the next scrape picks it up. +func (s *Server) AddGatherer(g prometheus.Gatherer) { + if g == nil { + return + } + s.extraMu.Lock() + s.extraGatherers = append(s.extraGatherers, g) + s.extraMu.Unlock() +} + +// SetFederator merges declared local exporters into the main /metrics +// response. Must be called before Start. +func (s *Server) SetFederator(f *Federator) *Server { + s.federator = f + return s +} + +// gatherer snapshots the default registry plus any extra gatherers per scrape. +func (s *Server) gatherer() prometheus.Gatherer { + return prometheus.GathererFunc(func() ([]*dto.MetricFamily, error) { + s.extraMu.RLock() + gs := make(prometheus.Gatherers, 0, len(s.extraGatherers)+1) + gs = append(gs, prometheus.DefaultGatherer) + gs = append(gs, s.extraGatherers...) + s.extraMu.RUnlock() + return gs.Gather() + }) +} + +// metricsHandler serves the merged exposition. Without federation the standard +// promhttp handler (with its content negotiation) runs over the merged +// gatherer; with federation the response is always plain text, because the +// federated bodies are appended verbatim and must not disagree with a +// negotiated encoding. +func (s *Server) metricsHandler() http.Handler { + g := s.gatherer() + if s.federator == nil { + return promhttp.HandlerFor(g, promhttp.HandlerOpts{}) + } + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4; charset=utf-8") + mfs, err := g.Gather() + if err != nil { + // Gatherers returns partial families alongside a MultiError; a + // half-full scrape beats an empty one, so log and keep going. + s.logger.Warn("Metrics gather reported errors", "error", err) + } + enc := expfmt.NewEncoder(w, expfmt.NewFormat(expfmt.TypeTextPlain)) + for _, mf := range mfs { + if encErr := enc.Encode(mf); encErr != nil { + s.logger.Warn("Metrics encode failed", "error", encErr) + return + } + } + s.federator.Append(r.Context(), w) + }) } // NewServer creates a new metrics server @@ -90,7 +160,7 @@ func (s *Server) Start(ctx context.Context) error { mux := http.NewServeMux() // Prometheus metrics endpoint - mux.Handle(s.path, promhttp.Handler()) + mux.Handle(s.path, s.metricsHandler()) // Health endpoint for the metrics server itself mux.HandleFunc("/health", func(w http.ResponseWriter, r *http.Request) { diff --git a/sbom.json b/sbom.json index b22b36f..b05d29e 100644 --- a/sbom.json +++ b/sbom.json @@ -1752,6 +1752,8 @@ "pkg:golang/github.com/charmbracelet/lipgloss@v1.1.0?type=module", "pkg:golang/github.com/fsnotify/fsnotify@v1.10.1?type=module", "pkg:golang/github.com/prometheus/client_golang@v1.24.1?type=module", + "pkg:golang/github.com/prometheus/client_model@v0.6.2?type=module", + "pkg:golang/github.com/prometheus/common@v0.70.1?type=module", "pkg:golang/github.com/robfig/cron/v3@v3.0.1?type=module", "pkg:golang/github.com/shirou/gopsutil/v4@v4.26.7?type=module", "pkg:golang/github.com/spf13/cobra@v1.10.2?type=module",