GeezerVison is showing github.com — [pipeline-manager] Check memory and storage autoscaling limits in the… · feldera/feldera@e9013fb
File treeExpand file treeCollapse file tree Open diff view settings crates/pipeline-manager/src
Expand file treeCollapse file tree Open diff view settings Collapse file Copy file name to clipboardExpand all lines: crates/pipeline-manager/src/api/endpoints/pipeline_management.rs +91-18Lines changed: 91 additions & 18 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -37,7 +37,9 @@ use actix_web::{ | 37 | 37 | }; | 38 | 38 | use chrono::{DateTime, Utc}; | 39 | 39 | use feldera_types::adapter_stats::PipelineStatsErrorsResponse; | 40 | | -use feldera_types::config::{InputEndpointConfig, OutputEndpointConfig, RuntimeConfig}; | | 40 | +use feldera_types::config::{ | | 41 | + InputEndpointConfig, OutputEndpointConfig, ResourceConfig, RuntimeConfig, | | 42 | +}; | 41 | 43 | use feldera_types::error::ErrorResponse; | 42 | 44 | use feldera_types::pipeline_diff::{PipelineDiff, compute_pipeline_diff}; | 43 | 45 | use feldera_types::program_schema::ProgramSchema; | @@ -1130,7 +1132,7 @@ pub(crate) async fn post_pipeline( | 1130 | 1132 | if let Some(deployment_config) = &body.deployment_config { | 1131 | 1133 | check_unchanged_deployment_config(None, deployment_config)?; | 1132 | 1134 | } | 1133 | | - check_runtime_config_storage(&state, body.runtime_config.as_ref())?; | | 1135 | + check_runtime_config_limits(&state, body.runtime_config.as_ref())?; | 1134 | 1136 | let pipeline_descr: PipelineDescr = body.into(); | 1135 | 1137 | let pipeline = state | 1136 | 1138 | .db | @@ -1212,7 +1214,7 @@ pub(crate) async fn put_pipeline( | 1212 | 1214 | }; | 1213 | 1215 | check_unchanged_deployment_config(stored.as_ref(), deployment_config)?; | 1214 | 1216 | } | 1215 | | - check_runtime_config_storage(&state, body.runtime_config.as_ref())?; | | 1217 | + check_runtime_config_limits(&state, body.runtime_config.as_ref())?; | 1216 | 1218 | let pipeline_descr: PipelineDescr = body.into(); | 1217 | 1219 | let (is_new, pipeline) = db | 1218 | 1220 | .new_or_update_pipeline( | @@ -1314,7 +1316,7 @@ pub(crate) async fn patch_pipeline( | 1314 | 1316 | ) | 1315 | 1317 | .await; | 1316 | 1318 | } | 1317 | | - check_runtime_config_storage(&state, body.runtime_config.as_ref())?; | | 1319 | + check_runtime_config_limits(&state, body.runtime_config.as_ref())?; | 1318 | 1320 | let pipeline = state | 1319 | 1321 | .db | 1320 | 1322 | .lock() | @@ -1348,9 +1350,9 @@ pub(crate) async fn patch_pipeline( | 1348 | 1350 | .json(returned_pipeline)) | 1349 | 1351 | } | 1350 | 1352 |
| 1351 | | -/// Refuses a `runtime_config` with storage that the runner refuses at start, so that it fails | 1352 | | -/// now instead of at the next start. Only new values are checked: stored ones still load. | 1353 | | -fn check_runtime_config_storage( | | 1353 | +/// Refuses a `runtime_config` that the runner refuses at start, so that it fails now instead of | | 1354 | +/// at the next start. Only new values are checked: stored ones still load. | | 1355 | +fn check_runtime_config_limits( | 1354 | 1356 | state: &ServerState, | 1355 | 1357 | runtime_config: Option<&serde_json::Value>, | 1356 | 1358 | ) -> Result<(), DBError> { | @@ -1361,13 +1363,23 @@ fn check_runtime_config_storage( | 1361 | 1363 | let Ok(runtime_config) = serde_json::from_value::<RuntimeConfig>(value.clone()) else { | 1362 | 1364 | return Ok(()); | 1363 | 1365 | }; | | 1366 | + let refused = |error| DBError::InvalidRuntimeConfig { | | 1367 | + value: value.clone(), | | 1368 | + error, | | 1369 | + }; | 1364 | 1370 | state | 1365 | 1371 | .config | 1366 | 1372 | .check_storage(&runtime_config.resources) | 1367 | | - .map_err(|error| DBError::InvalidRuntimeConfig { | 1368 | | - value: value.clone(), | 1369 | | - error: ValidationError::UnsupportedStorage(error), | | 1373 | + .map_err(|error| refused(ValidationError::UnsupportedStorage(error)))?; | | 1374 | + state | | 1375 | + .config | | 1376 | + .check_memory(&runtime_config.resources) | | 1377 | + .and_then(|()| { | | 1378 | + state | | 1379 | + .config | | 1380 | + .check_storage_autoscaling(&runtime_config.resources) | 1370 | 1381 | }) | | 1382 | + .map_err(|error| refused(ValidationError::NotAllowedOnInstallation(error))) | 1371 | 1383 | } | 1372 | 1384 |
| 1373 | 1385 | /// Handles a `PATCH` of `deployment_config`, which resizes a running pipeline. | @@ -1398,6 +1410,15 @@ async fn patch_deployment_config( | 1398 | 1410 | })?; | 1399 | 1411 | } | 1400 | 1412 | let patch = parse_deployment_config_patch(deployment_config)?; | | 1413 | + // Only the patched memory is checked, as in `check_runtime_config_limits`. | | 1414 | + state | | 1415 | + .config | | 1416 | + .check_memory(&ResourceConfig { | | 1417 | + memory_min: patch.resources.memory_min, | | 1418 | + memory_max: patch.resources.memory_max, | | 1419 | + ..Default::default() | | 1420 | + }) | | 1421 | + .map_err(|reason| DBError::InvalidDeploymentPatch { reason })?; | 1401 | 1422 | let pipeline = state | 1402 | 1423 | .db | 1403 | 1424 | .lock() | @@ -2513,10 +2534,11 @@ mod tests { | 2513 | 2534 | } | 2514 | 2535 | } | 2515 | 2536 |
| 2516 | | - /// `POST`, `PUT`, and `PATCH` refuse a `runtime_config` with storage that the runner refuses at | 2517 | | - /// start. A stored `runtime_config` still loads, and other fields can still be edited. | | 2537 | + /// `POST`, `PUT`, and `PATCH` refuse a `runtime_config` that the runner refuses at start, and a | | 2538 | + /// `PATCH` of `deployment_config` refuses such memory. A stored `runtime_config` still loads, | | 2539 | + /// and other fields can still be edited. | 2518 | 2540 | #[actix_web::test] | 2519 | | - async fn runtime_config_storage_requests() { | | 2541 | + async fn runtime_config_limit_requests() { | 2520 | 2542 | use crate::api::main::ServerState; | 2521 | 2543 | use crate::db::test::setup_pg; | 2522 | 2544 | use crate::db::types::tenant::TenantId; | @@ -2531,6 +2553,8 @@ mod tests { | 2531 | 2553 | let pipeline = |name: &str, resources: serde_json::Value| json!({"name": name, "program_code": "", "runtime_config": {"resources": resources}}); | 2532 | 2554 | let patch = | 2533 | 2555 | |resources: serde_json::Value| json!({"runtime_config": {"resources": resources}}); | | 2556 | + let resize = | | 2557 | + |resources: serde_json::Value| json!({"deployment_config": {"resources": resources}}); | 2534 | 2558 | let cases = [ | 2535 | 2559 | // (limits set, method, uri, body, expected status) | 2536 | 2560 | // Stored before the limits were set. | @@ -2612,10 +2636,50 @@ mod tests { | 2612 | 2636 | patch(json!({"storage_class": "io2"})), | 2613 | 2637 | 400, | 2614 | 2638 | ), | | 2639 | + ( | | 2640 | + true, | | 2641 | + "POST", | | 2642 | + "/pipelines", | | 2643 | + pipeline("m", json!({"memory_max": "2G"})), | | 2644 | + 400, | | 2645 | + ), | | 2646 | + ( | | 2647 | + true, | | 2648 | + "POST", | | 2649 | + "/pipelines", | | 2650 | + pipeline("a", json!({"storage_min": "500M", "storage_max": "1G"})), | | 2651 | + 400, | | 2652 | + ), | | 2653 | + // Unset, storage autoscaling is not checked. | | 2654 | + ( | | 2655 | + false, | | 2656 | + "POST", | | 2657 | + "/pipelines", | | 2658 | + pipeline("a", json!({"storage_min": "500M", "storage_max": "1G"})), | | 2659 | + 201, | | 2660 | + ), | | 2661 | + // A resize is checked before the pipeline is looked up. | | 2662 | + ( | | 2663 | + true, | | 2664 | + "PATCH", | | 2665 | + "/pipelines/missing", | | 2666 | + resize(json!({"memory_max": "2G"})), | | 2667 | + 400, | | 2668 | + ), | | 2669 | + ( | | 2670 | + true, | | 2671 | + "PATCH", | | 2672 | + "/pipelines/missing", | | 2673 | + resize(json!({"memory_max": "500M"})), | | 2674 | + 404, | | 2675 | + ), | 2615 | 2676 | ]; | 2616 | 2677 | for (limits, method, uri, body, expected_status) in cases { | 2617 | 2678 | let mut state = ServerState::test_state(db.clone()).await; | | 2679 | + state.config.enable_pipeline_resize = true; | 2618 | 2680 | if limits { | | 2681 | + state.config.upper_limit_memory_megabyte = Some(1000.0); | | 2682 | + state.config.enable_storage_autoscaling = Some(false); | 2619 | 2683 | state.config.upper_limit_storage_megabyte = Some(1000.0); | 2620 | 2684 | state.config.enable_storage_classes_allowlist = true; | 2621 | 2685 | state.config.storage_classes = vec!["gp3".to_string()]; | @@ -2644,12 +2708,21 @@ mod tests { | 2644 | 2708 | let response: serde_json::Value = test::read_body_json(resp).await; | 2645 | 2709 | assert_eq!(status, expected_status, "{method} {uri} {body}: {response}"); | 2646 | 2710 | if expected_status == 400 { | 2647 | | - assert_eq!(response["error_code"], "InvalidRuntimeConfig", "{response}"); | 2648 | 2711 | let message = response["message"].as_str().unwrap(); | 2649 | | - assert!( | 2650 | | - message.contains("unsupported storage: resources."), | 2651 | | - "{message}" | 2652 | | - ); | | 2712 | + if body.get("deployment_config").is_some() { | | 2713 | + assert_eq!( | | 2714 | + response["error_code"], "InvalidDeploymentPatch", | | 2715 | + "{response}" | | 2716 | + ); | | 2717 | + assert!(message.contains("resources.memory_max (2G)"), "{message}"); | | 2718 | + } else { | | 2719 | + assert_eq!(response["error_code"], "InvalidRuntimeConfig", "{response}"); | | 2720 | + assert!( | | 2721 | + message.contains("unsupported storage: resources.") | | 2722 | + || message.contains("not allowed on this installation: "), | | 2723 | + "{message}" | | 2724 | + ); | | 2725 | + } | 2653 | 2726 | } | 2654 | 2727 | } | 2655 | 2728 | } | |
Collapse file Copy file name to clipboardExpand all lines: crates/pipeline-manager/src/auth.rs +2Lines changed: 2 additions & 0 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -1966,6 +1966,8 @@ mod test { | 1966 | 1966 | enable_volume_attributes_classes_allowlist: false, | 1967 | 1967 | volume_attributes_classes: vec![], | 1968 | 1968 | upper_limit_storage_megabyte: None, | | 1969 | + upper_limit_memory_megabyte: None, | | 1970 | + enable_storage_autoscaling: None, | 1969 | 1971 | } | 1970 | 1972 | } | 1971 | 1973 |
| |
Collapse file Copy file name to clipboardExpand all lines: crates/pipeline-manager/src/config.rs +139Lines changed: 139 additions & 0 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -1543,6 +1543,18 @@ pub struct ApiServerConfig { | 1543 | 1543 | #[serde(default)] | 1544 | 1544 | #[arg(long, env = "FELDERA_UPPER_LIMIT_STORAGE_MEGABYTE")] | 1545 | 1545 | pub upper_limit_storage_megabyte: Option<f64>, | | 1546 | + | | 1547 | + /// Maximum `resources.memory_min` and `resources.memory_max`, in megabytes (M). Unset is no | | 1548 | + /// maximum. Set it the same as the runner. | | 1549 | + #[serde(default)] | | 1550 | + #[arg(long, env = "FELDERA_UPPER_LIMIT_MEMORY_MEGABYTE")] | | 1551 | + pub upper_limit_memory_megabyte: Option<f64>, | | 1552 | + | | 1553 | + /// Whether a pipeline can set both `resources.storage_min` and `resources.storage_max`, which | | 1554 | + /// turns on storage autoscaling. Unset is not checked. Set it the same as the runner. | | 1555 | + #[serde(default)] | | 1556 | + #[arg(long, action = clap::ArgAction::Set, env = "FELDERA_ENABLE_STORAGE_AUTOSCALING")] | | 1557 | + pub enable_storage_autoscaling: Option<bool>, | 1546 | 1558 | } | 1547 | 1559 |
| 1548 | 1560 | /// A trust relationship granting the platform-wide `owner` role, declared at | @@ -1686,6 +1698,48 @@ impl ApiServerConfig { | 1686 | 1698 | Ok(()) | 1687 | 1699 | } | 1688 | 1700 |
| | 1701 | + /// Refuses the memory in `resources` that the runner refuses at start. The runner raises a | | 1702 | + /// value below its minimum, so that is not refused. | | 1703 | + pub(crate) fn check_memory(&self, resources: &ResourceConfig) -> Result<(), String> { | | 1704 | + if let (Some(min), Some(max)) = (resources.memory_min, resources.memory_max) | | 1705 | + && min > max | | 1706 | + { | | 1707 | + return Err(format!( | | 1708 | + "resources.memory_min ({min}) is more than resources.memory_max ({max}). Lower memory_min or raise memory_max." | | 1709 | + )); | | 1710 | + } | | 1711 | + if let Some(upper_limit) = self.upper_limit_memory_megabyte { | | 1712 | + for (field, amount) in [ | | 1713 | + ("memory_min", resources.memory_min), | | 1714 | + ("memory_max", resources.memory_max), | | 1715 | + ] { | | 1716 | + if let Some(amount) = amount | | 1717 | + && amount > ConfigSize::from_mb(upper_limit as u64) | | 1718 | + { | | 1719 | + return Err(format!( | | 1720 | + "resources.{field} ({amount}) is more than the maximum of {upper_limit}M on this installation. Lower it, or ask an administrator to raise kubernetesRunner.upperLimitMemoryMegabyte." | | 1721 | + )); | | 1722 | + } | | 1723 | + } | | 1724 | + } | | 1725 | + Ok(()) | | 1726 | + } | | 1727 | + | | 1728 | + /// Refuses storage autoscaling, both `storage_min` and `storage_max`, when the runner does not | | 1729 | + /// enable it. | | 1730 | + pub(crate) fn check_storage_autoscaling( | | 1731 | + &self, | | 1732 | + resources: &ResourceConfig, | | 1733 | + ) -> Result<(), String> { | | 1734 | + if self.enable_storage_autoscaling == Some(false) | | 1735 | + && resources.storage_min.is_some() | | 1736 | + && resources.storage_max.is_some() | | 1737 | + { | | 1738 | + return Err("resources.storage_min and resources.storage_max are both set, which turns on storage autoscaling, and it is not enabled on this installation. Set only one of them for a fixed size, or ask an administrator to set kubernetesRunner.enableStorageAutoscaling.".to_string()); | | 1739 | + } | | 1740 | + Ok(()) | | 1741 | + } | | 1742 | + | 1689 | 1743 | /// Where a trust registered through the API may point. | 1690 | 1744 | pub(crate) fn tenant_issuer_policy(&self) -> TenantIssuerPolicy { | 1691 | 1745 | if self.allow_internal_tenant_trust_issuers { | @@ -1770,6 +1824,8 @@ impl ApiServerConfig { | 1770 | 1824 | enable_volume_attributes_classes_allowlist: false, | 1771 | 1825 | volume_attributes_classes: vec![], | 1772 | 1826 | upper_limit_storage_megabyte: None, | | 1827 | + upper_limit_memory_megabyte: None, | | 1828 | + enable_storage_autoscaling: None, | 1773 | 1829 | } | 1774 | 1830 | } | 1775 | 1831 | } | @@ -2635,6 +2691,89 @@ mod tests { | 2635 | 2691 | } | 2636 | 2692 | } | 2637 | 2693 |
| | 2694 | + /// `check_memory` and `check_storage_autoscaling` refuse what the Kubernetes runner refuses at | | 2695 | + /// start, and only when the installation says so. | | 2696 | + #[test] | | 2697 | + fn memory_and_storage_autoscaling_are_refused_as_the_runner_refuses_them() { | | 2698 | + let resources = | | 2699 | + |value: serde_json::Value| -> ResourceConfig { serde_json::from_value(value).unwrap() }; | | 2700 | + let check = |config: &ApiServerConfig, value: serde_json::Value| { | | 2701 | + let resources = resources(value); | | 2702 | + config | | 2703 | + .check_memory(&resources) | | 2704 | + .and_then(|()| config.check_storage_autoscaling(&resources)) | | 2705 | + }; | | 2706 | + let both = serde_json::json!({ | | 2707 | + "memory_min": "64G", "memory_max": "64G", "storage_min": "1G", "storage_max": "8G", | | 2708 | + }); | | 2709 | + let mut config = ApiServerConfig::test_config(); | | 2710 | + | | 2711 | + // Unset, nothing but `memory_min` > `memory_max` is refused. | | 2712 | + assert_eq!(check(&config, both.clone()), Ok(())); | | 2713 | + let error = check( | | 2714 | + &config, | | 2715 | + serde_json::json!({"memory_min": "2G", "memory_max": "1G"}), | | 2716 | + ) | | 2717 | + .unwrap_err(); | | 2718 | + assert!(error.contains("resources.memory_min (2G)"), "{error}"); | | 2719 | + | | 2720 | + // Allowed, as set by Helm. | | 2721 | + config.upper_limit_memory_megabyte = Some(64_000.0); | | 2722 | + config.enable_storage_autoscaling = Some(true); | | 2723 | + assert_eq!(check(&config, both.clone()), Ok(())); | | 2724 | + | | 2725 | + config.upper_limit_memory_megabyte = Some(1000.0); | | 2726 | + for (value, expected) in [ | | 2727 | + ( | | 2728 | + serde_json::json!({"memory_min": "2G"}), | | 2729 | + "resources.memory_min (2G) is more than the maximum of 1000M", | | 2730 | + ), | | 2731 | + ( | | 2732 | + serde_json::json!({"memory_max": "1001M"}), | | 2733 | + "resources.memory_max (1001M) is more than the maximum of 1000M", | | 2734 | + ), | | 2735 | + ] { | | 2736 | + let error = check(&config, value).unwrap_err(); | | 2737 | + assert!(error.contains(expected), "{error}"); | | 2738 | + } | | 2739 | + assert_eq!( | | 2740 | + check(&config, serde_json::json!({"memory_max": "1G"})), | | 2741 | + Ok(()) | | 2742 | + ); | | 2743 | + | | 2744 | + config.enable_storage_autoscaling = Some(false); | | 2745 | + let error = check( | | 2746 | + &config, | | 2747 | + serde_json::json!({"storage_min": "1G", "storage_max": "8G"}), | | 2748 | + ) | | 2749 | + .unwrap_err(); | | 2750 | + assert!( | | 2751 | + error.contains("kubernetesRunner.enableStorageAutoscaling"), | | 2752 | + "{error}" | | 2753 | + ); | | 2754 | + // A fixed size is not autoscaling. | | 2755 | + assert_eq!( | | 2756 | + check(&config, serde_json::json!({"storage_max": "8G"})), | | 2757 | + Ok(()) | | 2758 | + ); | | 2759 | + } | | 2760 | + | | 2761 | + /// The limits parse as the Helm chart passes them, and are unset without them. | | 2762 | + #[test] | | 2763 | + fn runner_limits_parse_from_helm_arguments() { | | 2764 | + let unset = ApiServerConfig::try_parse_from(["api-server"]).unwrap(); | | 2765 | + assert_eq!(unset.upper_limit_memory_megabyte, None); | | 2766 | + assert_eq!(unset.enable_storage_autoscaling, None); | | 2767 | + let config = ApiServerConfig::try_parse_from([ | | 2768 | + "api-server", | | 2769 | + "--upper-limit-memory-megabyte=32000", | | 2770 | + "--enable-storage-autoscaling=false", | | 2771 | + ]) | | 2772 | + .unwrap(); | | 2773 | + assert_eq!(config.upper_limit_memory_megabyte, Some(32000.0)); | | 2774 | + assert_eq!(config.enable_storage_autoscaling, Some(false)); | | 2775 | + } | | 2776 | + | 2638 | 2777 | /// The platform version suffix matches the edition of the build. | 2639 | 2778 | #[test] | 2640 | 2779 | fn platform_version_matches_edition() { | |
Collapse file Copy file name to clipboardExpand all lines: crates/pipeline-manager/src/db/types/utils.rs +2Lines changed: 2 additions & 0 deletions | Original file line number | Diff line number | Diff line change |
|---|
@@ -189,6 +189,8 @@ pub enum ValidationError { | 189 | 189 | InvalidDevTweaks(String), | 190 | 190 | #[error("unsupported storage: {0}")] | 191 | 191 | UnsupportedStorage(String), | | 192 | + #[error("not allowed on this installation: {0}")] | | 193 | + NotAllowedOnInstallation(String), | 192 | 194 | } | 193 | 195 |
| 194 | 196 | /// Deserializes generic JSON value into [`RuntimeConfig`] and performs any additional validation. | |
|