diff --git a/cmd/cbox-init/serve.go b/cmd/cbox-init/serve.go index 2a4bc59..b70f7e9 100644 --- a/cmd/cbox-init/serve.go +++ b/cmd/cbox-init/serve.go @@ -171,7 +171,7 @@ func runServe(cmd *cobra.Command, args []string) { Enabled: cfg.Global.TracingEnabled, Exporter: cfg.Global.TracingExporter, Endpoint: cfg.Global.TracingEndpoint, - SampleRate: cfg.Global.TracingSampleRate, + SampleRate: cfg.Global.TracingSampleRateValue(), ServiceName: cfg.Global.TracingServiceName, Version: version, UseTLS: cfg.Global.TracingUseTLS, diff --git a/docs/configuration/global-settings.md b/docs/configuration/global-settings.md index e15d2b1..b9cb106 100644 --- a/docs/configuration/global-settings.md +++ b/docs/configuration/global-settings.md @@ -158,7 +158,7 @@ global: **Settings:** - `api_enabled` - Enable/disable TCP REST API (default: `false`) - `api_port` - HTTP port for API endpoints (default: `9180`) -- `api_host` - Bind host for the API listener (default: all interfaces). **Recommended:** set to `127.0.0.1` when you only need local access, so the management API is never reachable from the pod/host network. +- `api_host` - Bind host for the API listener (default: `127.0.0.1`, loopback only). Binding beyond loopback exposes a control plane that can start, stop and reconfigure every process in the container, so cbox-init refuses to start unless you also set `api_auth` or an `api_acl` with `mode: allow` and a non-empty `allow_list`. - `api_auth` - Optional Bearer token for authentication **API Endpoints:** diff --git a/internal/config/config.go b/internal/config/config.go index 9d0ebef..7577713 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -107,6 +107,9 @@ func (c *Config) validateGlobal() error { if err := validateACL("metrics_acl", c.Global.MetricsACL); err != nil { return err } + if err := c.validateTracing(); err != nil { + return err + } return nil } @@ -127,6 +130,26 @@ func validateACL(key string, acl *ACLConfig) error { } } +// validateTracing rejects an exporter the tracing provider cannot build. +// +// otlp-http, jaeger and zipkin were advertised in the field's own comment and +// given default endpoints, but createExporter implements only otlp-grpc and +// stdout — so check-config passed a config that made serve exit 1 at startup +// with "unsupported trace exporter". A config the validator calls valid must +// boot. +func (c *Config) validateTracing() error { + if !c.Global.TracingEnabled { + return nil + } + + switch c.Global.TracingExporter { + case "", "otlp-grpc", "stdout": + return nil + default: + return fmt.Errorf("invalid tracing_exporter %q (valid: otlp-grpc, stdout)", c.Global.TracingExporter) + } +} + // validateAPIExposure refuses to start an unauthenticated management API on a // non-loopback interface. The API can add and stop processes and rewrite the // config, so exposing it beyond localhost without a bearer token or an IP ACL is diff --git a/internal/config/tracing_config_test.go b/internal/config/tracing_config_test.go new file mode 100644 index 0000000..acd459a --- /dev/null +++ b/internal/config/tracing_config_test.go @@ -0,0 +1,86 @@ +package config + +import ( + "os" + "path/filepath" + "testing" +) + +// TestTracingSampleRateZeroSamplesNothing: SetDefaults treated 0 as "unset" and +// replaced it with 1.0, so tracing_sample_rate: 0.0 — the documented way to +// sample nothing — produced 100% sampling instead. The docs even advise +// "ensure not 0.0", which only makes sense if 0.0 means what it says. +func TestTracingSampleRateZeroSamplesNothing(t *testing.T) { + tests := []struct { + name string + yaml string + want float64 + }{ + {"explicit zero", " tracing_sample_rate: 0.0\n", 0.0}, + {"explicit fraction", " tracing_sample_rate: 0.25\n", 0.25}, + {"absent defaults to full", "", 1.0}, + {"explicit one", " tracing_sample_rate: 1.0\n", 1.0}, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := loadTracingConfig(t, " tracing_enabled: true\n tracing_exporter: stdout\n"+tt.yaml) + + if got := cfg.Global.TracingSampleRateValue(); got != tt.want { + t.Errorf("sample rate = %v, want %v", got, tt.want) + } + }) + } +} + +// TestUnsupportedTracingExporterIsRejected: otlp-http, jaeger and zipkin were +// advertised in the field's comment and given default endpoints, but the +// provider implements only otlp-grpc and stdout. check-config passed, and then +// serve exited 1 at startup. A config the validator calls valid must boot. +func TestUnsupportedTracingExporterIsRejected(t *testing.T) { + for _, exporter := range []string{"otlp-http", "jaeger", "zipkin", "otlp"} { + t.Run(exporter, func(t *testing.T) { + path := writeTracingConfig(t, " tracing_enabled: true\n tracing_exporter: "+exporter+"\n") + + if _, err := LoadWithEnvExpansion(path); err == nil { + t.Errorf("tracing_exporter %q validated, but the provider cannot build it "+ + "— serve would exit 1 at startup", exporter) + } + }) + } + + for _, exporter := range []string{"otlp-grpc", "stdout"} { + t.Run(exporter, func(t *testing.T) { + path := writeTracingConfig(t, " tracing_enabled: true\n tracing_exporter: "+exporter+"\n") + + if _, err := LoadWithEnvExpansion(path); err != nil { + t.Errorf("tracing_exporter %q rejected: %v", exporter, err) + } + }) + } +} + +func writeTracingConfig(t *testing.T, globalExtra string) string { + t.Helper() + + body := "version: \"1.0\"\nglobal:\n log_level: info\n" + globalExtra + + "processes:\n app:\n command: [\"/bin/true\"]\n" + + path := filepath.Join(t.TempDir(), "cbox-init.yaml") + if err := os.WriteFile(path, []byte(body), 0o600); err != nil { + t.Fatal(err) + } + + return path +} + +func loadTracingConfig(t *testing.T, globalExtra string) *Config { + t.Helper() + + cfg, err := LoadWithEnvExpansion(writeTracingConfig(t, globalExtra)) + if err != nil { + t.Fatalf("load: %v", err) + } + + return cfg +} diff --git a/internal/config/types.go b/internal/config/types.go index 32ee524..de73857 100644 --- a/internal/config/types.go +++ b/internal/config/types.go @@ -18,52 +18,56 @@ 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 - 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 | otlp-http | stdout | jaeger | zipkin - TracingEndpoint string `yaml:"tracing_endpoint" json:"tracing_endpoint"` // Exporter endpoint (e.g., localhost:4317) - TracingSampleRate float64 `yaml:"tracing_sample_rate" json:"tracing_sample_rate"` // 0.0-1.0 (default: 1.0 = 100%) - TracingServiceName string `yaml:"tracing_service_name" json:"tracing_service_name"` // Service name for traces (default: cbox-init) - TracingUseTLS bool `yaml:"tracing_use_tls" json:"tracing_use_tls"` // Enable TLS for production (default: false) - ScheduleHistorySize int `yaml:"schedule_history_size" json:"schedule_history_size"` // Max execution history entries per job (default: 100) - OneshotHistoryMaxEntries int `yaml:"oneshot_history_max_entries" json:"oneshot_history_max_entries"` // Max oneshot history entries per process (default: 5000) - OneshotHistoryMaxAge time.Duration `yaml:"oneshot_history_max_age" json:"oneshot_history_max_age"` // Max age of oneshot history entries (default: 24h) - Readiness *ReadinessConfig `yaml:"readiness" json:"readiness"` // Container readiness file config for K8s - HealthCheckStrict bool `yaml:"health_check_strict" json:"health_check_strict"` // Fail process startup if health monitor creation fails (default: false) - DependencyTimeout time.Duration `yaml:"dependency_timeout" json:"dependency_timeout"` // Max time to wait for dependencies to become ready (default: 5m) - ProcessStartTimeout time.Duration `yaml:"process_start_timeout" json:"process_start_timeout"` // Timeout for starting a single process (default: 30s) - ProcessStopTimeout time.Duration `yaml:"process_stop_timeout" json:"process_stop_timeout"` // Timeout for stopping a single process (default: 60s) - MaxProcessScale int `yaml:"max_process_scale" json:"max_process_scale"` // Maximum instances per process (default: 100) - APIMaxRequestBody int64 `yaml:"api_max_request_body" json:"api_max_request_body"` // Max request body size in bytes (default: 8MB) - ZombieReapInterval time.Duration `yaml:"zombie_reap_interval" json:"zombie_reap_interval"` // Interval for zombie process reaping (default: 1s) + 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 + 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) + // 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. + // Read it through TracingSampleRateValue, which applies the 1.0 default. + TracingSampleRate *float64 `yaml:"tracing_sample_rate" json:"tracing_sample_rate"` // 0.0-1.0 (default: 1.0 = 100%) + TracingServiceName string `yaml:"tracing_service_name" json:"tracing_service_name"` // Service name for traces (default: cbox-init) + TracingUseTLS bool `yaml:"tracing_use_tls" json:"tracing_use_tls"` // Enable TLS for production (default: false) + ScheduleHistorySize int `yaml:"schedule_history_size" json:"schedule_history_size"` // Max execution history entries per job (default: 100) + OneshotHistoryMaxEntries int `yaml:"oneshot_history_max_entries" json:"oneshot_history_max_entries"` // Max oneshot history entries per process (default: 5000) + OneshotHistoryMaxAge time.Duration `yaml:"oneshot_history_max_age" json:"oneshot_history_max_age"` // Max age of oneshot history entries (default: 24h) + Readiness *ReadinessConfig `yaml:"readiness" json:"readiness"` // Container readiness file config for K8s + HealthCheckStrict bool `yaml:"health_check_strict" json:"health_check_strict"` // Fail process startup if health monitor creation fails (default: false) + DependencyTimeout time.Duration `yaml:"dependency_timeout" json:"dependency_timeout"` // Max time to wait for dependencies to become ready (default: 5m) + ProcessStartTimeout time.Duration `yaml:"process_start_timeout" json:"process_start_timeout"` // Timeout for starting a single process (default: 30s) + ProcessStopTimeout time.Duration `yaml:"process_stop_timeout" json:"process_stop_timeout"` // Timeout for stopping a single process (default: 60s) + MaxProcessScale int `yaml:"max_process_scale" json:"max_process_scale"` // Maximum instances per process (default: 100) + APIMaxRequestBody int64 `yaml:"api_max_request_body" json:"api_max_request_body"` // Max request body size in bytes (default: 8MB) + ZombieReapInterval time.Duration `yaml:"zombie_reap_interval" json:"zombie_reap_interval"` // Interval for zombie process reaping (default: 1s) } // HooksConfig contains lifecycle hooks @@ -437,23 +441,16 @@ func (c *Config) setGlobalTracingDefaults() { if c.Global.TracingExporter == "" { c.Global.TracingExporter = "stdout" } - if c.Global.TracingSampleRate == 0 { - c.Global.TracingSampleRate = 1.0 - } + if c.Global.TracingServiceName == "" { c.Global.TracingServiceName = "cbox-init" } - if c.Global.TracingEndpoint == "" { - switch c.Global.TracingExporter { - case "otlp-grpc": - c.Global.TracingEndpoint = "localhost:4317" - case "otlp-http": - c.Global.TracingEndpoint = "localhost:4318" - case "jaeger": - c.Global.TracingEndpoint = "localhost:14268" - case "zipkin": - c.Global.TracingEndpoint = "http://localhost:9411/api/v2/spans" - } + // Only the exporters the tracing provider actually implements get a default + // endpoint. Handing otlp-http, jaeger and zipkin a plausible one made them + // look supported: check-config passed, and then serve exited 1 on + // "unsupported trace exporter". Validate rejects them by name instead. + if c.Global.TracingEndpoint == "" && c.Global.TracingExporter == "otlp-grpc" { + c.Global.TracingEndpoint = "localhost:4317" } } @@ -705,6 +702,17 @@ func (g *GlobalConfig) SetResourceMetricsEnabled(v bool) { } // APIEnabledValue returns true if API enabled (default false) +// TracingSampleRateValue returns the effective sample rate: the configured +// value if the key is present (including an explicit 0.0, which samples +// nothing), or 1.0 when it was never set. +func (g *GlobalConfig) TracingSampleRateValue() float64 { + if g == nil || g.TracingSampleRate == nil { + return 1.0 + } + + return *g.TracingSampleRate +} + func (g *GlobalConfig) APIEnabledValue() bool { if g == nil || g.APIEnabled == nil { return false diff --git a/internal/config/types_test.go b/internal/config/types_test.go index 3cd0270..3462d3f 100644 --- a/internal/config/types_test.go +++ b/internal/config/types_test.go @@ -83,8 +83,8 @@ func TestSetDefaults(t *testing.T) { if c.Global.TracingExporter != "stdout" { t.Errorf("TracingExporter = %v, want stdout", c.Global.TracingExporter) } - if c.Global.TracingSampleRate != 1.0 { - t.Errorf("TracingSampleRate = %v, want 1.0", c.Global.TracingSampleRate) + if c.Global.TracingSampleRateValue() != 1.0 { + t.Errorf("TracingSampleRate = %v, want 1.0", c.Global.TracingSampleRateValue()) } if c.Global.TracingServiceName != "cbox-init" { t.Errorf("TracingServiceName = %v, want cbox-init", c.Global.TracingServiceName) @@ -389,7 +389,12 @@ func TestSetDefaults(t *testing.T) { }, }, { - name: "tracing endpoint defaults for otlp-http", + // Exporters the provider cannot build get NO default endpoint. A + // plausible-looking one made otlp-http, jaeger and zipkin appear + // supported: check-config passed and then serve exited 1 on + // "unsupported trace exporter". Validate rejects them by name now — + // see TestUnsupportedTracingExporterIsRejected. + name: "no endpoint default for an unsupported exporter", config: &Config{ Global: GlobalConfig{ TracingExporter: "otlp-http", @@ -399,40 +404,9 @@ func TestSetDefaults(t *testing.T) { }, }, validate: func(t *testing.T, c *Config) { - if c.Global.TracingEndpoint != "localhost:4318" { - t.Errorf("TracingEndpoint = %v, want localhost:4318", c.Global.TracingEndpoint) - } - }, - }, - { - name: "tracing endpoint defaults for jaeger", - config: &Config{ - Global: GlobalConfig{ - TracingExporter: "jaeger", - }, - Processes: map[string]*Process{ - "test": {Command: []string{"sleep", "1"}}, - }, - }, - validate: func(t *testing.T, c *Config) { - if c.Global.TracingEndpoint != "localhost:14268" { - t.Errorf("TracingEndpoint = %v, want localhost:14268", c.Global.TracingEndpoint) - } - }, - }, - { - name: "tracing endpoint defaults for zipkin", - config: &Config{ - Global: GlobalConfig{ - TracingExporter: "zipkin", - }, - Processes: map[string]*Process{ - "test": {Command: []string{"sleep", "1"}}, - }, - }, - validate: func(t *testing.T, c *Config) { - if c.Global.TracingEndpoint != "http://localhost:9411/api/v2/spans" { - t.Errorf("TracingEndpoint = %v, want zipkin default", c.Global.TracingEndpoint) + if c.Global.TracingEndpoint != "" { + t.Errorf("TracingEndpoint = %v, want empty for an unsupported exporter", + c.Global.TracingEndpoint) } }, }, diff --git a/internal/readiness/manager.go b/internal/readiness/manager.go index efa69f1..6b90229 100644 --- a/internal/readiness/manager.go +++ b/internal/readiness/manager.go @@ -379,6 +379,19 @@ func (m *Manager) isProcessReady(status ProcessStatus) bool { // // Must be called with the mutex lock held. func (m *Manager) setReady(ready bool) { + // A stopped manager never becomes ready again. IsReady and Snapshot already + // return false once stopped, but the FILE was still being written — so a + // late evaluation racing shutdown recreated the readiness file after Stop + // had removed it, and a file-based probe went on reporting the container + // ready while it was tearing itself down. Traffic kept arriving. + if m.stopped { + if ready { + m.logger.Debug("Ignoring readiness transition after stop") + } + + return + } + // Reconcile the file on every evaluation, not only on a change. Acting only // on transitions meant a readiness file removed from underneath us — a // tmpfs cleaner, a sidecar, an operator — was never recreated, so the diff --git a/internal/readiness/stopped_file_test.go b/internal/readiness/stopped_file_test.go new file mode 100644 index 0000000..4d0d018 --- /dev/null +++ b/internal/readiness/stopped_file_test.go @@ -0,0 +1,53 @@ +package readiness + +import ( + "log/slog" + "os" + "path/filepath" + "testing" + + "github.com/cboxdk/init/internal/config" +) + +// TestStoppedManagerDoesNotRecreateTheReadinessFile: IsReady and Snapshot +// already refuse to report ready once stopped, but setReady still wrote the +// FILE. A late evaluation racing shutdown therefore recreated the readiness file +// after Stop had removed it, and a file-based probe went on reporting the +// container ready while it was tearing itself down — so traffic kept arriving. +func TestStoppedManagerDoesNotRecreateTheReadinessFile(t *testing.T) { + path := filepath.Join(t.TempDir(), "ready") + + m := NewManager(&config.ReadinessConfig{ + Enabled: true, + Path: path, + Mode: "all_healthy", + }, slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError}))) + + m.mu.Lock() + m.setReady(true) + m.mu.Unlock() + + if _, err := os.Stat(path); err != nil { + t.Fatalf("readiness file was not created while running: %v", err) + } + + if err := m.Stop(); err != nil { + t.Fatalf("Stop: %v", err) + } + if _, err := os.Stat(path); !os.IsNotExist(err) { + t.Fatalf("Stop did not remove the readiness file: %v", err) + } + + // A late evaluation lands after Stop. + m.mu.Lock() + m.setReady(true) + m.mu.Unlock() + + if _, err := os.Stat(path); !os.IsNotExist(err) { + t.Error("the readiness file was recreated after Stop; a file-based probe " + + "keeps routing traffic to a container that is shutting down") + } + if m.IsReady() { + t.Error("a stopped manager reported ready") + } +} diff --git a/internal/schedule/job.go b/internal/schedule/job.go index af8bd14..35d0cb5 100644 --- a/internal/schedule/job.go +++ b/internal/schedule/job.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "log/slog" + "runtime/debug" "strings" "sync" "time" @@ -351,6 +352,42 @@ func (j *ScheduledJob) executeSync(ctx context.Context, triggered string) (int, j.CurrentExecID = execID j.mu.Unlock() + // Restore the state whatever happens below, including a panic. + // + // The state was reset only on the normal path, so a panicking executor left + // the job in JobStateExecuting for good — and the overlap check above then + // refused every subsequent run with "already executing". A recovered panic + // (the cron chain recovers, so the process survives) silently retired the + // job. The flag says whether the normal path already ran, so the happy case + // is untouched. + completed := false + defer func() { + if completed { + return + } + + r := recover() + + j.mu.Lock() + j.State = JobStateIdle + j.CurrentExecID = 0 + j.mu.Unlock() + + if r != nil { + j.logger.Error("job execution panicked", + "execution_id", execID, + "panic", r, + "stack", string(debug.Stack()), + ) + j.History.EndExecution(execID, -1, false, fmt.Sprintf("panic: %v", r)) + + panic(r) // let the cron chain's recoverer log and contain it + } + + // Not a panic: an early return that skipped the normal completion path. + j.History.EndExecution(execID, -1, false, "execution ended without recording a result") + }() + j.logger.Info("job execution started", "execution_id", execID, "triggered", triggered, @@ -374,6 +411,8 @@ func (j *ScheduledJob) executeSync(ctx context.Context, triggered string) (int, j.CurrentExecID = 0 j.mu.Unlock() + completed = true + success := execErr == nil && exitCode == 0 errMsg := "" if execErr != nil { diff --git a/internal/schedule/panic_state_test.go b/internal/schedule/panic_state_test.go new file mode 100644 index 0000000..48dadea --- /dev/null +++ b/internal/schedule/panic_state_test.go @@ -0,0 +1,55 @@ +package schedule + +import ( + "context" + "log/slog" + "os" + "testing" +) + +type panickingExecutor struct{ calls int } + +func (p *panickingExecutor) Execute(ctx context.Context, name string) (int, error) { + p.calls++ + panic("executor blew up") +} + +// TestPanickingJobDoesNotWedgeInExecuting: the state was reset only on the +// normal return path, so a panicking executor left the job in +// JobStateExecuting for good. The overlap check then refused every subsequent +// run with "already executing", and because the cron chain recovers the panic, +// the process survived — the job was silently retired with nothing to explain +// it. +func TestPanickingJobDoesNotWedgeInExecuting(t *testing.T) { + lg := slog.New(slog.NewTextHandler(os.Stderr, &slog.HandlerOptions{Level: slog.LevelError})) + exec := &panickingExecutor{} + + job, err := NewScheduledJob("boom", "* * * * *", "UTC", 10, exec, lg) + if err != nil { + t.Fatalf("NewScheduledJob: %v", err) + } + + runAndRecover := func() { + defer func() { _ = recover() }() + _, _ = job.executeSync(context.Background(), "test") + } + + runAndRecover() + + if state := job.GetState(); state != JobStateIdle { + t.Errorf("after a panic the job is %v, want idle; every later run is refused as "+ + "\"already executing\"", state) + } + + // The job must actually run again. + runAndRecover() + + if exec.calls != 2 { + t.Errorf("executor called %d times; the job was retired after the first panic", exec.calls) + } + + // And the failed attempts are recorded, not lost. + if recent := job.History.GetRecent(2); len(recent) != 2 { + t.Errorf("history holds %d executions, want 2 — a panicking run left no record", len(recent)) + } +}