worterbuch 1.5.0-alpha.4

A message broker / database hybrid.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

/*
 *  Copyright (C) 2025 Michael Bachmann
 *
 *  This program is free software: you can redistribute it and/or modify
 *  it under the terms of the GNU Affero General Public License as published by
 *  the Free Software Foundation, either version 3 of the License, or
 *  (at your option) any later version.
 *
 *  This program is distributed in the hope that it will be useful,
 *  but WITHOUT ANY WARRANTY; without even the implied warranty of
 *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *  GNU Affero General Public License for more details.
 *
 *  You should have received a copy of the GNU Affero General Public License
 *  along with this program.  If not, see <https://www.gnu.org/licenses/>.
 */

use opentelemetry::{KeyValue, global, trace::TracerProvider};
use opentelemetry_otlp::WithExportConfig;
use opentelemetry_resource_detectors::{
    HostResourceDetector, OsResourceDetector, ProcessResourceDetector,
};
use opentelemetry_sdk::{Resource, propagation::TraceContextPropagator, trace::SdkTracerProvider};
use std::{env, io};
use supports_color::Stream;
use tracing::{info, level_filters::LevelFilter};
use tracing_subscriber::{EnvFilter, Layer, fmt, layer::SubscriberExt, util::SubscriberInitExt};
use worterbuch_common::error::ConfigResult;

pub async fn init(node_id: String, cluster_role: Option<String>) -> ConfigResult<()> {
    let subscriber = tracing_subscriber::registry().with(
        fmt::Layer::new()
            .with_ansi(supports_color::on(Stream::Stderr).is_some())
            .with_writer(io::stderr)
            .with_filter(
                EnvFilter::builder()
                    .with_default_directive(LevelFilter::INFO.into())
                    .with_env_var("WORTERBUCH_LOG")
                    .from_env_lossy(),
            ),
    );

    let endpoint = env::var("WORTERBUCH_OPENTELEMETRY_ENDPOINT").ok();

    if let Some(endpoint) = endpoint {
        global::set_text_map_propagator(TraceContextPropagator::new());

        let exporter = {
            let builder = opentelemetry_otlp::SpanExporter::builder();

            builder.with_tonic().with_endpoint(endpoint).build()?
        };

        let mut resource_builder = Resource::builder()
            .with_service_name("worterbuch")
            .with_detectors(&[
                Box::new(HostResourceDetector::default()),
                Box::new(OsResourceDetector),
                Box::new(ProcessResourceDetector),
            ])
            .with_attributes(vec![KeyValue::new("instance.name", node_id.clone())]);

        if let Some(cluster_role) = cluster_role {
            resource_builder =
                resource_builder.with_attributes(vec![KeyValue::new("cluster.role", cluster_role)]);
        }

        let tracer_provider = SdkTracerProvider::builder()
            .with_batch_exporter(exporter)
            .with_resource(resource_builder.build())
            .build();

        let tracer = tracer_provider.tracer(node_id);

        global::set_tracer_provider(tracer_provider);

        let opentelemetry = tracing_opentelemetry::layer()
            .with_tracer(tracer)
            .with_filter(
                EnvFilter::builder()
                    .with_default_directive(LevelFilter::INFO.into())
                    .with_env_var("WORTERBUCH_TRACING")
                    .from_env_lossy(),
            );

        subscriber.with(opentelemetry).init();
        info!("Telemetry enabled.");
    } else {
        subscriber.init();
        info!("Telemetry disabled.");
    }

    Ok(())
}