diff --git a/crates/core/src/observability/plugin_component.rs b/crates/core/src/observability/plugin_component.rs index 53820bd71..a32eb317f 100644 --- a/crates/core/src/observability/plugin_component.rs +++ b/crates/core/src/observability/plugin_component.rs @@ -215,6 +215,9 @@ pub struct AtofStreamSinkSectionConfig { /// Field name policy applied before sending events: `preserve` or `replace_dots`. #[serde(default = "default_atof_endpoint_field_name_policy")] pub field_name_policy: String, + /// Optional stable name used by other components to reference this endpoint. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub name: Option, } /// Per-trajectory ATIF exporter config. @@ -503,6 +506,7 @@ crate::editor_config! { header_env => { label: "header_env", kind: StringMap }, timeout_millis => { label: "timeout_millis", kind: Integer }, field_name_policy => { label: "field_name_policy", kind: Enum, values: ["preserve", "replace_dots"] }, + name => { label: "name", kind: String, optional: true }, } } @@ -1902,6 +1906,7 @@ fn validate_atof_values( "ATOF requires at least one configured sink when enabled".to_string(), ); } + let mut stream_sink_names = HashSet::new(); for (index, sink) in section.sinks.iter().enumerate() { match sink { AtofSinkSectionConfig::File(file) => { @@ -1917,6 +1922,30 @@ fn validate_atof_values( } } AtofSinkSectionConfig::Stream(stream) => { + if let Some(name) = stream.name.as_deref() { + let trimmed = name.trim(); + let message = if trimmed.is_empty() { + Some(format!("ATOF sinks[{index}].name must be non-empty")) + } else if name != trimmed { + Some(format!( + "ATOF sinks[{index}].name must not have leading or trailing whitespace" + )) + } else if !stream_sink_names.insert(name) { + Some(format!("ATOF stream sink name {name:?} must be unique")) + } else { + None + }; + if let Some(message) = message { + push_policy_diag( + diagnostics, + policy.unsupported_value, + "observability.unsupported_value", + Some("atof".to_string()), + Some(format!("sinks[{index}].name")), + message, + ); + } + } validate_atof_stream_sink_values(diagnostics, policy, index, stream); } } @@ -2450,11 +2479,11 @@ fn default_atof_mode() -> String { } fn default_atof_endpoint_transport() -> String { - "http_post".to_string() + AtofEndpointTransport::default().as_str().to_string() } fn default_atof_endpoint_field_name_policy() -> String { - "preserve".to_string() + AtofEndpointFieldNamePolicy::default().as_str().to_string() } fn default_agent_name() -> String { diff --git a/crates/core/tests/unit/observability/plugin_component_tests.rs b/crates/core/tests/unit/observability/plugin_component_tests.rs index cd4bebef2..1d6e6e9c0 100644 --- a/crates/core/tests/unit/observability/plugin_component_tests.rs +++ b/crates/core/tests/unit/observability/plugin_component_tests.rs @@ -266,12 +266,13 @@ fn default_config_and_component_conversion_cover_public_shape() { assert!(atof.sinks.is_empty()); let parsed_atof: AtofSectionConfig = serde_json::from_value(json!({ - "sinks": [{"type": "stream", "url": "http://localhost/events"}] + "sinks": [{"type": "stream", "name": "switchyard", "url": "http://localhost/events"}] })) .unwrap(); let AtofSinkSectionConfig::Stream(stream) = &parsed_atof.sinks[0] else { panic!("expected stream sink"); }; + assert_eq!(stream.name.as_deref(), Some("switchyard")); assert_eq!(stream.transport, "http_post"); assert_eq!(stream.field_name_policy, "preserve"); @@ -368,6 +369,7 @@ fn schema_contains_every_supported_observability_option() { "output_directory", "filename", "mode", + "name", "sinks", "type", "url", @@ -639,7 +641,10 @@ fn atof_endpoint_validation_rejects_bad_values() { {"type": "stream", "url": "not a url", "transport": "http_post"}, {"type": "stream", "url": "http://localhost/events", "transport": "http_post", "field_name_policy": "bogus"}, {"type": "stream", "url": "http://localhost/events", "transport": "websocket"}, - {"type": "stream", "url": "http://localhost/events", "headers": {"invalid header": "value", "x-api-key": "value"}, "header_env": {"X-Api-Key": "NEMO_RELAY_TEST_MISSING_ATOF_HEADER_ENV"}} + {"type": "stream", "url": "http://localhost/events", "headers": {"invalid header": "value", "x-api-key": "value"}, "header_env": {"X-Api-Key": "NEMO_RELAY_TEST_MISSING_ATOF_HEADER_ENV"}}, + {"type": "stream", "name": "switchyard", "url": "http://localhost/first"}, + {"type": "stream", "name": "switchyard", "url": "http://localhost/second"}, + {"type": "stream", "name": " ", "url": "http://localhost/blank"} ] } }))); @@ -695,6 +700,18 @@ fn atof_endpoint_validation_rejects_bad_values() { .iter() .any(|diag| { diag.field.as_deref() == Some("sinks[6].headers.invalid header") }) ); + assert!( + report + .diagnostics + .iter() + .any(|diag| { diag.field.as_deref() == Some("sinks[8].name") }) + ); + assert!( + report + .diagnostics + .iter() + .any(|diag| { diag.field.as_deref() == Some("sinks[9].name") }) + ); } #[test] @@ -704,6 +721,7 @@ fn build_atof_sink_config_maps_headers_timeout_and_rejects_transport() { let config = build_atof_sink_config( 2, AtofSinkSectionConfig::Stream(AtofStreamSinkSectionConfig { + name: Some("switchyard".into()), url: "ws://127.0.0.1:47632/events".into(), transport: "websocket".into(), headers: headers.clone(), @@ -739,6 +757,7 @@ fn build_atof_sink_config_maps_headers_timeout_and_rejects_transport() { let error = build_atof_sink_config( 3, AtofSinkSectionConfig::Stream(AtofStreamSinkSectionConfig { + name: None, url: "http://127.0.0.1:47632/events".into(), transport: "smtp".into(), headers: std::collections::HashMap::new(), @@ -753,6 +772,7 @@ fn build_atof_sink_config_maps_headers_timeout_and_rejects_transport() { let error = build_atof_sink_config( 4, AtofSinkSectionConfig::Stream(AtofStreamSinkSectionConfig { + name: None, url: "http://127.0.0.1:47632/events".into(), transport: "http_post".into(), headers: std::collections::HashMap::new(), diff --git a/crates/node/observability.d.ts b/crates/node/observability.d.ts index d083ae7ed..0d93f8fa1 100644 --- a/crates/node/observability.d.ts +++ b/crates/node/observability.d.ts @@ -28,6 +28,7 @@ export interface AtofStreamSinkConfig { header_env?: Record; timeout_millis?: number; field_name_policy?: 'preserve' | 'replace_dots' | string; + name?: string; } /** @deprecated Use AtofStreamSinkConfig. */ diff --git a/crates/node/tests/observability_plugin_tests.mjs b/crates/node/tests/observability_plugin_tests.mjs index dd4c63d71..e5300b052 100644 --- a/crates/node/tests/observability_plugin_tests.mjs +++ b/crates/node/tests/observability_plugin_tests.mjs @@ -56,7 +56,10 @@ describe('observability plugin helpers', () => { }), ], }); - assert.deepEqual(report.diagnostics.map((diagnostic) => diagnostic.field).sort(), ['filename_template', 'sinks[0].mode']); + assert.deepEqual(report.diagnostics.map((diagnostic) => diagnostic.field).sort(), [ + 'filename_template', + 'sinks[0].mode', + ]); }); it('serializes ATOF stream sinks', () => { @@ -64,6 +67,7 @@ describe('observability plugin helpers', () => { sinks: [ { type: 'stream', + name: 'switchyard', url: 'http://localhost:8080/events', transport: 'http_post', headers: { 'X-Test': 'yes' }, @@ -76,6 +80,7 @@ describe('observability plugin helpers', () => { assert.deepEqual(config.sinks, [ { type: 'stream', + name: 'switchyard', url: 'http://localhost:8080/events', transport: 'http_post', headers: { 'X-Test': 'yes' }, diff --git a/crates/switchyard/src/component.rs b/crates/switchyard/src/component.rs index ab48e258b..56cdc8cfa 100644 --- a/crates/switchyard/src/component.rs +++ b/crates/switchyard/src/component.rs @@ -21,6 +21,7 @@ use nemo_relay::codec::optimization::{ LlmOptimizationModelTransition, }; use nemo_relay::error::{FlowError, Result as FlowResult}; +use nemo_relay::observability::atof::{AtofEndpointFieldNamePolicy, AtofEndpointTransport}; use nemo_relay::plugin::{ ConfigDiagnostic, DiagnosticLevel, Plugin, PluginComponentSpec, PluginConfig, PluginError, PluginRegistrationContext, Result as PluginResult, deregister_plugin, register_plugin, @@ -217,9 +218,9 @@ pub struct SwitchyardConfig { pub targets: BTreeMap, /// Trusted per-protocol fallbacks. pub default_targets: ProtocolDefaults, - /// Optional explicit ATOF endpoint used by CLI cross-validation. + /// Named observability ATOF endpoint used by history-backed profiles. #[serde(default)] - pub atof_endpoint_url: Option, + pub atof_endpoint_name: Option, } impl Default for SwitchyardConfig { @@ -244,7 +245,7 @@ impl Default for SwitchyardConfig { openai_responses: String::new(), anthropic_messages: String::new(), }, - atof_endpoint_url: None, + atof_endpoint_name: None, } } } @@ -269,7 +270,7 @@ nemo_relay::editor_config! { enabled_inbound_profiles => { label: "Enabled inbound profiles", kind: Json }, targets => { label: "Backend target bindings", kind: Json }, default_targets => { label: "Trusted protocol defaults", kind: Json }, - atof_endpoint_url => { label: "ATOF endpoint URL", kind: String, optional: true } + atof_endpoint_name => { label: "ATOF endpoint name", kind: String, optional: true } } } @@ -401,67 +402,74 @@ pub fn validate_switchyard_atof_configuration(config: &PluginConfig) -> Result<( if switchyard.context_mode != ContextMode::AtofRequired { return Ok(()); } - let required_url = match &switchyard.atof_endpoint_url { - Some(url) => url.clone(), - None => derived_atof_url(&switchyard.decision_api_url)?, - }; + let required_name = validate_atof_endpoint_name(switchyard.atof_endpoint_name.as_deref())? + .ok_or_else(|| { + "atof_required Switchyard profiles require atof_endpoint_name".to_string() + })?; let observability = config .components .iter() .find(|component| component.enabled && component.kind == "observability") .ok_or_else(|| "atof_required Switchyard profiles require observability".to_string())?; - let endpoints = observability + let sinks = observability .config .get("atof") .filter(|atof| atof.get("enabled").and_then(Json::as_bool) == Some(true)) - .and_then(|atof| atof.get("endpoints")) + .and_then(|atof| atof.get("sinks")) .and_then(Json::as_array) .ok_or_else(|| { "atof_required Switchyard profiles require an enabled ATOF endpoint".to_string() })?; - let endpoint = endpoints + let matching_sinks = sinks .iter() - .find(|endpoint| { - endpoint.get("url").and_then(Json::as_str) == Some(required_url.as_str()) - && endpoint - .get("transport") - .and_then(Json::as_str) - .unwrap_or("http_post") - == "http_post" + .filter(|sink| { + sink.get("type").and_then(Json::as_str) == Some("stream") + && sink.get("name").and_then(Json::as_str) == Some(required_name) }) - .ok_or_else(|| { - format!("atof_required Switchyard profile requires HTTP ATOF endpoint {required_url}") - })?; - if endpoint - .get("field_name_policy") - .and_then(Json::as_str) - .unwrap_or("preserve") - != "preserve" - { - return Err("Switchyard ATOF endpoint must use field_name_policy = preserve".into()); + .collect::>(); + let endpoint = match matching_sinks.as_slice() { + [sink] => *sink, + [] => { + return Err(format!( + "atof_required Switchyard profile requires named ATOF endpoint {required_name:?}" + )); + } + _ => { + return Err(format!( + "ATOF endpoint name {required_name:?} must resolve to exactly one endpoint" + )); + } + }; + let transport = endpoint.get("transport").map_or_else( + || Some(AtofEndpointTransport::default()), + |value| value.as_str().and_then(AtofEndpointTransport::parse), + ); + if transport != Some(AtofEndpointTransport::HttpPost) { + return Err(format!( + "Switchyard ATOF endpoint {required_name:?} must use transport = http_post" + )); + } + let field_name_policy = endpoint.get("field_name_policy").map_or_else( + || Some(AtofEndpointFieldNamePolicy::default()), + |value| value.as_str().and_then(AtofEndpointFieldNamePolicy::parse), + ); + if field_name_policy != Some(AtofEndpointFieldNamePolicy::Preserve) { + return Err(format!( + "Switchyard ATOF endpoint {required_name:?} must use field_name_policy = preserve" + )); } if endpoint .get("header_env") .and_then(Json::as_object) .is_none_or(Map::is_empty) { - return Err( - "Switchyard ATOF endpoint authentication must use at least one environment-referenced header" - .into(), - ); + return Err(format!( + "Switchyard ATOF endpoint {required_name:?} authentication must use at least one environment-referenced header" + )); } Ok(()) } -fn derived_atof_url(decision_api_url: &str) -> Result { - let mut url = reqwest::Url::parse(decision_api_url) - .map_err(|error| format!("decision_api_url is invalid: {error}"))?; - url.set_path("/v1/atof/events"); - url.set_query(None); - url.set_fragment(None); - Ok(url.to_string().trim_end_matches('/').to_string()) -} - fn parse_config(config: &Map) -> Result { serde_json::from_value(Json::Object(config.clone())) .map_err(|error| format!("invalid Switchyard plugin config: {error}")) @@ -1218,6 +1226,18 @@ impl SwitchyardRuntime { } } +fn validate_atof_endpoint_name(name: Option<&str>) -> Result, String> { + if let Some(name) = name { + if name.trim().is_empty() { + return Err("atof_endpoint_name must be non-empty when configured".into()); + } + if name != name.trim() { + return Err("atof_endpoint_name must not have leading or trailing whitespace".into()); + } + } + Ok(name) +} + fn validate_config(config: &SwitchyardConfig) -> Result<(), String> { if config.version != 1 { return Err(format!( @@ -1237,6 +1257,10 @@ fn validate_config(config: &SwitchyardConfig) -> Result<(), String> { if config.recent_message_count == 0 { return Err("recent_message_count must be greater than zero".into()); } + let atof_endpoint_name = validate_atof_endpoint_name(config.atof_endpoint_name.as_deref())?; + if config.context_mode == ContextMode::AtofRequired && atof_endpoint_name.is_none() { + return Err("atof_required Switchyard profiles require atof_endpoint_name".into()); + } let url = reqwest::Url::parse(&config.decision_api_url) .map_err(|error| format!("decision_api_url is invalid: {error}"))?; if !matches!(url.scheme(), "http" | "https") { diff --git a/crates/switchyard/tests/unit/component_tests.rs b/crates/switchyard/tests/unit/component_tests.rs index a6e7d5e92..de0267073 100644 --- a/crates/switchyard/tests/unit/component_tests.rs +++ b/crates/switchyard/tests/unit/component_tests.rs @@ -254,6 +254,12 @@ fn configuration_validation_rejects_unsafe_or_ambiguous_bindings() { let mut candidate = config(url.clone()); candidate.recent_message_count = 0; assert_invalid(candidate, "recent_message_count"); + let mut candidate = config(url.clone()); + candidate.context_mode = ContextMode::AtofRequired; + assert_invalid(candidate, "atof_endpoint_name"); + let mut candidate = config(url.clone()); + candidate.atof_endpoint_name = Some(" switchyard ".into()); + assert_invalid(candidate, "leading or trailing whitespace"); let candidate = config("file:///tmp/decision".into()); assert_invalid(candidate, "must use http or https"); let mut candidate = config(url.clone()); @@ -328,6 +334,30 @@ fn disabled_protocol_defaults_are_optional() { SwitchyardRuntime::new(candidate).expect("Anthropic-only configuration should validate"); } +#[test] +fn atof_cross_component_validation_rejects_invalid_endpoint_names_before_lookup() { + for (name, expected) in [ + (" ", "atof_endpoint_name must be non-empty when configured"), + ( + " switchyard ", + "atof_endpoint_name must not have leading or trailing whitespace", + ), + ] { + let mut switchyard = config("http://switchyard.test/v1/routing/decision".into()); + switchyard.context_mode = ContextMode::AtofRequired; + switchyard.atof_endpoint_name = Some(name.into()); + let plugin_config = PluginConfig { + components: vec![switchyard.into()], + ..PluginConfig::default() + }; + + assert_eq!( + validate_switchyard_atof_configuration(&plugin_config).unwrap_err(), + expected + ); + } +} + #[test] fn atof_cross_component_validation_reports_each_activation_mismatch() { assert!(validate_switchyard_atof_configuration(&PluginConfig::default()).is_ok()); @@ -340,7 +370,7 @@ fn atof_cross_component_validation_reports_each_activation_mismatch() { let mut switchyard = config("http://switchyard.test/v1/routing/decision".into()); switchyard.context_mode = ContextMode::AtofRequired; - switchyard.atof_endpoint_url = Some("http://events.test/v1/atof/events".into()); + switchyard.atof_endpoint_name = Some("switchyard".into()); let mut plugin_config = PluginConfig { components: vec![switchyard.into()], ..PluginConfig::default() @@ -348,8 +378,10 @@ fn atof_cross_component_validation_reports_each_activation_mismatch() { plugin_config.components.push(PluginComponentSpec { kind: "observability".into(), enabled: true, - config: json!({"atof": {"enabled": true, "endpoints": [{ - "url": "http://wrong.test/v1/atof/events", + config: json!({"atof": {"enabled": true, "sinks": [{ + "type": "stream", + "name": "other", + "url": "http://events.test/v1/atof/events", "header_env": {"authorization": "TOKEN"} }]}}) .as_object() @@ -359,19 +391,40 @@ fn atof_cross_component_validation_reports_each_activation_mismatch() { assert!( validate_switchyard_atof_configuration(&plugin_config) .unwrap_err() - .contains("requires HTTP ATOF endpoint") + .contains("requires named ATOF endpoint") ); - plugin_config.components[1].config["atof"]["endpoints"][0]["url"] = - json!("http://events.test/v1/atof/events"); - plugin_config.components[1].config["atof"]["endpoints"][0]["field_name_policy"] = + plugin_config.components[1].config["atof"]["sinks"][0]["name"] = json!("switchyard"); + assert!(validate_switchyard_atof_configuration(&plugin_config).is_ok()); + plugin_config.components[1].config["atof"]["sinks"][0]["transport"] = json!("websocket"); + assert!( + validate_switchyard_atof_configuration(&plugin_config) + .unwrap_err() + .contains("transport = http_post") + ); + plugin_config.components[1].config["atof"]["sinks"][0]["transport"] = json!("http_post"); + let duplicate = plugin_config.components[1].config["atof"]["sinks"][0].clone(); + plugin_config.components[1].config["atof"]["sinks"] + .as_array_mut() + .unwrap() + .push(duplicate); + assert!( + validate_switchyard_atof_configuration(&plugin_config) + .unwrap_err() + .contains("exactly one endpoint") + ); + plugin_config.components[1].config["atof"]["sinks"] + .as_array_mut() + .unwrap() + .pop(); + plugin_config.components[1].config["atof"]["sinks"][0]["field_name_policy"] = json!("snake_case"); assert!( validate_switchyard_atof_configuration(&plugin_config) .unwrap_err() .contains("field_name_policy = preserve") ); - let endpoint = &mut plugin_config.components[1].config["atof"]["endpoints"][0]; + let endpoint = &mut plugin_config.components[1].config["atof"]["sinks"][0]; endpoint["field_name_policy"] = json!("preserve"); endpoint.as_object_mut().unwrap().remove("header_env"); assert!( @@ -677,6 +730,7 @@ fn identity_policy_requires_stable_request_scope_only_for_atof_profiles() { assert_eq!(synthetic.identity.quality, "synthetic"); config.context_mode = ContextMode::AtofRequired; + config.atof_endpoint_name = Some("switchyard".into()); let atof_runtime = SwitchyardRuntime::new(config).unwrap(); assert!( atof_runtime @@ -808,6 +862,7 @@ fn routing_decision_mark_has_canonical_shape_and_mirrored_identity() { fn atof_required_cross_component_validation_is_context_sensitive() { let mut switchyard = config("http://switchyard.test:8080/v1/routing/decision".into()); switchyard.context_mode = ContextMode::AtofRequired; + switchyard.atof_endpoint_name = Some("switchyard".into()); let mut plugin_config = PluginConfig { components: vec![switchyard.into()], ..PluginConfig::default() @@ -818,7 +873,9 @@ fn atof_required_cross_component_validation_is_context_sensitive() { enabled: true, config: json!({"atof": { "enabled": true, - "endpoints": [{ + "sinks": [{ + "type": "stream", + "name": "switchyard", "url": "http://switchyard.test:8080/v1/atof/events", "transport": "http_post", "field_name_policy": "preserve", diff --git a/examples/switchyard/hermes-ollama-plugins.toml b/examples/switchyard/hermes-ollama-plugins.toml index 6a7d128b0..f562e2c72 100644 --- a/examples/switchyard/hermes-ollama-plugins.toml +++ b/examples/switchyard/hermes-ollama-plugins.toml @@ -13,6 +13,7 @@ decision_api_url = "http://127.0.0.1:4000/v1/routing/decision" decision_profile_id = "hermes-ollama-stage-router" request_materialization = "full_body" context_mode = "atof_required" +atof_endpoint_name = "switchyard" decision_timeout_millis = 120000 max_retries = 3 recent_message_count = 8 @@ -55,17 +56,22 @@ enabled = true [components.config.atof] enabled = true + +[[components.config.atof.sinks]] +type = "file" mode = "append" output_directory = "." filename = "trajectory.atof.jsonl" -[[components.config.atof.endpoints]] +[[components.config.atof.sinks]] +type = "stream" +name = "switchyard" url = "http://127.0.0.1:4000/v1/atof/events" transport = "http_post" field_name_policy = "preserve" timeout_millis = 1000 -[components.config.atof.endpoints.header_env] +[components.config.atof.sinks.header_env] authorization = "SWITCHYARD_AUTHORIZATION" [components.config.atif] diff --git a/examples/switchyard/plugins.toml b/examples/switchyard/plugins.toml index f99d9e4c3..d6a1a69ed 100644 --- a/examples/switchyard/plugins.toml +++ b/examples/switchyard/plugins.toml @@ -14,6 +14,7 @@ decision_api_url = "http://127.0.0.1:4000/v1/routing/decision" decision_profile_id = "smart-stage-router" request_materialization = "recent_message_window" context_mode = "atof_required" +atof_endpoint_name = "switchyard" decision_timeout_millis = 25 max_retries = 3 recent_message_count = 8 @@ -57,13 +58,18 @@ enabled = true [components.config.atof] enabled = true + +[[components.config.atof.sinks]] +type = "file" mode = "append" -[[components.config.atof.endpoints]] +[[components.config.atof.sinks]] +type = "stream" +name = "switchyard" url = "http://127.0.0.1:4000/v1/atof/events" transport = "http_post" field_name_policy = "preserve" timeout_millis = 3000 -[components.config.atof.endpoints.header_env] +[components.config.atof.sinks.header_env] authorization = "SWITCHYARD_AUTHORIZATION" diff --git a/examples/switchyard/real-e2e-plugins.toml b/examples/switchyard/real-e2e-plugins.toml index 9c6351a6d..541beb057 100644 --- a/examples/switchyard/real-e2e-plugins.toml +++ b/examples/switchyard/real-e2e-plugins.toml @@ -13,6 +13,7 @@ decision_api_url = "http://127.0.0.1:4000/v1/routing/decision" decision_profile_id = "remote-stage-router" request_materialization = "summary_only" context_mode = "atof_required" +atof_endpoint_name = "switchyard" decision_timeout_millis = 1000 max_retries = 3 recent_message_count = 8 @@ -55,13 +56,18 @@ enabled = true [components.config.atof] enabled = true + +[[components.config.atof.sinks]] +type = "file" mode = "append" -[[components.config.atof.endpoints]] +[[components.config.atof.sinks]] +type = "stream" +name = "switchyard" url = "http://127.0.0.1:4000/v1/atof/events" transport = "http_post" field_name_policy = "preserve" timeout_millis = 1000 -[components.config.atof.endpoints.header_env] +[components.config.atof.sinks.header_env] authorization = "SWITCHYARD_AUTHORIZATION" diff --git a/go/nemo_relay/observability_plugin.go b/go/nemo_relay/observability_plugin.go index b25baee77..2885eaad3 100644 --- a/go/nemo_relay/observability_plugin.go +++ b/go/nemo_relay/observability_plugin.go @@ -67,6 +67,7 @@ type ObservabilityAtofStreamSinkConfig struct { HeaderEnv map[string]string `json:"header_env,omitempty"` TimeoutMillis uint64 `json:"timeout_millis,omitempty"` FieldNamePolicy string `json:"field_name_policy,omitempty"` + Name string `json:"name,omitempty"` } func (ObservabilityAtofStreamSinkConfig) atofSinkConfig() {} @@ -80,14 +81,9 @@ func (config ObservabilityAtofStreamSinkConfig) MarshalJSON() ([]byte, error) { }{Type: "stream", alias: alias(config)}) } -// ObservabilityAtofEndpoint configures one streaming destination for raw ATOF events. -type ObservabilityAtofEndpoint struct { - URL string `json:"url"` - Transport string `json:"transport,omitempty"` - Headers map[string]string `json:"headers,omitempty"` - TimeoutMillis uint64 `json:"timeout_millis,omitempty"` - FieldNamePolicy string `json:"field_name_policy,omitempty"` -} +// ObservabilityAtofEndpoint is the deprecated name for an ATOF stream sink. +// Deprecated: Use ObservabilityAtofStreamSinkConfig. +type ObservabilityAtofEndpoint = ObservabilityAtofStreamSinkConfig // ObservabilityAtifConfig configures per-top-level-agent ATIF file export. type ObservabilityAtifConfig struct { diff --git a/go/nemo_relay/observability_plugin_test.go b/go/nemo_relay/observability_plugin_test.go index 664b93cca..c30ac9fa2 100644 --- a/go/nemo_relay/observability_plugin_test.go +++ b/go/nemo_relay/observability_plugin_test.go @@ -35,7 +35,8 @@ func TestObservabilityConfigHelpers(t *testing.T) { if atof.Enabled || len(atof.Sinks) != 0 { t.Fatalf("unexpected ATOF defaults: %#v", atof) } - atof.Sinks = []ObservabilityAtofSinkConfigurer{ObservabilityAtofStreamSinkConfig{ + atof.Sinks = []ObservabilityAtofSinkConfigurer{ObservabilityAtofEndpoint{ + Name: "archive", URL: "http://localhost:8080/events", Transport: "http_post", Headers: map[string]string{"X-Test": "yes"}, @@ -85,17 +86,17 @@ func TestObservabilityConfigHelpers(t *testing.T) { if !ok { t.Fatalf("expected serialized ATOF sinks, got %#v", atofConfig) } - firstEndpoint, ok := sinks[0].(map[string]any) - if !ok || firstEndpoint["field_name_policy"] != "replace_dots" || - firstEndpoint["header_env"].(map[string]any)["authorization"] != "NEMO_RELAY_ATOF_AUTH" { + firstSink, ok := sinks[0].(map[string]any) + if !ok || firstSink["name"] != "archive" || firstSink["field_name_policy"] != "replace_dots" || + firstSink["header_env"].(map[string]any)["authorization"] != "NEMO_RELAY_ATOF_AUTH" { t.Fatalf("expected serialized ATOF stream sink settings, got %#v", sinks) } serialized, err := json.Marshal(wrapped) if err != nil { t.Fatalf("marshal observability component failed: %v", err) } - if !strings.Contains(string(serialized), `"field_name_policy":"replace_dots"`) { - t.Fatalf("expected field_name_policy in serialized component, got %s", serialized) + if !strings.Contains(string(serialized), `"name":"archive"`) || !strings.Contains(string(serialized), `"field_name_policy":"replace_dots"`) { + t.Fatalf("expected named ATOF endpoint in serialized component, got %s", serialized) } assertWrappedAtifStorageConfig(t, wrapped.Config["atif"].(map[string]any)) if wrapped.Config["opentelemetry"].(map[string]any)["mark_projection"] != "tool" { diff --git a/python/nemo_relay/observability.py b/python/nemo_relay/observability.py index 1495ba9f7..1fc60eae4 100644 --- a/python/nemo_relay/observability.py +++ b/python/nemo_relay/observability.py @@ -64,12 +64,14 @@ class AtofStreamSinkConfig: header_env: dict[str, str] = field(default_factory=dict) timeout_millis: int = 3000 field_name_policy: Literal["preserve", "replace_dots"] = "preserve" + name: str | None = None def to_dict(self) -> JsonObject: """Serialize this ATOF stream sink to the canonical JSON object shape.""" return _normalize_object( { "type": "stream", + "name": self.name, "url": self.url, "transport": self.transport, "headers": self.headers, @@ -85,7 +87,7 @@ class AtofConfig: """Multi-sink raw ATOF export settings.""" enabled: bool = False - sinks: list["AtofFileSinkConfig | AtofStreamSinkConfig"] | None = None + sinks: list[AtofFileSinkConfig | AtofStreamSinkConfig] | None = None def to_dict(self) -> JsonObject: """Serialize this ATOF config to the canonical JSON object shape.""" diff --git a/python/nemo_relay/observability.pyi b/python/nemo_relay/observability.pyi index bdeaf236b..a1c421632 100644 --- a/python/nemo_relay/observability.pyi +++ b/python/nemo_relay/observability.pyi @@ -27,6 +27,7 @@ class AtofStreamSinkConfig: header_env: dict[str, str] = field(default_factory=dict) timeout_millis: int = ... field_name_policy: Literal["preserve", "replace_dots"] = ... + name: str | None = ... def to_dict(self) -> JsonObject: ... @dataclass(slots=True) diff --git a/python/tests/test_observability_plugin.py b/python/tests/test_observability_plugin.py index 7839f16e3..77a527c49 100644 --- a/python/tests/test_observability_plugin.py +++ b/python/tests/test_observability_plugin.py @@ -15,6 +15,7 @@ OBSERVABILITY_PLUGIN_KIND, AtifConfig, AtofConfig, + AtofEndpointConfig, AtofFileSinkConfig, AtofStreamSinkConfig, ComponentSpec, @@ -104,6 +105,7 @@ def test_s3_storage_config_serializes_credential_fields(self): def test_atof_sink_config_serializes_streaming_fields(self): sink = AtofStreamSinkConfig( url="http://localhost:8080/events", + name="switchyard", transport="http_post", headers={"X-Test": "yes"}, header_env={"authorization": "NEMO_RELAY_ATOF_AUTH"}, @@ -112,6 +114,7 @@ def test_atof_sink_config_serializes_streaming_fields(self): ) assert sink.to_dict() == { "type": "stream", + "name": "switchyard", "url": "http://localhost:8080/events", "transport": "http_post", "headers": {"X-Test": "yes"}, @@ -120,6 +123,12 @@ def test_atof_sink_config_serializes_streaming_fields(self): "field_name_policy": "replace_dots", } assert AtofConfig(sinks=[sink]).to_dict()["sinks"] == [sink.to_dict()] + assert "name" not in AtofStreamSinkConfig(url="http://localhost:8080/events").to_dict() + + def test_atof_endpoint_alias_preserves_positional_transport(self): + endpoint = AtofEndpointConfig("http://localhost:8080/events", "websocket") + assert endpoint.transport == "websocket" + assert endpoint.name is None def test_http_storage_config_serializes_headers(self): s3 = S3StorageConfig(bucket="archive")