feat: add OTLP gRPC exporter

Cargo.toml

@@ -403,6 +403,7 @@ opentelemetry = [
     "rama-core/opentelemetry",
     "rama-http?/opentelemetry",
     "rama-net?/opentelemetry",
+    "rama-grpc?/opentelemetry",
     "dep:opentelemetry-otlp",
 ]
 

docs/book/src/intro/telemetry.md

@@ -36,6 +36,8 @@ popular ingestion tools such as [Prometheus](https://prometheus.io/).
 ### Rama Telemetry Example
 
 > Source Code: [/examples/http_telemetry.rs](https://github.com/plabayo/rama/tree/main/examples/http_telemetry.rs)
+>
+> Source Code (gRPC OTLP): [/examples/grpc_example.rs](https://github.com/plabayo/rama/tree/main/examples/grpc_example.rs)
 
 In this example you can see a web service which keeps track of a visitor counter as a custom opentelemetry counter metric. It also makes use of the rama provided [`RequestMetricsLayer`](https://ramaproxy.org/docs/rama/http/layer/opentelemetry/struct.RequestMetricsLayer.html) and [`NetworkMetricsLayer`](https://ramaproxy.org/docs/rama/net/stream/layer/opentelemetry/struct.NetworkMetricsLayer.html) layers to also some insights in the traffic both on the network- and application (http) layers. These metrics are exported using the <https://crates.io/crates/opentelemetry-otlp> dependency.
 

examples/grpc/Cargo.toml

@@ -44,7 +44,13 @@ path = "src/compression/client.rs"
 name = "health-server"
 path = "src/health/server.rs"
 
+[[bin]]
+name = "otel-exporter"
+path = "src/otel_exporter.rs"
+required-features = ["opentelemetry"]
+
 [features]
+opentelemetry = ["rama/opentelemetry"]
 
 [dependencies]
 base64 = { workspace = true }

examples/grpc/README.md

@@ -48,3 +48,19 @@ service unhealthy (responded with "NOT_SERVING")
 status: SERVING
 ...
 ```
+
+## OTLP Exporter
+
+Demonstrates exporting OpenTelemetry metrics over gRPC using the OTLP protocol.
+
+Requires a running OpenTelemetry collector:
+
+```bash
+docker run -p 127.0.0.1:4317:4317 otel/opentelemetry-collector:latest
+```
+
+Then run:
+
+```bash
+cargo run -p rama-grpc-examples --features opentelemetry --bin otel-exporter
+```

examples/grpc/src/otel_exporter.rs

@@ -0,0 +1,170 @@
+//! An example to show how to export your [`opentelemetry`] metrics over gRPC.
+//! It also sets up [`tracing`] in a basic manner.
+//!
+//! Learn more about telemetry at <https://ramaproxy.org/book/intro/telemetry.html>.
+//! In this book chapter you'll also find more information on how you can
+//! consume the metrics of this example in tools such as Prometheus and Grafana.
+//!
+//! [`opentelemetry`]: https://opentelemetry.io/
+//! [`tracing`]: https://tracing.rs/
+//!
+//! This example will create a server that listens on `127.0.0.1:62013`.
+//!
+//! It also expects you to run the OT collector, e.g.:
+//!
+//! ```
+//! docker run \
+//!   -p 127.0.0.1:4317:4317 \
+//!   otel/opentelemetry-collector:latest
+//! ```
+//!
+//! # Run the example
+//!
+//! ```sh
+//! cargo run -p rama-grpc-examples --bin otel-exporter
+//! ```
+//!
+//! # Expected output
+//!
+//! The server will start and listen on `:62013`. You can use `curl`:
+//!
+//! ```sh
+//! curl -v http://127.0.0.1:62013
+//! ```
+//!
+//! With the response you should see a response with `HTTP/1.1 200` and a greeting.
+//!
+//! You can now use tools like grafana to collect metrics from the collector running at 127.0.0.1:4317 over GRPC.
+
+use rama::{
+    Layer,
+    extensions::Extensions,
+    http::{
+        client::EasyHttpWebClient,
+        grpc,
+        layer::{opentelemetry::RequestMetricsLayer, trace::TraceLayer},
+        server::HttpServer,
+        service::web::{WebService, response::Html},
+    },
+    layer::AddInputExtensionLayer,
+    net::stream::layer::opentelemetry::NetworkMetricsLayer,
+    rt::Executor,
+    tcp::server::TcpListener,
+    telemetry::{
+        opentelemetry::{
+            self, InstrumentationScope, KeyValue,
+            metrics::UpDownCounter,
+            sdk::{
+                Resource,
+                metrics::{PeriodicReader, SdkMeterProvider},
+            },
+            semantic_conventions::{
+                self,
+                resource::{HOST_ARCH, OS_NAME, SERVICE_NAME, SERVICE_VERSION},
+            },
+        },
+        tracing::{
+            self,
+            level_filters::LevelFilter,
+            subscriber::{EnvFilter, fmt, layer::SubscriberExt, util::SubscriberInitExt},
+        },
+    },
+};
+
+use std::{sync::Arc, time::Duration};
+
+#[derive(Debug)]
+struct Metrics {
+    counter: UpDownCounter<i64>,
+}
+
+impl Metrics {
+    fn new() -> Self {
+        let meter = opentelemetry::global::meter_with_scope(
+            InstrumentationScope::builder("example.otel_exporter")
+                .with_version(env!("CARGO_PKG_VERSION"))
+                .with_schema_url(semantic_conventions::SCHEMA_URL)
+                .with_attributes(vec![
+                    KeyValue::new(OS_NAME, std::env::consts::OS),
+                    KeyValue::new(HOST_ARCH, std::env::consts::ARCH),
+                ])
+                .build(),
+        );
+
+        let counter = meter.i64_up_down_counter("visitor_counter").build();
+        Self { counter }
+    }
+}
+
+#[tokio::main]
+async fn main() {
+    // tracing setup
+    tracing::subscriber::registry()
+        .with(fmt::layer())
+        .with(
+            EnvFilter::builder()
+                .with_default_directive(LevelFilter::DEBUG.into())
+                .from_env_lossy(),
+        )
+        .init();
+
+    let exporter_grpc_svc = EasyHttpWebClient::default();
+    let exporter_grpc_client = grpc::service::opentelemetry::OtelExporter::new(exporter_grpc_svc)
+        .with_endpoint(rama::http::Uri::from_static("http://localhost:4317"))
+        .with_timeout(Duration::from_secs(10));
+
+    let meter_reader = PeriodicReader::builder(exporter_grpc_client)
+        .with_interval(Duration::from_secs(3))
+        .build();
+
+    let resource = Resource::builder()
+        .with_attribute(KeyValue::new(SERVICE_NAME, "otel_exporter"))
+        .with_attribute(KeyValue::new(SERVICE_VERSION, rama::utils::info::VERSION))
+        .build();
+
+    let meter = SdkMeterProvider::builder()
+        .with_resource(resource)
+        .with_reader(meter_reader)
+        .build();
+
+    opentelemetry::global::set_meter_provider(meter);
+
+    // state for our custom app metrics
+    let state = Arc::new(Metrics::new());
+
+    let graceful = rama::graceful::Shutdown::default();
+
+    // http web service
+    graceful.spawn_task_fn(async |guard| {
+        // http service
+        let exec = Executor::graceful(guard);
+        let http_service = HttpServer::auto(exec.clone()).service(
+            (TraceLayer::new_for_http(), RequestMetricsLayer::default()).into_layer(
+                WebService::default().with_get("/", async |ext: Extensions| {
+                    ext.get::<Arc<Metrics>>().unwrap().counter.add(1, &[]);
+                    Html("<h1>Hello!</h1>")
+                }),
+            ),
+        );
+
+        // service setup & go
+        TcpListener::build(exec)
+            .bind("127.1:62013")
+            .await
+            .unwrap()
+            .serve(
+                (
+                    AddInputExtensionLayer::new(state),
+                    NetworkMetricsLayer::default(),
+                )
+                    .into_layer(http_service),
+            )
+            .await;
+    });
+
+    // wait for graceful shutdown
+    graceful
+        .shutdown_with_limit(Duration::from_secs(30))
+        .await
+        .unwrap();
+}

rama-grpc/Cargo.toml

@@ -23,10 +23,22 @@ full = ["protobuf", "transport", "compression"]
 protobuf = ["dep:prost", "dep:prost-types", "rama-grpc-build/protobuf"]
 transport = ["dep:rama-http-core"]
 compression = ["dep:flate2", "dep:zstd"]
+opentelemetry = [
+    "protobuf",
+    "dep:opentelemetry",
+    "dep:opentelemetry_sdk",
+    "rama-core/opentelemetry",
+]
+internal-logs = ["opentelemetry", "opentelemetry/internal-logs"]
 
 [dependencies]
 base64 = { workspace = true }
 flate2 = { workspace = true, optional = true }
+opentelemetry = { workspace = true, optional = true }
+opentelemetry_sdk = { workspace = true, optional = true, features = [
+    "metrics",
+    "trace",
+] }
 pin-project-lite = { workspace = true }
 prost = { workspace = true, optional = true }
 prost-types = { workspace = true, optional = true }
@@ -43,6 +55,8 @@ tokio = { workspace = true, features = ["macros", "net"] }
 zstd = { workspace = true, optional = true }
 
 [build-dependencies]
+prost-build = { workspace = true }
+protoc-bin-vendored = { workspace = true }
 rama-grpc-build = { workspace = true }
 
 [dev-dependencies]

rama-grpc/build.rs

@@ -4,4 +4,27 @@ fn main() {
         rama_grpc_build::protobuf::compile_protos("proto/health.proto").unwrap();
         println!("cargo::rerun-if-changed=proto");
     }
+
+    #[cfg(feature = "opentelemetry")]
+    {
+        // Proto files vendored from https://github.com/open-telemetry/opentelemetry-proto
+        // tag: v1.5.0, commit: 2bd940b2b77c1ab57c27166af21384906da7bb2b
+        let mut config = prost_build::Config::new();
+        config.disable_comments(["."]);
+
+        // Use vendored protoc to avoid requiring system protoc installation
+        if let Ok(path) = protoc_bin_vendored::protoc_bin_path() {
+            config.protoc_executable(path);
+        }
+
+        config
+            .compile_protos(
+                &[
+                    "proto/opentelemetry/proto/collector/trace/v1/trace_service.proto",
+                    "proto/opentelemetry/proto/collector/metrics/v1/metrics_service.proto",
+                ],
+                &["proto/"],
+            )
+            .unwrap();
+    }
 }

rama-grpc/proto/opentelemetry/proto/collector/metrics/v1/metrics_service.proto

@@ -0,0 +1,79 @@
+// Copyright 2019, OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+syntax = "proto3";
+
+package opentelemetry.proto.collector.metrics.v1;
+
+import "opentelemetry/proto/metrics/v1/metrics.proto";
+
+option csharp_namespace = "OpenTelemetry.Proto.Collector.Metrics.V1";
+option java_multiple_files = true;
+option java_package = "io.opentelemetry.proto.collector.metrics.v1";
+option java_outer_classname = "MetricsServiceProto";
+option go_package = "go.opentelemetry.io/proto/otlp/collector/metrics/v1";
+
+// Service that can be used to push metrics between one Application
+// instrumented with OpenTelemetry and a collector, or between a collector and a
+// central collector.
+service MetricsService {
+  // For performance reasons, it is recommended to keep this RPC
+  // alive for the entire life of the application.
+  rpc Export(ExportMetricsServiceRequest) returns (ExportMetricsServiceResponse) {}
+}
+
+message ExportMetricsServiceRequest {
+  // An array of ResourceMetrics.
+  // For data coming from a single resource this array will typically contain one
+  // element. Intermediary nodes (such as OpenTelemetry Collector) that receive
+  // data from multiple origins typically batch the data before forwarding further and
+  // in that case this array will contain multiple elements.
+  repeated opentelemetry.proto.metrics.v1.ResourceMetrics resource_metrics = 1;
+}
+
+message ExportMetricsServiceResponse {
+  // The details of a partially successful export request.
+  //
+  // If the request is only partially accepted
+  // (i.e. when the server accepts only parts of the data and rejects the rest)
+  // the server MUST initialize the `partial_success` field and MUST
+  // set the `rejected_<signal>` with the number of items it rejected.
+  //
+  // Servers MAY also make use of the `partial_success` field to convey
+  // warnings/suggestions to senders even when the request was fully accepted.
+  // In such cases, the `rejected_<signal>` MUST have a value of `0` and
+  // the `error_message` MUST be non-empty.
+  //
+  // A `partial_success` message with an empty value (rejected_<signal> = 0 and
+  // `error_message` = "") is equivalent to it not being set/present. Senders
+  // SHOULD interpret it the same way as in the full success case.
+  ExportMetricsPartialSuccess partial_success = 1;
+}
+
+message ExportMetricsPartialSuccess {
+  // The number of rejected data points.
+  //
+  // A `rejected_<signal>` field holding a `0` value indicates that the
+  // request was fully accepted.
+  int64 rejected_data_points = 1;
+
+  // A developer-facing human-readable message in English. It should be used
+  // either to explain why the server rejected parts of the data during a partial
+  // success or to convey warnings/suggestions during a full success. The message
+  // should offer guidance on how users can address such issues.
+  //
+  // error_message is an optional field. An error_message with an empty value
+  // is equivalent to it not being set.
+  string error_message = 2;
+}