diff --git a/Cargo.lock b/Cargo.lock index f1e318a8ef7..7e3d27b4672 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5421,6 +5421,7 @@ dependencies = [ "reqwest 0.12.24", "sentry", "serde_json", + "sysinfo 0.38.4", "tracing", "tracing-subscriber", ] diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs index b9cd2600227..b815ff27866 100644 --- a/crates/adapters/src/controller.rs +++ b/crates/adapters/src/controller.rs @@ -5577,15 +5577,22 @@ impl ControllerInit { if max_rss_mb.is_none() && let Some(memory_mb_max) = &pipeline_config.global.resources.memory_mb_max { - warn!( - "RSS memory limit ('max_rss_mb') is not set, but a Kubernetes memory limit \ + if feldera_observability::system::running_in_kubernetes() { + warn!( + "RSS memory limit ('max_rss_mb') is not set, but a Kubernetes pod memory limit \ ('resources.memory_mb_max' = {memory_mb_max} MB) is configured. \ -Using the Kubernetes limit as the RSS memory limit." - ); +Using the pod limit as the RSS memory limit." + ); + } else { + info!( + "RSS memory limit ('max_rss_mb') is not set; using 'resources.memory_mb_max' \ +({memory_mb_max} MB) as the RSS memory limit." + ); + } max_rss_mb = Some(*memory_mb_max); } else if max_rss_mb.is_none() && pipeline_config.global.resources.memory_mb_max.is_none() { warn!( - "RSS memory limit ('max_rss_mb') is not set, and no Kubernetes memory limit \ + "RSS memory limit ('max_rss_mb') is not set, and no deployment memory limit \ ('resources.memory_mb_max') is configured. We recommend configuring at least one of these settings to avoid out-of-memory failures." ); } else if let Some(max_rss_mb) = max_rss_mb @@ -5593,8 +5600,8 @@ Using the Kubernetes limit as the RSS memory limit." && max_rss_mb > *memory_mb_max { warn!( - "RSS memory limit ('max_rss_mb') is set to {max_rss_mb} MB exceeds the Kubernetes memory limit \ -('resources.memory_mb_max' = {memory_mb_max} MB) is configured. This will likely cause out-of-memory failures." + "RSS memory limit ('max_rss_mb' = {max_rss_mb} MB) exceeds the deployment memory limit \ +('resources.memory_mb_max' = {memory_mb_max} MB). This will likely cause out-of-memory failures." ); } @@ -8415,7 +8422,7 @@ impl ControllerInner { } TransactionState::Started { tid, - start, + start: _, processed_records, } if next.is_none() || next != open => { // The next step belongs to a different transaction, or the diff --git a/crates/adapters/src/server.rs b/crates/adapters/src/server.rs index b6bc9ca73ca..5f33e7d2655 100644 --- a/crates/adapters/src/server.rs +++ b/crates/adapters/src/server.rs @@ -1024,7 +1024,7 @@ fn parse_config(config_file: impl AsRef) -> Result) -> Result bool { + let enabled = self + .runtime + .as_ref() + .is_some_and(|runtime| runtime.runtime().adaptive_joins()); + if !enabled { + warn!( + "ignoring balancer hints: adaptive joins are disabled (dev_tweaks.adaptive_joins)" + ); + } + enabled + } + pub fn set_balancer_hint_by_global_id( &mut self, global_node_id: &GlobalNodeId, diff --git a/crates/dbsp/src/circuit/runtime.rs b/crates/dbsp/src/circuit/runtime.rs index 97243c0a868..822207ec004 100644 --- a/crates/dbsp/src/circuit/runtime.rs +++ b/crates/dbsp/src/circuit/runtime.rs @@ -1088,6 +1088,14 @@ impl Runtime { self.inner().allow_input_during_commit } + /// Whether adaptive (dynamically balanced) joins are enabled, per + /// `dev_tweaks.adaptive_joins`. Unlike [`Self::with_dev_tweaks`], this reads the + /// runtime's configured tweaks, so it is valid off the worker threads (e.g. on + /// the thread that owns the `DBSPHandle`). + pub fn adaptive_joins(&self) -> bool { + self.inner().dev_tweaks.adaptive_joins() + } + /// Returns the worker index as a string. /// /// This is useful for metric labels. diff --git a/crates/dbsp/src/operator/dynamic/balance/balancer.rs b/crates/dbsp/src/operator/dynamic/balance/balancer.rs index e4156af223f..7bc28ff8a6b 100644 --- a/crates/dbsp/src/operator/dynamic/balance/balancer.rs +++ b/crates/dbsp/src/operator/dynamic/balance/balancer.rs @@ -776,6 +776,9 @@ impl BalancerInner { /// Set a hint for a stream. /// + /// Fails with `NotRegisteredWithBalancer` if the stream is not managed by the balancer, + /// i.e., it is not an input to an adaptively balanced join. + /// /// Fails if the hint specifies a fixed policy that cannot be enforced in the current state, i.e., /// it either contradicts the set of existing hints for other streams (e.g., we refuse to join /// two streams both of which have Broadcast policies) or the policy cannot be enforced during the @@ -789,7 +792,15 @@ impl BalancerInner { // Runtime::worker_index(), // hint // ); - let cluster_index = *self.stream_to_cluster.get(&node_id).unwrap(); + // A stream is absent from `stream_to_cluster` when it is not an input to an + // adaptively balanced join (e.g. an input to a plain stream join, such as a + // `now()` cross join, or a join the compiler could not make adaptive). Setting + // a policy hint on such a stream is meaningless, so reject it with an error + // instead of panicking and taking down the worker thread. + let cluster_index = *self + .stream_to_cluster + .get(&node_id) + .ok_or(BalancerError::NotRegisteredWithBalancer(node_id))?; // Apply the hint to the stored hints, remembering the previous value so we // can roll back if the hint turns out to be unenforceable. The hint must be diff --git a/crates/feldera-observability/Cargo.toml b/crates/feldera-observability/Cargo.toml index 8a852aa8823..4a106f99abd 100644 --- a/crates/feldera-observability/Cargo.toml +++ b/crates/feldera-observability/Cargo.toml @@ -20,3 +20,4 @@ serde_json = { workspace = true } tracing = { workspace = true } tracing-subscriber = { workspace = true, features = ["env-filter", "json"] } colored = { workspace = true } +sysinfo = { workspace = true } diff --git a/crates/feldera-observability/src/lib.rs b/crates/feldera-observability/src/lib.rs index 975473a81ff..63ecb33e2da 100644 --- a/crates/feldera-observability/src/lib.rs +++ b/crates/feldera-observability/src/lib.rs @@ -69,6 +69,7 @@ pub fn actix_middleware() -> sentry::integrations::actix::Sentry { pub mod fips; pub mod json_logging; +pub mod system; fn trace_header_value() -> Option { if !sentry_enabled() { diff --git a/crates/feldera-observability/src/system.rs b/crates/feldera-observability/src/system.rs new file mode 100644 index 00000000000..6ca57b07126 --- /dev/null +++ b/crates/feldera-observability/src/system.rs @@ -0,0 +1,39 @@ +//! System introspection utilities. + +/// True when this process runs inside a Kubernetes pod. +/// +/// Kubernetes injects `KUBERNETES_SERVICE_HOST` into every pod, and the +/// pipeline manager rejects user-supplied `KUBERNETES_*` variables in the +/// pipeline's `env` config. +pub fn running_in_kubernetes() -> bool { + std::env::var_os("KUBERNETES_SERVICE_HOST").is_some() +} + +/// Returns the memory, in megabytes (MB), of the current system, or `None` +/// when the operating system does not report it. +/// +/// This is the min of the host's available memory and the cgroup budget +/// (what a container or Kubernetes pod memory limit sets). +pub fn total_memory_megabyte() -> Option { + let mut system = sysinfo::System::new(); + system.refresh_memory(); + let mut available = system.total_memory(); + if let Some(limits) = system.cgroup_limits() { + available = available.min(limits.total_memory); + } + let mb = available / 1_000_000; + // `sysinfo` reports 0 on platforms it does not support. + if mb > 0 { Some(mb) } else { None } +} + +#[cfg(test)] +mod tests { + /// Guards the `sysinfo` call sequence (`cgroup_limits` panics when memory + /// was not refreshed first). + #[cfg(target_os = "linux")] + #[test] + fn reports_nonzero_memory() { + let mb = super::total_memory_megabyte().expect("Linux reports available memory"); + assert!(mb > 0, "available memory must be positive, got {mb} MB"); + } +} diff --git a/crates/pipeline-manager/src/runner/local_runner.rs b/crates/pipeline-manager/src/runner/local_runner.rs index b741b04fe80..949deef9951 100644 --- a/crates/pipeline-manager/src/runner/local_runner.rs +++ b/crates/pipeline-manager/src/runner/local_runner.rs @@ -13,9 +13,10 @@ use crate::runner::error::RunnerError; use crate::runner::pipeline_executor::{PipelineExecutor, ProvisionStatus}; use crate::runner::pipeline_logs::{LogMessage, LogsSender}; use async_trait::async_trait; +use feldera_observability::system::total_memory_megabyte; use feldera_observability::ReqwestTracingExt; use feldera_types::config::{ - PipelineConfig, PipelineConfigProgramInfo, StorageCacheConfig, StorageConfig, + PipelineConfig, PipelineConfigProgramInfo, RuntimeConfig, StorageCacheConfig, StorageConfig, }; use feldera_types::runtime_status::{BootstrapConfig, RuntimeDesiredStatus}; use reqwest::StatusCode; @@ -791,6 +792,13 @@ impl LocalRunner { base_config.outputs = program_info.outputs; base_config.program_ir = program_info.program_ir; + // The memory budget is per host, and all host processes share this + // machine, so each host gets an equal share of the available memory. + apply_default_memory_limit( + &mut base_config.global, + per_host_share(total_memory_megabyte(), n_hosts), + ); + // Retrieve the pipeline binary once; every host process runs the same // executable. let binary_file_path = self @@ -1105,6 +1113,10 @@ impl PipelineExecutor for LocalRunner { deployment_config.outputs = program_info.outputs; deployment_config.program_ir = program_info.program_ir; + // Unlike Kubernetes, a local deployment has no pod memory limit for + // the pipeline to fall back on, so derive one from this machine. + apply_default_memory_limit(&mut deployment_config.global, total_memory_megabyte()); + // Write config as YAML and JSON // // Newer pipelines will read the JSON, older ones will read the YAML. @@ -1390,6 +1402,42 @@ impl PipelineExecutor for LocalRunner { } } +/// Splits `available_mb` evenly among `n_hosts` colocated host processes; +/// `None` when the share rounds down to nothing. +fn per_host_share(available_mb: Option, n_hosts: usize) -> Option { + available_mb + .map(|mb| mb / n_hosts.max(1) as u64) + .filter(|mb| *mb > 0) +} + +/// Sets `global.resources.memory_mb_max` to `available_mb` when the pipeline +/// has no memory budget. +fn apply_default_memory_limit(global: &mut RuntimeConfig, available_mb: Option) { + if global.effective_memory_mb().is_none() { + if let Some(available_mb) = available_mb { + info!( + "pipeline has no memory limit ('max_rss_mb' or 'resources.memory_mb_max'): \ +defaulting 'resources.memory_mb_max' to {available_mb} MB" + ); + global.resources.memory_mb_max = Some(available_mb); + } + } +} + +#[cfg(test)] +mod memory_limit_tests { + use super::per_host_share; + + /// Colocated hosts split the machine evenly; a share that rounds down to + /// nothing must yield `None`, not a 0 MB limit. + #[test] + fn hosts_split_available_memory() { + assert_eq!(per_host_share(Some(4000), 4), Some(1000)); + assert_eq!(per_host_share(Some(3), 4), None); + assert_eq!(per_host_share(None, 4), None); + } +} + #[cfg(test)] mod multihost_tests { use super::{ diff --git a/docs.feldera.com/docs/operations/memory.md b/docs.feldera.com/docs/operations/memory.md index 276d5c372e4..ea58bf92936 100644 --- a/docs.feldera.com/docs/operations/memory.md +++ b/docs.feldera.com/docs/operations/memory.md @@ -13,6 +13,13 @@ If `max_rss_mb` is not set but `memory_mb_max` is configured in the `resources` [Runtime configuration], the latter is used as the effective memory cap. We **strongly recommend** setting at least one of these parameters to prevent out-of-memory failures. +On single-host deployments (for example, in Docker), if neither +parameter is set, `resources.memory_mb_max` is set to the smaller +value of host memory size, and container memory limit. In a +multi-host deployment each pipeline process that finds neither +parameter set derives `max_rss_mb` from its own host's available +memory at startup. + When either `max_rss_mb` or `resources.memory_mb_max` is configured, the pipeline reports its current memory pressure level (`low`, `moderate`, `high`, or `critical`) via the [`memory_pressure` metric](/operations/metrics). High or critical pressure indicates that memory usage diff --git a/python/tests/workloads/test_now.py b/python/tests/workloads/test_now.py index c96189c6d91..a2a75c8e59a 100644 --- a/python/tests/workloads/test_now.py +++ b/python/tests/workloads/test_now.py @@ -88,7 +88,7 @@ def test_now(self): ViewSpec( "v", """ - select + select /*+ BROADCAST(v_now) */ t.*, v_now.now_ts as now_ts from t, v_now @@ -101,7 +101,10 @@ def test_now(self): tables, views, # This test uses >2GB of storage in the ad hoc query. - resources=Resources(storage_mb_max=8192), + resources=Resources( + storage_mb_max=8192, + memory_mb_max=1024, + ), ) pipeline.start() diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java index 22da616ce46..07752d34d41 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/ToRustVisitor.java @@ -303,8 +303,7 @@ public static void registerPreAndPostprocessors(DBSPCompiler compiler, IIndentSt } void emitBalancerHints() { - for (var hint: this.postfix.balancerHints) - hint.emit(this.builder); + CircuitPostfix.emitBalancerHints(this.builder, this.postfix.balancerHints); } @Override diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/multi/CircuitWriter.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/multi/CircuitWriter.java index d436fecc869..31e982b0d66 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/multi/CircuitWriter.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/backend/rust/multi/CircuitWriter.java @@ -220,8 +220,7 @@ void writeCircuit(DBSPCompiler compiler, DBSPCircuit circuit) { } void emitBalancerHints() { - for (var hint: this.materializations.balancerHints) - hint.emit(this.builder()); + CircuitPostfix.emitBalancerHints(this.builder(), this.materializations.balancerHints); } @Override diff --git a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitPostfix.java b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitPostfix.java index 4a0d7f5ee85..c70ae12be06 100644 --- a/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitPostfix.java +++ b/sql-to-dbsp-compiler/SQL-compiler/src/main/java/org/dbsp/sqlCompiler/compiler/visitors/outer/CircuitPostfix.java @@ -81,6 +81,19 @@ public void recordHint(DBSPBinaryOperator operator, JoinStrategy strategy) { this.balancerHints.add(hint); } + /** Emit the recorded balancer hints, gated on adaptive joins being enabled. + * When adaptive joins are disabled (`dev_tweaks.adaptive_joins`), the streams a + * hint references are never registered with the balancer; `adaptive_joins_enabled` + * then returns false (logging a warning) and the hints are skipped. */ + public static void emitBalancerHints(IIndentStream builder, List hints) { + if (hints.isEmpty()) + return; + builder.append("if circuit.adaptive_joins_enabled() {").increase().newline(); + for (BalancerHint hint: hints) + hint.emit(builder); + builder.decrease().append("}").newline(); + } + public void clearRegions() { this.regionsCreated.clear(); }