1use serde::{Deserialize, Serialize};
26
27use crate::capability_report::LogsCapabilities;
28
29pub const STANDING_WORK_STATUSES: [&str; 4] = ["active", "paused", "superseded", "revoked"];
34
35pub const ONE_SHOT_WORK_STATUSES: [&str; 9] = [
37 "dispatched",
38 "accepted",
39 "running",
40 "paused",
41 "succeeded",
42 "failed",
43 "timed_out",
44 "canceled",
45 "expired",
46];
47
48pub const AGENT_REPORTABLE_WORK_STATUSES: [&str; 3] = ["running", "succeeded", "failed"];
54
55pub const ONE_SHOT_TERMINAL_STATUSES: [&str; 5] =
57 ["succeeded", "failed", "timed_out", "canceled", "expired"];
58
59#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, ::jumo_derive::Jumo)]
61#[jumo(kind = "state", domain = "Control", module = "Control.Agent.Work")]
62#[serde(rename_all = "PascalCase")]
63pub enum WorkKind {
64 Standing,
66 OneShot,
68}
69
70#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ::jumo_derive::Jumo)]
76#[jumo(kind = "struct", domain = "Control", module = "Control.Agent.Work")]
77#[serde(deny_unknown_fields)]
78pub struct StandingWork {
79 pub work_id: String,
80 pub agent_id: String,
81 pub family: String,
83 pub spec: String,
85 pub catalog_version: i64,
87 #[serde(default)]
89 pub proposal_id: Option<String>,
90 pub plan_version: i64,
92 pub effective_from: String,
93 pub status: String,
95 pub updated_by: String,
96 pub updated_at: String,
97}
98
99#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ::jumo_derive::Jumo)]
101#[jumo(kind = "struct", domain = "Control", module = "Control.Agent.Work")]
102#[serde(deny_unknown_fields)]
103pub struct OneShotWork {
104 pub work_id: String,
105 pub agent_id: String,
106 pub action: String,
108 pub spec: String,
109 pub scheduled_at: String,
111 pub deadline_at: String,
113 pub timeout_seconds: i64,
115 pub interruptible: bool,
117 pub status: String,
119 #[serde(default)]
121 pub paused_at: Option<String>,
122 pub paused_total_seconds: i64,
124 #[serde(default)]
126 pub current_step: Option<String>,
127 #[serde(default)]
128 pub completed_steps: Vec<String>,
129 pub attempt: i64,
131 pub issued_by: String,
132 pub issued_at: String,
133}
134
135impl OneShotWork {
136 pub fn is_outstanding(&self) -> bool {
138 !ONE_SHOT_TERMINAL_STATUSES.contains(&self.status.as_str())
139 }
140
141 pub fn can_pause(&self) -> bool {
146 self.interruptible && matches!(self.status.as_str(), "accepted" | "running")
147 }
148}
149
150#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, ::jumo_derive::Jumo)]
152#[jumo(kind = "struct", domain = "Control", module = "Control.Agent.Work")]
153#[serde(deny_unknown_fields)]
154pub struct WorkReceipt {
155 pub work_id: String,
156 pub agent_id: String,
157 pub work_kind: WorkKind,
158 pub status: String,
160 pub plan_version: i64,
161 pub created_at: String,
162}
163
164#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
181#[serde(deny_unknown_fields)]
182pub struct WorkSpec {
183 #[serde(default)]
185 pub units: Vec<WorkSpecUnit>,
186}
187
188#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
194#[serde(deny_unknown_fields)]
195pub struct WorkSpecUnit {
196 pub unit_id: String,
197 pub capability: String,
199 #[serde(default)]
201 pub rule_ref: String,
202 #[serde(default)]
205 pub requires_privilege: String,
206 #[serde(default)]
207 pub sources: Vec<WorkSpecSource>,
208}
209
210#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
215#[serde(deny_unknown_fields)]
216pub struct WorkSpecSource {
217 pub kind: String,
218 pub target: String,
219 #[serde(default = "default_multiline")]
225 pub multiline: String,
226 #[serde(default, skip_serializing_if = "Vec::is_empty")]
236 pub exclude_globs: Vec<String>,
237 #[serde(default, skip_serializing_if = "ContainerFilters::is_empty")]
247 pub filters: ContainerFilters,
248 #[serde(default, skip_serializing_if = "Vec::is_empty")]
260 pub drop_if_contains: Vec<String>,
261}
262
263pub const CONTAINER_LOG_ROOT: &str = "/var/log/containers";
268
269#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
278#[serde(deny_unknown_fields)]
279pub struct ContainerFilters {
280 #[serde(default, skip_serializing_if = "Vec::is_empty")]
282 pub namespace_in: Vec<String>,
283 #[serde(default, skip_serializing_if = "Vec::is_empty")]
285 pub pod_not_in: Vec<String>,
286}
287
288pub const FILTER_NAMESPACE_IN: &str = "namespace_in";
293pub const FILTER_POD_NOT_IN: &str = "pod_not_in";
295
296pub const CONTAINER_FILTERS: &[&str] = &[FILTER_NAMESPACE_IN, FILTER_POD_NOT_IN];
301
302pub const DROP_IF_CONTAINS_SUBSTRING: &str = "substring";
307pub const DROP_IF_CONTAINS_MODES: &[&str] = &[DROP_IF_CONTAINS_SUBSTRING];
309
310pub const DROP_IF_CONTAINS_MAX_ITEMS: usize = 8;
312pub const DROP_IF_CONTAINS_MAX_BYTES: usize = 200;
314
315pub fn validate_drop_if_contains(needles: &[String]) -> Result<(), String> {
320 if needles.len() > DROP_IF_CONTAINS_MAX_ITEMS {
321 return Err(format!(
322 "drop_if_contains 条数过多({},上限 {DROP_IF_CONTAINS_MAX_ITEMS})",
323 needles.len()
324 ));
325 }
326 for (index, needle) in needles.iter().enumerate() {
327 if needle.trim().is_empty() {
328 return Err(format!("drop_if_contains[{index}] 是空串"));
329 }
330 if needle.trim() != needle {
331 return Err(format!("drop_if_contains[{index}] {needle:?} 带首尾空白"));
332 }
333 if needle.chars().any(char::is_control) {
334 return Err(format!("drop_if_contains[{index}] {needle:?} 含控制字符"));
335 }
336 if needle.len() > DROP_IF_CONTAINS_MAX_BYTES {
337 return Err(format!(
338 "drop_if_contains[{index}] 过长({} 字节,上限 {DROP_IF_CONTAINS_MAX_BYTES})",
339 needle.len()
340 ));
341 }
342 if needles[..index].contains(needle) {
343 return Err(format!("drop_if_contains[{index}] {needle:?} 重复"));
344 }
345 }
346 Ok(())
347}
348
349impl ContainerFilters {
350 pub fn is_empty(&self) -> bool {
352 self.namespace_in.is_empty() && self.pod_not_in.is_empty()
353 }
354
355 pub fn used_dimensions(&self) -> Vec<&'static str> {
360 let mut used = Vec::new();
361 if !self.namespace_in.is_empty() {
362 used.push(FILTER_NAMESPACE_IN);
363 }
364 if !self.pod_not_in.is_empty() {
365 used.push(FILTER_POD_NOT_IN);
366 }
367 used
368 }
369
370 pub fn validate(&self) -> Result<(), String> {
376 for (field, values) in [
377 ("namespace_in", &self.namespace_in),
378 ("pod_not_in", &self.pod_not_in),
379 ] {
380 for value in values {
381 if value.trim().is_empty() {
382 return Err(format!("{field} 里有空值"));
383 }
384 if value.trim() != value {
385 return Err(format!("{field} 里 {value:?} 带首尾空白"));
386 }
387 if value.contains(['*', '?', '[', ']']) {
388 return Err(format!(
389 "{field} 里 {value:?} 含通配符(这里是精确名字,不是 pattern)"
390 ));
391 }
392 if value.len() > 253 {
393 return Err(format!("{field} 里 {value:?} 过长(>253)"));
394 }
395 }
396 }
397 Ok(())
398 }
399}
400
401fn default_multiline() -> String {
402 "none".to_string()
403}
404
405impl WorkSpec {
406 pub fn parse(spec: &str) -> Result<Self, serde_json::Error> {
411 serde_json::from_str(spec)
412 }
413
414 pub fn encode(&self) -> Result<String, serde_json::Error> {
417 serde_json::to_string(self)
418 }
419
420 pub fn metric_interval_seconds(&self) -> Option<i64> {
425 self.units
426 .iter()
427 .flat_map(|unit| unit.sources.iter())
428 .filter(|source| source.kind == "MetricInterval")
429 .filter_map(|source| parse_interval_seconds(&source.target))
430 .min()
431 }
432
433 pub fn collects_logs(&self) -> bool {
435 self.units
436 .iter()
437 .any(|unit| unit.capability == "collect_logs")
438 }
439}
440
441pub const EXECUTABLE_SOURCE_KINDS: &[&str] = &["FileGlob", "MetricInterval", "Exporter"];
447
448pub fn has_glob_meta(target: &str) -> bool {
450 target.contains(['*', '?', '['])
451}
452
453pub fn is_explicit_path(target: &str) -> bool {
463 target.starts_with('/') && !has_glob_meta(target)
464}
465
466pub fn is_glob_path(target: &str) -> bool {
472 target.starts_with('/') && has_glob_meta(target) && !target.contains("**")
473}
474
475pub fn is_file_target(target: &str) -> bool {
480 is_explicit_path(target) || is_glob_path(target)
481}
482
483pub const KERNEL_LOG_EXPORTER: &str = "journalctl-kernel";
492
493pub fn missing_log_capabilities(
511 source: &WorkSpecSource,
512 declared: &LogsCapabilities,
513) -> Option<Vec<String>> {
514 let declared_has = |bucket: &[String], value: &str| bucket.iter().any(|item| item == value);
515 let kind = source.kind.as_str();
516 let target = source.target.as_str();
517
518 match kind {
519 "FileGlob" => {
520 if !is_file_target(target) {
522 return None;
523 }
524 let mut missing = Vec::new();
525 let need = if is_glob_path(target) {
526 FILE_INPUT_GLOB
527 } else {
528 FILE_INPUT_ABSOLUTE_PATH
529 };
530 if !declared_has(&declared.file_inputs, need) {
531 missing.push(format!("file_inputs:{need}"));
532 }
533 if source.multiline != "none"
535 && !declared_has(&declared.multiline_modes, &source.multiline)
536 {
537 missing.push(format!("multiline_modes:{}", source.multiline));
538 }
539 for dimension in source.filters.used_dimensions() {
542 if !declared_has(&declared.filters, dimension) {
543 missing.push(format!("filters:{dimension}"));
544 }
545 }
546 if !source.drop_if_contains.is_empty()
550 && !declared_has(&declared.drop_if_contains, DROP_IF_CONTAINS_SUBSTRING)
551 {
552 missing.push(format!("drop_if_contains:{DROP_IF_CONTAINS_SUBSTRING}"));
553 }
554 Some(missing)
555 }
556 "Exporter" => {
557 let (id, _) = parse_exporter_target(target)?;
558 Some(if declared_has(&declared.exporters, id) {
559 Vec::new()
560 } else {
561 vec![format!("exporters:{id}")]
562 })
563 }
564 _ => None,
565 }
566}
567
568pub const FILE_INPUT_ABSOLUTE_PATH: &str = "absolute-path";
572pub const FILE_INPUT_GLOB: &str = "glob";
574
575pub const MULTILINE_MODES: &[&str] = &["none", "indented", "mysql_slow", "postgres_csv"];
581
582pub const EXPORTER_IDS: &[&str] = &[
590 "journalctl-unit",
591 "journalctl-shutdown",
592 "last-reboot",
593 "nft-ruleset",
594 "iptables-save",
595 "smartctl",
596 "dmesg",
597 "auditd-execve",
598 KERNEL_LOG_EXPORTER,
599];
600
601pub fn parse_exporter_target(target: &str) -> Option<(&str, Option<&str>)> {
606 let (id, arg) = match target.split_once(':') {
607 Some((id, arg)) => (id, Some(arg)),
608 None => (target, None),
609 };
610 let id = id.trim();
611 if id.is_empty() || id.contains(char::is_whitespace) {
612 return None;
613 }
614 Some((id, arg))
615}
616
617pub fn is_known_exporter(target: &str) -> bool {
619 match parse_exporter_target(target) {
620 Some((id, _)) => EXPORTER_IDS.contains(&id),
621 None => false,
622 }
623}
624
625pub fn is_executable_source(kind: &str, target: &str) -> bool {
631 match kind {
632 "FileGlob" => is_file_target(target),
634 "MetricInterval" => true,
636 "Exporter" => is_known_exporter(target),
638 _ => false,
639 }
640}
641
642impl WorkSpecUnit {
643 pub fn file_sources(&self) -> Vec<&WorkSpecSource> {
648 self.sources
649 .iter()
650 .filter(|source| source.kind == "FileGlob")
651 .collect()
652 }
653
654 pub fn unsupported_sources(&self) -> Vec<&WorkSpecSource> {
656 self.sources
657 .iter()
658 .filter(|source| !is_executable_source(&source.kind, &source.target))
659 .collect()
660 }
661
662 pub fn has_executable_source(&self) -> bool {
667 self.sources
668 .iter()
669 .any(|source| is_executable_source(&source.kind, &source.target))
670 }
671}
672
673pub fn parse_interval_seconds(raw: &str) -> Option<i64> {
678 let raw = raw.trim();
679 if let Some(value) = raw.strip_suffix('s') {
680 return value.trim().parse::<i64>().ok().filter(|value| *value > 0);
681 }
682 if let Some(value) = raw.strip_suffix('m') {
683 let minutes = value.trim().parse::<i64>().ok()?;
684 return minutes.checked_mul(60).filter(|value| *value > 0);
688 }
689 None
690}
691
692#[cfg(test)]
693mod tests {
694 use super::*;
695 use crate::capability_report::LogsCapabilities;
696
697 fn one_shot(status: &str, interruptible: bool) -> OneShotWork {
698 OneShotWork {
699 work_id: "work-1".to_string(),
700 agent_id: "agent-1".to_string(),
701 action: "upgrade".to_string(),
702 spec: "0.1.4".to_string(),
703 scheduled_at: "2026-09-23T00:00:00Z".to_string(),
704 deadline_at: "2026-09-24T00:00:00Z".to_string(),
705 timeout_seconds: 600,
706 interruptible,
707 status: status.to_string(),
708 paused_at: None,
709 paused_total_seconds: 0,
710 current_step: None,
711 completed_steps: vec![],
712 attempt: 0,
713 issued_by: "admin".to_string(),
714 issued_at: "2026-09-23T00:00:00Z".to_string(),
715 }
716 }
717
718 #[test]
719 fn terminal_one_shot_work_is_not_outstanding() {
720 for status in ONE_SHOT_TERMINAL_STATUSES {
722 assert!(!one_shot(status, true).is_outstanding(), "{status}");
723 }
724 for status in ["dispatched", "accepted", "running", "paused"] {
725 assert!(one_shot(status, true).is_outstanding(), "{status}");
726 }
727 }
728
729 #[test]
730 fn only_interruptible_running_work_can_pause() {
731 assert!(!one_shot("running", false).can_pause());
733 assert!(!one_shot("dispatched", true).can_pause());
735 assert!(!one_shot("succeeded", true).can_pause());
736 assert!(one_shot("accepted", true).can_pause());
737 assert!(one_shot("running", true).can_pause());
738 }
739
740 #[test]
741 fn work_kind_serializes_as_the_model_names() {
742 assert_eq!(
743 serde_json::to_string(&WorkKind::Standing).expect("serialize"),
744 "\"Standing\""
745 );
746 assert_eq!(
747 serde_json::to_string(&WorkKind::OneShot).expect("serialize"),
748 "\"OneShot\""
749 );
750 }
751
752 #[test]
753 fn only_explicit_absolute_paths_are_collectable_file_targets() {
754 assert!(is_explicit_path("/var/log/install.log"));
757 assert!(!is_explicit_path("/var/log/wifi.log*"));
758 assert!(!is_explicit_path("/Library/Logs/DiagnosticReports/*.ips"));
759 assert!(!is_explicit_path("~/Library/Logs/Homebrew/*"));
760 assert!(!is_explicit_path("relative/app.log"));
761 }
762
763 #[test]
764 fn file_targets_accept_single_level_globs_and_reject_recursion() {
765 assert!(is_glob_path("/var/log/containers/*.log"));
767 assert!(is_glob_path("/var/log/pods/*/*/*.log"));
768 assert!(is_glob_path("/var/lib/postgresql/*/data/log/*.csv"));
769 assert!(is_file_target("/var/log/containers/*.log"));
770 assert!(is_file_target("/var/log/auth.log"));
772 assert!(!is_glob_path("/var/log/**/*.log"));
774 assert!(!is_file_target("/var/log/**/*.log"));
775 assert!(!is_glob_path("~/Library/Logs/*.log"));
777 assert!(!is_file_target("relative/*.log"));
778 assert!(!is_file_target(""));
780 }
781
782 #[test]
783 fn executability_needs_both_a_supported_kind_and_a_supported_target() {
784 assert!(is_executable_source("MetricInterval", "15s"));
786 assert!(is_executable_source("FileGlob", "/var/log/app.log"));
788 assert!(is_executable_source(
789 "FileGlob",
790 "/var/log/containers/*.log"
791 ));
792 assert!(!is_executable_source("FileGlob", "/var/log/**/*.log"));
793 assert!(!is_executable_source("FileGlob", "~/logs/*.log"));
794 assert!(is_executable_source("Exporter", "smartctl"));
796 assert!(is_executable_source("Exporter", "dmesg:panic"));
797 assert!(!is_executable_source("Exporter", "last,lastb"));
798 assert!(!is_executable_source("UnifiedLogPredicate", "syspolicyd"));
800 }
801
802 #[test]
803 fn exporter_targets_split_into_a_known_id_and_an_optional_arg() {
804 assert_eq!(parse_exporter_target("smartctl"), Some(("smartctl", None)));
805 assert_eq!(
806 parse_exporter_target("dmesg:panic"),
807 Some(("dmesg", Some("panic")))
808 );
809 assert_eq!(parse_exporter_target(":panic"), None);
811 assert_eq!(parse_exporter_target(" dm esg"), None);
812 assert!(is_known_exporter("dmesg:nvidia-xid"));
814 assert!(!is_known_exporter("nope"));
815 }
816
817 #[test]
818 fn the_kernel_log_exporter_is_in_the_contract_vocabulary() {
819 assert!(EXPORTER_IDS.contains(&KERNEL_LOG_EXPORTER));
822 assert!(is_known_exporter(KERNEL_LOG_EXPORTER));
823 assert!(is_executable_source("Exporter", KERNEL_LOG_EXPORTER));
824 assert!(!is_known_exporter("journalctl-kernelx"));
826 }
827
828 fn source(kind: &str, target: &str, multiline: &str) -> WorkSpecSource {
832 WorkSpecSource {
833 kind: kind.to_string(),
834 target: target.to_string(),
835 multiline: multiline.to_string(),
836 exclude_globs: Vec::new(),
837 filters: ContainerFilters::default(),
838 drop_if_contains: Vec::new(),
839 }
840 }
841
842 fn container_source(target: &str, filters: ContainerFilters) -> WorkSpecSource {
844 WorkSpecSource {
845 filters,
846 ..source("FileGlob", target, "none")
847 }
848 }
849
850 fn missing(
852 kind: &str,
853 target: &str,
854 multiline: &str,
855 declared: &LogsCapabilities,
856 ) -> Option<Vec<String>> {
857 missing_log_capabilities(&source(kind, target, multiline), declared)
858 }
859
860 fn full_logs() -> LogsCapabilities {
862 LogsCapabilities {
863 file_inputs: vec![
864 FILE_INPUT_ABSOLUTE_PATH.to_string(),
865 FILE_INPUT_GLOB.to_string(),
866 ],
867 multiline_modes: MULTILINE_MODES
868 .iter()
869 .map(|mode| mode.to_string())
870 .collect(),
871 exporters: EXPORTER_IDS.iter().map(|id| id.to_string()).collect(),
872 filters: CONTAINER_FILTERS.iter().map(|di| di.to_string()).collect(),
873 drop_if_contains: DROP_IF_CONTAINS_MODES
874 .iter()
875 .map(|m| m.to_string())
876 .collect(),
877 ..LogsCapabilities::default()
878 }
879 }
880
881 #[test]
882 fn capability_gaps_name_the_bucket_and_value() {
883 let old_agent = LogsCapabilities {
884 file_inputs: vec![FILE_INPUT_ABSOLUTE_PATH.to_string()],
886 ..LogsCapabilities::default()
887 };
888 assert_eq!(
889 missing("FileGlob", "/var/log/app.log", "none", &old_agent),
890 Some(Vec::new())
891 );
892 assert_eq!(
893 missing("FileGlob", "/var/log/app/*.log", "none", &old_agent),
894 Some(vec!["file_inputs:glob".to_string()])
895 );
896 assert_eq!(
898 missing("FileGlob", "/var/log/a.log", "indented", &old_agent),
899 Some(vec!["multiline_modes:indented".to_string()])
900 );
901 assert_eq!(
903 missing("Exporter", KERNEL_LOG_EXPORTER, "none", &old_agent),
904 Some(vec![format!("exporters:{KERNEL_LOG_EXPORTER}")])
905 );
906 assert_eq!(
908 missing("Exporter", "dmesg:panic", "none", &full_logs()),
909 Some(Vec::new())
910 );
911 assert_eq!(
913 missing("FileGlob", "/var/log/containers/*.log", "none", &old_agent),
914 Some(vec!["file_inputs:glob".to_string()])
915 );
916 }
917
918 #[test]
919 fn container_filter_dimensions_are_negotiated_separately() {
920 let before_w2 = LogsCapabilities {
923 filters: Vec::new(),
924 ..full_logs()
925 };
926 let namespace_only = ContainerFilters {
927 namespace_in: vec!["prod".to_string()],
928 pod_not_in: Vec::new(),
929 };
930 let target = "/var/log/containers/*.log";
931 assert_eq!(
932 missing_log_capabilities(&container_source(target, namespace_only), &before_w2),
933 Some(vec!["filters:namespace_in".to_string()])
934 );
935 let both = ContainerFilters {
937 namespace_in: vec!["prod".to_string()],
938 pod_not_in: vec!["debug-pod".to_string()],
939 };
940 assert_eq!(
941 missing_log_capabilities(&container_source(target, both.clone()), &before_w2),
942 Some(vec![
943 "filters:namespace_in".to_string(),
944 "filters:pod_not_in".to_string(),
945 ])
946 );
947 assert_eq!(
949 missing_log_capabilities(&container_source(target, both), &full_logs()),
950 Some(Vec::new())
951 );
952 assert_eq!(
954 ContainerFilters {
955 namespace_in: vec!["a".to_string()],
956 pod_not_in: vec!["b".to_string()],
957 }
958 .used_dimensions(),
959 vec![FILTER_NAMESPACE_IN, FILTER_POD_NOT_IN]
960 );
961 assert!(ContainerFilters::default().used_dimensions().is_empty());
962 }
963
964 #[test]
965 fn noise_drop_needles_are_a_capability_of_their_own() {
966 let before_w4 = LogsCapabilities {
969 drop_if_contains: Vec::new(),
970 ..full_logs()
971 };
972 let mut source = source("FileGlob", "/var/log/auth.log", "none");
973 source.drop_if_contains = vec!["pam_unix(cron:session)".to_string()];
974 assert_eq!(
975 missing_log_capabilities(&source, &before_w4),
976 Some(vec!["drop_if_contains:substring".to_string()])
977 );
978 assert_eq!(
980 missing_log_capabilities(&source, &full_logs()),
981 Some(Vec::new())
982 );
983 }
984
985 #[test]
986 fn drop_if_contains_is_substring_shaped_and_stays_out_of_the_wire_when_empty() {
987 let mut source = source("FileGlob", "/var/log/auth.log", "none");
989 assert!(
990 !serde_json::to_string(&source)
991 .expect("serialize")
992 .contains("drop_if_contains")
993 );
994
995 source.drop_if_contains = vec!["pam_unix(cron:session)".to_string()];
996 let json = serde_json::to_string(&source).expect("serialize");
997 assert!(
998 json.contains(r#""drop_if_contains":["pam_unix(cron:session)"]"#),
999 "{json}"
1000 );
1001 assert_eq!(
1002 serde_json::from_str::<WorkSpecSource>(&json).expect("parse"),
1003 source
1004 );
1005
1006 assert!(validate_drop_if_contains(&["ok".to_string()]).is_ok());
1008 assert!(validate_drop_if_contains(&[]).is_ok());
1009 for bad in ["", " x", "x ", "line\nbreak", " nul\0"] {
1010 assert!(
1011 validate_drop_if_contains(&[bad.to_string()]).is_err(),
1012 "{bad:?} should be rejected"
1013 );
1014 }
1015 assert!(validate_drop_if_contains(&["x".repeat(DROP_IF_CONTAINS_MAX_BYTES + 1)]).is_err());
1016 let too_many: Vec<String> = (0..=DROP_IF_CONTAINS_MAX_ITEMS)
1017 .map(|i| format!("needle-{i}"))
1018 .collect();
1019 assert!(validate_drop_if_contains(&too_many).is_err());
1020 assert!(validate_drop_if_contains(&["dup".to_string(), "dup".to_string()]).is_err());
1021 }
1022
1023 #[test]
1024 fn the_new_multiline_modes_are_in_the_contract_vocabulary() {
1025 for mode in ["mysql_slow", "postgres_csv"] {
1027 assert!(MULTILINE_MODES.contains(&mode), "{mode}");
1028 }
1029 let old_agent = LogsCapabilities {
1031 multiline_modes: vec!["none".to_string(), "indented".to_string()],
1032 ..full_logs()
1033 };
1034 assert_eq!(
1035 missing(
1036 "FileGlob",
1037 "/var/log/postgresql/x.csv",
1038 "postgres_csv",
1039 &old_agent
1040 ),
1041 Some(vec!["multiline_modes:postgres_csv".to_string()])
1042 );
1043 assert_eq!(
1044 missing(
1045 "FileGlob",
1046 "/var/log/mysql/slow.log",
1047 "mysql_slow",
1048 &old_agent
1049 ),
1050 Some(vec!["multiline_modes:mysql_slow".to_string()])
1051 );
1052 }
1053
1054 #[test]
1055 fn an_exporter_id_outside_the_vocabulary_is_reported_as_missing() {
1056 assert_eq!(
1059 missing("Exporter", "journalctl-kernelx", "none", &full_logs()),
1060 Some(vec!["exporters:journalctl-kernelx".to_string()])
1061 );
1062 assert_eq!(
1064 missing("Exporter", "dmesg:panic", "none", &old_style_logs()),
1065 Some(vec!["exporters:dmesg".to_string()])
1066 );
1067 }
1068
1069 fn old_style_logs() -> LogsCapabilities {
1071 LogsCapabilities {
1072 file_inputs: vec![FILE_INPUT_ABSOLUTE_PATH.to_string()],
1073 exporters: vec!["smartctl".to_string()],
1074 ..LogsCapabilities::default()
1075 }
1076 }
1077
1078 #[test]
1079 fn capability_checks_say_nothing_about_things_they_do_not_own() {
1080 assert_eq!(missing("MetricInterval", "15s", "none", &full_logs()), None);
1082 assert_eq!(missing("FileGlob", "~/a/*.log", "none", &full_logs()), None);
1083 assert_eq!(
1084 missing("UnifiedLogPredicate", "syspolicyd", "none", &full_logs()),
1085 None
1086 );
1087 }
1088
1089 #[test]
1090 fn a_fully_declared_agent_is_missing_nothing_the_contract_calls_executable() {
1091 let full = full_logs();
1094 for target in ["/var/log/a.log", "/var/log/containers/*.log"] {
1095 assert_eq!(
1096 missing("FileGlob", target, "none", &full),
1097 Some(Vec::new()),
1098 "{target}"
1099 );
1100 }
1101 for mode in MULTILINE_MODES {
1102 assert_eq!(
1103 missing("FileGlob", "/var/log/a.log", mode, &full),
1104 Some(Vec::new()),
1105 "{mode}"
1106 );
1107 }
1108 for id in EXPORTER_IDS {
1109 assert_eq!(
1110 missing("Exporter", id, "none", &full),
1111 Some(Vec::new()),
1112 "{id}"
1113 );
1114 }
1115 let filters = ContainerFilters {
1117 namespace_in: vec!["prod".to_string()],
1118 pod_not_in: vec!["debug-pod".to_string()],
1119 };
1120 assert_eq!(
1121 filters.used_dimensions(),
1122 CONTAINER_FILTERS,
1123 "`used_dimensions` 与 `CONTAINER_FILTERS` 漂了"
1124 );
1125 assert_eq!(
1126 missing_log_capabilities(
1127 &container_source("/var/log/containers/*.log", filters),
1128 &full
1129 ),
1130 Some(Vec::new())
1131 );
1132 }
1133
1134 fn spec() -> WorkSpec {
1137 WorkSpec {
1138 units: vec![
1139 WorkSpecUnit {
1140 unit_id: "mac-host-metrics".to_string(),
1141 capability: "collect_metrics".to_string(),
1142 rule_ref: "agent_uplink".to_string(),
1143 requires_privilege: "none".to_string(),
1144 sources: vec![WorkSpecSource {
1145 kind: "MetricInterval".to_string(),
1146 target: "15s".to_string(),
1147 multiline: "none".to_string(),
1148 exclude_globs: Vec::new(),
1149 filters: ContainerFilters::default(),
1150 drop_if_contains: Vec::new(),
1151 }],
1152 },
1153 WorkSpecUnit {
1154 unit_id: "mac-privacy-tcc".to_string(),
1155 capability: "collect_logs".to_string(),
1156 rule_ref: "macos/tcc".to_string(),
1157 requires_privilege: "fda".to_string(),
1158 sources: vec![
1159 WorkSpecSource {
1160 kind: "Exporter".to_string(),
1161 target: "sqlite-snapshot(TCC.db)".to_string(),
1162 multiline: "none".to_string(),
1163 exclude_globs: Vec::new(),
1164 filters: ContainerFilters::default(),
1165 drop_if_contains: Vec::new(),
1166 },
1167 WorkSpecSource {
1168 kind: "FileGlob".to_string(),
1169 target: "/var/log/tccd.log".to_string(),
1171 multiline: "indented".to_string(),
1173 exclude_globs: Vec::new(),
1174 filters: ContainerFilters::default(),
1175 drop_if_contains: Vec::new(),
1176 },
1177 ],
1178 },
1179 ],
1180 }
1181 }
1182
1183 #[test]
1184 fn spec_round_trips_and_reports_what_agentd_can_do() {
1185 let encoded = spec().encode().expect("encode");
1186 let decoded = WorkSpec::parse(&encoded).expect("parse");
1187 assert_eq!(decoded, spec());
1188
1189 assert!(decoded.collects_logs());
1190 assert_eq!(decoded.metric_interval_seconds(), Some(15));
1192 let files = decoded.units[1].file_sources();
1194 assert_eq!(files.len(), 1);
1195 assert_eq!(files[0].target, "/var/log/tccd.log");
1196 assert_eq!(files[0].multiline, "indented");
1197 assert_eq!(decoded.units[1].unsupported_sources().len(), 1);
1198 assert_eq!(decoded.units[1].unsupported_sources()[0].kind, "Exporter");
1199 }
1200
1201 #[test]
1202 fn a_glob_file_source_is_collectable_but_recursive_globs_are_not() {
1203 let mut unit = WorkSpecUnit {
1205 unit_id: "linux-container-stdout".to_string(),
1206 capability: "collect_logs".to_string(),
1207 rule_ref: String::new(),
1208 requires_privilege: "root".to_string(),
1209 sources: vec![WorkSpecSource {
1210 kind: "FileGlob".to_string(),
1211 target: "/var/log/containers/*.log".to_string(),
1212 multiline: "none".to_string(),
1213 exclude_globs: vec!["/var/log/containers/*_kube-system_*.log".to_string()],
1214 filters: ContainerFilters::default(),
1215 drop_if_contains: Vec::new(),
1216 }],
1217 };
1218 assert_eq!(unit.file_sources().len(), 1, "它是一条文件来源");
1219 assert!(unit.unsupported_sources().is_empty());
1220 assert!(unit.has_executable_source());
1221
1222 unit.sources[0].target = "/var/log/**/*.log".to_string();
1224 assert_eq!(unit.unsupported_sources().len(), 1);
1225 assert!(!unit.has_executable_source());
1226 }
1227
1228 #[test]
1229 fn exclude_globs_round_trip_and_stay_out_of_the_wire_when_empty() {
1230 let mut source = WorkSpecSource {
1231 kind: "FileGlob".to_string(),
1232 target: "/var/log/containers/*.log".to_string(),
1233 multiline: "none".to_string(),
1234 exclude_globs: Vec::new(),
1235 filters: ContainerFilters::default(),
1236 drop_if_contains: Vec::new(),
1237 };
1238 assert_eq!(
1240 serde_json::to_string(&source).expect("serialize"),
1241 r#"{"kind":"FileGlob","target":"/var/log/containers/*.log","multiline":"none"}"#
1242 );
1243
1244 source.exclude_globs = vec!["/var/log/containers/*_kube-system_*.log".to_string()];
1245 let json = serde_json::to_string(&source).expect("serialize");
1246 assert_eq!(
1247 serde_json::from_str::<WorkSpecSource>(&json).expect("parse"),
1248 source
1249 );
1250 let old = r#"{"kind":"FileGlob","target":"/a/*"}"#;
1252 let parsed: WorkSpecSource = serde_json::from_str(old).expect("parse legacy source");
1253 assert!(parsed.exclude_globs.is_empty());
1254 assert!(parsed.filters.is_empty());
1255 assert_eq!(parsed.multiline, "none");
1256 }
1257
1258 #[test]
1259 fn container_filters_round_trip_and_stay_out_of_the_wire_when_empty() {
1260 let mut source = WorkSpecSource {
1261 kind: "FileGlob".to_string(),
1262 target: "/var/log/containers/*.log".to_string(),
1263 multiline: "none".to_string(),
1264 exclude_globs: Vec::new(),
1265 filters: ContainerFilters::default(),
1266 drop_if_contains: Vec::new(),
1267 };
1268 let json = serde_json::to_string(&source).expect("serialize");
1270 assert!(!json.contains("filters"), "{json}");
1271
1272 source.filters = ContainerFilters {
1273 namespace_in: vec!["prod".to_string()],
1274 pod_not_in: vec!["debug-pod".to_string()],
1275 };
1276 let json = serde_json::to_string(&source).expect("serialize");
1277 assert!(
1278 json.contains(r#""filters":{"namespace_in":["prod"],"pod_not_in":["debug-pod"]}"#),
1279 "{json}"
1280 );
1281 assert_eq!(
1282 serde_json::from_str::<WorkSpecSource>(&json).expect("parse"),
1283 source
1284 );
1285
1286 let half = r#"{"kind":"FileGlob","target":"/var/log/containers/*.log","filters":{"namespace_in":["prod"]}}"#;
1288 let parsed: WorkSpecSource = serde_json::from_str(half).expect("parse half filters");
1289 assert_eq!(parsed.filters.namespace_in, vec!["prod".to_string()]);
1290 assert!(parsed.filters.pod_not_in.is_empty());
1291 }
1292
1293 #[test]
1294 fn an_explicitly_empty_filters_object_is_not_serialized() {
1295 let source = WorkSpecSource {
1298 kind: "FileGlob".to_string(),
1299 target: "/var/log/containers/*.log".to_string(),
1300 multiline: "none".to_string(),
1301 exclude_globs: Vec::new(),
1302 filters: ContainerFilters {
1303 namespace_in: Vec::new(),
1304 pod_not_in: Vec::new(),
1305 },
1306 drop_if_contains: Vec::new(),
1307 };
1308 let json = serde_json::to_string(&source).expect("serialize");
1309 assert!(!json.contains("filters"), "{json}");
1310 }
1311
1312 #[test]
1313 fn container_filters_keep_values_verbatim() {
1314 let filters = ContainerFilters {
1317 namespace_in: vec!["Prod".to_string()],
1318 pod_not_in: vec!["debug-pod".to_string()],
1319 };
1320 let json = serde_json::to_string(&filters).expect("serialize");
1321 let parsed: ContainerFilters = serde_json::from_str(&json).expect("parse");
1322 assert_eq!(parsed.namespace_in, vec!["Prod".to_string()]);
1323 assert_eq!(parsed.pod_not_in, vec!["debug-pod".to_string()]);
1324 }
1325
1326 #[test]
1327 fn container_filters_reject_unknown_keys() {
1328 assert!(serde_json::from_str::<ContainerFilters>(r#"{"namespaces":["prod"]}"#).is_err());
1331 assert!(serde_json::from_str::<ContainerFilters>(r#"{"namespace_in":["prod"]}"#).is_ok());
1332 }
1333
1334 #[test]
1335 fn container_filters_reject_values_that_can_never_match() {
1336 let ok = |namespace: &[&str], pods: &[&str]| ContainerFilters {
1337 namespace_in: namespace.iter().map(|v| v.to_string()).collect(),
1338 pod_not_in: pods.iter().map(|v| v.to_string()).collect(),
1339 };
1340
1341 assert!(
1342 ok(&["prod", "kube-system"], &["debug-pod"])
1343 .validate()
1344 .is_ok()
1345 );
1346 assert!(ok(&[], &[]).is_empty());
1347 assert!(ok(&[""], &[]).validate().is_err());
1349 assert!(ok(&[" prod"], &[]).validate().is_err());
1350 assert!(ok(&["prod*"], &[]).validate().is_err());
1351 assert!(ok(&[], &["debug?pod"]).validate().is_err());
1352 assert!(ok(&[&"x".repeat(254)], &[]).validate().is_err());
1353 }
1354
1355 #[test]
1356 fn a_source_without_multiline_decodes_as_single_line() {
1357 let json = r#"{"units":[{"unit_id":"u","capability":"collect_logs","sources":[{"kind":"FileGlob","target":"/a/*"}]}]}"#;
1360 let spec = WorkSpec::parse(json).expect("parse");
1361 assert_eq!(spec.units[0].sources[0].multiline, "none");
1362 assert_eq!(
1364 spec.encode().expect("encode"),
1365 r#"{"units":[{"unit_id":"u","capability":"collect_logs","rule_ref":"","requires_privilege":"","sources":[{"kind":"FileGlob","target":"/a/*","multiline":"none"}]}]}"#
1366 );
1367 }
1368
1369 #[test]
1370 fn a_broken_spec_is_an_error_not_an_empty_work() {
1371 assert!(WorkSpec::parse("mac-host-metrics").is_err());
1373 assert!(WorkSpec::parse("{").is_err());
1374 assert!(WorkSpec::parse(r#"{"units":[]}"#).unwrap().units.is_empty());
1376 assert_eq!(
1377 WorkSpec::parse(r#"{"units":[]}"#)
1378 .unwrap()
1379 .metric_interval_seconds(),
1380 None
1381 );
1382 }
1383
1384 #[test]
1385 fn interval_parsing_accepts_seconds_and_minutes_only() {
1386 assert_eq!(parse_interval_seconds("15s"), Some(15));
1387 assert_eq!(parse_interval_seconds(" 5s "), Some(5));
1388 assert_eq!(parse_interval_seconds("5m"), Some(300));
1389 assert_eq!(parse_interval_seconds("15"), None);
1391 assert_eq!(parse_interval_seconds("0s"), None);
1392 assert_eq!(parse_interval_seconds(""), None);
1393 }
1394
1395 #[test]
1396 fn interval_parsing_never_overflows_on_huge_values() {
1397 assert_eq!(parse_interval_seconds("922337203685477580m"), None);
1400 assert_eq!(parse_interval_seconds("153722867280912931m"), None);
1401 assert_eq!(parse_interval_seconds("99999999999999999999s"), None);
1403 assert_eq!(parse_interval_seconds("600m"), Some(36_000));
1405 }
1406
1407 #[test]
1408 fn metric_interval_ignores_unparseable_targets_instead_of_treating_them_as_zero() {
1409 let unit = |target: &str| WorkSpecUnit {
1411 unit_id: "u".to_string(),
1412 capability: "collect_metrics".to_string(),
1413 rule_ref: String::new(),
1414 requires_privilege: String::new(),
1415 sources: vec![WorkSpecSource {
1416 kind: "MetricInterval".to_string(),
1417 target: target.to_string(),
1418 multiline: "none".to_string(),
1419 exclude_globs: Vec::new(),
1420 filters: ContainerFilters::default(),
1421 drop_if_contains: Vec::new(),
1422 }],
1423 };
1424 let mixed = WorkSpec {
1425 units: vec![unit("garbage"), unit("30s")],
1426 };
1427 assert_eq!(mixed.metric_interval_seconds(), Some(30));
1428
1429 let all_bad = WorkSpec {
1430 units: vec![unit("nope")],
1431 };
1432 assert_eq!(all_bad.metric_interval_seconds(), None);
1433 }
1434}