Skip to content
139 changes: 98 additions & 41 deletions crates/core/src/observability/plugin_component.rs
Original file line number Diff line number Diff line change
Expand Up @@ -63,8 +63,9 @@ use crate::observability::{
use crate::plugin::{
ATIF_RUNTIME_DELIVERY_FAILURE_MARKER, ConfigDiagnostic, ConfigPolicy, DiagnosticLevel,
OTEL_RUNTIME_DELIVERY_FAILURE_MARKER, Plugin, PluginComponentSpec, PluginError,
PluginRegistration, PluginRegistrationContext, Result as PluginResult, UnsupportedBehavior,
apply_global_config_policy, deregister_plugin, register_builtin_plugin,
PluginRegistration, PluginRegistrationCleanupOutcome, PluginRegistrationContext,
Result as PluginResult, UnsupportedBehavior, apply_global_config_policy, deregister_plugin,
register_builtin_plugin,
};
use crate::plugin::{RuntimeDiagnostic, record_active_plugin_runtime_diagnostic};

Expand Down Expand Up @@ -892,41 +893,62 @@ fn register_atif_dispatcher(
);
ctx.register_subscriber("atif", dispatcher)?;
let shutdown_storage = Arc::clone(&storage);
ctx.add_registration(PluginRegistration::new(
ctx.add_registration(PluginRegistration::new_with_outcome(
"observability",
ctx.qualify_name("atif.shutdown"),
Box::new(move || {
let work = {
let mut guard = manager.lock().map_err(|err| {
PluginError::Internal(format!("ATIF dispatcher lock poisoned: {err}"))
})?;
guard.flush_open_agents()
};
for (scope_uuid, name) in work.scope_subscribers {
deregister_atif_shutdown_subscriber(&scope_uuid, &name)?;
}
for export in work.exports {
let write = prepare_atif_shutdown_file(&export, Arc::clone(&manager))
.map_err(observability_registration_error)?;
let agent_uuid = write.agent_uuid;
let targets = {
let guard = manager.lock().map_err(|err| {
let work = match (|| -> PluginResult<_> {
let work = {
let mut guard = manager.lock().map_err(|err| {
PluginError::Internal(format!("ATIF dispatcher lock poisoned: {err}"))
})?;
guard.sink_targets()
guard.flush_open_agents()
};
let results = write_atif(&write, shutdown_storage.as_slice(), &targets);
let mut guard = manager.lock().map_err(|err| {
PluginError::Internal(format!("ATIF dispatcher lock poisoned: {err}"))
})?;
let _ = guard.complete_scope_write(agent_uuid, results);
for (scope_uuid, name) in &work.scope_subscribers {
deregister_atif_shutdown_subscriber(scope_uuid, name)?;
}
Ok(work)
})() {
Ok(work) => work,
Err(error) => return PluginRegistrationCleanupOutcome::NotRemoved(error),
};

let delivery = (|| -> PluginResult<()> {
for export in work.exports {
let write = prepare_atif_shutdown_file(&export, Arc::clone(&manager))
.map_err(observability_registration_error)?;
let agent_uuid = write.agent_uuid;
let targets = {
let guard = manager.lock().map_err(|err| {
PluginError::Internal(format!("ATIF dispatcher lock poisoned: {err}"))
})?;
guard.sink_targets()
};
let results = write_atif(&write, shutdown_storage.as_slice(), &targets);
let mut guard = manager.lock().map_err(|err| {
PluginError::Internal(format!("ATIF dispatcher lock poisoned: {err}"))
})?;
let _ = guard.complete_scope_write(agent_uuid, results);
}
Ok(())
})();
if let Err(error) = delivery {
return PluginRegistrationCleanupOutcome::RemovedWithError(error);
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
let guard = match manager.lock() {
Ok(guard) => guard,
Err(error) => {
return PluginRegistrationCleanupOutcome::RemovedWithError(
PluginError::Internal(format!("ATIF dispatcher lock poisoned: {error}")),
);
}
};
match guard.last_error_result() {
Ok(()) => PluginRegistrationCleanupOutcome::Removed,
Err(error) => PluginRegistrationCleanupOutcome::RemovedWithError(
observability_registration_error(error),
),
}
let guard = manager.lock().map_err(|err| {
PluginError::Internal(format!("ATIF dispatcher lock poisoned: {err}"))
})?;
guard
.last_error_result()
.map_err(observability_registration_error)
}),
));
Ok(())
Expand Down Expand Up @@ -999,10 +1021,20 @@ fn register_opentelemetry(
// Retain the subscribers as long as the registered fan-out callback exists.
// Their tracer providers and exporter runtimes must outlive event delivery.
let delivery_subscribers = subscribers.clone();
ctx.add_registration(PluginRegistration::new(
ctx.add_registration(PluginRegistration::new_with_outcome(
"observability",
ctx.qualify_name("opentelemetry.shutdown"),
Box::new(move || shutdown_opentelemetry_subscribers(&subscribers).map_or(Ok(()), Err)),
Box::new(
move || match shutdown_opentelemetry_subscribers(&subscribers) {
None => PluginRegistrationCleanupOutcome::Removed,
Some(OpenTelemetryShutdownFailure::Delivery(error)) => {
PluginRegistrationCleanupOutcome::RemovedWithError(error)
}
Some(OpenTelemetryShutdownFailure::Other(error)) => {
PluginRegistrationCleanupOutcome::NotRemoved(error)
}
},
),
Comment thread
coderabbitai[bot] marked this conversation as resolved.
));
ctx.register_subscriber(
"opentelemetry",
Expand Down Expand Up @@ -1059,9 +1091,14 @@ fn build_opentelemetry_subscribers(
Ok(subscribers)
}

enum OpenTelemetryShutdownFailure {
Delivery(PluginError),
Other(PluginError),
}

fn shutdown_opentelemetry_subscribers(
subscribers: &[Arc<OpenTelemetrySubscriber>],
) -> Option<PluginError> {
) -> Option<OpenTelemetryShutdownFailure> {
let mut errors = Vec::new();
if let Err(error) = flush_subscribers() {
errors.push(crate::observability::otel::OpenTelemetryError::Core(error));
Expand All @@ -1086,7 +1123,12 @@ fn shutdown_opentelemetry_subscribers(
} else {
format!("OpenTelemetry shutdown failures: {summary}")
};
Some(PluginError::RegistrationFailed(message))
let error = PluginError::RegistrationFailed(message);
Some(if all_delivery_failures {
OpenTelemetryShutdownFailure::Delivery(error)
} else {
OpenTelemetryShutdownFailure::Other(error)
})
}

fn shutdown_opentelemetry_providers(
Expand Down Expand Up @@ -1624,16 +1666,31 @@ fn render_atif_filename(
})?;
let expression = rendered[selector_start..end].to_string();
let (selector, fallback) = parse_atif_metadata_expression(&expression)?;
let value = selector
.split('.')
.fold(metadata, |value, segment| value?.get(segment))
.and_then(Json::as_str)
.or(fallback)
.ok_or_else(|| {
let mut resolved = metadata;
for segment in selector.split('.') {
resolved = match resolved {
Some(Json::Object(object)) => object.get(segment),
None | Some(Json::Null) => break,
Some(_) => {
return Err(format!(
"filename_template placeholder '{{metadata.{selector}}}' traversed a non-object value"
));
}
};
}
Comment thread
coderabbitai[bot] marked this conversation as resolved.
let value = match resolved {
Some(Json::String(value)) => value.as_str(),
None | Some(Json::Null) => fallback.ok_or_else(|| {
format!(
"filename_template placeholder '{{metadata.{selector}}}' must resolve to a string"
)
})?;
})?,
Some(_) => {
return Err(format!(
"filename_template placeholder '{{metadata.{selector}}}' resolved to a non-string value"
));
}
};
if !is_safe_atif_metadata_path(value) {
return Err(format!(
"metadata path '{selector}' must be a path-safe relative fragment"
Expand Down
Loading
Loading