Update opentelemetry in controller

This moves to opentelemetry 0.22 with related changes in prometheus and
actix-web-opentelemetry. Some of the logic around how one should
configure opentelemetry changed and these changes bring it up to the
latest recommendations.

controller/Cargo.toml

@@ -21,8 +21,9 @@ serde_plain = "1"
 k8s-openapi = { version = "0.21", default-features = false, features = ["v1_24"] }
 kube = { version = "0.88", default-features = false, features = [ "derive", "runtime", "rustls-tls" ] }
 models = { path = "../models", version = "0.1.0" }
-opentelemetry = { version = "0.18", features = ["rt-tokio-current-thread"] }
-opentelemetry-prometheus = "0.11"
+opentelemetry = { version = "0.22"}
+opentelemetry_sdk = { version = "0.22"}
+opentelemetry-prometheus = "0.15"
 prometheus = "0.13.0"
 
 snafu = "0.7"

controller/src/controller.rs

@@ -15,7 +15,8 @@ use kube::api::DeleteParams;
 use kube::runtime::reflector::Store;
 use kube::Api;
 use kube::ResourceExt;
-use opentelemetry::global;
+use opentelemetry::metrics::MeterProvider;
+use opentelemetry_sdk::metrics::SdkMeterProvider;
 use snafu::ResultExt;
 use std::collections::BTreeMap;
 use std::env;
@@ -52,10 +53,11 @@ impl<T: BottlerocketShadowClient> BrupopController<T> {
         brs_reader: Store<BottlerocketShadow>,
         node_reader: Store<Node>,
         namespace: &str,
+        provider: &SdkMeterProvider,
     ) -> Self {
         // Creates brupop-controller meter via the configured
         // GlobalMeterProvider which is setup in PrometheusExporter
-        let meter = global::meter("brupop-controller");
+        let meter = provider.meter("brupop-controller");
         let metrics = BrupopControllerMetrics::new(meter);
         BrupopController {
             k8s_client,

controller/src/main.rs

@@ -21,8 +21,7 @@ use kube::{
     ResourceExt,
 };
 
-use opentelemetry::sdk::export::metrics::aggregation;
-use opentelemetry::sdk::metrics::{controllers, processors, selectors};
+use opentelemetry_sdk::metrics::SdkMeterProvider;
 use snafu::ResultExt;
 use tracing::{event, Level};
 
@@ -50,18 +49,26 @@ async fn main() -> Result<()> {
 
     let node_client = K8SBottlerocketShadowClient::new(k8s_client.clone(), &namespace);
 
-    let controller = controllers::basic(
-        processors::factory(
-            selectors::simple::histogram([1.0, 2.0, 5.0, 10.0, 20.0, 50.0]),
-            aggregation::cumulative_temporality_selector(),
-        )
-        .with_memory(false),
-    )
-    .build();
+    // let controller = controllers::basic(
+    //     processors::factory(
+    //         selectors::simple::histogram([1.0, 2.0, 5.0, 10.0, 20.0, 50.0]),
+    //         aggregation::cumulative_temporality_selector(),
+    //     )
+    //     .with_memory(false),
+    // )
+    // .build();
+
+    let registry = prometheus::Registry::new();
 
     // Exporter has to be initialized before BrupopController
     // in order to setup global meter provider properly
-    let exporter = opentelemetry_prometheus::exporter(controller).init();
+    let exporter = opentelemetry_prometheus::exporter()
+        .with_registry(registry.clone())
+        .build()
+        .context(controller_error::PrometheusRegsitrySnafu)?;
+    // let exporter = opentelemetry_prometheus::exporter(controller).init();
+
+    let provider = SdkMeterProvider::builder().with_reader(exporter).build();
 
     // Setup and run a reflector, ensuring that `BottlerocketShadow` updates are reflected to the controller.
     let brs_reflector = reflector::reflector(
@@ -96,8 +103,14 @@ async fn main() -> Result<()> {
         });
 
     // Setup and run the controller.
-    let controller =
-        BrupopController::new(k8s_client, node_client, brs_reader, node_reader, &namespace);
+    let controller = BrupopController::new(
+        k8s_client,
+        node_client,
+        brs_reader,
+        node_reader,
+        &namespace,
+        &provider,
+    );
     let controller_runner = controller.run();
 
     let k8s_service_addr = env::var("KUBERNETES_SERVICE_HOST")
@@ -113,7 +126,7 @@ async fn main() -> Result<()> {
     // Setup Http server to vend prometheus metrics
     let prometheus_server = HttpServer::new(move || {
         App::new()
-            .app_data(Data::new(exporter.clone()))
+            .app_data(Data::new(registry.clone()))
             .service(vending_metrics)
     })
     .bind(format!("{}:{}", bindaddress, CONTROLLER_INTERNAL_PORT))
@@ -178,5 +191,10 @@ pub mod controller_error {
         TelemetryInit {
             source: telemetry::TelemetryConfigError,
         },
+
+        #[snafu(display("Error creating prometheus registry: '{}'", source))]
+        PrometheusRegsitry {
+            source: opentelemetry::metrics::MetricsError,
+        },
     }
 }

controller/src/metrics.rs

@@ -81,19 +81,19 @@ impl BrupopControllerMetrics {
             .with_description("Brupop host's state")
             .init();
 
-        let _ = meter.register_callback(move |cx| {
+        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())];
-                brupop_hosts_version_observer.observe(cx, *count, &labels);
+                cx.observe_u64(&brupop_hosts_version_observer, *count, &labels);
             }
         });
 
-        let _ = meter.register_callback(move |cx| {
+        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())];
-                brupop_hosts_state_observer.observe(cx, *count, &labels);
+                cx.observe_u64(&brupop_hosts_state_observer, *count, &labels);
             }
         });
 

controller/src/telemetry.rs

@@ -3,11 +3,9 @@ use opentelemetry::{global, metrics::MetricsError};
 use prometheus::{Encoder, TextEncoder};
 
 #[get("/metrics")]
-pub async fn vending_metrics(
-    exporter: Data<opentelemetry_prometheus::PrometheusExporter>,
-) -> HttpResponse {
+pub async fn vending_metrics(registry: Data<prometheus::Registry>) -> HttpResponse {
     let encoder = TextEncoder::new();
-    let metric_families = exporter.registry().gather();
+    let metric_families = registry.gather();
     let mut buf = Vec::new();
     if let Err(err) = encoder.encode(&metric_families[..], &mut buf) {
         global::handle_error(MetricsError::Other(err.to_string()));