scrape-manager: add bounded concurrent and sequential scrape scheduler

Cargo.toml

@@ -22,14 +22,18 @@ arrow-array = { version = "59", default-features = false }
 arrow-buffer = { version = "59", default-features = false }
 arrow-schema = { version = "59", default-features = false }
 benchfmt = { git = "https://github.com/bwplotka/benchfmt.rs" }
+flate2 = { version = "1", default-features = false, features = ["rust_backend"] }
+futures-util = { version = "0.3", default-features = false, features = ["alloc"] }
 prost = { version = "0.14" }
 prost-build = "=0.14.4"
 prost-types = { version = "0.14" }
 protox = "0.10"
 regex = { version = "1" }
+reqwest = { version = "0.13", default-features = false, features = ["http2", "rustls"] }
 serde = { version = "1", features = ["derive"] }
 serde_json = "1"
 thiserror = "2"
+tokio = { version = "1", default-features = false, features = ["rt", "sync", "time"] }
 
 [workspace.lints.rust]
 missing_docs = "warn"

README.md

@@ -30,7 +30,7 @@ The project serves two main use cases:
 | :--- | :--- | :--- | :--- | :--- |
 | `prometheus-expfmt` | [`crates/expfmt`](./crates/expfmt) | In progress | [specs/SPEC.md](./specs/SPEC.md) | `Content-Type` and `Accept` negotiation, zero-copy streaming parsers for Prometheus Text 0.0.4, OpenMetrics 1.0, OpenMetrics 2.0, and Prometheus Protobuf, primary Apache Arrow / OTAP decoder (`ArrowDecoder`), and [`MetricFamily`](https://github.com/prometheus/prometheus/blob/main/prompb/io/prometheus/client/metrics.proto) protobuf DTO decoder. |
 | `prometheus-relabel` | [`crates/relabel`](./crates/relabel) | In progress | [specs/SPEC.md](./specs/SPEC.md#target-and-metric-relabeling-architecture) | Prometheus-compatible target and metric relabeling (`replace`, `keep`, `drop`, `hashmod`, `labelmap`, `labeldrop`, `labelkeep`, `lowercase`, `uppercase`, `keepequal`, `dropequal`), Stage 1 `__name__` filter pushdown, and Stage 2 dictionary-vectorized relabeling and `honor_labels` resolution. |
-| `prometheus-scrape-manager` | [`crates/scrape-manager`](./crates/scrape-manager) | Planned | TBD | Prometheus HTTP/TLS client configuration (`http_config`), service discovery mechanisms (`discovery`: static, file, HTTP, DNS, Kubernetes), and bounded target scrape pool scheduling (`scrape`). |
+| `prometheus-scrape-manager` | [`crates/scrape-manager`](./crates/scrape-manager) | In progress | [specs/SPEC.md](./specs/SPEC.md#bounded-scrape-scheduling-and-memory-pools) | Bounded concurrent and sequential target scrape scheduling (`scrape`), Prometheus HTTP/TLS client configuration (`http_config`), and service discovery mechanisms (`discovery`: static, file, HTTP, DNS, Kubernetes). |
 
 ## Development and testing
 
@@ -52,9 +52,10 @@ Run the Go differential fixture extractor and tests (requires Go):
 cd tests/go-harness && go test ./...
 ```
 
-Run the decoding and relabeling benchmarks in Go `benchfmt` format for [`benchstat`](https://pkg.go.dev/golang.org/x/perf/cmd/benchstat):
+Run the decoding, relabeling, and scrape scheduling benchmarks in Go `benchfmt` format for [`benchstat`](https://pkg.go.dev/golang.org/x/perf/cmd/benchstat):
 
 ```sh
 cargo bench -p prometheus-expfmt --bench decode -- -bench '^BenchmarkDecode' -benchtime 1s -count 6 -cpu 2 -benchmem | tee decode.txt
 cargo bench -p prometheus-relabel --bench relabel -- -bench '^BenchmarkRelabel' -benchtime 1s -count 6 -cpu 2 -benchmem | tee relabel.txt
+cargo bench -p prometheus-scrape-manager --bench scrape -- -bench '^BenchmarkScrape' -benchtime 500ms -count 6 -cpu 2 -benchmem | tee scrape.txt
 ```

crates/scrape-manager/Cargo.toml

@@ -9,6 +9,10 @@ description = "Prometheus service discovery, HTTP client configuration, and scra
 readme = "README.md"
 keywords = ["prometheus", "openmetrics", "metrics", "scrape", "service-discovery"]
 categories = ["network-programming", "asynchronous"]
+exclude = [
+    "benches/**",
+    "tests/**",
+]
 # Kept unpublished while initial service discovery, HTTP config, and scrape pool modules land.
 publish = false
 
@@ -17,8 +21,22 @@ default = ["arrow"]
 arrow = ["prometheus-expfmt/arrow"]
 
 [dependencies]
+flate2 = { workspace = true }
+futures-util = { workspace = true }
 prometheus-expfmt = { workspace = true }
 prometheus-relabel = { workspace = true }
+reqwest = { workspace = true }
+thiserror = { workspace = true }
+tokio = { workspace = true }
+
+[dev-dependencies]
+benchfmt = { workspace = true }
+tokio = { workspace = true, features = ["io-util", "macros", "net", "rt-multi-thread", "test-util"] }
+
+[[bench]]
+name = "scrape"
+harness = false
+required-features = ["arrow"]
 
 [lints]
 workspace = true

crates/scrape-manager/README.md

@@ -1,5 +1,14 @@
 # prometheus-scrape-manager
 
-Prometheus service discovery (`discovery`), HTTP/TLS client configuration (`http_config`), and target scrape pool management (`scrape`) in Rust, built on [`prometheus-expfmt`](../expfmt) and [`prometheus-relabel`](../relabel).
+Prometheus target scrape pool management (`scrape`), HTTP/TLS client configuration (`http_config`), and service discovery (`discovery`) in Rust, built on [`prometheus-expfmt`](../expfmt) and [`prometheus-relabel`](../relabel).
+
+## Scrape scheduling modes
+
+`scrape::ScrapeManager` supports two scheduling strategies configured via `SchedulingMode`:
+
+- `SchedulingMode::Concurrent` schedules every target on its own deterministic 64-bit FNV-1a phase offset modulo `scrape_interval` inside a single min-heap. Up to `max_in_flight` HTTP requests wait for response headers in parallel while holding zero payload memory. Only after an HTTP `200 OK` header arrives does the fetcher lease one of `max_buffered_bodies` reusable `Vec<u8>` buffers from `BodyBufferPool`, and decoding leases one of `max_decoders` reusable `ArrowDecoder` instances from `DecoderPool`.
+- `SchedulingMode::Sequential` scrapes one target at a time using a single reusable body buffer and a single decoder, either on a job-wide tick via `SequentialSchedule::CycleTick` or across per-target FNV-1a phase offsets via `SequentialSchedule::Jittered`.
+
+The scrape loop drives concurrent futures with `FuturesUnordered` rather than `tokio::spawn`, so thread-per-core `!Send` `LocalSet` runtimes can pass non-`Send` sinks directly.
 
 Part of the [`prometheus/scrape.rs`](https://github.com/prometheus/scrape.rs) workspace.

crates/scrape-manager/benches/scrape.rs

@@ -0,0 +1,193 @@
+// Copyright 2026 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.
+
+//! Throughput and allocation benchmarks for `prometheus_scrape_manager::scrape::ScrapeManager`
+//! comparing sequential and bounded concurrent scrape modes over the upstream Prometheus
+//! `model/textparse` fixture corpus.
+//!
+//! Recommended CLI invocation:
+//! ```sh
+//! cargo bench -p prometheus-scrape-manager --bench scrape -- -bench '^BenchmarkScrape' -benchtime 500ms -count 6 -cpu 2 -benchmem | tee ${bench}.txt
+//! ```
+
+use std::fs;
+use std::hint::black_box;
+use std::num::NonZeroUsize;
+use std::ops::ControlFlow;
+use std::path::Path;
+use std::sync::Arc;
+use std::time::Duration;
+
+use benchfmt::{CountingAlloc, Suite};
+use prometheus_expfmt::arrow::OtapMetricRecords;
+use prometheus_expfmt::format::Format;
+use prometheus_scrape_manager::scrape::{
+    ArrowScrapeSink, BodyBufferPool, FetchedScrape, SchedulingMode, ScrapeConfig, ScrapeError,
+    ScrapeManager, ScrapeReport, ScrapeTarget, TargetFetcher,
+};
+
+#[global_allocator]
+static ALLOC: CountingAlloc = CountingAlloc::new();
+
+#[derive(Clone)]
+struct BenchFetcher {
+    payload: Arc<[u8]>,
+    header_wait: Duration,
+}
+
+impl TargetFetcher for BenchFetcher {
+    async fn fetch(
+        &self,
+        _target: &ScrapeTarget,
+        body_pool: &BodyBufferPool,
+    ) -> Result<FetchedScrape, ScrapeError> {
+        // Simulate waiting on target TCP connect and server-side `/metrics` generation before
+        // HTTP 200 headers arrive. No body buffer or decoder is leased during this wait.
+        if self.header_wait.is_zero() {
+            tokio::task::yield_now().await;
+        } else {
+            tokio::time::sleep(self.header_wait).await;
+        }
+
+        let mut body = body_pool.acquire(self.payload.len()).await;
+        body.extend_from_slice(&self.payload);
+
+        Ok(FetchedScrape {
+            body,
+            format: Format::TEXT,
+            scrape_timestamp_ms: 1_700_000_000_000,
+            wire_bytes: self.payload.len(),
+            fetch_duration: self.header_wait,
+        })
+    }
+}
+
+struct BlackBoxSink;
+
+impl ArrowScrapeSink for BlackBoxSink {
+    async fn on_batch(
+        &self,
+        target: &ScrapeTarget,
+        scrape_timestamp_ms: i64,
+        records: OtapMetricRecords,
+    ) -> ControlFlow<()> {
+        black_box((target, scrape_timestamp_ms, records));
+        ControlFlow::Continue(())
+    }
+
+    async fn on_scrape_complete(
+        &self,
+        target: &ScrapeTarget,
+        report: ScrapeReport,
+    ) -> ControlFlow<()> {
+        black_box((target, report));
+        ControlFlow::Continue(())
+    }
+}
+
+fn build_targets(count: usize) -> Vec<ScrapeTarget> {
+    (0..count)
+        .map(|i| {
+            ScrapeTarget::parse(&format!("http://10.0.{}.{}:9100/metrics", i / 256, i % 256))
+                .expect("valid target url")
+                .with_label("job", "node")
+                .with_label("instance", format!("node-{i}"))
+        })
+        .collect()
+}
+
+fn main() {
+    let root = Path::new(env!("CARGO_MANIFEST_DIR")).join("../../testdata/textparse");
+    let full = fs::read(root.join("alltypes.237mfs.prom.txt")).expect("read fixture");
+    let payload: Arc<[u8]> = Arc::from(full.into_boxed_slice());
+
+    let targets = Arc::new(build_targets(16));
+    let total_bytes = (payload.len() * targets.len()) as u64;
+
+    let modes: [(&str, SchedulingMode, usize, usize); 3] = [
+        ("sequential_tick", SchedulingMode::sequential(), 1, 1),
+        (
+            "concurrent_k8_b2_d1",
+            SchedulingMode::concurrent_with_pools(
+                NonZeroUsize::new(8).unwrap(),
+                NonZeroUsize::new(2).unwrap(),
+                NonZeroUsize::new(1).unwrap(),
+            ),
+            2,
+            1,
+        ),
+        (
+            "concurrent_k8_b8_d4",
+            SchedulingMode::concurrent_with_pools(
+                NonZeroUsize::new(8).unwrap(),
+                NonZeroUsize::new(8).unwrap(),
+                NonZeroUsize::new(4).unwrap(),
+            ),
+            8,
+            4,
+        ),
+    ];
+
+    let mut suite = Suite::from_env("prometheus-scrape-manager").with_alloc(&ALLOC);
+
+    // Tokio's 1ms timer-wheel rounding plus OS timer slack makes `Duration::from_millis(1)`
+    // wait ~2.2ms per fetch in practice, so label the dimension `2ms` to reflect actual wait time.
+    for (wait_label, header_wait) in [("none", Duration::ZERO), ("2ms", Duration::from_millis(1))] {
+        for (mode_label, scheduling, max_bodies, max_decoders) in modes {
+            let payload = Arc::clone(&payload);
+            let targets = Arc::clone(&targets);
+
+            suite = suite.bench(
+                format!("ScrapeCycle/targets=16/header_wait={wait_label}/mode={mode_label}"),
+                move |b| {
+                    let rt = tokio::runtime::Builder::new_current_thread()
+                        .enable_time()
+                        .build()
+                        .expect("current_thread runtime");
+
+                    let fetcher = BenchFetcher {
+                        payload: Arc::clone(&payload),
+                        header_wait,
+                    };
+                    let cfg = ScrapeConfig::new("bench")
+                        .with_scheduling(scheduling)
+                        .with_max_data_points_per_batch(256);
+                    let manager = ScrapeManager::new(cfg, fetcher).expect("manager");
+                    let sink = BlackBoxSink;
+
+                    // Warm pool buffers and decoder interner arenas once so B/op and allocs/op
+                    // reflect steady-state scrape reuse rather than first-pass capacity growth.
+                    rt.block_on(async {
+                        let _ = manager.run_once_arrow(&targets, &sink).await;
+                    });
+                    let warm_stats = manager.stats();
+                    assert!(warm_stats.total_body_buffers_created <= max_bodies);
+                    assert!(warm_stats.total_decoders_created <= max_decoders);
+
+                    b.set_bytes(total_bytes);
+                    b.report_allocs();
+                    while b.keep_running() {
+                        rt.block_on(async {
+                            let out = manager
+                                .run_once_arrow(black_box(targets.as_slice()), &sink)
+                                .await;
+                            let _ = black_box(out);
+                        });
+                    }
+                },
+            );
+        }
+    }
+
+    suite.run();
+}

crates/scrape-manager/src/lib.rs

@@ -14,7 +14,9 @@
 //! Prometheus service discovery, HTTP scrape client configuration, and target scrape management.
 //!
 //! Re-exports [`prometheus_expfmt`] and [`prometheus_relabel`] and hosts the async runtime I/O
-//! layers (`http_config`, `discovery`, and `scrape`) for [`prometheus/scrape.rs`](https://github.com/prometheus/scrape.rs).
+//! layers for [`prometheus/scrape.rs`](https://github.com/prometheus/scrape.rs).
 
 pub use prometheus_expfmt as expfmt;
 pub use prometheus_relabel as relabel;
+
+pub mod scrape;

crates/scrape-manager/src/scrape.rs

@@ -0,0 +1,50 @@
+// Copyright 2026 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.
+
+//! Bounded-memory HTTP target scraping and interval scheduling.
+//!
+//! [`ScrapeManager`] coordinates periodic target scraping in two user-selectable modes via
+//! [`SchedulingMode`]:
+//! - [`SchedulingMode::Sequential`]: scrapes one target at a time with a single reusable HTTP body
+//!   buffer and a single decoder, using either a job-wide tick ([`SequentialSchedule::CycleTick`])
+//!   or deterministic per-target FNV-1a phase offsets ([`SequentialSchedule::Jittered`]).
+//! - [`SchedulingMode::Concurrent`]: schedules every target on its own deterministic jittered
+//!   interval (`hash(target) % scrape_interval`) while decoupling in-flight HTTP header concurrency
+//!   (`max_in_flight`) from pooled response body buffers (`max_buffered_bodies`) and pooled
+//!   decoders (`max_decoders`). Slow targets stalled on TCP connect or server-side metric
+//!   generation neither delay sibling targets nor hold response payload buffers.
+
+mod config;
+mod http;
+mod manager;
+mod pool;
+mod schedule;
+
+pub use config::{
+    DEFAULT_MAX_BODY_BYTES, DEFAULT_MAX_DATA_POINTS_PER_BATCH, DEFAULT_MAX_RETAINED_BODY_BYTES,
+    DEFAULT_MAX_RETAINED_DECODER_BYTES, DEFAULT_SCRAPE_INTERVAL, DEFAULT_SCRAPE_TIMEOUT,
+    DEFAULT_USER_AGENT, SchedulingMode, ScrapeConfig, ScrapeTarget, SequentialSchedule,
+};
+pub use http::{
+    FetchedScrape, ReqwestFetcher, ScrapeError, TargetFetcher, build_accept_header,
+    format_scrape_timeout_seconds, resolve_scrape_format,
+};
+#[cfg(feature = "arrow")]
+pub use manager::ArrowScrapeSink;
+pub use manager::{
+    ManagerStats, RawScrapeSink, ScrapeCommand, ScrapeControl, ScrapeManager, ScrapeReport,
+};
+pub use pool::{BodyBufferPool, PooledBodyBuffer};
+#[cfg(feature = "arrow")]
+pub use pool::{DecoderPool, PooledDecoder};
+pub use schedule::target_offset;