Add OTel HTTP support (#853)

* Add OTel HTTP support

* temporal_sdk_temporal_num_pollers works, need to use more specific query for test

* added tests and CI action to run docker and run new test

* Remove stale println's, move variables to #[values()], assert specific metrics

Author: yuandrew

.github/workflows/per-pr.yml

@@ -84,3 +84,31 @@ jobs:
 
       - uses: Swatinem/rust-cache@v2
       - run: cargo integ-test
+
+  docker-integ-tests:
+    name: Docker integ tests
+    env:
+      TEMPORAL_CLOUD_ADDRESS: https://${{ vars.TEMPORAL_CLIENT_NAMESPACE }}.tmprl.cloud:7233
+      TEMPORAL_CLOUD_NAMESPACE: ${{ vars.TEMPORAL_CLIENT_NAMESPACE }}
+      TEMPORAL_CLIENT_CERT: ${{ secrets.TEMPORAL_CLIENT_CERT }}
+      TEMPORAL_CLIENT_KEY: ${{ secrets.TEMPORAL_CLIENT_KEY }}
+      DOCKER_PROMETHEUS_RUNNING: true
+    timeout-minutes: 20
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: dtolnay/rust-toolchain@stable
+        with:
+          toolchain: 1.80.0
+      - name: Install protoc
+        uses: arduino/setup-protoc@v3
+        with:
+          # TODO: Upgrade proto once https://github.com/arduino/setup-protoc/issues/99 is fixed
+          version: '23.x'
+          repo-token: ${{ secrets.GITHUB_TOKEN }}
+      - name: Start container for otel-collector and prometheus
+        uses: hoverkraft-tech/[email protected]
+        with:
+          compose-file: ./docker/docker-compose-ci.yaml
+      - uses: Swatinem/rust-cache@v2
+      - run: cargo integ-test test_docker_

core-api/src/telemetry.rs

@@ -71,6 +71,9 @@ pub struct OtelCollectorOptions {
     /// Overrides for histogram buckets. Units depend on the value of `use_seconds_for_durations`.
     #[builder(default)]
     pub histogram_bucket_overrides: HistogramBucketOverrides,
+    /// Protocol to use for communication with the collector
+    #[builder(default = "OtlpProtocol::Grpc")]
+    pub protocol: OtlpProtocol,
 }
 
 /// Options for exporting metrics to Prometheus
@@ -143,6 +146,15 @@ pub enum MetricTemporality {
     Delta,
 }
 
+/// Options for configuring telemetry
+#[derive(Debug, Clone, Copy)]
+pub enum OtlpProtocol {
+    /// Use gRPC to communicate with the collector
+    Grpc,
+    /// Use HTTP to communicate with the collector
+    Http,
+}
+
 impl Default for TelemetryOptions {
     fn default() -> Self {
         TelemetryOptionsBuilder::default().build().unwrap()

core/Cargo.toml

@@ -44,7 +44,7 @@ lru = "0.12"
 mockall = "0.13"
 opentelemetry = { workspace = true, features = ["metrics"], optional = true }
 opentelemetry_sdk = { version = "0.26", features = ["rt-tokio", "metrics"], optional = true }
-opentelemetry-otlp = { version = "0.26", features = ["tokio", "metrics", "tls"], optional = true }
+opentelemetry-otlp = { version = "0.26", features = ["tokio", "metrics", "tls", "http-proto", "reqwest-client",], optional = true }
 opentelemetry-prometheus = { git = "https://github.com/open-telemetry/opentelemetry-rust.git", rev = "e911383", optional = true }
 parking_lot = { version = "0.12", features = ["send_guard"] }
 pid = "4.0"

core/src/telemetry/otel.rs

@@ -30,7 +30,8 @@ use temporal_sdk_core_api::telemetry::{
         CoreMeter, Counter, Gauge, GaugeF64, Histogram, HistogramDuration, HistogramF64,
         MetricAttributes, MetricParameters, NewAttributes,
     },
-    HistogramBucketOverrides, MetricTemporality, OtelCollectorOptions, PrometheusExporterOptions,
+    HistogramBucketOverrides, MetricTemporality, OtelCollectorOptions, OtlpProtocol,
+    PrometheusExporterOptions,
 };
 use tokio::task::AbortHandle;
 use tonic::{metadata::MetadataMap, transport::ClientTlsConfig};
@@ -120,16 +121,26 @@ pub(super) fn augment_meter_provider_with_defaults(
 pub fn build_otlp_metric_exporter(
     opts: OtelCollectorOptions,
 ) -> Result<CoreOtelMeter, anyhow::Error> {
-    let mut exporter =
-        opentelemetry_otlp::TonicExporterBuilder::default().with_endpoint(opts.url.to_string());
-    if opts.url.scheme() == "https" || opts.url.scheme() == "grpcs" {
-        exporter = exporter.with_tls_config(ClientTlsConfig::new().with_native_roots());
-    }
-    let exporter = exporter
-        .with_metadata(MetadataMap::from_headers((&opts.headers).try_into()?))
-        .build_metrics_exporter(Box::new(metric_temporality_to_selector(
-            opts.metric_temporality,
-        )))?;
+    let exporter = match opts.protocol {
+        OtlpProtocol::Grpc => {
+            let mut exporter = opentelemetry_otlp::TonicExporterBuilder::default()
+                .with_endpoint(opts.url.to_string());
+            if opts.url.scheme() == "https" || opts.url.scheme() == "grpcs" {
+                exporter = exporter.with_tls_config(ClientTlsConfig::new().with_native_roots());
+            }
+            exporter
+                .with_metadata(MetadataMap::from_headers((&opts.headers).try_into()?))
+                .build_metrics_exporter(Box::new(metric_temporality_to_selector(
+                    opts.metric_temporality,
+                )))?
+        }
+        OtlpProtocol::Http => opentelemetry_otlp::HttpExporterBuilder::default()
+            .with_endpoint(opts.url.to_string())
+            .with_headers(opts.headers)
+            .build_metrics_exporter(Box::new(metric_temporality_to_selector(
+                opts.metric_temporality,
+            )))?,
+    };
     let reader = PeriodicReader::builder(exporter, runtime::Tokio)
         .with_interval(opts.metric_periodicity)
         .build();

docker/docker-compose-ci.yaml

@@ -0,0 +1,25 @@
+version: '3.5'
+
+services:
+  otel-collector:
+    image: otel/opentelemetry-collector:latest
+    command: [ '--config=/etc/otel-collector-ci.yaml' ]
+    volumes:
+      - ../etc/otel-collector-ci.yaml:/etc/otel-collector-ci.yaml
+    ports:
+      #      - "1888:1888"   # pprof extension
+      # It's useful to be able to manually inspect metrics during dev
+      - '8888:8888' # Prometheus metrics exposed by the collector
+      - '8889:8889' # Prometheus exporter metrics
+      #      - "13133:13133" # health_check extension
+      - '4317:4317' # OTLP gRPC receiver
+      - '4318:4318' # OTLP HTTP receiver
+    #      - "55679:55679" # zpages extension
+
+  prometheus:
+    container_name: prometheus
+    image: prom/prometheus:latest
+    volumes:
+      - ../etc/prometheus.yaml:/etc/prometheus/prometheus.yml
+    ports:
+      - '9090:9090'

etc/otel-collector-ci.yaml

@@ -0,0 +1,36 @@
+receivers:
+  otlp:
+    protocols:
+      grpc:
+        endpoint: 0.0.0.0:4317
+      http:
+        endpoint: 0.0.0.0:4318
+
+exporters:
+  prometheus:
+    endpoint: '0.0.0.0:8889'
+    namespace: temporal_sdk
+
+  debug:
+
+processors:
+  batch:
+
+extensions:
+  health_check:
+  pprof:
+    endpoint: :1888
+  zpages:
+    endpoint: :55679
+
+service:
+  extensions: [ pprof, zpages, health_check ]
+  pipelines:
+    traces:
+      receivers: [ otlp ]
+      processors: [ batch ]
+      exporters: [ debug, ]
+    metrics:
+      receivers: [ otlp ]
+      processors: [ batch ]
+      exporters: [ debug, prometheus ]

etc/otel-collector-config.yaml

@@ -11,7 +11,7 @@ exporters:
     endpoint: '0.0.0.0:8889'
     namespace: temporal_sdk
 
-  logging:
+  debug:
 
   otlp/jaeger:
     endpoint: jaeger:14250
@@ -32,8 +32,8 @@ service:
     traces:
       receivers: [ otlp ]
       processors: [ batch ]
-      exporters: [ logging, otlp/jaeger ]
+      exporters: [ debug, otlp/jaeger ]
     metrics:
       receivers: [ otlp ]
       processors: [ batch ]
-      exporters: [ logging, prometheus ]
+      exporters: [ debug, prometheus ]

etc/prometheus.yaml

@@ -1,6 +1,6 @@
 scrape_configs:
   - job_name: 'otel-collector'
-    scrape_interval: 3s
+    scrape_interval: 1s
     static_configs:
       - targets: ['otel-collector:8889']
       - targets: ['otel-collector:8888']

test-utils/src/lib.rs

@@ -80,6 +80,8 @@ pub static SEARCH_ATTR_INT: &str = "CustomIntField";
 pub const OTEL_URL_ENV_VAR: &str = "TEMPORAL_INTEG_OTEL_URL";
 /// If set, enable direct scraping of prom metrics on the specified port
 pub const PROM_ENABLE_ENV_VAR: &str = "TEMPORAL_INTEG_PROM_PORT";
+/// This should match the prometheus port exposed in docker-compose-ci.yaml
+pub const PROMETHEUS_QUERY_API: &str = "http://localhost:9090/api/v1/query";
 #[macro_export]
 macro_rules! prost_dur {
     ($dur_call:ident $args:tt) => {

tests/integ_tests/metrics_tests.rs

@@ -1,5 +1,6 @@
 use anyhow::anyhow;
 use assert_matches::assert_matches;
+use std::string::ToString;
 use std::{collections::HashMap, env, net::SocketAddr, sync::Arc, time::Duration};
 use temporal_client::{
     WorkflowClientTrait, WorkflowOptions, WorkflowService, REQUEST_LATENCY_HISTOGRAM_NAME,
@@ -15,8 +16,9 @@ use temporal_sdk_core::{
 use temporal_sdk_core_api::{
     telemetry::{
         metrics::{CoreMeter, MetricAttributes, MetricParameters},
-        HistogramBucketOverrides, OtelCollectorOptionsBuilder, PrometheusExporterOptions,
-        PrometheusExporterOptionsBuilder, TelemetryOptions, TelemetryOptionsBuilder,
+        HistogramBucketOverrides, OtelCollectorOptionsBuilder, OtlpProtocol,
+        PrometheusExporterOptions, PrometheusExporterOptionsBuilder, TelemetryOptions,
+        TelemetryOptionsBuilder,
     },
     worker::WorkerConfigBuilder,
     Worker,
@@ -43,6 +45,7 @@ use temporal_sdk_core_protos::{
 };
 use temporal_sdk_core_test_utils::{
     get_integ_server_options, get_integ_telem_options, CoreWfStarter, NAMESPACE, OTEL_URL_ENV_VAR,
+    PROMETHEUS_QUERY_API,
 };
 use tokio::{join, sync::Barrier, task::AbortHandle};
 use url::Url;
@@ -651,6 +654,88 @@ async fn request_fail_codes_otel() {
     }
 }
 
+// Tests that rely on Prometheus running in a docker container need to start
+// with `docker_` and set the `DOCKER_PROMETHEUS_RUNNING` env variable to run
+#[rstest::rstest]
+#[tokio::test]
+async fn docker_metrics_with_prometheus(
+    #[values(
+        ("http://localhost:4318/v1/metrics", OtlpProtocol::Http),
+        ("http://localhost:4317", OtlpProtocol::Grpc)
+    )]
+    otel_collector: (&str, OtlpProtocol),
+) {
+    if std::env::var("DOCKER_PROMETHEUS_RUNNING").is_err() {
+        return;
+    }
+    let (otel_collector_addr, otel_protocol) = otel_collector;
+    let test_uid = format!(
+        "test_{}_",
+        uuid::Uuid::new_v4().to_string().replace("-", "")
+    );
+
+    // Configure the OTLP exporter with HTTP
+    let opts = OtelCollectorOptionsBuilder::default()
+        .url(otel_collector_addr.parse().unwrap())
+        .protocol(otel_protocol)
+        .global_tags(HashMap::from([("test_id".to_string(), test_uid.clone())]))
+        .build()
+        .unwrap();
+    let exporter = Arc::new(build_otlp_metric_exporter(opts).unwrap());
+    let telemopts = TelemetryOptionsBuilder::default()
+        .metrics(exporter as Arc<dyn CoreMeter>)
+        .metric_prefix(test_uid.clone())
+        .build()
+        .unwrap();
+    let rt = CoreRuntime::new_assume_tokio(telemopts).unwrap();
+    let test_name = "docker_metrics_with_prometheus";
+    let mut starter = CoreWfStarter::new_with_runtime(test_name, rt);
+    let worker = starter.get_worker().await;
+    starter.start_wf().await;
+
+    // Immediately finish the workflow
+    let task = worker.poll_workflow_activation().await.unwrap();
+    worker
+        .complete_workflow_activation(WorkflowActivationCompletion::from_cmd(
+            task.run_id,
+            CompleteWorkflowExecution { result: None }.into(),
+        ))
+        .await
+        .unwrap();
+
+    let client = starter.get_client().await;
+    client.list_namespaces().await.unwrap();
+
+    // Give Prometheus time to scrape metrics
+    tokio::time::sleep(std::time::Duration::from_secs(2)).await;
+
+    // Query Prometheus API for metrics
+    let client = reqwest::Client::new();
+    let query = format!("temporal_sdk_{}num_pollers", test_uid.clone());
+    let response = client
+        .get(PROMETHEUS_QUERY_API)
+        .query(&[("query", query)])
+        .send()
+        .await
+        .unwrap()
+        .json::<serde_json::Value>()
+        .await
+        .unwrap();
+
+    // Validate the Prometheus response
+    if let Some(data) = response["data"]["result"].as_array() {
+        assert!(!data.is_empty(), "No metrics found for query: {test_uid}");
+        assert_eq!(data[0]["metric"]["exported_job"], "temporal-core-sdk");
+        assert_eq!(data[0]["metric"]["job"], "otel-collector");
+        assert!(data[0]["metric"]["task_queue"]
+            .as_str()
+            .unwrap()
+            .starts_with(test_name));
+    } else {
+        panic!("Invalid Prometheus response: {:?}", response);
+    }
+}
+
 #[tokio::test]
 async fn activity_metrics() {
     let (telemopts, addr, _aborter) = prom_metrics(None);