Skip to content

feat(observability)!: add multi-sink ATOF export #416

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
rapids-bot merged 3 commits into from
Jul 14, 2026
Merged

feat(observability)!: add multi-sink ATOF export #416

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ac34d2e
feat!: add multi-sink ATOF export
willkill07 Jul 14, 2026
f2b2a95
fix: align ATOF v2 fixtures with CI
willkill07 Jul 14, 2026
dbb1105
fix(observability): address ATOF sink review feedback
willkill07 Jul 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
160 changes: 101 additions & 59 deletions crates/cli/src/doctor.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -736,29 +736,67 @@ async fn collect_observability(gateway: &GatewayConfig) -> Vec<Check> {
}

async fn collect_observability_component_checks(checks: &mut Vec<Check>, config: &Value) {
for section in ["atof", "atif"] {
if let Some(check) = observability_file_exporter_check(config, section) {
checks.push(check);
}
checks.extend(observability_atof_file_checks(config));
if let Some(check) = observability_file_exporter_check(config, "atif") {
checks.push(check);
}
for section in ["opentelemetry", "openinference"] {
if let Some(check) = observability_http_exporter_check(config, section).await {
checks.push(check);
}
}
if section_enabled(config, "atof") && atof_endpoint_count(config) > 0 {
if section_enabled(config, "atof") && !atof_stream_sinks(config).is_empty() {
if atof_streaming_supported() {
checks.extend(observability_atof_endpoint_checks(config).await);
checks.extend(observability_atof_stream_checks(config).await);
} else {
checks.push(Check {
name: "ATOF endpoint",
name: "ATOF stream sink",
status: Status::Fail,
details: "ATOF streaming endpoints are not available in this binary".into(),
details: "ATOF stream sinks are not available in this binary".into(),
});
}
}
}

fn observability_atof_file_checks(config: &Value) -> Vec<Check> {
if !section_enabled(config, "atof") {
return Vec::new();
}
let sinks = config
.get("atof")
.and_then(|section| section.get("sinks"))
.and_then(Value::as_array)
.into_iter()
.flatten()
.enumerate()
.filter(|(_, sink)| sink.get("type").and_then(Value::as_str) == Some("file"));
let checks = sinks
.map(
|(index, sink)| match sink.get("output_directory").and_then(Value::as_str) {
Some(path) => {
let mut check = check_directory("ATOF file sink", Path::new(path));
check.details = format!("sinks[{index}]: {}", check.details);
check
}
None => Check {
name: "ATOF file sink",
status: Status::Info,
details: format!("sinks[{index}] uses the runtime default output directory"),
},
},
)
.collect::<Vec<_>>();
if checks.is_empty() {
vec![Check {
name: "ATOF file sink",
status: Status::Info,
details: "no file sinks configured".into(),
}]
} else {
checks
}
}

fn observability_file_exporter_check(config: &Value, section: &str) -> Option<Check> {
if !section_enabled(config, section) {
return None;
Expand Down Expand Up @@ -919,40 +957,41 @@ fn section_endpoint(config: &Value, section: &str) -> Option<String> {
.map(str::to_string)
}

fn atof_endpoint_count(config: &Value) -> usize {
fn atof_stream_sinks(config: &Value) -> Vec<(usize, &Value)> {
config
.get("atof")
.and_then(|section| section.get("endpoints"))
.and_then(|section| section.get("sinks"))
.and_then(Value::as_array)
.map_or(0, Vec::len)
.map(|sinks| {
sinks
.iter()
.enumerate()
.filter(|(_, sink)| sink.get("type").and_then(Value::as_str) == Some("stream"))
.collect()
})
.unwrap_or_default()
}

fn atof_streaming_supported() -> bool {
cfg!(feature = "atof-streaming")
}

async fn observability_atof_endpoint_checks(config: &Value) -> Vec<Check> {
let Some(endpoints) = config
.get("atof")
.and_then(|section| section.get("endpoints"))
.and_then(Value::as_array)
else {
return Vec::new();
};
let mut checks = Vec::with_capacity(endpoints.len());
for (index, endpoint) in endpoints.iter().enumerate() {
checks.push(probe_atof_endpoint(index, endpoint).await);
async fn observability_atof_stream_checks(config: &Value) -> Vec<Check> {
let streams = atof_stream_sinks(config);
let mut checks = Vec::with_capacity(streams.len());
for (index, sink) in streams {
checks.push(probe_atof_stream_sink(index, sink).await);
}
checks
}

async fn probe_atof_endpoint(index: usize, endpoint: &Value) -> Check {
let name = "ATOF endpoint";
async fn probe_atof_stream_sink(index: usize, endpoint: &Value) -> Check {
let name = "ATOF stream sink";
let Some(url) = endpoint.get("url").and_then(Value::as_str) else {
return Check {
name,
status: Status::Fail,
details: format!("endpoints[{index}]: missing url"),
details: format!("sinks[{index}]: missing url"),
};
};
let transport = endpoint
Expand All @@ -967,7 +1006,7 @@ async fn probe_atof_endpoint(index: usize, endpoint: &Value) -> Check {
return Check {
name,
status: Status::Fail,
details: format!("endpoints[{index}] {transport} {url}: timeout_millis must be > 0"),
details: format!("sinks[{index}] {transport} {url}: timeout_millis must be > 0"),
};
}
let headers = match endpoint_headers(endpoint) {
Expand All @@ -976,7 +1015,7 @@ async fn probe_atof_endpoint(index: usize, endpoint: &Value) -> Check {
return Check {
name,
status: Status::Fail,
details: format!("endpoints[{index}] {transport} {url}: {err}"),
details: format!("sinks[{index}] {transport} {url}: {err}"),
};
}
};
Expand All @@ -986,7 +1025,7 @@ async fn probe_atof_endpoint(index: usize, endpoint: &Value) -> Check {
return Check {
name,
status: Status::Fail,
details: format!("endpoints[{index}] {transport} {url}: {err}"),
details: format!("sinks[{index}] {transport} {url}: {err}"),
};
}
};
Expand All @@ -998,11 +1037,16 @@ async fn probe_atof_endpoint(index: usize, endpoint: &Value) -> Check {
_ => Check {
name,
status: Status::Fail,
details: format!("endpoints[{index}] {transport} {url}: unsupported transport"),
details: format!("sinks[{index}] {transport} {url}: unsupported transport"),
},
}
}

#[cfg(test)]
async fn probe_atof_endpoint(index: usize, endpoint: &Value) -> Check {
probe_atof_stream_sink(index, endpoint).await
}

fn endpoint_headers(endpoint: &Value) -> Result<Vec<(String, String)>, String> {
let mut out = Vec::new();
let mut names = std::collections::HashSet::new();
Expand Down Expand Up @@ -1096,11 +1140,9 @@ async fn probe_atof_http_upload(
Ok(client) => client,
Err(err) => {
return Check {
name: "ATOF endpoint",
name: "ATOF stream sink",
status: Status::Fail,
details: format!(
"endpoints[{index}] {transport} {url}: could not build client: {err}"
),
details: format!("sinks[{index}] {transport} {url}: could not build client: {err}"),
};
}
};
Expand All @@ -1113,25 +1155,25 @@ async fn probe_atof_http_upload(
}
match request.send().await {
Ok(response) if response.status().is_success() => Check {
name: "ATOF endpoint",
name: "ATOF stream sink",
status: Status::Pass,
details: format!(
"endpoints[{index}] {transport} {url} (HTTP {})",
"sinks[{index}] {transport} {url} (HTTP {})",
response.status()
),
},
Ok(response) => Check {
name: "ATOF endpoint",
name: "ATOF stream sink",
status: Status::Fail,
details: format!(
"endpoints[{index}] {transport} {url} (HTTP {})",
"sinks[{index}] {transport} {url} (HTTP {})",
response.status()
),
},
Err(err) => Check {
name: "ATOF endpoint",
name: "ATOF stream sink",
status: Status::Fail,
details: format!("endpoints[{index}] {transport} {url}: {err}"),
details: format!("sinks[{index}] {transport} {url}: {err}"),
},
}
}
Expand All @@ -1147,28 +1189,28 @@ async fn probe_atof_websocket(
Ok(parsed) if matches!(parsed.scheme(), "ws" | "wss") => {}
Ok(_) => {
return Check {
name: "ATOF endpoint",
name: "ATOF stream sink",
status: Status::Fail,
details: format!(
"endpoints[{index}] websocket {url}: invalid scheme (must be ws or wss)"
"sinks[{index}] websocket {url}: invalid scheme (must be ws or wss)"
),
};
}
Err(err) => {
return Check {
name: "ATOF endpoint",
name: "ATOF stream sink",
status: Status::Fail,
details: format!("endpoints[{index}] websocket {url}: {err}"),
details: format!("sinks[{index}] websocket {url}: {err}"),
};
}
}
let mut request = match url.into_client_request() {
Ok(request) => request,
Err(err) => {
return Check {
name: "ATOF endpoint",
name: "ATOF stream sink",
status: Status::Fail,
details: format!("endpoints[{index}] websocket {url}: {err}"),
details: format!("sinks[{index}] websocket {url}: {err}"),
};
}
};
Expand All @@ -1179,9 +1221,9 @@ async fn probe_atof_websocket(
Ok(name) => name,
Err(err) => {
return Check {
name: "ATOF endpoint",
name: "ATOF stream sink",
status: Status::Fail,
details: format!("endpoints[{index}] websocket {url}: {err}"),
details: format!("sinks[{index}] websocket {url}: {err}"),
};
}
};
Expand All @@ -1190,9 +1232,9 @@ async fn probe_atof_websocket(
Ok(value) => value,
Err(err) => {
return Check {
name: "ATOF endpoint",
name: "ATOF stream sink",
status: Status::Fail,
details: format!("endpoints[{index}] websocket {url}: {err}"),
details: format!("sinks[{index}] websocket {url}: {err}"),
};
}
};
Expand All @@ -1210,33 +1252,33 @@ async fn probe_atof_websocket(
let _ = timeout(timeout_duration, socket.close(None)).await;
match send {
Ok(Ok(())) => Check {
name: "ATOF endpoint",
name: "ATOF stream sink",
status: Status::Pass,
details: format!("endpoints[{index}] websocket {url}"),
details: format!("sinks[{index}] websocket {url}"),
},
Ok(Err(err)) => Check {
name: "ATOF endpoint",
name: "ATOF stream sink",
status: Status::Fail,
details: format!("endpoints[{index}] websocket {url}: {err}"),
details: format!("sinks[{index}] websocket {url}: {err}"),
},
Err(_) => Check {
name: "ATOF endpoint",
name: "ATOF stream sink",
status: Status::Fail,
details: format!(
"endpoints[{index}] websocket {url}: timed out sending probe payload"
"sinks[{index}] websocket {url}: timed out sending probe payload"
),
},
}
}
Ok(Err(err)) => Check {
name: "ATOF endpoint",
name: "ATOF stream sink",
status: Status::Fail,
details: format!("endpoints[{index}] websocket {url}: {err}"),
details: format!("sinks[{index}] websocket {url}: {err}"),
},
Err(_) => Check {
name: "ATOF endpoint",
name: "ATOF stream sink",
status: Status::Fail,
details: format!("endpoints[{index}] websocket {url}: timed out"),
details: format!("sinks[{index}] websocket {url}: timed out"),
},
}
}
Expand Down
32 changes: 21 additions & 11 deletions crates/cli/src/launcher.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -671,17 +671,27 @@ pub(crate) fn exporter_destinations(config: &GatewayConfig) -> Vec<String> {
fn observability_exporter_destinations(config: &ObservabilityConfig) -> Vec<String> {
let mut destinations = Vec::new();
if let Some(section) = config.atof.as_ref().filter(|section| section.enabled) {
let directory = section
.output_directory
.clone()
.unwrap_or_else(current_output_directory);
let path = directory.join(
section
.filename
.clone()
.unwrap_or_else(|| "nemo-relay-events-<timestamp>.jsonl".into()),
);
destinations.push(format!("ATOF {}", path.display()));
for sink in &section.sinks {
match sink {
nemo_relay::observability::plugin_component::AtofSinkSectionConfig::File(file) => {
let directory = file
.output_directory
.clone()
.unwrap_or_else(current_output_directory);
let path = directory.join(
file.filename
.clone()
.unwrap_or_else(|| "nemo-relay-events-<timestamp>.jsonl".into()),
);
destinations.push(format!("ATOF {}", path.display()));
}
nemo_relay::observability::plugin_component::AtofSinkSectionConfig::Stream(
stream,
) => {
destinations.push(format!("ATOF {}", stream.url));
}
}
}
}
if let Some(section) = config.atif.as_ref().filter(|section| section.enabled) {
if section.storage.is_empty() {
Expand Down
5 changes: 4 additions & 1 deletion crates/cli/tests/cli_tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1549,10 +1549,13 @@ kind = "observability"
enabled = true

[components.config]
version = 1
version = 2

[components.config.atof]
enabled = true

[[components.config.atof.sinks]]
type = "file"
output_directory = "logs"
filename = "events.jsonl"
mode = "append"
Expand Down
Loading
Loading