diff --git a/.hyperi-ci.yaml b/.hyperi-ci.yaml index 7e16778..8f89a92 100644 --- a/.hyperi-ci.yaml +++ b/.hyperi-ci.yaml @@ -84,7 +84,7 @@ release: helm: enabled: true contract: emit - library: "2.14.28" + library: "2.14.29" container: enabled: true dockerfile: Dockerfile diff --git a/Cargo.lock b/Cargo.lock index ff17288..07261db 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4304,9 +4304,9 @@ dependencies = [ [[package]] name = "scalo" -version = "2.14.28" +version = "2.14.29" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d6a5354f244c12085ea4935a9d915324a2203f58ac6d5f2bd3ee02364552e2a8" +checksum = "feceebb26b38815f5f1b78b03f235630d011240810c75af6e230dd7518245a1f" dependencies = [ "aes-gcm", "backon", diff --git a/Cargo.toml b/Cargo.toml index 2d92499..c26e22e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -34,7 +34,7 @@ tokio-util = { version = ">=0.7.20, <0.8", features = ["rt"] } # Shared library with transport abstraction. The floor moves with # `release.helm.library` in .hyperi-ci.yaml: the library renders only the # contract version its scalo release writes. -scalo = { version = ">=2.14.28, <3", features = ["config-schema", "transport-kafka", "transport-grpc", "dlq-kafka", "config", "config-reload", "deployment", "version-check", "scaling", "cli-service", "top", "logger", "metrics", "service-metrics", "expression", "memory", "worker-batch", "governor", "secrets", "sink-stack", "geoip-download"] } +scalo = { version = ">=2.14.29, <3", features = ["config-schema", "transport-kafka", "transport-grpc", "dlq-kafka", "config", "config-reload", "deployment", "version-check", "scaling", "cli-service", "top", "logger", "metrics", "service-metrics", "expression", "memory", "worker-batch", "governor", "secrets", "sink-stack", "geoip-download"] } # Steps scalo's jittered retry schedule (sink_stack backoff). Range MUST track # scalo's: BackoffBuilder only applies to the ExponentialBuilder scalo returns. @@ -126,7 +126,7 @@ reqwest = { version = ">=0.12, <0.13", default-features = false, features = ["bl # The Kafka broker image pin and the TLS / mTLS / SASL_SSL broker fixture the # Kafka tests share, so neither is copied here. The production dependency above # does not carry this feature. -scalo = { version = ">=2.14.28, <3", features = ["transport-kafka-test-support"] } +scalo = { version = ">=2.14.29, <3", features = ["transport-kafka-test-support"] } # The status code a raw Push is refused with, through scalo's generated gRPC # client. Same range as the copy scalo resolves, so cargo unifies to one. diff --git a/src/kafka/transport.rs b/src/kafka/transport.rs index 03e46f2..9d3e701 100644 --- a/src/kafka/transport.rs +++ b/src/kafka/transport.rs @@ -1287,10 +1287,13 @@ mod discovery_tests { use super::*; - /// A Strimzi broker: two loader topics plus everything else that lives on it. + /// A Strimzi broker: two loader topics, two a `topic_regex` can pick, plus + /// everything else that lives on it. const BROKER_TOPICS: &[&str] = &[ "main_land", "x_load", + "audit.events", + "audit.logins", "strimzi.cruisecontrol.metrics", "strimzi.cruisecontrol.modeltrainingsamples", "dlq_land", @@ -1359,25 +1362,18 @@ mod discovery_tests { fn discovery_subscribes_each_dlq_load_topic_and_no_other_dlq_topic() { let loaded = crate::config::DlqLoadConfig::default().topics; let broker: Vec<&str> = BROKER_TOPICS.iter().chain(DLQ_TOPICS).copied().collect(); - let strimzi: &[&str] = &[ - "strimzi.cruisecontrol.metrics", - "strimzi.cruisecontrol.modeltrainingsamples", - ]; - // Every broker topic no exclude takes out, under a broad include. - let broad: &[&str] = &[ - "_schemas", - "connect-offsets", - "main_land", - "strimzi.cruisecontrol.metrics", - "strimzi.cruisecontrol.modeltrainingsamples", - "x_load", - ]; + let audit: &[&str] = &["audit.events", "audit.logins"]; + // Every broker topic no exclude takes out, under a broad include. scalo's + // defaults drop the Strimzi, Connect and schema-registry topics. + let broad: &[&str] = &["audit.events", "audit.logins", "main_land", "x_load"]; let broad_and_dlq_land: Vec<&str> = broad.iter().copied().chain(["dlq_land"]).collect(); // (the loader's own DLQ topic, topic_regex, what discovery subscribes to // beside the five loaded topics) - let cases: [(&str, Option<&str>, &[&str]); 4] = [ + let cases: [(&str, Option<&str>, &[&str]); 5] = [ ("dlq_land", None, &["main_land", "x_load"]), - ("dlq_land", Some("^strimzi\\."), strimzi), + ("dlq_land", Some("^audit\\."), audit), + // A regex naming a system topic still cannot reach it. + ("dlq_land", Some("^strimzi\\."), &[]), ("dlq_land", Some(".*"), broad), // The loader's own DLQ topic is one it loads. ("dfe_loader_dlq", Some(".*"), &broad_and_dlq_land), @@ -1494,19 +1490,16 @@ mod discovery_tests { #[test] fn topic_regex_replaces_the_include_and_keeps_the_dlq_exclude() { let kafka = KafkaConfig { - topic_regex: Some("^strimzi\\.".to_string()), + topic_regex: Some("^audit\\.".to_string()), ..KafkaConfig::default() }; let cfg = TransportAdapter::convert_config(&kafka, &dlq_land(), &[]); - assert_eq!(cfg.topic_include, ["^strimzi\\."]); + assert_eq!(cfg.topic_include, ["^audit\\."]); assert!(cfg.topic_exclude.iter().any(|p| p == "^dlq_land$")); assert_eq!( resolve(&cfg, BROKER_TOPICS), - [ - "strimzi.cruisecontrol.metrics", - "strimzi.cruisecontrol.modeltrainingsamples" - ] + ["audit.events", "audit.logins"] ); }