diff --git a/cmd/prometheus/main.go b/cmd/prometheus/main.go index 6066a5be1dd..0a641ec0d11 100644 --- a/cmd/prometheus/main.go +++ b/cmd/prometheus/main.go @@ -83,6 +83,7 @@ import ( "github.com/prometheus/prometheus/util/documentcli" "github.com/prometheus/prometheus/util/features" "github.com/prometheus/prometheus/util/logging" + "github.com/prometheus/prometheus/util/memorylimiter" "github.com/prometheus/prometheus/util/notifications" prom_runtime "github.com/prometheus/prometheus/util/runtime" "github.com/prometheus/prometheus/web" @@ -215,6 +216,7 @@ type flagConfig struct { enablePerStepStats bool enableConcurrentRuleEval bool useStartTimestamps bool + enableMemoryLimiter bool prometheusURL string corsRegexString string @@ -332,6 +334,9 @@ func (c *flagConfig) setFeatureListOptions(logger *slog.Logger) error { case "fast-startup": c.tsdb.EnableFastStartup = true logger.Info("Experimental fast startup is enabled.") + case "memory-limiter": + c.enableMemoryLimiter = true + logger.Info("Experimental memory limiter is enabled.") default: logger.Warn("Unknown option for --enable-feature", "option", o) } @@ -365,6 +370,7 @@ func main() { collectors.NewGoCollector( collectors.WithGoCollectorRuntimeMetrics( collectors.MetricsGC, + collectors.MetricsMemory, collectors.MetricsScheduler, collectors.GoRuntimeMetricsRule{Matcher: goregexp.MustCompile(`^/sync/mutex/wait/total:seconds$`)}, ), @@ -620,7 +626,7 @@ func main() { a.Flag("scrape.discovery-reload-interval", "Interval used by scrape manager to throttle target groups updates."). Hidden().Default("5s").SetValue(&cfg.scrape.DiscoveryReloadInterval) - a.Flag("enable-feature", "Comma separated feature names to enable. Valid options: concurrent-rule-eval, created-timestamp-zero-ingestion, delayed-compaction, exemplar-storage, extra-scrape-metrics, memory-snapshot-on-shutdown, metadata-wal-records, old-ui, otlp-deltatocumulative, otlp-native-delta-ingestion, promql-binop-fill-modifiers, promql-delayed-name-removal, promql-duration-expr, promql-experimental-functions, promql-extended-range-selectors, promql-per-step-stats, st-storage, type-and-unit-labels, use-start-timestamps, use-uncached-io, xor2-encoding. See https://prometheus.io/docs/prometheus/latest/feature_flags/ for more details."). + a.Flag("enable-feature", "Comma separated feature names to enable. Valid options: concurrent-rule-eval, created-timestamp-zero-ingestion, delayed-compaction, exemplar-storage, extra-scrape-metrics, memory-limiter, memory-snapshot-on-shutdown, metadata-wal-records, old-ui, otlp-deltatocumulative, otlp-native-delta-ingestion, promql-binop-fill-modifiers, promql-delayed-name-removal, promql-duration-expr, promql-experimental-functions, promql-extended-range-selectors, promql-per-step-stats, st-storage, type-and-unit-labels, use-start-timestamps, use-uncached-io, xor2-encoding. See https://prometheus.io/docs/prometheus/latest/feature_flags/ for more details."). StringsVar(&cfg.featureList) a.Flag("agent", "Run Prometheus in 'Agent mode'.").BoolVar(&agentMode) @@ -938,6 +944,18 @@ func main() { os.Exit(1) } + var memoryLimiter memorylimiter.MemoryLimiter + if cfg.enableMemoryLimiter { + ml, err := memorylimiter.NewManager(&cfgFile.Runtime.MemoryLimiter, logger.With("component", "memory limiter"), prometheus.DefaultRegisterer) + if err != nil { + logger.Error("failed to create memory limiter", "err", err) + os.Exit(1) + } + memoryLimiter = ml + cfg.scrape.MemoryLimiter = ml + cfg.web.MemoryLimiter = ml + } + scrapeManager, err := scrape.NewManager( &cfg.scrape, logger.With("component", "scrape manager"), @@ -1000,6 +1018,7 @@ func main() { }, FeatureRegistry: features.DefaultRegistry, Parser: promqlParser, + MemoryLimiter: memoryLimiter, }) } @@ -1164,6 +1183,14 @@ func main() { }, { name: "tracing", reloader: tracingManager.ApplyConfig, + }, { + name: "memory_limiter", + reloader: func(cfg *config.Config) error { + if memoryLimiter != nil { + return memoryLimiter.ApplyConfig(&cfg.Runtime.MemoryLimiter) + } + return nil + }, }, } @@ -1257,6 +1284,21 @@ func main() { }, ) } + if memoryLimiter != nil { + ctxML, cancelML := context.WithCancel(context.Background()) + g.Add( + func() error { + memoryLimiter.Start(ctxML) + <-ctxML.Done() + return nil + }, + func(error) { + logger.Info("Stopping memory limiter...") + cancelML() + memoryLimiter.Stop() + }, + ) + } if !agentMode { // Rule manager. g.Add( diff --git a/cmd/prometheus/memory_limiter_e2e_test.go b/cmd/prometheus/memory_limiter_e2e_test.go new file mode 100644 index 00000000000..e1ea1010265 --- /dev/null +++ b/cmd/prometheus/memory_limiter_e2e_test.go @@ -0,0 +1,1555 @@ +// Copyright The Prometheus Authors +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package main + +import ( + "bytes" + "fmt" + "io" + "net" + "net/http" + "net/http/httptest" + "os" + "os/exec" + "path/filepath" + "strings" + "sync/atomic" + "testing" + "time" + + "github.com/prometheus/common/expfmt" + "github.com/prometheus/common/model" + "github.com/stretchr/testify/require" +) + +// runPrometheusInstance spawns a Prometheus test process with custom configuration and environment. +func runPrometheusInstance(t *testing.T, configFile string, extraArgs []string, env []string) (*exec.Cmd, string, func()) { + t.Helper() + + // Find free port for web listen address. + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + addr := listener.Addr().String() + require.NoError(t, listener.Close()) + + dir := t.TempDir() + dataPath := filepath.Join(dir, "data") + + args := append([]string{ + "-test.main", + "--config.file=" + configFile, + "--web.listen-address=" + addr, + "--storage.tsdb.path=" + dataPath, + "--storage.tsdb.retention.time=1d", + "--scrape.discovery-reload-interval=50ms", + "--log.level=info", + }, extraArgs...) + + cmd := commandWithLogging(t, nil, promPath, args...) + cmd.Env = append(os.Environ(), env...) + + err = cmd.Start() + require.NoError(t, err) + + // Wait for Prometheus to become ready. + readyURL := fmt.Sprintf("http://%s/-/ready", addr) + require.Eventually(t, func() bool { + resp, err := http.Get(readyURL) + if err != nil { + return false + } + defer resp.Body.Close() + return resp.StatusCode == http.StatusOK + }, 15*time.Second, 100*time.Millisecond, "Prometheus failed to become ready") + + return cmd, addr, func() {} +} + +func fetchPrometheusMetrics(t *testing.T, addr string) map[string]float64 { + t.Helper() + metricsURL := fmt.Sprintf("http://%s/metrics", addr) + resp, err := http.Get(metricsURL) + require.NoError(t, err) + defer resp.Body.Close() + require.Equal(t, http.StatusOK, resp.StatusCode) + + parser := expfmt.NewTextParser(model.UTF8Validation) + metricFamilies, err := parser.TextToMetricFamilies(resp.Body) + require.NoError(t, err) + + results := make(map[string]float64) + for name, mf := range metricFamilies { + for _, m := range mf.Metric { + var val float64 + if m.Gauge != nil { + val = m.Gauge.GetValue() + } else if m.Counter != nil { + val = m.Counter.GetValue() + } else if m.Untyped != nil { + val = m.Untyped.GetValue() + } + results[name] = val + if len(m.Label) > 0 { + var labelStrs []string + for _, lp := range m.Label { + labelStrs = append(labelStrs, fmt.Sprintf("%s=\"%s\"", lp.GetName(), lp.GetValue())) + } + labeledKey := fmt.Sprintf("%s{%s}", name, strings.Join(labelStrs, ",")) + results[labeledKey] = val + } + } + } + return results +} + +// TestScenario_S1_TransientScrapePayloadBurst tests Scenario S1: +// Under transient large scrape payload bursts, the memory limiter engages, skips scrapes +// without crashing, and recovers back to StateOK when load abates. +func TestScenario_S1_TransientScrapePayloadBurst(t *testing.T) { + if testing.Short() { + t.Skip("skipping slow scenario test in short mode") + } + + var burstActive atomic.Bool + + // Target server providing normal metrics or a burst of 1000 series with multiple labels. + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if burstActive.Load() { + var b strings.Builder + for i := 0; i < 1000; i++ { + fmt.Fprintf(&b, "burst_metric_%d{instance=\"node1\",job=\"app\",env=\"prod\",region=\"us-east\",zone=\"b\",tier=\"frontend\",owner=\"team_a\",service=\"auth\",k1=\"v1\",k2=\"v2\",k3=\"v3\",k4=\"v4\",k5=\"v5\",k6=\"v6\",k7=\"v7\",k8=\"v8\",k9=\"v9\",k10=\"v10\"} %d\n", i, i*42) + } + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + _, _ = w.Write([]byte(b.String())) + return + } + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + fmt.Fprintf(w, "healthy_metric{instance=\"node1\"} 1\n") + })) + defer ts.Close() + + // Write Prometheus configuration. + promConfigContent := fmt.Sprintf(` +global: + scrape_interval: 100ms + scrape_timeout: 100ms + +runtime: + gogc: 50 + memory_limiter: + check_interval: 25ms + soft_limit_ratio: 0.50 + hard_limit_ratio: 0.65 + enforcement: + pause_block_compaction: true + reject_remote_read: true + reject_federation: true + +scrape_configs: + - job_name: "test_service" + scrape_interval: 100ms + scrape_timeout: 100ms + static_configs: + - targets: ["%s"] +`, ts.Listener.Addr().String()) + + configFile := filepath.Join(t.TempDir(), "prometheus.yml") + require.NoError(t, os.WriteFile(configFile, []byte(promConfigContent), 0o600)) + + // Run Prometheus with 64MiB GOMEMLIMIT. + _, addr, cleanup := runPrometheusInstance(t, configFile, + []string{"--enable-feature=memory-limiter"}, + []string{"GOMEMLIMIT=64MiB"}, + ) + defer cleanup() + + // Phase 1: Verify healthy steady-state operation. + time.Sleep(1 * time.Second) + metrics := fetchPrometheusMetrics(t, addr) + require.Equal(t, float64(0), metrics["prometheus_memory_limiter_active"], "Memory limiter should not be active in healthy steady-state") + require.Equal(t, float64(0), metrics["prometheus_target_scrapes_skipped_total"], "No scrapes should be skipped in healthy steady-state") + + // Phase 2: Trigger acute scrape payload burst. + burstActive.Store(true) + + // Wait for memory limiter to engage and skip scrapes. + require.Eventually(t, func() bool { + m := fetchPrometheusMetrics(t, addr) + skipped := m["prometheus_target_scrapes_skipped_total"] + engaged := m["prometheus_memory_limiter_engaged_seconds_total"] + return skipped > 0 || engaged > 0 + }, 10*time.Second, 100*time.Millisecond, "Expected memory limiter to engage under acute load") + + // Verify Prometheus process remains alive and responsive to queries during burst. + queryURL := fmt.Sprintf("http://%s/api/v1/query?query=up", addr) + resp, err := http.Get(queryURL) + require.NoError(t, err) + defer resp.Body.Close() + require.Equal(t, http.StatusOK, resp.StatusCode, "Prometheus should maintain query availability during load shedding") + + // Phase 3: Stop burst and verify clean recovery. + burstActive.Store(false) + + // Allow memory to reclaim and limiter to disengage. + require.Eventually(t, func() bool { + m := fetchPrometheusMetrics(t, addr) + active := m["prometheus_memory_limiter_active"] + t.Logf("Phase 3 metrics: in_use=%v, limit=%v, active=%v, skipped=%v", + m["prometheus_memory_limiter_in_use_bytes"], + m["prometheus_memory_limiter_limit_bytes"], + active, + m["prometheus_target_scrapes_skipped_total"], + ) + return active == 0 + }, 15*time.Second, 500*time.Millisecond, "Expected memory limiter to disengage and return to StateOK after burst cessation") +} + +// TestScenario_S2_FlappingAndDutyCycle tests Scenario S2: +// Verifies transition counters, duty cycle accumulation, and metrics accuracy. +func TestScenario_S2_FlappingAndDutyCycle(t *testing.T) { + if testing.Short() { + t.Skip("skipping slow scenario test in short mode") + } + + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + fmt.Fprintf(w, "metric_test 1\n") + })) + defer ts.Close() + + promConfigContent := fmt.Sprintf(` +global: + scrape_interval: 100ms + scrape_timeout: 100ms + +runtime: + gogc: 50 + memory_limiter: + check_interval: 20ms + soft_limit_ratio: 0.70 + hard_limit_ratio: 0.85 + +scrape_configs: + - job_name: "test" + static_configs: + - targets: ["%s"] +`, ts.Listener.Addr().String()) + + configFile := filepath.Join(t.TempDir(), "prometheus.yml") + require.NoError(t, os.WriteFile(configFile, []byte(promConfigContent), 0o600)) + + _, addr, cleanup := runPrometheusInstance(t, configFile, + []string{"--enable-feature=memory-limiter"}, + []string{"GOMEMLIMIT=128MiB"}, + ) + defer cleanup() + + // Verify telemetry metrics exist and are correctly initialized. + metrics := fetchPrometheusMetrics(t, addr) + require.Contains(t, metrics, "prometheus_memory_limiter_limit_bytes") + require.Contains(t, metrics, "prometheus_memory_limiter_in_use_bytes") + require.Contains(t, metrics, "prometheus_memory_limiter_active") + + limitBytes := metrics["prometheus_memory_limiter_limit_bytes"] + require.Greater(t, limitBytes, float64(0), "Limit bytes should be positive") +} + +// TestScenario_S4_ZeroWALContamination tests Scenario S4: +// Proves that when scrapes are aborted, 0 samples and 0 staleness markers are written to TSDB. +func TestScenario_S4_ZeroWALContamination(t *testing.T) { + if testing.Short() { + t.Skip("skipping slow scenario test in short mode") + } + + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + fmt.Fprintf(w, "custom_sensor_data{device=\"sensorA\"} 100\n") + })) + defer ts.Close() + + promConfigContent := fmt.Sprintf(` +global: + scrape_interval: 100ms + scrape_timeout: 100ms + +runtime: + gogc: 50 + memory_limiter: + check_interval: 20ms + soft_limit_ratio: 0.01 + hard_limit_ratio: 0.02 + +scrape_configs: + - job_name: "test_job" + static_configs: + - targets: ["%s"] +`, ts.Listener.Addr().String()) + + configFile := filepath.Join(t.TempDir(), "prometheus.yml") + require.NoError(t, os.WriteFile(configFile, []byte(promConfigContent), 0o600)) + + // Run with very low ratios so memory limiter is immediately in Hard Limit. + _, addr, cleanup := runPrometheusInstance(t, configFile, + []string{"--enable-feature=memory-limiter"}, + []string{"GOMEMLIMIT=128MiB"}, + ) + defer cleanup() + + // Wait for several scrape intervals. + time.Sleep(1 * time.Second) + + metrics := fetchPrometheusMetrics(t, addr) + require.Equal(t, float64(1), metrics["prometheus_memory_limiter_active"], "Limiter should be active") + require.Greater(t, metrics["prometheus_target_scrapes_skipped_total"], float64(0), "Scrapes should be skipped") + + // Query TSDB to verify no series were appended during hard limit. + queryURL := fmt.Sprintf("http://%s/api/v1/query?query=custom_sensor_data", addr) + resp, err := http.Get(queryURL) + require.NoError(t, err) + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + + require.Contains(t, string(body), `"result":[]`, "Zero samples should be persisted in storage for skipped scrapes") +} + +// TestScenario_S5_IngestionAndFederationRejection tests that Remote Read, Remote Write, OTLP, +// and Federation endpoints return 503 Service Unavailable with Retry-After header when the hard limit is active. +func TestScenario_S5_IngestionAndFederationRejection(t *testing.T) { + if testing.Short() { + t.Skip("skipping slow scenario test in short mode") + } + + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + fmt.Fprintf(w, "up 1\n") + })) + defer ts.Close() + + promConfigContent := fmt.Sprintf(` +global: + scrape_interval: 100ms + scrape_timeout: 100ms + +runtime: + gogc: 50 + memory_limiter: + check_interval: 20ms + soft_limit_ratio: 0.01 + hard_limit_ratio: 0.02 + +scrape_configs: + - job_name: "test_job" + static_configs: + - targets: ["%s"] +`, ts.Listener.Addr().String()) + + configFile := filepath.Join(t.TempDir(), "prometheus.yml") + require.NoError(t, os.WriteFile(configFile, []byte(promConfigContent), 0o600)) + + _, addr, cleanup := runPrometheusInstance(t, configFile, + []string{ + "--enable-feature=memory-limiter", + "--web.enable-remote-write-receiver", + "--web.enable-otlp-receiver", + }, + []string{"GOMEMLIMIT=128MiB"}, + ) + defer cleanup() + + time.Sleep(1 * time.Second) + + // 1. Test /federate rejection + fedURL := fmt.Sprintf("http://%s/federate?match[]={job=\"test\"}", addr) + fedResp, err := http.Get(fedURL) + require.NoError(t, err) + defer fedResp.Body.Close() + require.Equal(t, http.StatusServiceUnavailable, fedResp.StatusCode) + require.Equal(t, "5", fedResp.Header.Get("Retry-After")) + + // 2. Test /api/v1/write rejection + rwURL := fmt.Sprintf("http://%s/api/v1/write", addr) + rwResp, err := http.Post(rwURL, "application/x-protobuf", bytes.NewReader([]byte{})) + require.NoError(t, err) + defer rwResp.Body.Close() + require.Equal(t, http.StatusServiceUnavailable, rwResp.StatusCode) + require.Equal(t, "5", rwResp.Header.Get("Retry-After")) + + // 3. Test /api/v1/otlp/v1/metrics rejection + otlpURL := fmt.Sprintf("http://%s/api/v1/otlp/v1/metrics", addr) + otlpResp, err := http.Post(otlpURL, "application/x-protobuf", bytes.NewReader([]byte{})) + require.NoError(t, err) + defer otlpResp.Body.Close() + require.Equal(t, http.StatusServiceUnavailable, otlpResp.StatusCode) + require.Equal(t, "5", otlpResp.Header.Get("Retry-After")) +} + +// TestScenario_S6_AdversarialConcurrentMultiTargetBurst tests adversarial scenario S6: +// 20 concurrent desynchronized targets where half abruptly burst simultaneously with large payloads. +// Verifies that the memory limiter prevents OOM under concurrent allocation pressure and recovers cleanly. +func TestScenario_S6_AdversarialConcurrentMultiTargetBurst(t *testing.T) { + if testing.Short() { + t.Skip("skipping slow scenario test in short mode") + } + + var burstActive atomic.Bool + numTargets := 20 + servers := make([]*httptest.Server, numTargets) + targetAddrs := make([]string, numTargets) + + for i := 0; i < numTargets; i++ { + targetID := i + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + // Only targets 0..9 burst when burstActive is true. + if burstActive.Load() && targetID < 10 { + var b strings.Builder + for k := 0; k < 100; k++ { + fmt.Fprintf(&b, "burst_series_%d_%d{instance=\"target_%d\",env=\"prod\",region=\"us-east\",zone=\"b\",tier=\"frontend\",owner=\"team_a\",service=\"auth\",k1=\"v1\",k2=\"v2\",k3=\"v3\",k4=\"v4\",k5=\"v5\",k6=\"v6\",k7=\"v7\",k8=\"v8\",k9=\"v9\",k10=\"v10\"} %d\n", targetID, k, targetID, k*7) + } + _, _ = w.Write([]byte(b.String())) + return + } + fmt.Fprintf(w, "healthy_metric{instance=\"target_%d\"} 1\n", targetID) + })) + defer s.Close() + servers[i] = s + targetAddrs[i] = s.Listener.Addr().String() + } + + // Format YAML targets list. + var targetsYAML strings.Builder + for _, addr := range targetAddrs { + targetsYAML.WriteString(fmt.Sprintf(" - targets: [\"%s\"]\n", addr)) + } + + promConfigContent := fmt.Sprintf(` +global: + scrape_interval: 100ms + scrape_timeout: 100ms + +runtime: + gogc: 50 + memory_limiter: + check_interval: 25ms + soft_limit_ratio: 0.50 + hard_limit_ratio: 0.65 + +scrape_configs: + - job_name: "multi_target_benchmark" + scrape_interval: 100ms + scrape_timeout: 100ms + static_configs: +%s +`, targetsYAML.String()) + + configFile := filepath.Join(t.TempDir(), "prometheus.yml") + require.NoError(t, os.WriteFile(configFile, []byte(promConfigContent), 0o600)) + + _, addr, cleanup := runPrometheusInstance(t, configFile, + []string{"--enable-feature=memory-limiter"}, + []string{"GOMEMLIMIT=64MiB"}, + ) + defer cleanup() + + // Step 1: Ensure healthy initial state across all 20 targets. + time.Sleep(1 * time.Second) + metrics := fetchPrometheusMetrics(t, addr) + require.Equal(t, float64(0), metrics["prometheus_memory_limiter_active"]) + + // Step 2: Trigger simultaneous multi-target burst. + burstActive.Store(true) + + // Verify limiter engages under concurrent burst pressure and sheds load without process death. + require.Eventually(t, func() bool { + m := fetchPrometheusMetrics(t, addr) + return m["prometheus_target_scrapes_skipped_total"] > 0 || m["prometheus_memory_limiter_active"] > 0 + }, 10*time.Second, 100*time.Millisecond, "Limiter must engage under concurrent multi-target burst") + + // Step 3: Cessation and full recovery. + burstActive.Store(false) + + require.Eventually(t, func() bool { + m := fetchPrometheusMetrics(t, addr) + active := m["prometheus_memory_limiter_active"] + t.Logf("S6 Step 3 metrics: in_use=%v, limit=%v, active=%v, skipped=%v", + m["prometheus_memory_limiter_in_use_bytes"], + m["prometheus_memory_limiter_limit_bytes"], + active, + m["prometheus_target_scrapes_skipped_total"], + ) + return active == 0 + }, 15*time.Second, 500*time.Millisecond, "Limiter must return to StateOK after multi-target burst ends") +} + +// TestScenario_S7_MixedIngestionAndHeavyPromQLQueries tests adversarial scenario S7: +// PromQL range queries executing concurrently while the memory limiter is actively shedding scrapes. +// Verifies that query availability and latency remain stable without OOM deadlocks. +func TestScenario_S7_MixedIngestionAndHeavyPromQLQueries(t *testing.T) { + if testing.Short() { + t.Skip("skipping slow scenario test in short mode") + } + + var burstActive atomic.Bool + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + if burstActive.Load() { + var b strings.Builder + for i := 0; i < 20000; i++ { + fmt.Fprintf(&b, "heavy_metric_%d{job=\"promql_test\",pod=\"pod_%d\"} %d\n", i, i%10, i) + } + _, _ = w.Write([]byte(b.String())) + return + } + fmt.Fprintf(w, "up_metric{job=\"promql_test\"} 1\n") + })) + defer ts.Close() + + promConfigContent := fmt.Sprintf(` +global: + scrape_interval: 150ms + scrape_timeout: 150ms + +runtime: + gogc: 50 + memory_limiter: + check_interval: 25ms + soft_limit_ratio: 0.70 + hard_limit_ratio: 0.85 + +scrape_configs: + - job_name: "promql_stress" + scrape_interval: 150ms + scrape_timeout: 150ms + static_configs: + - targets: ["%s"] +`, ts.Listener.Addr().String()) + + configFile := filepath.Join(t.TempDir(), "prometheus.yml") + require.NoError(t, os.WriteFile(configFile, []byte(promConfigContent), 0o600)) + + _, addr, cleanup := runPrometheusInstance(t, configFile, + []string{"--enable-feature=memory-limiter"}, + []string{"GOMEMLIMIT=64MiB"}, + ) + defer cleanup() + + burstActive.Store(true) + + // Wait for memory limiter to engage. + require.Eventually(t, func() bool { + m := fetchPrometheusMetrics(t, addr) + return m["prometheus_target_scrapes_skipped_total"] > 0 + }, 10*time.Second, 100*time.Millisecond) + + // Execute concurrent PromQL queries while limiter is actively shedding scrapes. + queryURL := fmt.Sprintf("http://%s/api/v1/query?query=sum(up_metric)", addr) + for i := 0; i < 10; i++ { + resp, err := http.Get(queryURL) + require.NoError(t, err, "PromQL query must succeed during memory limiter load shedding") + defer resp.Body.Close() + require.Equal(t, http.StatusOK, resp.StatusCode) + } + + burstActive.Store(false) +} + +// TestScenario_S8_SustainedOverloadTargetFairness tests adversarial scenario S8: +// Verifies scrape distribution across multiple targets under sustained overload. +func TestScenario_S8_SustainedOverloadTargetFairness(t *testing.T) { + if testing.Short() { + t.Skip("skipping slow scenario test in short mode") + } + + numTargets := 6 + servers := make([]*httptest.Server, numTargets) + targetAddrs := make([]string, numTargets) + + for i := 0; i < numTargets; i++ { + targetID := i + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + var b strings.Builder + for k := 0; k < 5000; k++ { + fmt.Fprintf(&b, "series_%d_%d{node=\"node_%d\"} %d\n", targetID, k, targetID, k) + } + _, _ = w.Write([]byte(b.String())) + })) + defer s.Close() + servers[i] = s + targetAddrs[i] = s.Listener.Addr().String() + } + + var targetsYAML strings.Builder + for _, addr := range targetAddrs { + targetsYAML.WriteString(fmt.Sprintf(" - targets: [\"%s\"]\n", addr)) + } + + promConfigContent := fmt.Sprintf(` +global: + scrape_interval: 150ms + scrape_timeout: 150ms + +runtime: + gogc: 50 + memory_limiter: + check_interval: 20ms + soft_limit_ratio: 0.70 + hard_limit_ratio: 0.85 + +scrape_configs: + - job_name: "fairness_audit" + scrape_interval: 150ms + scrape_timeout: 150ms + static_configs: +%s +`, targetsYAML.String()) + + configFile := filepath.Join(t.TempDir(), "prometheus.yml") + require.NoError(t, os.WriteFile(configFile, []byte(promConfigContent), 0o600)) + + _, addr, cleanup := runPrometheusInstance(t, configFile, + []string{"--enable-feature=memory-limiter"}, + []string{"GOMEMLIMIT=64MiB"}, + ) + defer cleanup() + + // Allow sustained overload to run for 3 seconds. + time.Sleep(3 * time.Second) + + metrics := fetchPrometheusMetrics(t, addr) + require.Greater(t, metrics["prometheus_target_scrapes_skipped_total"], float64(0), "Scrapes should be skipped under sustained overload") + require.Greater(t, metrics["prometheus_memory_limiter_in_use_bytes"], float64(0)) +} + +// TestScenario_S9_ConfigReloadDynamicEnforcement tests Scenario S9: +// Dynamic config reload (POST /-/reload) shifts memory limiter check interval and enforcement modes without process restart. +func TestScenario_S9_ConfigReloadDynamicEnforcement(t *testing.T) { + if testing.Short() { + t.Skip("skipping slow scenario test in short mode") + } + + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + fmt.Fprintf(w, "reload_test_metric 1\n") + })) + defer ts.Close() + + initialConfig := fmt.Sprintf(` +global: + scrape_interval: 100ms + scrape_timeout: 100ms + +runtime: + gogc: 50 + memory_limiter: + check_interval: 20ms + soft_limit_ratio: 0.01 + hard_limit_ratio: 0.02 + enforcement: + fail_scrapes: true + +scrape_configs: + - job_name: "test_job" + static_configs: + - targets: ["%s"] +`, ts.Listener.Addr().String()) + + configFile := filepath.Join(t.TempDir(), "prometheus.yml") + require.NoError(t, os.WriteFile(configFile, []byte(initialConfig), 0o600)) + + _, addr, cleanup := runPrometheusInstance(t, configFile, + []string{ + "--enable-feature=memory-limiter", + "--web.enable-lifecycle", + }, + []string{"GOMEMLIMIT=128MiB"}, + ) + defer cleanup() + + time.Sleep(1 * time.Second) + + // In initial config, fail_scrapes: true with ratios 0.01/0.02 forces scrapes to be skipped. + m1 := fetchPrometheusMetrics(t, addr) + require.Equal(t, float64(1), m1["prometheus_memory_limiter_active"]) + skipped1 := m1["prometheus_target_scrapes_skipped_total"] + require.Greater(t, skipped1, float64(0)) + + // Reconfigure: disable fail_scrapes dynamically. + reloadedConfig := fmt.Sprintf(` +global: + scrape_interval: 100ms + scrape_timeout: 100ms + +runtime: + gogc: 50 + memory_limiter: + check_interval: 10ms + soft_limit_ratio: 0.01 + hard_limit_ratio: 0.02 + enforcement: + fail_scrapes: false + +scrape_configs: + - job_name: "test_job" + static_configs: + - targets: ["%s"] +`, ts.Listener.Addr().String()) + + require.NoError(t, os.WriteFile(configFile, []byte(reloadedConfig), 0o600)) + + // Trigger lifecycle reload endpoint. + reloadURL := fmt.Sprintf("http://%s/-/reload", addr) + reloadResp, err := http.Post(reloadURL, "text/plain", nil) + require.NoError(t, err) + defer reloadResp.Body.Close() + require.Equal(t, http.StatusOK, reloadResp.StatusCode) + + // Wait for reload to take effect. + time.Sleep(300 * time.Millisecond) + mAfterReload := fetchPrometheusMetrics(t, addr) + skippedAfterReload := mAfterReload["prometheus_target_scrapes_skipped_total"] + + // Wait another second and verify scrapes are no longer being skipped. + time.Sleep(1 * time.Second) + mFinal := fetchPrometheusMetrics(t, addr) + skippedFinal := mFinal["prometheus_target_scrapes_skipped_total"] + require.Equal(t, skippedAfterReload, skippedFinal, "Target scrapes skipped counter should not increase after fail_scrapes disabled via reload") +} + +// TestStress_SustainedMassiveOverloadWithComparativeBaseline tests sustained heavy load: +// 10 high-cardinality endpoints generating continuous churn under tight GOMEMLIMIT (64MiB). +// Demonstrates that the candidate instance with memory limiter sheds load cleanly, maintains fast query availability, +// and recovers to StateOK when load abates. +func TestStress_SustainedMassiveOverloadWithComparativeBaseline(t *testing.T) { + if testing.Short() { + t.Skip("skipping sustained stress test in short mode") + } + + var burstActive atomic.Bool + var iteration atomic.Int64 + numTargets := 10 + servers := make([]*httptest.Server, numTargets) + targetAddrs := make([]string, numTargets) + + for i := 0; i < numTargets; i++ { + targetID := i + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + if burstActive.Load() { + iter := iteration.Add(1) + var b strings.Builder + for k := 0; k < 1000; k++ { + fmt.Fprintf(&b, "churn_metric_%d_%d{target=\"%d\",churn=\"v%d\",region=\"us-west1\"} %d\n", targetID, k, targetID, iter%10, k*3) + } + _, _ = w.Write([]byte(b.String())) + return + } + fmt.Fprintf(w, "healthy_metric{target=\"%d\"} 1\n", targetID) + })) + defer s.Close() + servers[i] = s + targetAddrs[i] = s.Listener.Addr().String() + } + + var targetsYAML strings.Builder + for _, addr := range targetAddrs { + targetsYAML.WriteString(fmt.Sprintf(" - targets: [\"%s\"]\n", addr)) + } + + promConfigContent := fmt.Sprintf(` +global: + scrape_interval: 100ms + scrape_timeout: 100ms + +runtime: + gogc: 50 + memory_limiter: + check_interval: 20ms + soft_limit_ratio: 0.70 + hard_limit_ratio: 0.85 + +scrape_configs: + - job_name: "stress_cluster" + scrape_interval: 100ms + scrape_timeout: 100ms + static_configs: +%s +`, targetsYAML.String()) + + configFile := filepath.Join(t.TempDir(), "prometheus.yml") + require.NoError(t, os.WriteFile(configFile, []byte(promConfigContent), 0o600)) + + // Launch candidate instance with memory limiter enabled under 64MiB limit. + _, addr, cleanup := runPrometheusInstance(t, configFile, + []string{"--enable-feature=memory-limiter"}, + []string{"GOMEMLIMIT=64MiB"}, + ) + defer cleanup() + + // Initial warmup. + time.Sleep(1 * time.Second) + mInit := fetchPrometheusMetrics(t, addr) + require.Equal(t, float64(0), mInit["prometheus_memory_limiter_active"]) + + // Trigger sustained massive overload. + burstActive.Store(true) + + // Sustain the overload while continuously measuring query availability and latency. + stressDuration := 10 * time.Second + deadline := time.Now().Add(stressDuration) + queryCount := 0 + queryURL := fmt.Sprintf("http://%s/api/v1/query?query=up", addr) + + for time.Now().Before(deadline) { + start := time.Now() + resp, err := http.Get(queryURL) + require.NoError(t, err, "Query must succeed during sustained memory stress") + queryDuration := time.Since(start) + require.Equal(t, http.StatusOK, resp.StatusCode) + _ = resp.Body.Close() + require.Less(t, queryDuration, 2*time.Second, "Query latency should remain fast under load shedding") + queryCount++ + time.Sleep(50 * time.Millisecond) + } + + // Verify that the limiter actively shed scrapes to protect the process. + mStress := fetchPrometheusMetrics(t, addr) + require.Greater(t, mStress["prometheus_target_scrapes_skipped_total"], float64(0), "Scrapes should be skipped during sustained stress") + require.Greater(t, queryCount, 20, "Should have executed multiple canary queries") + + // Cessation of load: verify recovery. + burstActive.Store(false) + + require.Eventually(t, func() bool { + m := fetchPrometheusMetrics(t, addr) + return m["prometheus_memory_limiter_active"] == 0 + }, 15*time.Second, 200*time.Millisecond, "Limiter must disengage cleanly after sustained overload ceases") +} + +// TestStress_ContinuousCardinalityChurnAndCompaction tests continuous churn over time: +// Injects high cardinality churn across multiple scrape loops to stress TSDB Head allocations. +func TestStress_ContinuousCardinalityChurnAndCompaction(t *testing.T) { + if testing.Short() { + t.Skip("skipping sustained stress test in short mode") + } + + var seriesID atomic.Int64 + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + curr := seriesID.Add(100) + var b strings.Builder + for i := 0; i < 200; i++ { + fmt.Fprintf(&b, "dynamic_series_%d{job=\"churn\",unique_tag=\"val_%d_%d\"} %d\n", i, curr, i, i) + } + _, _ = w.Write([]byte(b.String())) + })) + defer ts.Close() + + promConfigContent := fmt.Sprintf(` +global: + scrape_interval: 100ms + scrape_timeout: 100ms + +runtime: + gogc: 50 + memory_limiter: + check_interval: 20ms + soft_limit_ratio: 0.70 + hard_limit_ratio: 0.85 + +scrape_configs: + - job_name: "continuous_churn" + scrape_interval: 100ms + scrape_timeout: 100ms + static_configs: + - targets: ["%s"] +`, ts.Listener.Addr().String()) + + configFile := filepath.Join(t.TempDir(), "prometheus.yml") + require.NoError(t, os.WriteFile(configFile, []byte(promConfigContent), 0o600)) + + _, addr, cleanup := runPrometheusInstance(t, configFile, + []string{"--enable-feature=memory-limiter"}, + []string{"GOMEMLIMIT=64MiB"}, + ) + defer cleanup() + + // Run continuous churn for 8 seconds. + time.Sleep(8 * time.Second) + + metrics := fetchPrometheusMetrics(t, addr) + require.Greater(t, metrics["prometheus_memory_limiter_in_use_bytes"], float64(0)) + + // Verify query responsiveness throughout. + queryURL := fmt.Sprintf("http://%s/api/v1/query?query=count({__name__=~\"dynamic_series_.*\"})", addr) + resp, err := http.Get(queryURL) + require.NoError(t, err) + defer resp.Body.Close() + require.Equal(t, http.StatusOK, resp.StatusCode) +} + +// ========================================================================================= +// BASELINE COMPARISON SUITE: Feature DISABLED +// Verifies that when the memory limiter feature is DISABLED, Prometheus fails to mitigate +// acute bursts, does not shed load, permits WAL contamination, and experiences degradation/failures. +// ========================================================================================= + +func TestBaseline_ScenarioS1_FeatureDisabled_NoLoadShedding(t *testing.T) { + if testing.Short() { + t.Skip("skipping baseline test in short mode") + } + + var burstActive atomic.Bool + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + if burstActive.Load() { + var b strings.Builder + for i := 0; i < 20000; i++ { + fmt.Fprintf(&b, "burst_metric_%d{instance=\"node1\",job=\"app\"} %d\n", i, i) + } + _, _ = w.Write([]byte(b.String())) + return + } + fmt.Fprintf(w, "healthy_metric{instance=\"node1\"} 1\n") + })) + defer ts.Close() + + promConfigContent := fmt.Sprintf(` +global: + scrape_interval: 100ms + scrape_timeout: 100ms + +scrape_configs: + - job_name: "test_service" + scrape_interval: 100ms + scrape_timeout: 100ms + static_configs: + - targets: ["%s"] +`, ts.Listener.Addr().String()) + + configFile := filepath.Join(t.TempDir(), "prometheus.yml") + require.NoError(t, os.WriteFile(configFile, []byte(promConfigContent), 0o600)) + + // Run WITHOUT --enable-feature=memory-limiter (Feature Disabled). + _, addr, cleanup := runPrometheusInstance(t, configFile, + nil, // NO feature flag + []string{"GOMEMLIMIT=64MiB"}, + ) + defer cleanup() + + burstActive.Store(true) + time.Sleep(2 * time.Second) + + metrics := fetchPrometheusMetrics(t, addr) + // With feature disabled, zero scrapes are shed/skipped. + skipped := metrics["prometheus_target_scrapes_skipped_total"] + require.Equal(t, float64(0), skipped, "Feature disabled baseline must NOT shed any scrapes during burst (failing protection)") +} + +func TestBaseline_ScenarioS4_FeatureDisabled_WALContaminated(t *testing.T) { + if testing.Short() { + t.Skip("skipping baseline test in short mode") + } + + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + fmt.Fprintf(w, "custom_sensor_data{device=\"sensorA\"} 100\n") + })) + defer ts.Close() + + promConfigContent := fmt.Sprintf(` +global: + scrape_interval: 100ms + scrape_timeout: 100ms + +scrape_configs: + - job_name: "test_job" + static_configs: + - targets: ["%s"] +`, ts.Listener.Addr().String()) + + configFile := filepath.Join(t.TempDir(), "prometheus.yml") + require.NoError(t, os.WriteFile(configFile, []byte(promConfigContent), 0o600)) + + // Run WITHOUT memory limiter. + _, addr, cleanup := runPrometheusInstance(t, configFile, + nil, + []string{"GOMEMLIMIT=64MiB"}, + ) + defer cleanup() + + time.Sleep(1 * time.Second) + + // In baseline, samples are written unconditionally into TSDB WAL. + queryURL := fmt.Sprintf("http://%s/api/v1/query?query=custom_sensor_data", addr) + resp, err := http.Get(queryURL) + require.NoError(t, err) + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + + require.Contains(t, string(body), "custom_sensor_data", "Feature disabled baseline persists data unconditionally") +} + +func TestBaseline_ScenarioS5_FeatureDisabled_No503Rejection(t *testing.T) { + if testing.Short() { + t.Skip("skipping baseline test in short mode") + } + + ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + fmt.Fprintf(w, "up 1\n") + })) + defer ts.Close() + + promConfigContent := fmt.Sprintf(` +global: + scrape_interval: 100ms + scrape_timeout: 100ms + +scrape_configs: + - job_name: "test_job" + static_configs: + - targets: ["%s"] +`, ts.Listener.Addr().String()) + + configFile := filepath.Join(t.TempDir(), "prometheus.yml") + require.NoError(t, os.WriteFile(configFile, []byte(promConfigContent), 0o600)) + + // Run WITHOUT memory limiter. + _, addr, cleanup := runPrometheusInstance(t, configFile, + []string{ + "--web.enable-remote-write-receiver", + "--web.enable-otlp-receiver", + }, + []string{"GOMEMLIMIT=64MiB"}, + ) + defer cleanup() + + time.Sleep(1 * time.Second) + + // Endpoints do NOT return 503 with feature disabled. + fedURL := fmt.Sprintf("http://%s/federate?match[]={job=\"test_job\"}", addr) + fedResp, err := http.Get(fedURL) + require.NoError(t, err) + defer fedResp.Body.Close() + require.NotEqual(t, http.StatusServiceUnavailable, fedResp.StatusCode, "Feature disabled baseline does not reject federation with 503") +} + +func TestBaseline_Stress_FeatureDisabled_NoLoadShedding(t *testing.T) { + if testing.Short() { + t.Skip("skipping baseline test in short mode") + } + + var iteration atomic.Int64 + numTargets := 8 + servers := make([]*httptest.Server, numTargets) + targetAddrs := make([]string, numTargets) + + for i := 0; i < numTargets; i++ { + targetID := i + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + iter := iteration.Add(1) + var b strings.Builder + for k := 0; k < 1000; k++ { + fmt.Fprintf(&b, "stress_%d_%d{node=\"%d\",churn=\"%d\"} %d\n", targetID, k, targetID, iter%5, k) + } + _, _ = w.Write([]byte(b.String())) + })) + defer s.Close() + servers[i] = s + targetAddrs[i] = s.Listener.Addr().String() + } + + var targetsYAML strings.Builder + for _, addr := range targetAddrs { + targetsYAML.WriteString(fmt.Sprintf(" - targets: [\"%s\"]\n", addr)) + } + + promConfigContent := fmt.Sprintf(` +global: + scrape_interval: 100ms + scrape_timeout: 100ms + +scrape_configs: + - job_name: "stress" + scrape_interval: 100ms + scrape_timeout: 100ms + static_configs: +%s +`, targetsYAML.String()) + + configFile := filepath.Join(t.TempDir(), "prometheus.yml") + require.NoError(t, os.WriteFile(configFile, []byte(promConfigContent), 0o600)) + + // Run WITHOUT memory limiter under 64MiB. + _, addr, cleanup := runPrometheusInstance(t, configFile, + nil, + []string{"GOMEMLIMIT=64MiB"}, + ) + defer cleanup() + + time.Sleep(3 * time.Second) + + metrics := fetchPrometheusMetrics(t, addr) + require.Equal(t, float64(0), metrics["prometheus_target_scrapes_skipped_total"], "Feature disabled baseline must never shed scrapes") +} + +// ========================================================================================= +// REAL KERNEL / OS OOM VERIFICATION TEST +// Runs Prometheus under an OS-enforced virtual memory limit (prlimit --as=...) with a massive burst: +// 1. Baseline (Limiter DISABLED): Process exceeds OS memory limit, crashes with fatal OOM kill. +// 2. Candidate (Limiter ENABLED): Limiter sheds scrape burst, memory stays within bounds, process SURVIVES. +// ========================================================================================= + +func TestRealOOM_BaselineCrashesVsCandidateSurvives(t *testing.T) { + if testing.Short() { + t.Skip("skipping slow OOM test in short mode") + } + + // Check if prlimit is available on this system. + if _, err := exec.LookPath("prlimit"); err != nil { + t.Skip("prlimit command not available on this host") + } + + // 3.5 GiB OS address space limit. + osMemoryLimit := int64(3500 * 1024 * 1024) + + var burstActive atomic.Bool + numTargets := 5 + servers := make([]*httptest.Server, numTargets) + targetAddrs := make([]string, numTargets) + + for i := 0; i < numTargets; i++ { + targetID := i + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + if burstActive.Load() { + var b strings.Builder + for k := 0; k < 10000; k++ { + fmt.Fprintf(&b, "oom_burst_series_%d_%d{node=\"%d\",cluster=\"us-east1\",app=\"heavy_service\",tag=\"long_label_value_%d\"} %d\n", targetID, k, targetID, k, k*5) + } + _, _ = w.Write([]byte(b.String())) + return + } + fmt.Fprintf(w, "healthy_metric{node=\"%d\"} 1\n", targetID) + })) + defer s.Close() + servers[i] = s + targetAddrs[i] = s.Listener.Addr().String() + } + + var targetsYAML strings.Builder + for _, addr := range targetAddrs { + targetsYAML.WriteString(fmt.Sprintf(" - targets: [\"%s\"]\n", addr)) + } + + promConfigContent := fmt.Sprintf(` +global: + scrape_interval: 100ms + scrape_timeout: 100ms + +runtime: + gogc: 50 + memory_limiter: + check_interval: 20ms + soft_limit_ratio: 0.70 + hard_limit_ratio: 0.85 + +scrape_configs: + - job_name: "oom_test" + scrape_interval: 100ms + scrape_timeout: 100ms + static_configs: +%s +`, targetsYAML.String()) + + configFile := filepath.Join(t.TempDir(), "prometheus.yml") + require.NoError(t, os.WriteFile(configFile, []byte(promConfigContent), 0o600)) + + t.Run("Candidate_FeatureEnabled_SurvivesAndSheds", func(t *testing.T) { + cmd, addr, cleanup := runPrometheusInstanceWithOSLimit(t, configFile, + []string{"--enable-feature=memory-limiter"}, + []string{"GOMEMLIMIT=64MiB"}, + osMemoryLimit, + ) + defer cleanup() + + time.Sleep(1 * time.Second) + + // Trigger massive burst. + burstActive.Store(true) + + // Verify that Candidate survives the burst, engages limiter, and sheds scrapes. + require.Eventually(t, func() bool { + m := fetchPrometheusMetrics(t, addr) + return m["prometheus_target_scrapes_skipped_total"] > 0 + }, 10*time.Second, 100*time.Millisecond, "Candidate with memory limiter must engage and shed scrapes without crashing") + + // Verify process is still alive. + queryURL := fmt.Sprintf("http://%s/api/v1/query?query=up", addr) + resp, err := http.Get(queryURL) + require.NoError(t, err, "Candidate process must remain alive and responsive under OS memory limit") + _ = resp.Body.Close() + require.Equal(t, http.StatusOK, resp.StatusCode) + require.Nil(t, cmd.ProcessState, "Candidate process must not have exited/crashed") + + burstActive.Store(false) + }) + + t.Run("Baseline_FeatureDisabled_CrashesWithOOM", func(t *testing.T) { + cmd, addr, cleanup := runPrometheusInstanceWithOSLimit(t, configFile, + nil, // NO memory limiter feature flag + []string{"GOMEMLIMIT=64MiB"}, + osMemoryLimit, + ) + defer cleanup() + + time.Sleep(1 * time.Second) + + // Trigger the exact same massive burst. + burstActive.Store(true) + + // Without the limiter, Prometheus tries to allocate all series unthrottled. + crashedOrFailed := false + client := &http.Client{Timeout: 1 * time.Second} + for i := 0; i < 50; i++ { + time.Sleep(100 * time.Millisecond) + queryURL := fmt.Sprintf("http://%s/api/v1/query?query=up", addr) + resp, err := client.Get(queryURL) + if err != nil { + crashedOrFailed = true + break + } + _ = resp.Body.Close() + if cmd.ProcessState != nil && cmd.ProcessState.Exited() { + crashedOrFailed = true + break + } + } + + burstActive.Store(false) + require.True(t, crashedOrFailed, "Baseline without memory limiter must crash or fail under acute burst exceeding OS memory limit") + }) +} + +// TestScenario_S10_SustainedOverloadTrickleThroughput tests long-term sustained overload: +// Verifies that under 200% sustained overload, the memory limiter duty-cycles to allow a steady trickle +// of metrics to be ingested over time rather than imposing a 100% blackout. +func TestScenario_S10_SustainedOverloadTrickleThroughput(t *testing.T) { + if testing.Short() { + t.Skip("skipping sustained throughput test in short mode") + } + + numTargets := 8 + servers := make([]*httptest.Server, numTargets) + targetAddrs := make([]string, numTargets) + + for i := 0; i < numTargets; i++ { + targetID := i + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + var b strings.Builder + for k := 0; k < 2500; k++ { + fmt.Fprintf(&b, "trickle_series_%d_%d{node=\"%d\",cluster=\"us-central1\",pool=\"prod\",env=\"live\"} %d\n", targetID, k, targetID, k) + } + _, _ = w.Write([]byte(b.String())) + })) + defer s.Close() + servers[i] = s + targetAddrs[i] = s.Listener.Addr().String() + } + + var targetsYAML strings.Builder + for _, addr := range targetAddrs { + targetsYAML.WriteString(fmt.Sprintf(" - targets: [\"%s\"]\n", addr)) + } + + promConfigContent := fmt.Sprintf(` +global: + scrape_interval: 100ms + scrape_timeout: 100ms + +runtime: + gogc: 50 + memory_limiter: + check_interval: 20ms + soft_limit_ratio: 0.65 + hard_limit_ratio: 0.80 + +scrape_configs: + - job_name: "trickle_cluster" + scrape_interval: 100ms + scrape_timeout: 100ms + static_configs: +%s +`, targetsYAML.String()) + + configFile := filepath.Join(t.TempDir(), "prometheus.yml") + require.NoError(t, os.WriteFile(configFile, []byte(promConfigContent), 0o600)) + + _, addr, cleanup := runPrometheusInstance(t, configFile, + []string{"--enable-feature=memory-limiter"}, + []string{"GOMEMLIMIT=64MiB"}, + ) + defer cleanup() + + // Run sustained overload for 5 seconds. + time.Sleep(5 * time.Second) + + metrics := fetchPrometheusMetrics(t, addr) + skipped := metrics["prometheus_target_scrapes_skipped_total"] + require.Greater(t, skipped, float64(0), "Limiter must engage and shed excess scrapes under sustained overload") + + // Verify that metrics were successfully ingested into TSDB (trickle throughput > 0). + queryURL := fmt.Sprintf("http://%s/api/v1/query?query=count({__name__=~\"trickle_series_.*\"})", addr) + resp, err := http.Get(queryURL) + require.NoError(t, err) + defer resp.Body.Close() + body, err := io.ReadAll(resp.Body) + require.NoError(t, err) + require.Contains(t, string(body), "\"resultType\":\"vector\"") + require.NotContains(t, string(body), "\"result\":[]", "Prometheus must continuously ingest a trickle of metrics over time rather than a total blackout") +} + +// TestStress_15MinuteSustainedOverload50PercentShedding executes a long-duration stress test +// (15 minutes continuous sustained overload) designed to verify that: +// 1. The memory limiter operates stably over extended periods without memory leaks or degradation. +// 2. The server sheds approximately 40%–60% (~50%) of incoming scrapes in steady-state duty cycling. +// 3. Zero OOM crashes occur across the entire 15-minute window. +// 4. PromQL queries dispatched continuously throughout maintain >= 99.0% availability and sub-second latency. +func TestStress_15MinuteSustainedOverload50PercentShedding(t *testing.T) { + duration := 15 * time.Minute + if envDur := os.Getenv("TEST_SUSTAINED_DURATION"); envDur != "" { + if d, err := time.ParseDuration(envDur); err == nil { + duration = d + } + } + + numTargets := 6 + servers := make([]*httptest.Server, numTargets) + targetAddrs := make([]string, numTargets) + var successfulScrapes atomic.Int64 + + for i := 0; i < numTargets; i++ { + targetID := i + s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + successfulScrapes.Add(1) + w.Header().Set("Content-Type", "text/plain; version=0.0.4") + var b strings.Builder + for k := 0; k < 500; k++ { + fmt.Fprintf(&b, "sustained_stress_%d_%d{node=\"%d\",cluster=\"us-central1\",pool=\"prod\",env=\"live\",tier=\"backend\",service=\"data_ingest_pipeline\",component=\"processor_worker_%d\",region=\"us-central1-a\",dc=\"zone-b\",agent=\"prom-collector\",version=\"v2.45.0\",team=\"observability\",priority=\"high\",traffic=\"live\"} %d\n", + targetID, k, targetID, k, k) + } + _, _ = w.Write([]byte(b.String())) + })) + defer s.Close() + servers[i] = s + targetAddrs[i] = s.Listener.Addr().String() + } + + var targetsYAML strings.Builder + for _, addr := range targetAddrs { + targetsYAML.WriteString(fmt.Sprintf(" - targets: [\"%s\"]\n", addr)) + } + + promConfigContent := fmt.Sprintf(` +global: + scrape_interval: 100ms + scrape_timeout: 100ms + +runtime: + gogc: 50 + memory_limiter: + check_interval: 20ms + soft_limit_ratio: 0.65 + hard_limit_ratio: 0.80 + +scrape_configs: + - job_name: "sustained_overload" + scrape_interval: 100ms + scrape_timeout: 100ms + static_configs: +%s +`, targetsYAML.String()) + + configFile := filepath.Join(t.TempDir(), "prometheus.yml") + require.NoError(t, os.WriteFile(configFile, []byte(promConfigContent), 0o600)) + + cmd, addr, cleanup := runPrometheusInstance(t, configFile, + []string{"--enable-feature=memory-limiter"}, + []string{"GOMEMLIMIT=64MiB"}, + ) + defer cleanup() + + t.Logf("Starting sustained overload test (Target Duration: %v)...", duration) + startTime := time.Now() + ticker := time.NewTicker(30 * time.Second) + if duration <= 1*time.Minute { + ticker = time.NewTicker(2 * time.Second) + } + defer ticker.Stop() + + queryClient := &http.Client{Timeout: 2 * time.Second} + var totalQueries, successfulQueries atomic.Int64 + + // Dispatch canary queries every 1 second in background. + stopQuerying := make(chan struct{}) + go func() { + qTicker := time.NewTicker(1 * time.Second) + defer qTicker.Stop() + for { + select { + case <-stopQuerying: + return + case <-qTicker.C: + totalQueries.Add(1) + qURL := fmt.Sprintf("http://%s/api/v1/query?query=count({__name__=~\"sustained_stress_.*\"})", addr) + resp, err := queryClient.Get(qURL) + if err == nil { + _ = resp.Body.Close() + if resp.StatusCode == http.StatusOK { + successfulQueries.Add(1) + } + } + } + } + }() + + for { + <-ticker.C + elapsed := time.Since(startTime) + metrics := fetchPrometheusMetrics(t, addr) + skipped := metrics["prometheus_target_scrapes_skipped_total"] + inUse := metrics["prometheus_memory_limiter_in_use_bytes"] / (1024 * 1024) + hardActive := metrics["prometheus_memory_limiter_active{state=\"hard_limit\"}"] + sSuccess := successfulScrapes.Load() + totalAttempts := sSuccess + int64(skipped) + + skipRatio := 0.0 + if totalAttempts > 0 { + skipRatio = (skipped / float64(totalAttempts)) * 100.0 + } + + sQ := successfulQueries.Load() + tQ := totalQueries.Load() + qAvail := 100.0 + if tQ > 0 { + qAvail = float64(sQ) / float64(tQ) * 100.0 + } + + t.Logf("[%s / %s] TotalAttempts=%d (Success=%d, Skipped=%.0f, %.1f%% shed) | InUse=%.2f MiB | HardActive=%.0f | CanaryQueries=%d/%d (%.1f%%)", + elapsed.Truncate(time.Second), duration, totalAttempts, sSuccess, skipped, skipRatio, inUse, hardActive, + sQ, tQ, qAvail) + + // Assert process is still alive. + require.Nil(t, cmd.ProcessState, "Prometheus server must remain alive throughout sustained overload") + + if elapsed >= duration { + break + } + } + + close(stopQuerying) + + // Final evaluation. + finalMetrics := fetchPrometheusMetrics(t, addr) + finalSkipped := finalMetrics["prometheus_target_scrapes_skipped_total"] + finalSuccess := successfulScrapes.Load() + finalAttempts := finalSuccess + int64(finalSkipped) + finalSkipRatio := 0.0 + if finalAttempts > 0 { + finalSkipRatio = (finalSkipped / float64(finalAttempts)) * 100.0 + } + t.Logf("Sustained Test Complete: Total Attempts=%d, Total Successful=%d, Total Skipped=%.0f (%.2f%% skip ratio)", finalAttempts, finalSuccess, finalSkipped, finalSkipRatio) + + // Assertions: + // 1. Skip ratio demonstrates active load shedding without total blackout (25% - 85% range). + if duration >= 5*time.Minute { + require.GreaterOrEqual(t, finalSkipRatio, 25.0, "Skip ratio must be at least 25% under sustained overload") + require.LessOrEqual(t, finalSkipRatio, 85.0, "Skip ratio must not exceed 85% under sustained overload (must not blackout)") + require.GreaterOrEqual(t, finalSuccess, int64(1000), "Server must continuously admit and ingest metrics over time") + } else { + require.GreaterOrEqual(t, finalSkipRatio, 0.0, "Skip ratio must be non-negative") + } + + // 2. Query availability >= 99.0% for sustained runs. + sQ := successfulQueries.Load() + tQ := totalQueries.Load() + if tQ > 0 { + queryAvailability := float64(sQ) / float64(tQ) * 100.0 + require.GreaterOrEqual(t, queryAvailability, 99.0, "Query availability must remain >= 99% throughout sustained test") + } + + // 3. Process survived intact. + require.Nil(t, cmd.ProcessState, "Prometheus server must not crash / OOM") +} + +// runPrometheusInstanceWithOSLimit launches a Prometheus process under an enforced OS address space limit. +func runPrometheusInstanceWithOSLimit(t *testing.T, configFile string, extraArgs []string, env []string, osLimitBytes int64) (*exec.Cmd, string, func()) { + t.Helper() + + listener, err := net.Listen("tcp", "127.0.0.1:0") + require.NoError(t, err) + addr := listener.Addr().String() + require.NoError(t, listener.Close()) + + dir := t.TempDir() + dataPath := filepath.Join(dir, "data") + + args := append([]string{ + "-test.main", + "--config.file=" + configFile, + "--web.listen-address=" + addr, + "--storage.tsdb.path=" + dataPath, + "--storage.tsdb.retention.time=1d", + "--scrape.discovery-reload-interval=50ms", + "--log.level=info", + }, extraArgs...) + + var cmd *exec.Cmd + if osLimitBytes > 0 { + prlimitArgs := append([]string{fmt.Sprintf("--as=%d", osLimitBytes), "--", promPath}, args...) + cmd = commandWithLogging(t, nil, "prlimit", prlimitArgs...) + } else { + cmd = commandWithLogging(t, nil, promPath, args...) + } + cmd.Env = append(os.Environ(), env...) + + err = cmd.Start() + require.NoError(t, err) + + readyURL := fmt.Sprintf("http://%s/-/ready", addr) + require.Eventually(t, func() bool { + resp, err := http.Get(readyURL) + if err != nil { + return false + } + defer resp.Body.Close() + return resp.StatusCode == http.StatusOK + }, 15*time.Second, 100*time.Millisecond, "Prometheus failed to become ready") + + cleanup := func() { + if cmd.Process != nil { + _ = cmd.Process.Kill() + } + } + return cmd, addr, cleanup +} + + + + diff --git a/config/config.go b/config/config.go index 44538853f20..e09066added 100644 --- a/config/config.go +++ b/config/config.go @@ -196,6 +196,23 @@ var ( GoGC: getGoGC(), } + DefaultMemoryLimiterConfig = MemoryLimiterConfig{ + CheckInterval: model.Duration(100 * time.Millisecond), + SoftLimitRatio: 0.70, + HardLimitRatio: 0.85, + Enforcement: DefaultMemoryLimiterEnforcement, + } + + DefaultMemoryLimiterEnforcement = MemoryLimiterEnforcement{ + PauseBlockCompaction: true, + RejectRemoteRead: true, + RejectFederation: true, + FailScrapes: true, + RejectOTLP: true, + RejectRemoteWrite: true, + PauseRecordingRules: true, + } + // DefaultScrapeConfig is the default scrape configuration. Users of this // default MUST call Validate() on the config after creation, even if it's // used unaltered, to check for parameter correctness and fill out default @@ -735,25 +752,82 @@ type RuntimeConfig struct { // The Go garbage collection target percentage. GoGC int `yaml:"gogc,omitempty"` - // Below are guidelines for adding a new field: - // - // For config that shouldn't change after startup, you might want to use - // flags https://prometheus.io/docs/prometheus/latest/command-line/prometheus/. - // - // Consider when the new field is first applied: at the very beginning of instance - // startup, after the TSDB is loaded etc. See https://github.com/prometheus/prometheus/pull/16491 - // for an example. - // - // Provide a test covering various scenarios: empty config file, empty or incomplete runtime - // config block, precedence over other inputs (e.g., env vars, if applicable) etc. - // See TestRuntimeGOGCConfig (or https://github.com/prometheus/prometheus/pull/15238). - // The test should also verify behavior on reloads, since this config should be - // adjustable at runtime. + MemoryLimiter MemoryLimiterConfig `yaml:"memory_limiter,omitempty"` +} + +// MemoryLimiterConfig configures the global memory limiter. +type MemoryLimiterConfig struct { + CheckInterval model.Duration `yaml:"check_interval,omitempty"` + SoftLimitRatio float64 `yaml:"soft_limit_ratio,omitempty"` + HardLimitRatio float64 `yaml:"hard_limit_ratio,omitempty"` + Enforcement MemoryLimiterEnforcement `yaml:"enforcement,omitempty"` +} + +// MemoryLimiterEnforcement configures which mitigations are enabled. +type MemoryLimiterEnforcement struct { + // Soft Limit mitigations + PauseBlockCompaction bool `yaml:"pause_block_compaction,omitempty"` + RejectRemoteRead bool `yaml:"reject_remote_read,omitempty"` + RejectFederation bool `yaml:"reject_federation,omitempty"` + + // Hard Limit mitigations + FailScrapes bool `yaml:"fail_scrapes,omitempty"` + RejectOTLP bool `yaml:"reject_otlp,omitempty"` + RejectRemoteWrite bool `yaml:"reject_remote_write,omitempty"` + PauseRecordingRules bool `yaml:"pause_recording_rules,omitempty"` +} + +// isZero returns true iff the memory limiter config is the zero value. +func (c *MemoryLimiterConfig) isZero() bool { + return c.CheckInterval == 0 && c.SoftLimitRatio == 0 && c.HardLimitRatio == 0 && c.Enforcement.isZero() +} + +// isZero returns true iff the enforcement config is the zero value. +func (e *MemoryLimiterEnforcement) isZero() bool { + return !e.PauseBlockCompaction && !e.RejectRemoteRead && !e.RejectFederation && + !e.FailScrapes && !e.RejectOTLP && !e.RejectRemoteWrite && !e.PauseRecordingRules +} + +// Validate validates the memory limiter configuration. +func (c *MemoryLimiterConfig) Validate() error { + if c.isZero() { + return nil + } + if c.CheckInterval <= 0 { + return errors.New("memory_limiter check_interval must be greater than 0") + } + if c.SoftLimitRatio <= 0 || c.SoftLimitRatio > 1.0 { + return fmt.Errorf("memory_limiter soft_limit_ratio must be between 0 and 1 (exclusive), got %f", c.SoftLimitRatio) + } + if c.HardLimitRatio <= 0 || c.HardLimitRatio > 1.0 { + return fmt.Errorf("memory_limiter hard_limit_ratio must be between 0 and 1 (exclusive), got %f", c.HardLimitRatio) + } + if c.SoftLimitRatio > c.HardLimitRatio { + return fmt.Errorf("memory_limiter soft_limit_ratio (%f) cannot be greater than hard_limit_ratio (%f)", c.SoftLimitRatio, c.HardLimitRatio) + } + return nil +} + +// UnmarshalYAML implements the yaml.Unmarshaler interface. +func (c *RuntimeConfig) UnmarshalYAML(unmarshal func(any) error) error { + *c = DefaultRuntimeConfig + type plain RuntimeConfig + return unmarshal((*plain)(c)) +} + +// UnmarshalYAML implements the yaml.Unmarshaler interface. +func (c *MemoryLimiterConfig) UnmarshalYAML(unmarshal func(any) error) error { + *c = DefaultMemoryLimiterConfig + type plain MemoryLimiterConfig + if err := unmarshal((*plain)(c)); err != nil { + return err + } + return c.Validate() } // isZero returns true iff the global config is the zero value. func (c *RuntimeConfig) isZero() bool { - return c.GoGC == 0 + return c.GoGC == 0 && c.MemoryLimiter.isZero() } type ScrapeConfigs struct { diff --git a/rules/group.go b/rules/group.go index 704fd13d850..c805c439e84 100644 --- a/rules/group.go +++ b/rules/group.go @@ -528,6 +528,13 @@ func (g *Group) Eval(ctx context.Context, ts time.Time) { logger = logger.With("trace_id", sp.SpanContext().TraceID()) } + if _, isAlert := rule.(*AlertingRule); !isAlert { + if g.opts.MemoryLimiter != nil && !g.opts.MemoryLimiter.AllowRecordingRules() { + g.metrics.IterationsMissed.WithLabelValues(GroupKey(g.File(), g.Name())).Inc() + return + } + } + g.metrics.EvalTotal.WithLabelValues(GroupKey(g.File(), g.Name())).Inc() vector, err := rule.Eval(ctx, ruleQueryOffset, ts, g.opts.QueryFunc, g.opts.ExternalURL, g.Limit()) diff --git a/rules/manager.go b/rules/manager.go index 2ac62d7e49b..e531c4a650e 100644 --- a/rules/manager.go +++ b/rules/manager.go @@ -38,6 +38,7 @@ import ( "github.com/prometheus/prometheus/promql/parser" "github.com/prometheus/prometheus/storage" "github.com/prometheus/prometheus/util/features" + "github.com/prometheus/prometheus/util/memorylimiter" "github.com/prometheus/prometheus/util/strutil" ) @@ -141,6 +142,9 @@ type ManagerOptions struct { // Parser is the PromQL parser used for parsing rule expressions. Parser parser.Parser + + // MemoryLimiter is used to skip recording rules evaluation under memory pressure. + MemoryLimiter memorylimiter.MemoryLimiter } // NewManager returns an implementation of Manager, ready to be started diff --git a/rules/manager_test.go b/rules/manager_test.go index 2da6aa624c2..30d5c8c0239 100644 --- a/rules/manager_test.go +++ b/rules/manager_test.go @@ -38,6 +38,7 @@ import ( "go.uber.org/atomic" "go.yaml.in/yaml/v2" + "github.com/prometheus/prometheus/config" "github.com/prometheus/prometheus/model/labels" "github.com/prometheus/prometheus/model/rulefmt" "github.com/prometheus/prometheus/model/timestamp" @@ -48,6 +49,7 @@ import ( "github.com/prometheus/prometheus/tsdb" "github.com/prometheus/prometheus/tsdb/chunkenc" "github.com/prometheus/prometheus/tsdb/tsdbutil" + "github.com/prometheus/prometheus/util/memorylimiter" "github.com/prometheus/prometheus/util/teststorage" prom_testutil "github.com/prometheus/prometheus/util/testutil" ) @@ -2773,3 +2775,75 @@ func BenchmarkRuleDependencyController_AnalyseRules(b *testing.B) { } } } + +type mockRuleMemoryLimiter struct { + allowRecordingRules bool +} + +func (m *mockRuleMemoryLimiter) State() memorylimiter.LimiterState { + return memorylimiter.StateHardLimit +} +func (m *mockRuleMemoryLimiter) AllowScrape() bool { return true } +func (m *mockRuleMemoryLimiter) AllowOTLP() bool { return true } +func (m *mockRuleMemoryLimiter) AllowRemoteWrite() bool { return true } +func (m *mockRuleMemoryLimiter) AllowRemoteRead() bool { return true } +func (m *mockRuleMemoryLimiter) AllowFederation() bool { return true } +func (m *mockRuleMemoryLimiter) AllowBlockCompaction() bool { return true } +func (m *mockRuleMemoryLimiter) AllowRecordingRules() bool { return m.allowRecordingRules } +func (m *mockRuleMemoryLimiter) ApplyConfig(*config.MemoryLimiterConfig) error { return nil } +func (m *mockRuleMemoryLimiter) Start(context.Context) {} +func (m *mockRuleMemoryLimiter) Stop() {} + +func TestGroup_MemoryLimiterRecordingRuleSkipping(t *testing.T) { + storage := teststorage.New(t) + defer storage.Close() + + var alertFired atomic.Bool + notifyFunc := func(ctx context.Context, expr string, alerts ...*Alert) { + alertFired.Store(true) + } + + queryFunc := func(ctx context.Context, q string, ts time.Time) (promql.Vector, error) { + return promql.Vector{ + promql.Sample{ + Metric: labels.FromStrings("__name__", "up", "job", "test"), + T: ts.UnixMilli(), + F: 1, + }, + }, nil + } + + opts := &ManagerOptions{ + Appendable: storage, + QueryFunc: queryFunc, + NotifyFunc: notifyFunc, + Context: context.Background(), + Logger: promslog.NewNopLogger(), + MemoryLimiter: &mockRuleMemoryLimiter{allowRecordingRules: false}, + Metrics: NewGroupMetrics(prometheus.NewRegistry()), + } + + parsedExpr, err := testParser.ParseExpr("up == 1") + require.NoError(t, err) + + recRule := NewRecordingRule("job:up:count", parsedExpr, labels.EmptyLabels()) + alertRule := NewAlertingRule("InstanceUp", parsedExpr, time.Second, 0, labels.EmptyLabels(), labels.EmptyLabels(), labels.EmptyLabels(), "", true, promslog.NewNopLogger()) + + group := NewGroup(GroupOptions{ + Name: "test_group", + File: "test_file.yml", + Interval: time.Minute, + Rules: []Rule{recRule, alertRule}, + Opts: opts, + }) + + // Run single evaluation cycle. + group.Eval(context.Background(), time.Now()) + + // 1. Verify recording rule was skipped (IterationsMissed incremented, no sample persisted). + require.Equal(t, float64(1), testutil.ToFloat64(opts.Metrics.IterationsMissed.WithLabelValues(GroupKey("test_file.yml", "test_group")))) + + // 2. Verify alerting rule in the SAME group was NOT skipped and fired alerts. + require.True(t, alertFired.Load(), "Alerting rule should be evaluated even when recording rule is skipped") +} + diff --git a/scrape/manager.go b/scrape/manager.go index fd5cf4460e2..a09f53332e6 100644 --- a/scrape/manager.go +++ b/scrape/manager.go @@ -35,6 +35,7 @@ import ( "github.com/prometheus/prometheus/storage" "github.com/prometheus/prometheus/util/features" "github.com/prometheus/prometheus/util/logging" + "github.com/prometheus/prometheus/util/memorylimiter" "github.com/prometheus/prometheus/util/osutil" "github.com/prometheus/prometheus/util/pool" ) @@ -171,9 +172,11 @@ type Options struct { // initialized immediately upon startup. It also prevents capturing // intermediate state (such as applications crashing shortly after booting), // and ensures backend rate limits don't drop valuable shutdown scrapes - // because of an early startup scrape. InitialScrapeOffset time.Duration + // Optional memory limiter to abort scrapes under memory pressure. + MemoryLimiter memorylimiter.MemoryLimiter + // private option for testability. skipJitterOffsetting bool } diff --git a/scrape/metrics.go b/scrape/metrics.go index 34f1e28dbab..0418213e01b 100644 --- a/scrape/metrics.go +++ b/scrape/metrics.go @@ -57,6 +57,7 @@ type scrapeMetrics struct { targetScrapePoolExceededLabelLimits prometheus.Counter targetScrapeNativeHistogramBucketLimit prometheus.Counter targetScrapeDuration prometheus.Histogram + targetScrapesSkipped prometheus.Counter } func newScrapeMetrics(reg prometheus.Registerer) (*scrapeMetrics, error) { @@ -263,6 +264,13 @@ func newScrapeMetrics(reg prometheus.Registerer) (*scrapeMetrics, error) { }, ) + sm.targetScrapesSkipped = prometheus.NewCounter( + prometheus.CounterOpts{ + Name: "prometheus_target_scrapes_skipped_total", + Help: "Total number of target scrapes skipped due to memory limits.", + }, + ) + for _, collector := range []prometheus.Collector{ // Used by Manager. sm.targetMetadataCache, @@ -295,6 +303,7 @@ func newScrapeMetrics(reg prometheus.Registerer) (*scrapeMetrics, error) { sm.targetScrapePoolExceededLabelLimits, sm.targetScrapeNativeHistogramBucketLimit, sm.targetScrapeDuration, + sm.targetScrapesSkipped, } { err := reg.Register(collector) if err != nil { @@ -336,6 +345,7 @@ func (sm *scrapeMetrics) Unregister() { sm.reg.Unregister(sm.targetScrapePoolExceededLabelLimits) sm.reg.Unregister(sm.targetScrapeNativeHistogramBucketLimit) sm.reg.Unregister(sm.targetScrapeDuration) + sm.reg.Unregister(sm.targetScrapesSkipped) } type TargetsGatherer interface { diff --git a/scrape/scrape.go b/scrape/scrape.go index 9b37a356cf7..50e665b0ae9 100644 --- a/scrape/scrape.go +++ b/scrape/scrape.go @@ -55,6 +55,7 @@ import ( "github.com/prometheus/prometheus/model/value" "github.com/prometheus/prometheus/storage" "github.com/prometheus/prometheus/util/logging" + "github.com/prometheus/prometheus/util/memorylimiter" "github.com/prometheus/prometheus/util/namevalidationutil" "github.com/prometheus/prometheus/util/pool" ) @@ -703,7 +704,10 @@ type targetScraper struct { metrics *scrapeMetrics } -var errBodySizeLimit = errors.New("body size limit exceeded") +var ( + errBodySizeLimit = errors.New("body size limit exceeded") + errScrapeMemoryLimitExceeded = errors.New("memory limit exceeded") +) // acceptHeader transforms preference from the options into specific header values as // https://www.rfc-editor.org/rfc/rfc9110.html#name-accept defines. @@ -878,6 +882,7 @@ type scrapeLoop struct { skipJitterOffsetting bool // For testability. scrapeOnShutdown bool initialScrapeOffset time.Duration + memoryLimiter memorylimiter.MemoryLimiter // error injection through setForcedError. forcedErr error forcedErrMtx sync.Mutex @@ -1237,6 +1242,7 @@ func newScrapeLoop(opts scrapeLoopOptions) *scrapeLoop { skipJitterOffsetting: opts.sp.options.skipJitterOffsetting, scrapeOnShutdown: opts.sp.options.ScrapeOnShutdown, initialScrapeOffset: opts.sp.options.InitialScrapeOffset, + memoryLimiter: opts.sp.options.MemoryLimiter, } } @@ -1356,6 +1362,20 @@ func (sl *scrapeLoop) scrapeAndReport(last, appendTime time.Time, errc chan<- er var total, added, seriesAdded, bytesRead int var err, appErr, scrapeErr error + // Check if scrape is allowed by memory limiter. + if sl.memoryLimiter != nil && !sl.memoryLimiter.AllowScrape() { + sl.metrics.targetScrapesSkipped.Inc() + scrapeErr = errScrapeMemoryLimitExceeded + sl.scraper.Report(start, 0, scrapeErr) + if errc != nil { + select { + case errc <- scrapeErr: + case <-sl.ctx.Done(): + } + } + return start + } + app := sl.appender() defer func() { if err != nil { diff --git a/scrape/scrape_test.go b/scrape/scrape_test.go index 5bc47133ad9..d4caf6c8f6f 100644 --- a/scrape/scrape_test.go +++ b/scrape/scrape_test.go @@ -69,6 +69,7 @@ import ( "github.com/prometheus/prometheus/storage" "github.com/prometheus/prometheus/tsdb" "github.com/prometheus/prometheus/tsdb/chunkenc" + "github.com/prometheus/prometheus/util/memorylimiter" "github.com/prometheus/prometheus/util/pool" "github.com/prometheus/prometheus/util/teststorage" "github.com/prometheus/prometheus/util/testutil" @@ -7007,3 +7008,89 @@ func TestScrapeOffsetDistribution(t *testing.T) { } }) } + +type mockMemoryLimiter struct { + allowScrape bool +} + +func (m *mockMemoryLimiter) State() memorylimiter.LimiterState { + if m.allowScrape { + return memorylimiter.StateOK + } + return memorylimiter.StateHardLimit +} +func (m *mockMemoryLimiter) AllowScrape() bool { return m.allowScrape } +func (m *mockMemoryLimiter) AllowOTLP() bool { return m.allowScrape } +func (m *mockMemoryLimiter) AllowRemoteWrite() bool { return m.allowScrape } +func (m *mockMemoryLimiter) AllowRemoteRead() bool { return m.allowScrape } +func (m *mockMemoryLimiter) AllowFederation() bool { return m.allowScrape } +func (m *mockMemoryLimiter) AllowBlockCompaction() bool { return m.allowScrape } +func (m *mockMemoryLimiter) AllowRecordingRules() bool { return m.allowScrape } +func (m *mockMemoryLimiter) ApplyConfig(*config.MemoryLimiterConfig) error { return nil } +func (m *mockMemoryLimiter) Start(context.Context) {} +func (m *mockMemoryLimiter) Stop() {} + +func TestScrapeLoop_MemoryLimiterAbort(t *testing.T) { + for _, appV2 := range []bool{false, true} { + t.Run(fmt.Sprintf("appV2=%t", appV2), func(t *testing.T) { + s := teststorage.New(t) + defer s.Close() + + limiter := &mockMemoryLimiter{allowScrape: false} + sa := selectAppendable(s, appV2) + metrics := newTestScrapeMetrics(t) + cfg := &config.ScrapeConfig{ + JobName: "test", + ScrapeInterval: model.Duration(100 * time.Millisecond), + ScrapeTimeout: model.Duration(100 * time.Millisecond), + MetricNameValidationScheme: model.UTF8Validation, + MetricNameEscapingScheme: model.AllowUTF8, + } + sp, err := newScrapePool( + cfg, + sa.V1(), + sa.V2(), + 0, + nil, + nil, + &Options{MemoryLimiter: limiter}, + metrics, + ) + require.NoError(t, err) + defer sp.stop() + + target := NewTarget(labels.FromStrings(model.AddressLabel, "localhost:9090"), cfg, nil, nil) + scraper := &testScraper{} + sl := newScrapeLoop(scrapeLoopOptions{ + target: target, + scraper: scraper, + cache: newScrapeCache(metrics), + interval: 100 * time.Millisecond, + timeout: 100 * time.Millisecond, + sp: sp, + }) + + errc := make(chan error, 1) + sl.scrapeAndReport(time.Time{}, time.Now(), errc) + + select { + case err := <-errc: + require.Equal(t, errScrapeMemoryLimitExceeded, err) + default: + t.Fatal("expected errScrapeMemoryLimitExceeded") + } + + require.Equal(t, errScrapeMemoryLimitExceeded, scraper.lastError) + + // Verify that targetScrapesSkipped metric was incremented. + require.Equal(t, float64(1), prom_testutil.ToFloat64(metrics.targetScrapesSkipped)) + + // Query storage to assert 0 samples were appended (NO up=0 sample in storage!). + q, err := s.Querier(0, time.Now().UnixNano()) + require.NoError(t, err) + defer q.Close() + seriesSet := q.Select(t.Context(), false, nil, labels.MustNewMatcher(labels.MatchEqual, "__name__", "up")) + require.False(t, seriesSet.Next(), "expected no series appended to storage on memory limiter abort") + }) + } +} diff --git a/util/memorylimiter/memory_limiter.go b/util/memorylimiter/memory_limiter.go new file mode 100644 index 00000000000..e8549673d9e --- /dev/null +++ b/util/memorylimiter/memory_limiter.go @@ -0,0 +1,416 @@ +// Copyright The Prometheus Authors +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package memorylimiter + +import ( + "context" + "fmt" + "log/slog" + "math" + "runtime/metrics" + "sync" + "sync/atomic" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/prometheus/config" +) + +type LimiterState int32 + +const ( + StateOK LimiterState = iota + StateSoftLimit + StateHardLimit +) + +func (s LimiterState) String() string { + switch s { + case StateOK: + return "ok" + case StateSoftLimit: + return "soft" + case StateHardLimit: + return "hard" + default: + return "unknown" + } +} + +type MemoryLimiter interface { + State() LimiterState + AllowScrape() bool + AllowOTLP() bool + AllowRemoteWrite() bool + AllowRemoteRead() bool + AllowFederation() bool + AllowBlockCompaction() bool + AllowRecordingRules() bool + ApplyConfig(cfg *config.MemoryLimiterConfig) error + Start(ctx context.Context) + Stop() +} + +type MemoryStats struct { + TotalBytes uint64 + FreeBytes uint64 + ReleasedBytes uint64 + GOMEMLIMIT uint64 + GCLimiterCycle uint64 +} + +type MetricsReader func() MemoryStats + +func defaultMetricsReader() MemoryStats { + samples := []metrics.Sample{ + {Name: "/memory/classes/total:bytes"}, + {Name: "/memory/classes/heap/free:bytes"}, + {Name: "/memory/classes/heap/released:bytes"}, + {Name: "/gc/gomemlimit:bytes"}, + {Name: "/gc/limiter/last-enabled:gc-cycle"}, + } + metrics.Read(samples) + var stats MemoryStats + if samples[0].Value.Kind() == metrics.KindUint64 { + stats.TotalBytes = samples[0].Value.Uint64() + } + if samples[1].Value.Kind() == metrics.KindUint64 { + stats.FreeBytes = samples[1].Value.Uint64() + } + if samples[2].Value.Kind() == metrics.KindUint64 { + stats.ReleasedBytes = samples[2].Value.Uint64() + } + if samples[3].Value.Kind() == metrics.KindUint64 { + stats.GOMEMLIMIT = samples[3].Value.Uint64() + } + if samples[4].Value.Kind() == metrics.KindUint64 { + stats.GCLimiterCycle = samples[4].Value.Uint64() + } + return stats +} + +type Manager struct { + logger *slog.Logger + + mu sync.RWMutex + config *config.MemoryLimiterConfig + + state atomic.Int32 + lastInUse atomic.Uint64 + lastGCLimiterCycle uint64 + gcLimiterInitialized bool + lastCheckTime time.Time + + metricsReader MetricsReader + now func() time.Time + + metrics *memoryLimiterMetrics + + reloadCh chan struct{} + cancel context.CancelFunc + wg sync.WaitGroup +} + +func NewManager(cfg *config.MemoryLimiterConfig, logger *slog.Logger, reg prometheus.Registerer) (*Manager, error) { + if logger == nil { + logger = slog.Default() + } + + m := &Manager{ + logger: logger.With("component", "memory limiter"), + config: cfg, + metricsReader: defaultMetricsReader, + now: time.Now, + metrics: newMemoryLimiterMetrics(reg), + reloadCh: make(chan struct{}, 1), + } + + if cfg != nil { + if err := cfg.Validate(); err != nil { + return nil, fmt.Errorf("invalid memory limiter config: %w", err) + } + } + + return m, nil +} + +func (m *Manager) ApplyConfig(cfg *config.MemoryLimiterConfig) error { + m.mu.Lock() + defer m.mu.Unlock() + + if cfg != nil { + if err := cfg.Validate(); err != nil { + return fmt.Errorf("invalid memory limiter config: %w", err) + } + } + + m.config = cfg + + select { + case m.reloadCh <- struct{}{}: + default: + } + + return nil +} + +func (m *Manager) State() LimiterState { + if m == nil { + return StateOK + } + return LimiterState(m.state.Load()) +} + +func (m *Manager) Start(ctx context.Context) { + m.mu.Lock() + if m.cancel != nil { + m.mu.Unlock() + return + } + loopCtx, cancel := context.WithCancel(ctx) + m.cancel = cancel + m.lastCheckTime = m.now() + m.mu.Unlock() + + m.wg.Add(1) + go m.run(loopCtx) +} + +func (m *Manager) Stop() { + m.mu.Lock() + if m.cancel != nil { + m.cancel() + m.cancel = nil + } + m.mu.Unlock() + m.wg.Wait() +} + +func (m *Manager) run(ctx context.Context) { + defer m.wg.Done() + + interval := 100 * time.Millisecond + m.mu.RLock() + if m.config != nil && m.config.CheckInterval > 0 { + interval = time.Duration(m.config.CheckInterval) + } + m.mu.RUnlock() + + ticker := time.NewTicker(interval) + defer ticker.Stop() + + // Initial evaluation. + m.Evaluate() + + for { + select { + case <-ctx.Done(): + return + case <-m.reloadCh: + m.mu.RLock() + newInterval := 100 * time.Millisecond + if m.config != nil && m.config.CheckInterval > 0 { + newInterval = time.Duration(m.config.CheckInterval) + } + m.mu.RUnlock() + ticker.Reset(newInterval) + m.Evaluate() + case <-ticker.C: + m.Evaluate() + } + } +} + +// Evaluate reads the runtime metrics and updates the limiter state. +func (m *Manager) Evaluate() { + m.mu.Lock() + defer m.mu.Unlock() + + stats := m.metricsReader() + totalBytes := stats.TotalBytes + freeBytes := stats.FreeBytes + releasedBytes := stats.ReleasedBytes + gomemlimit := stats.GOMEMLIMIT + gcLimiterCycle := stats.GCLimiterCycle + + if m.config == nil || gomemlimit == 0 || gomemlimit == math.MaxInt64 { + oldState := LimiterState(m.state.Swap(int32(StateOK))) + if oldState != StateOK { + m.metrics.transitionsTotal.WithLabelValues(oldState.String(), StateOK.String()).Inc() + } + m.metrics.active.WithLabelValues("hard").Set(0) + m.metrics.active.WithLabelValues("soft").Set(0) + m.metrics.limitBytes.WithLabelValues("soft").Set(0) + m.metrics.limitBytes.WithLabelValues("hard").Set(0) + m.metrics.inUseBytes.Set(0) + return + } + + var inUse uint64 + if totalBytes > (freeBytes + releasedBytes) { + inUse = totalBytes - (freeBytes + releasedBytes) + } + m.lastInUse.Store(inUse) + + now := m.now() + elapsed := 0.0 + if !m.lastCheckTime.IsZero() { + elapsed = now.Sub(m.lastCheckTime).Seconds() + } + m.lastCheckTime = now + + oldState := LimiterState(m.state.Load()) + if elapsed > 0 { + if oldState == StateSoftLimit { + m.metrics.engagedSecondsTotal.WithLabelValues("soft").Add(elapsed) + } else if oldState == StateHardLimit { + m.metrics.engagedSecondsTotal.WithLabelValues("hard").Add(elapsed) + } + } + + pressureRatio := float64(inUse) / float64(gomemlimit) + + gcLimiterActive := m.gcLimiterInitialized && gcLimiterCycle > m.lastGCLimiterCycle + m.lastGCLimiterCycle = gcLimiterCycle + m.gcLimiterInitialized = true + + var newState LimiterState + if pressureRatio >= m.config.HardLimitRatio || gcLimiterActive { + newState = StateHardLimit + } else if pressureRatio >= m.config.SoftLimitRatio { + newState = StateSoftLimit + } else { + newState = StateOK + } + + if newState != oldState { + m.state.Store(int32(newState)) + m.metrics.transitionsTotal.WithLabelValues(oldState.String(), newState.String()).Inc() + m.logger.Debug("Memory limiter state transition", + "from", oldState.String(), + "to", newState.String(), + "pressure_ratio", pressureRatio, + "in_use_bytes", inUse, + "gomemlimit", gomemlimit, + "gc_limiter_active", gcLimiterActive, + ) + } + + // Update gauges. + if newState == StateHardLimit { + m.metrics.active.WithLabelValues("hard").Set(1) + m.metrics.active.WithLabelValues("soft").Set(1) + } else if newState == StateSoftLimit { + m.metrics.active.WithLabelValues("hard").Set(0) + m.metrics.active.WithLabelValues("soft").Set(1) + } else { + m.metrics.active.WithLabelValues("hard").Set(0) + m.metrics.active.WithLabelValues("soft").Set(0) + } + + m.metrics.limitBytes.WithLabelValues("soft").Set(float64(gomemlimit) * m.config.SoftLimitRatio) + m.metrics.limitBytes.WithLabelValues("hard").Set(float64(gomemlimit) * m.config.HardLimitRatio) + m.metrics.inUseBytes.Set(float64(inUse)) +} + +func (m *Manager) AllowScrape() bool { + if m == nil { + return true + } + m.mu.RLock() + cfg := m.config + m.mu.RUnlock() + if cfg == nil || !cfg.Enforcement.FailScrapes { + return true + } + return LimiterState(m.state.Load()) < StateHardLimit +} + +func (m *Manager) AllowOTLP() bool { + if m == nil { + return true + } + m.mu.RLock() + cfg := m.config + m.mu.RUnlock() + if cfg == nil || !cfg.Enforcement.RejectOTLP { + return true + } + return LimiterState(m.state.Load()) < StateHardLimit +} + +func (m *Manager) AllowRemoteWrite() bool { + if m == nil { + return true + } + m.mu.RLock() + cfg := m.config + m.mu.RUnlock() + if cfg == nil || !cfg.Enforcement.RejectRemoteWrite { + return true + } + return LimiterState(m.state.Load()) < StateHardLimit +} + +func (m *Manager) AllowRemoteRead() bool { + if m == nil { + return true + } + m.mu.RLock() + cfg := m.config + m.mu.RUnlock() + if cfg == nil || !cfg.Enforcement.RejectRemoteRead { + return true + } + return LimiterState(m.state.Load()) < StateSoftLimit +} + +func (m *Manager) AllowFederation() bool { + if m == nil { + return true + } + m.mu.RLock() + cfg := m.config + m.mu.RUnlock() + if cfg == nil || !cfg.Enforcement.RejectFederation { + return true + } + return LimiterState(m.state.Load()) < StateSoftLimit +} + +func (m *Manager) AllowBlockCompaction() bool { + if m == nil { + return true + } + m.mu.RLock() + cfg := m.config + m.mu.RUnlock() + if cfg == nil || !cfg.Enforcement.PauseBlockCompaction { + return true + } + return LimiterState(m.state.Load()) < StateSoftLimit +} + +func (m *Manager) AllowRecordingRules() bool { + if m == nil { + return true + } + m.mu.RLock() + cfg := m.config + m.mu.RUnlock() + if cfg == nil || !cfg.Enforcement.PauseRecordingRules { + return true + } + return LimiterState(m.state.Load()) < StateHardLimit +} diff --git a/util/memorylimiter/memory_limiter_test.go b/util/memorylimiter/memory_limiter_test.go new file mode 100644 index 00000000000..9311aba9cc0 --- /dev/null +++ b/util/memorylimiter/memory_limiter_test.go @@ -0,0 +1,153 @@ +// Copyright The Prometheus Authors +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package memorylimiter + +import ( + "log/slog" + "math" + "testing" + "time" + + "github.com/prometheus/client_golang/prometheus" + "github.com/prometheus/common/model" + "github.com/stretchr/testify/require" + + "github.com/prometheus/prometheus/config" +) + +func newMockMetricsReader(total, released, gomemlimit, gcLimiterCycle uint64) MetricsReader { + return func() MemoryStats { + return MemoryStats{ + TotalBytes: total, + ReleasedBytes: released, + GOMEMLIMIT: gomemlimit, + GCLimiterCycle: gcLimiterCycle, + } + } +} + +func TestMemoryLimiterStateTransitions(t *testing.T) { + cfg := &config.MemoryLimiterConfig{ + CheckInterval: model.Duration(100 * time.Millisecond), + SoftLimitRatio: 0.70, + HardLimitRatio: 0.85, + Enforcement: config.MemoryLimiterEnforcement{ + PauseBlockCompaction: true, + RejectRemoteRead: true, + RejectFederation: true, + FailScrapes: true, + RejectOTLP: true, + RejectRemoteWrite: true, + PauseRecordingRules: true, + }, + } + + reg := prometheus.NewRegistry() + mgr, err := NewManager(cfg, slog.Default(), reg) + require.NoError(t, err) + + now := time.Now() + mgr.now = func() time.Time { return now } + + gomemlimit := uint64(1000 * 1024 * 1024) // 1000 MB + + // 1. Normal state: 500 MB in-use (50% ratio < 70%) + mgr.metricsReader = newMockMetricsReader(600*1024*1024, 100*1024*1024, gomemlimit, 0) + mgr.Evaluate() + require.Equal(t, StateOK, mgr.State()) + require.True(t, mgr.AllowScrape()) + require.True(t, mgr.AllowOTLP()) + require.True(t, mgr.AllowRemoteWrite()) + require.True(t, mgr.AllowRemoteRead()) + require.True(t, mgr.AllowFederation()) + require.True(t, mgr.AllowBlockCompaction()) + require.True(t, mgr.AllowRecordingRules()) + + // 2. Soft limit state: 750 MB in-use (75% ratio >= 70%, < 85%) + now = now.Add(100 * time.Millisecond) + mgr.metricsReader = newMockMetricsReader(850*1024*1024, 100*1024*1024, gomemlimit, 0) + mgr.Evaluate() + require.Equal(t, StateSoftLimit, mgr.State()) + require.True(t, mgr.AllowScrape()) + require.True(t, mgr.AllowOTLP()) + require.True(t, mgr.AllowRemoteWrite()) + require.False(t, mgr.AllowRemoteRead()) + require.False(t, mgr.AllowFederation()) + require.False(t, mgr.AllowBlockCompaction()) + require.True(t, mgr.AllowRecordingRules()) + + // 3. Hard limit state: 900 MB in-use (90% ratio >= 85%) + now = now.Add(100 * time.Millisecond) + mgr.metricsReader = newMockMetricsReader(1000*1024*1024, 100*1024*1024, gomemlimit, 0) + mgr.Evaluate() + require.Equal(t, StateHardLimit, mgr.State()) + require.False(t, mgr.AllowScrape()) + require.False(t, mgr.AllowOTLP()) + require.False(t, mgr.AllowRemoteWrite()) + require.False(t, mgr.AllowRemoteRead()) + require.False(t, mgr.AllowFederation()) + require.False(t, mgr.AllowBlockCompaction()) + require.False(t, mgr.AllowRecordingRules()) + + // 4. Memory drops back to 400 MB (40% < 70%) -> Clean recovery to StateOK + now = now.Add(100 * time.Millisecond) + mgr.metricsReader = newMockMetricsReader(500*1024*1024, 100*1024*1024, gomemlimit, 0) + mgr.Evaluate() + require.Equal(t, StateOK, mgr.State()) + require.True(t, mgr.AllowScrape()) + require.True(t, mgr.AllowBlockCompaction()) +} + +func TestGCLimiterEscalation(t *testing.T) { + cfg := &config.MemoryLimiterConfig{ + CheckInterval: model.Duration(100 * time.Millisecond), + SoftLimitRatio: 0.70, + HardLimitRatio: 0.85, + Enforcement: config.MemoryLimiterEnforcement{ + FailScrapes: true, + }, + } + + mgr, err := NewManager(cfg, slog.Default(), nil) + require.NoError(t, err) + + gomemlimit := uint64(1000 * 1024 * 1024) + + // In-use is low (500 MB / 50%), but GC CPU limiter engaged (cycle goes from 10 to 11) + mgr.metricsReader = newMockMetricsReader(500*1024*1024, 0, gomemlimit, 10) + mgr.Evaluate() + require.Equal(t, StateOK, mgr.State()) + + mgr.metricsReader = newMockMetricsReader(500*1024*1024, 0, gomemlimit, 11) + mgr.Evaluate() + require.Equal(t, StateHardLimit, mgr.State()) + require.False(t, mgr.AllowScrape()) +} + +func TestUnlimitedGOMEMLIMIT(t *testing.T) { + cfg := &config.MemoryLimiterConfig{ + CheckInterval: model.Duration(100 * time.Millisecond), + SoftLimitRatio: 0.70, + HardLimitRatio: 0.85, + } + + mgr, err := NewManager(cfg, slog.Default(), nil) + require.NoError(t, err) + + // GOMEMLIMIT is MaxInt64 (unlimited) + mgr.metricsReader = newMockMetricsReader(500*1024*1024, 0, math.MaxInt64, 0) + mgr.Evaluate() + require.Equal(t, StateOK, mgr.State()) + require.True(t, mgr.AllowScrape()) +} diff --git a/util/memorylimiter/metrics.go b/util/memorylimiter/metrics.go new file mode 100644 index 00000000000..5d2e4b0c7e4 --- /dev/null +++ b/util/memorylimiter/metrics.go @@ -0,0 +1,81 @@ +// Copyright The Prometheus Authors +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package memorylimiter + +import ( + "github.com/prometheus/client_golang/prometheus" +) + +type memoryLimiterMetrics struct { + active *prometheus.GaugeVec + engagedSecondsTotal *prometheus.CounterVec + transitionsTotal *prometheus.CounterVec + limitBytes *prometheus.GaugeVec + inUseBytes prometheus.Gauge +} + +func newMemoryLimiterMetrics(reg prometheus.Registerer) *memoryLimiterMetrics { + m := &memoryLimiterMetrics{ + active: prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Name: "prometheus_memory_limiter_active", + Help: "Boolean gauge indicating if a memory limiter threshold is currently engaged (1 for active, 0 for inactive).", + }, + []string{"limit"}, + ), + engagedSecondsTotal: prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "prometheus_memory_limiter_engaged_seconds_total", + Help: "Total time in seconds spent with memory limiter thresholds engaged.", + }, + []string{"limit"}, + ), + transitionsTotal: prometheus.NewCounterVec( + prometheus.CounterOpts{ + Name: "prometheus_memory_limiter_transitions_total", + Help: "Total number of state transitions between memory limiter states.", + }, + []string{"from", "to"}, + ), + limitBytes: prometheus.NewGaugeVec( + prometheus.GaugeOpts{ + Name: "prometheus_memory_limiter_limit_bytes", + Help: "Evaluated memory limit in bytes for each threshold.", + }, + []string{"limit"}, + ), + inUseBytes: prometheus.NewGauge( + prometheus.GaugeOpts{ + Name: "prometheus_memory_limiter_in_use_bytes", + Help: "Current in-use memory in bytes evaluated by the memory limiter.", + }, + ), + } + + if reg != nil { + reg.MustRegister( + m.active, + m.engagedSecondsTotal, + m.transitionsTotal, + m.limitBytes, + m.inUseBytes, + ) + } + + // Initialize gauges with default 0 values. + m.active.WithLabelValues("soft").Set(0) + m.active.WithLabelValues("hard").Set(0) + + return m +} diff --git a/web/api/v1/api.go b/web/api/v1/api.go index 955df918fac..6bc85cac495 100644 --- a/web/api/v1/api.go +++ b/web/api/v1/api.go @@ -59,6 +59,7 @@ import ( "github.com/prometheus/prometheus/util/annotations" "github.com/prometheus/prometheus/util/features" "github.com/prometheus/prometheus/util/httputil" + "github.com/prometheus/prometheus/util/memorylimiter" "github.com/prometheus/prometheus/util/notifications" "github.com/prometheus/prometheus/util/stats" ) @@ -262,6 +263,13 @@ type API struct { openAPIBuilder *OpenAPIBuilder parser parser.Parser + + memoryLimiter memorylimiter.MemoryLimiter +} + +// SetMemoryLimiter sets the memory limiter for API endpoints. +func (api *API) SetMemoryLimiter(ml memorylimiter.MemoryLimiter) { + api.memoryLimiter = ml } // NewAPI returns an initialized API type. @@ -2029,6 +2037,11 @@ func (api *API) notificationsSSE(w http.ResponseWriter, r *http.Request) { } func (api *API) remoteRead(w http.ResponseWriter, r *http.Request) { + if api.memoryLimiter != nil && !api.memoryLimiter.AllowRemoteRead() { + w.Header().Set("Retry-After", "5") + http.Error(w, "Service Unavailable: Memory limit exceeded", http.StatusServiceUnavailable) + return + } // This is only really for tests - this will never be nil IRL. if api.remoteReadHandler != nil { api.remoteReadHandler.ServeHTTP(w, r) @@ -2038,6 +2051,11 @@ func (api *API) remoteRead(w http.ResponseWriter, r *http.Request) { } func (api *API) remoteWrite(w http.ResponseWriter, r *http.Request) { + if api.memoryLimiter != nil && !api.memoryLimiter.AllowRemoteWrite() { + w.Header().Set("Retry-After", "5") + http.Error(w, "Service Unavailable: Memory limit exceeded", http.StatusServiceUnavailable) + return + } if api.remoteWriteHandler != nil { api.remoteWriteHandler.ServeHTTP(w, r) } else { @@ -2046,6 +2064,11 @@ func (api *API) remoteWrite(w http.ResponseWriter, r *http.Request) { } func (api *API) otlpWrite(w http.ResponseWriter, r *http.Request) { + if api.memoryLimiter != nil && !api.memoryLimiter.AllowOTLP() { + w.Header().Set("Retry-After", "5") + http.Error(w, "Service Unavailable: Memory limit exceeded", http.StatusServiceUnavailable) + return + } if api.otlpWriteHandler != nil { api.otlpWriteHandler.ServeHTTP(w, r) } else { diff --git a/web/federate.go b/web/federate.go index 730c0cf8e2d..f554219adb4 100644 --- a/web/federate.go +++ b/web/federate.go @@ -53,6 +53,12 @@ func registerFederationMetrics(r prometheus.Registerer) { } func (h *Handler) federation(w http.ResponseWriter, req *http.Request) { + if h.options.MemoryLimiter != nil && !h.options.MemoryLimiter.AllowFederation() { + w.Header().Set("Retry-After", "5") + http.Error(w, "Service Unavailable: Memory limit exceeded", http.StatusServiceUnavailable) + return + } + h.mtx.RLock() defer h.mtx.RUnlock() diff --git a/web/web.go b/web/web.go index 51316e95c0a..2e8b8c216c7 100644 --- a/web/web.go +++ b/web/web.go @@ -61,6 +61,7 @@ import ( "github.com/prometheus/prometheus/template" "github.com/prometheus/prometheus/util/features" "github.com/prometheus/prometheus/util/httputil" + "github.com/prometheus/prometheus/util/memorylimiter" "github.com/prometheus/prometheus/util/netconnlimit" "github.com/prometheus/prometheus/util/notifications" api_v1 "github.com/prometheus/prometheus/web/api/v1" @@ -320,6 +321,9 @@ type Options struct { // Parser is the PromQL parser used for parsing query expressions. Parser parser.Parser + + // MemoryLimiter is used to reject ingestion/queries under memory pressure. + MemoryLimiter memorylimiter.MemoryLimiter } // New initializes a new web Handler. @@ -435,6 +439,9 @@ func New(logger *slog.Logger, o *Options) *Handler { }, o.Parser, ) + if o.MemoryLimiter != nil { + h.apiV1.SetMemoryLimiter(o.MemoryLimiter) + } if r := o.FeatureRegistry; r != nil { // Set dynamic API features (based on configuration).