Merge pull request #777 from yeazelm/update_ot

Update  Open Telemetry dependencies

Author: yeazelm

Cargo.lock

@@ -41,13 +41,13 @@ nonzero_ext = "0.3"
 # these package versions should be moved in lockstep to match upstream
 actix-web = { version = "4.9", features = ["rustls-0_23"] }
 tracing-actix-web = "0.7"
-actix-web-opentelemetry = { version = "0.18", features = ["metrics", "metrics-prometheus"] }
+opentelemetry-instrumentation-actix-web = { version = "0.22", features = ["metrics"] }
 
 # opentelemetry dependencies
-opentelemetry = { version = "0.23"}
-opentelemetry_sdk = { version = "0.23", features = ["rt-tokio-current-thread"]}
-opentelemetry-prometheus = "0.16"
-prometheus = "0.13.0"
+opentelemetry = { version = "0.29", features = ["metrics"]}
+opentelemetry_sdk = { version = "0.29", features = ["rt-tokio-current-thread", "metrics"]}
+opentelemetry-prometheus = "0.29"
+prometheus = "0.14"
 
 # k8s-openapi must match the version required by kube and enable a k8s version feature
 k8s-openapi = { version = "0.21", default-features = false, features = ["v1_24"] }

Cargo.toml

@@ -29,7 +29,7 @@ DESTDIR ?= .
 # tarball.
 DISTFILE ?= $(DESTDIR:/=)/$(subst /,_,$(IMAGE_NAME)).tar.gz
 
-BOTTLEROCKET_SDK_VERSION = v0.61.0
+BOTTLEROCKET_SDK_VERSION = v0.62.0
 
 # Tools used during the chart release lifecycle
 export KUBECONFORM_VERSION = v0.6.3

Makefile

@@ -18,7 +18,7 @@ models = { path = "../models", version = "0.1.0" }
 
 actix-web = { workspace = true }
 awc = { workspace = true }
-actix-web-opentelemetry = { workspace = true }
+opentelemetry-instrumentation-actix-web = { workspace = true }
 rustls = { workspace = true }
 rustls-pemfile = { workspace = true }
 webpki = { workspace = true }

apiserver/Cargo.toml

@@ -5,28 +5,14 @@ pub mod error;
 mod node;
 mod ping;
 
-use crate::{
-    auth::{K8STokenAuthorizor, K8STokenReviewer, TokenAuthMiddleware},
-    constants::{
-        CRD_CONVERT_ENDPOINT, EXCLUDE_NODE_FROM_LB_ENDPOINT, HEADER_BRUPOP_K8S_AUTH_TOKEN,
-        HEADER_BRUPOP_NODE_NAME, HEADER_BRUPOP_NODE_UID, NODE_CORDON_AND_DRAIN_ENDPOINT,
-        NODE_RESOURCE_ENDPOINT, NODE_UNCORDON_ENDPOINT, REMOVE_NODE_EXCLUSION_TO_LB_ENDPOINT,
-    },
-    telemetry,
-};
-use models::constants::{
-    AGENT, APISERVER_HEALTH_CHECK_ROUTE, APISERVER_SERVICE_NAME, LABEL_COMPONENT, PRIVATE_KEY_NAME,
-    PUBLIC_KEY_NAME, TLS_KEY_MOUNT_PATH,
-};
-use models::node::{read_certificate, BottlerocketShadowClient, BottlerocketShadowSelector};
+use std::{env, fs::File, io::BufReader};
 
 use actix_web::{
     dev::ServerHandle,
     http::header::HeaderMap,
     web::{self, Data},
     App, HttpServer,
 };
-use actix_web_opentelemetry::{PrometheusMetricsHandler, RequestMetrics, RequestTracing};
 use futures::StreamExt;
 use k8s_openapi::api::core::v1::Pod;
 use kube::{
@@ -38,11 +24,25 @@ use kube::{
     },
     ResourceExt,
 };
-
+use opentelemetry_instrumentation_actix_web::{RequestMetrics, RequestTracing};
 use rustls::ServerConfig;
 use rustls_pemfile::{certs, pkcs8_private_keys};
 use snafu::{OptionExt, ResultExt};
-use std::{env, fs::File, io::BufReader};
+
+use crate::{
+    auth::{K8STokenAuthorizor, K8STokenReviewer, TokenAuthMiddleware},
+    constants::{
+        CRD_CONVERT_ENDPOINT, EXCLUDE_NODE_FROM_LB_ENDPOINT, HEADER_BRUPOP_K8S_AUTH_TOKEN,
+        HEADER_BRUPOP_NODE_NAME, HEADER_BRUPOP_NODE_UID, NODE_CORDON_AND_DRAIN_ENDPOINT,
+        NODE_RESOURCE_ENDPOINT, NODE_UNCORDON_ENDPOINT, REMOVE_NODE_EXCLUSION_TO_LB_ENDPOINT,
+    },
+    telemetry,
+};
+use models::constants::{
+    AGENT, APISERVER_HEALTH_CHECK_ROUTE, APISERVER_SERVICE_NAME, LABEL_COMPONENT, PRIVATE_KEY_NAME,
+    PUBLIC_KEY_NAME, TLS_KEY_MOUNT_PATH,
+};
+use models::node::{read_certificate, BottlerocketShadowClient, BottlerocketShadowSelector};
 use tokio::time::{sleep, Duration};
 use tracing::{event, Level};
 use tracing_actix_web::TracingLogger;
@@ -217,12 +217,10 @@ pub async fn run_server<T: 'static + BottlerocketShadowClient>(
             )
             .wrap(RequestTracing::new())
             .wrap(RequestMetrics::default())
-            .route(
-                "/metrics",
-                web::get().to(PrometheusMetricsHandler::new(prometheus_registry.clone())),
-            )
+            .route("/metrics", web::get().to(crate::telemetry::vending_metrics))
             .wrap(TracingLogger::<telemetry::BrupopApiserverRootSpanBuilder>::new())
             .app_data(Data::new(settings.clone()))
+            .app_data(Data::new(prometheus_registry.clone()))
             .service(
                 web::resource(NODE_RESOURCE_ENDPOINT)
                     .route(web::post().to(node::create_bottlerocket_shadow_resource::<T>))

apiserver/src/api/mod.rs

@@ -1,29 +1,43 @@
-use apiserver::api::{self, APIServerSettings};
-use apiserver_error::{StartServerSnafu, StartTelemetrySnafu};
-use models::node::K8SBottlerocketShadowClient;
-use models::telemetry;
-use opentelemetry::global;
-use tracing::{event, Level};
+use std::convert::TryFrom;
+use std::env;
+use std::fs;
+use std::sync::OnceLock;
 
+use opentelemetry::global;
 use opentelemetry::KeyValue;
 use opentelemetry_sdk::metrics::SdkMeterProvider;
 use opentelemetry_sdk::Resource;
 use snafu::ResultExt;
+use tracing::{event, Level};
 
-use std::convert::TryFrom;
-use std::env;
-use std::fs;
+use apiserver::api::{self, APIServerSettings};
+use apiserver_error::{StartServerSnafu, StartTelemetrySnafu};
+use models::node::K8SBottlerocketShadowClient;
+use models::telemetry;
 
 // By default, errors resulting in termination of the apiserver are written to this file,
 // which is the location kubernetes uses by default to surface termination-causing errors.
 const TERMINATION_LOG: &str = "/dev/termination-log";
 const APISERVER_INTERNAL_PORT_ENV_VAR: &str = "APISERVER_INTERNAL_PORT";
 
+// Store the meter provider so we can shut it down properly
+static METER_PROVIDER: OnceLock<SdkMeterProvider> = OnceLock::new();
+
+/// Custom error handler for OpenTelemetry operations
+fn handle_opentelemetry_error(operation: &str, error: impl std::fmt::Display) {
+    tracing::error!("OpenTelemetry {} error: {}", operation, error);
+}
+
 #[actix_web::main]
 async fn main() {
     main_inner().await;
 
-    opentelemetry::global::shutdown_tracer_provider();
+    // Properly shutdown the OpenTelemetry meter provider
+    if let Some(provider) = METER_PROVIDER.get() {
+        if let Err(e) = provider.shutdown() {
+            handle_opentelemetry_error("shutdown", e);
+        }
+    }
 }
 
 async fn main_inner() {
@@ -56,8 +70,17 @@ async fn run_server() -> Result<(), apiserver_error::Error> {
 
     let prometheus_provider = SdkMeterProvider::builder()
         .with_reader(prometheus_exporter)
-        .with_resource(Resource::new([KeyValue::new("service.name", "apiserver")]))
+        .with_resource(
+            Resource::builder()
+                .with_attribute(KeyValue::new("service.name", "apiserver"))
+                .build(),
+        )
         .build();
+
+    // Store the provider for shutdown
+    METER_PROVIDER
+        .set(prometheus_provider.clone())
+        .expect("Failed to set meter provider");
     global::set_meter_provider(prometheus_provider);
 
     let incluster_config =
@@ -130,7 +153,7 @@ pub mod apiserver_error {
 
         #[snafu(display("Error creating prometheus registry: '{}'", source))]
         PrometheusRegsitry {
-            source: opentelemetry::metrics::MetricsError,
+            source: opentelemetry_sdk::metrics::MetricError,
         },
     }
 }

apiserver/src/main.rs

@@ -1,13 +1,15 @@
-use crate::api::NO_TELEMETRY_ENDPOINTS;
-use crate::constants::HEADER_BRUPOP_NODE_NAME;
+use std::collections::HashSet;
 
 use actix_web::body::MessageBody;
 use actix_web::dev::{ServiceRequest, ServiceResponse};
+use actix_web::{http::header::ContentType, web::Data, HttpResponse};
 use lazy_static::lazy_static;
+use prometheus::{Encoder, TextEncoder};
 use tracing::Span;
 use tracing_actix_web::{DefaultRootSpanBuilder, RootSpanBuilder};
 
-use std::collections::HashSet;
+use crate::api::NO_TELEMETRY_ENDPOINTS;
+use crate::constants::HEADER_BRUPOP_NODE_NAME;
 
 // tracing-actix-web doesn't provide a convenient way to remove any routes from the logs, so we use a global
 // settings containing API paths to generate empty `tracing::Span`s on paths which we don't want logged.
@@ -48,3 +50,30 @@ impl RootSpanBuilder for BrupopApiserverRootSpanBuilder {
         DefaultRootSpanBuilder::on_request_end(span, response);
     }
 }
+
+/// Custom error handler for OpenTelemetry metrics encoding errors
+fn handle_metrics_error(err: prometheus::Error) {
+    tracing::error!("Metrics encoding error: {}", err);
+}
+
+pub async fn vending_metrics(registry: Data<prometheus::Registry>) -> HttpResponse {
+    let encoder = TextEncoder::new();
+    let metric_families = registry.gather();
+    let mut buf = Vec::new();
+
+    match encoder.encode(&metric_families[..], &mut buf) {
+        Ok(()) => {
+            let body = String::from_utf8(buf).unwrap_or_default();
+            HttpResponse::Ok()
+                .insert_header(ContentType::plaintext())
+                .body(body)
+        }
+        Err(err) => {
+            handle_metrics_error(err);
+            // Return empty metrics response on error
+            HttpResponse::InternalServerError()
+                .insert_header(ContentType::plaintext())
+                .body("# Metrics encoding error\n")
+        }
+    }
+}

apiserver/src/telemetry.rs

@@ -439,11 +439,11 @@ fn find_unlabeled_nodes(mut nodes: Vec<Node>) -> Vec<String> {
 
 #[instrument(skip(node))]
 fn node_has_label(node: &Node) -> bool {
-    return node.labels().get_key_value(LABEL_BRUPOP_INTERFACE_NAME)
+    node.labels().get_key_value(LABEL_BRUPOP_INTERFACE_NAME)
         == Some((
             &LABEL_BRUPOP_INTERFACE_NAME.to_string(),
             &BRUPOP_INTERFACE_VERSION.to_string(),
-        ));
+        ))
 }
 
 fn read_env_var(env_var: &str) -> Result<String> {

controller/src/controller.rs

@@ -1,14 +1,6 @@
 use std::{convert::TryFrom, env};
 
-use controller::{telemetry::vending_metrics, BrupopController};
-use models::{
-    constants::CONTROLLER_INTERNAL_PORT,
-    node::{BottlerocketShadow, K8SBottlerocketShadowClient},
-    telemetry,
-};
-
 use actix_web::{web::Data, App, HttpServer};
-
 use futures::StreamExt;
 use k8s_openapi::api::core::v1::Node;
 use kube::{
@@ -20,11 +12,17 @@ use kube::{
     },
     ResourceExt,
 };
-
 use opentelemetry_sdk::metrics::SdkMeterProvider;
 use snafu::ResultExt;
 use tracing::{event, Level};
 
+use controller::{telemetry::vending_metrics, BrupopController};
+use models::{
+    constants::CONTROLLER_INTERNAL_PORT,
+    node::{BottlerocketShadow, K8SBottlerocketShadowClient},
+    telemetry,
+};
+
 /// The module-wide result type.
 type Result<T> = std::result::Result<T, controller_error::Error>;
 
@@ -192,7 +190,7 @@ pub mod controller_error {
 
         #[snafu(display("Error creating prometheus registry: '{}'", source))]
         PrometheusRegsitry {
-            source: opentelemetry::metrics::MetricsError,
+            source: opentelemetry_sdk::metrics::MetricError,
         },
     }
 }

controller/src/main.rs

@@ -1,10 +1,12 @@
-use models::node::BottlerocketShadow;
-use opentelemetry::{metrics::Meter, Key};
-use snafu::ResultExt;
 use std::collections::HashMap;
 use std::sync::{Arc, Mutex};
+
+use opentelemetry::{metrics::Meter, Key, KeyValue};
+use snafu::ResultExt;
 use tracing::instrument;
 
+use models::node::BottlerocketShadow;
+
 const HOST_VERSION_KEY: Key = Key::from_static_str("bottlerocket_version");
 const HOST_STATE_KEY: Key = Key::from_static_str("state");
 
@@ -69,34 +71,31 @@ impl BrupopControllerMetrics {
         let hosts_data_clone_for_version = Arc::clone(&brupop_shared_hosts_data);
         let hosts_data_clone_for_state = Arc::clone(&brupop_shared_hosts_data);
 
-        // Observer for cluster host's bottlerocket version
-        let brupop_hosts_version_observer = meter
+        // Create observable gauges with callbacks using OpenTelemetry API
+        let _brupop_hosts_version_observer = meter
             .u64_observable_gauge("brupop_hosts_version")
             .with_description("Brupop host's bottlerocket version")
-            .init();
+            .with_callback(move |observer| {
+                let data = hosts_data_clone_for_version.lock().unwrap();
+                for (host_version, count) in &data.hosts_version_count {
+                    let attributes = [KeyValue::new(HOST_VERSION_KEY, host_version.to_string())];
+                    observer.observe(*count, &attributes);
+                }
+            })
+            .build();
 
         // Observer for cluster host's brupop state
-        let brupop_hosts_state_observer = meter
+        let _brupop_hosts_state_observer = meter
             .u64_observable_gauge("brupop_hosts_state")
             .with_description("Brupop host's state")
-            .init();
-
-        let _ = meter.register_callback(&[brupop_hosts_version_observer.as_any()], move |cx| {
-            let data = hosts_data_clone_for_version.lock().unwrap();
-            for (host_version, count) in &data.hosts_version_count {
-                let labels = vec![HOST_VERSION_KEY.string(host_version.to_string())];
-                cx.observe_u64(&brupop_hosts_version_observer, *count, &labels);
-            }
-        });
-
-        let _ = meter.register_callback(&[brupop_hosts_state_observer.as_any()], move |cx| {
-            let data = hosts_data_clone_for_state.lock().unwrap();
-            for (host_state, count) in &data.hosts_state_count {
-                let labels = vec![HOST_STATE_KEY.string(host_state.to_string())];
-                cx.observe_u64(&brupop_hosts_state_observer, *count, &labels);
-            }
-        });
-
+            .with_callback(move |observer| {
+                let data = hosts_data_clone_for_state.lock().unwrap();
+                for (host_state, count) in &data.hosts_state_count {
+                    let attributes = [KeyValue::new(HOST_STATE_KEY, host_state.to_string())];
+                    observer.observe(*count, &attributes);
+                }
+            })
+            .build();
         BrupopControllerMetrics {
             brupop_shared_hosts_data,
         }