otel: Add trace sampling

Adds code to rust to take a trace sampling ratio, and adds code on the
Unit side implementing the settings/telemetry/sample_ratio configuration
path that will be resolved into a number. Adds a validator that makes
sure that the sample ratio is in the range [0, 1].

Signed-off-by: Gabor Javorszky <g.javorszky@f5.com>

Author: javorszky

src/nxt_conf_validation.c

@@ -248,6 +248,9 @@ nxt_inline nxt_int_t nxt_otel_validate_endpoint(nxt_conf_validation_t *vldt,
 nxt_int_t nxt_otel_validate_batch_size(nxt_conf_validation_t *vldt,
                                        nxt_conf_value_t *value,
                                        void *data);
+nxt_int_t nxt_otel_validate_sample_ratio(nxt_conf_validation_t *vldt,
+                                         nxt_conf_value_t *value,
+                                         void *data);
 nxt_int_t nxt_otel_validate_protocol(nxt_conf_validation_t *vldt,
                                      nxt_conf_value_t *value,
                                      void *data);
@@ -336,6 +339,10 @@ static nxt_conf_vldt_object_t nxt_conf_vldt_otel_members[] = {
         .type      = NXT_CONF_VLDT_STRING,
         .validator = nxt_otel_validate_protocol,
         .flags     = NXT_CONF_VLDT_REQUIRED
+    }, {
+        .name      = nxt_string("sampling_ratio"),
+        .type      = NXT_CONF_VLDT_NUMBER,
+        .validator = nxt_otel_validate_sample_ratio,
     },
 
     NXT_CONF_VLDT_END
@@ -1534,6 +1541,20 @@ nxt_otel_validate_batch_size(nxt_conf_validation_t *vldt,
     return NXT_OK;
 }
 
+nxt_int_t
+nxt_otel_validate_sample_ratio(nxt_conf_validation_t *vldt,
+                               nxt_conf_value_t *value,
+                               void *data)
+{
+    double sample_ratio;
+
+    sample_ratio = nxt_conf_get_number(value);
+    if (sample_ratio < 0 || sample_ratio > 1) {
+        return NXT_ERROR;
+    }
+
+    return NXT_OK;
+}
 
 nxt_int_t
 nxt_otel_validate_protocol(nxt_conf_validation_t *vldt,

src/nxt_otel.h

@@ -19,6 +19,7 @@ extern void   * nxt_otel_rs_get_or_create_trace(u_char *trace_id);
 extern void     nxt_otel_rs_init(void (*log_callback)(u_char *log_string),
                                  const nxt_str_t *endpoint,
                                  const nxt_str_t *protocol,
+                                 double sample_fraction,
                                  double batch_size);
 extern void     nxt_otel_rs_copy_traceparent(u_char *buffer, void *span);
 extern void     nxt_otel_rs_add_event_to_trace(void *trace,

src/nxt_router.c

@@ -25,6 +25,7 @@
 #define NXT_SHARED_PORT_ID  0xFFFFu
 #if (NXT_HAVE_OTEL)
 #define NXT_OTEL_BATCH_DEFAULT 128
+#define NXT_OTEL_SAMPLING_DEFAULT 1
 #endif
 
 typedef struct {
@@ -1645,9 +1646,9 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
     nxt_conf_value_t            *js_module;
 #endif
 #if (NXT_HAVE_OTEL)
-    nxt_conf_value_t            *otel, *otel_endpoint, *otel_batching, *otel_proto;
+    nxt_conf_value_t            *otel, *otel_endpoint, *otel_sampling, *otel_batching, *otel_proto;
     nxt_str_t                   telemetry_endpoint, telemetry_proto;
-    double                      telemetry_batching;
+    double                      telemetry_sample_fraction, telemetry_batching;
 #endif
     nxt_conf_value_t            *root, *conf, *http, *value, *websocket;
     nxt_conf_value_t            *applications, *application, *settings;
@@ -1688,6 +1689,7 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
     static const nxt_str_t  telemetry_path = nxt_string("/settings/telemetry");
     static const nxt_str_t  telemetry_endpoint_path = nxt_string("/settings/telemetry/endpoint");
     static const nxt_str_t  telemetry_batch_path = nxt_string("/settings/telemetry/batch_size");
+    static const nxt_str_t  telemetry_sample_path = nxt_string("/settings/telemetry/sampling_ratio");
     static const nxt_str_t  telemetry_proto_path = nxt_string("/settings/telemetry/protocol");
 #endif
 
@@ -2195,16 +2197,19 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
     otel          = nxt_conf_get_path(root, &telemetry_path);
     otel_endpoint = nxt_conf_get_path(root, &telemetry_endpoint_path);
     otel_batching = nxt_conf_get_path(root, &telemetry_batch_path);
+    otel_sampling = nxt_conf_get_path(root, &telemetry_sample_path);
     otel_proto    = nxt_conf_get_path(root, &telemetry_proto_path);
 
     if (otel) {
         nxt_conf_get_string(otel_endpoint, &telemetry_endpoint);
         nxt_conf_get_string(otel_proto, &telemetry_proto);
         telemetry_batching = otel_batching ? nxt_conf_get_number(otel_batching) : NXT_OTEL_BATCH_DEFAULT;
+        telemetry_sample_fraction = otel_sampling ? nxt_conf_get_number(otel_sampling) : NXT_OTEL_SAMPLING_DEFAULT;
 
         nxt_otel_rs_init(&nxt_otel_log_callback,
                       &telemetry_endpoint,
                       &telemetry_proto,
+                      telemetry_sample_fraction,
                       telemetry_batching);
     } else {
       nxt_otel_rs_uninit();

src/otel/src/lib.rs

@@ -5,10 +5,10 @@ use opentelemetry::trace::{
 use opentelemetry::{global, KeyValue};
 use opentelemetry_otlp::Protocol;
 use opentelemetry_otlp::WithExportConfig;
-use opentelemetry_sdk::trace::{Config, BatchConfigBuilder};
+use opentelemetry_sdk::trace::{Config, BatchConfigBuilder, Sampler};
 use opentelemetry_sdk::{runtime, Resource};
 use std::ffi::{CStr, CString};
-use std::ptr;
+use std::{ptr, time};
 use std::ptr::addr_of;
 use std::slice;
 use std::sync::{Arc, OnceLock};
@@ -17,7 +17,7 @@ use tokio::sync::mpsc::{Receiver, Sender};
 
 
 const TRACEPARENT_HEADER_LEN: u8 = 55;
-
+const TIMEOUT: time::Duration = std::time::Duration::from_secs(10);
 
 #[repr(C)]
 pub struct nxt_str_t {
@@ -64,6 +64,7 @@ unsafe fn nxt_otel_rs_init(
     log_callback: unsafe extern "C" fn(*mut i8),
     endpoint: *const nxt_str_t,
     protocol: *const nxt_str_t,
+    sample_fraction: f64,
     batch_size: f64
 ) {
     if endpoint.is_null() ||
@@ -110,6 +111,7 @@ unsafe fn nxt_otel_rs_init(
                 ep,
                 proto,
                 batch_size,
+                sample_fraction,
                 rx
             ));
         },
@@ -132,16 +134,21 @@ async unsafe fn nxt_otel_rs_runtime(
     endpoint: String,
     proto: Protocol,
     batch_size: f64,
+    sample_fraction: f64,
     mut rx: Receiver<SpanMessage>
 ) {
     let pipeline = opentelemetry_otlp::new_pipeline()
         .tracing()
-        .with_trace_config(Config::default().with_resource(
-            Resource::new(vec![KeyValue::new(
-                opentelemetry_semantic_conventions::resource::SERVICE_NAME,
-                "NGINX Unit",
-            )]),
-        ))
+        .with_trace_config(
+            Config::default()
+                .with_resource(
+                    Resource::new(vec![KeyValue::new(
+                        opentelemetry_semantic_conventions::resource::SERVICE_NAME,
+                        "NGINX Unit",
+                    )])
+                )
+                .with_sampler(Sampler::TraceIdRatioBased(sample_fraction))
+        )
         .with_batch_config(
             BatchConfigBuilder::default()
                 .with_max_export_batch_size(batch_size as _)
@@ -156,15 +163,15 @@ async unsafe fn nxt_otel_rs_runtime(
                     .with_http_client(reqwest::Client::new()) // needed because rustls feature
                     .with_endpoint(endpoint)
                     .with_protocol(proto)
-                    .with_timeout(std::time::Duration::new(10, 0))
+                    .with_timeout(TIMEOUT)
             ).install_batch(runtime::Tokio),
         Protocol::Grpc => pipeline
             .with_exporter(
                 opentelemetry_otlp::new_exporter()
                     .tonic()
                     .with_endpoint(endpoint)
                     .with_protocol(proto)
-                    .with_timeout(std::time::Duration::new(10, 0))
+                    .with_timeout(TIMEOUT)
             ).install_batch(runtime::Tokio),
     };
 
@@ -211,7 +218,7 @@ pub unsafe fn nxt_otel_rs_copy_traceparent(buf: *mut i8, span: *const BoxedSpan)
         (*span).span_context().trace_flags()  // 1 char, 2 hex
     );
 
-    assert_eq!(traceparent.len(), TRACEPARENT_HEADER_LEN as _);
+    assert_eq!(traceparent.len(), TRACEPARENT_HEADER_LEN as usize);
 
     ptr::copy_nonoverlapping(
         traceparent.as_bytes().as_ptr(),
@@ -270,8 +277,7 @@ pub unsafe fn nxt_otel_rs_get_or_create_trace(trace_id: *mut i8) -> *mut BoxedSp
 }
 
 #[no_mangle]
-#[tokio::main]
-pub async unsafe fn nxt_otel_rs_send_trace(trace: *mut BoxedSpan) {
+pub unsafe fn nxt_otel_rs_send_trace(trace: *mut BoxedSpan) {
     // damage nothing on an improper call
     if trace.is_null() {
         eprintln!("trace was null, returning");