rt: &Runtime,

service_name: &'static str,

node_id: uuid::Uuid,

config: TelemetryConfig,

) -> Vec<Box<dyn Send + Sync + 'static>> {

let mut drop_guards = vec![];

if let Some(metrics) = &config.metrics {

drop_guards.extend(init_metrics(rt, service_name, metrics));

}

if let Some(traces) = &config.traces {

drop_guards.extend(init_traces(rt, service_name, traces));

}

drop_guards.extend(init_logs(rt, service_name, node_id, &config));

drop_guards

rt: &Runtime,

service_name: &'static str,

config: &MetricsConfig,

) -> Vec<Box<dyn Send + Sync + 'static>> {

let MetricsConfig {

opentelemetry: Some(config),

..

} = config

else {

return vec![];

};

rt.block_on(async {

let exporter = opentelemetry_otlp::MetricExporter::builder()

.with_tonic()

.with_protocol(opentelemetry_otlp::Protocol::Grpc)

.with_endpoint(&config.otlp_endpoint)

.build()

.expect("initialize oltp metrics exporter");

let reader = opentelemetry_sdk::metrics::PeriodicReader::builder(exporter)

.with_interval(Duration::from_secs_f64(config.push_interval.as_secs_f64()))

.build();

let resource = opentelemetry_sdk::Resource::builder()

.with_attributes([opentelemetry::KeyValue::new("service.name", service_name)])

.build();

let provider = opentelemetry_sdk::metrics::SdkMeterProvider::builder()

.with_reader(reader)

.with_resource(resource)

.build();

opentelemetry::global::set_meter_provider(provider.clone());

vec![Box::new(scopeguard::guard((), move |_| {

provider

.shutdown()

.inspect_err(|err| log::error!("failed to shutdown metrics provider: {err:?}"))

.ok();

})) as _]

})

rt: &Runtime,

service_name: &'static str,

config: &TracesConfig,

) -> Vec<Box<dyn Send + Sync + 'static>> {

let TracesConfig {

opentelemetry: Some(opentelemetry),

..

} = config

else {

return vec![];

};

let resource = opentelemetry_sdk::Resource::builder()

.with_attributes([opentelemetry::KeyValue::new("service.name", service_name)])

.build();

let otlp_reporter = rt.block_on(async move {

fastrace_opentelemetry::OpenTelemetryReporter::new(

opentelemetry_otlp::SpanExporter::builder()

.with_tonic()

.with_endpoint(&opentelemetry.otlp_endpoint)

.with_protocol(opentelemetry_otlp::Protocol::Grpc)

.with_timeout(opentelemetry_otlp::OTEL_EXPORTER_OTLP_TIMEOUT_DEFAULT)

.build()

.expect("initialize oltp trace exporter"),

Cow::Owned(resource),

opentelemetry::InstrumentationScope::builder(service_name).build(),

)

});

fastrace::set_reporter(otlp_reporter, fastrace::collector::Config::default());

vec![Box::new(scopeguard::guard((), |_| {

struct NoopReporter;

impl fastrace::collector::Reporter for NoopReporter {

fn report(&mut self, _batch: Vec<fastrace::prelude::SpanRecord>) {}

}

fastrace::flush();

fastrace::set_reporter(NoopReporter, fastrace::collector::Config::default());

}))]

rt: &Runtime,

service_name: &'static str,

node_id: uuid::Uuid,

config: &TelemetryConfig,

) -> Vec<Box<dyn Send + Sync + 'static>> {

let static_diagnostic = {

let mut static_diagnostic = StaticDiagnostic::default();

static_diagnostic.insert("node_id", node_id.to_string());

static_diagnostic

};

let drop_guards: Vec<Box<dyn Send + Sync + 'static>> = Vec::new();

let mut builder = logforth::starter_log::builder();

// fastrace appender

if let Some(TracesConfig {

capture_log_filter, ..

}) = &config.traces

{

builder = builder.dispatch(|b| {

b.filter(make_rust_log_filter(capture_log_filter))

.append(append::FastraceEvent::default())

});

}

// file appender

if let Some(file) = &config.logs.file {

let mut b = FileBuilder::new(&file.dir, service_name)

.layout(layout::JsonLayout::default())

.rollover_hourly()

.filename_suffix("log");

if let Some(max_files) = file.max_files {

b = b.max_log_files(max_files);

}

let append = b.build().expect("failed to init file appender");

builder = builder.dispatch(|b| {

b.filter(make_rust_log_filter(&file.filter))

.diagnostic(FastraceDiagnostic::default())

.diagnostic(static_diagnostic.clone())

.append(append)

});

}

// stderr appender

if let Ok(filter) = std::env::var("RUST_LOG") {

builder = builder.dispatch(|b| {

b.filter(make_rust_log_filter(&filter))

.diagnostic(FastraceDiagnostic::default())

.diagnostic(static_diagnostic.clone())

.append(append::Stderr::default().with_layout(layout::TextLayout::default()))

});

} else if let Some(stderr) = &config.logs.stderr {

builder = builder.dispatch(|b| {

b.filter(make_rust_log_filter(&stderr.filter))

.diagnostic(FastraceDiagnostic::default())

.diagnostic(static_diagnostic.clone())

.append(append::Stderr::default().with_layout(layout::TextLayout::default()))

});

}

// opentelemetry appender

if let Some(opentelemetry) = &config.logs.opentelemetry {

let filter = make_rust_log_filter(&opentelemetry.filter);

let appender = rt.block_on(async {

let exporter = opentelemetry_otlp::LogExporter::builder()

.with_tonic()

.with_endpoint(&opentelemetry.otlp_endpoint)

.with_protocol(opentelemetry_otlp::Protocol::Grpc)

.build()

.expect("failed to initialize opentelemetry logger");

append::opentelemetry::OpentelemetryLogBuilder::new(service_name, exporter)

.label("service.name", service_name)

.build()

});

builder = builder.dispatch(|b| {

b.filter(filter)

.diagnostic(FastraceDiagnostic::default())

.diagnostic(static_diagnostic.clone())

.append(appender)

});

}

// apply returns err if already set; ignored

let _ = builder.try_apply();

drop_guards

let builder = EnvFilterBuilder::try_from_spec(filter)

.unwrap_or_else(|_| panic!("failed to parse filter: {filter}"));

builder.build()