otel: configuration items and their validation

Adds code responsible for users to apply the `telemetry` configuration
options.

configuration snippet as follows:
{
    "settings": {
        "telemetry": {
            "batch_size": 20,
            "endpoint": "http://lgtm:4318/v1/traces",
            "protocol": "http",
            "sampling_ratio": 1
        }
    },
    "listeners": {
        "*:80": {
            "pass": "routes"
        }
    },
    "routes": [
        {
            "match": {
                "headers": {
                    "accept": "*text/html*"
                }
            },
            "action": {
                "share": "/usr/share/unit/welcome/welcome.html"
            }
        },
        {
            "action": {
                "share": "/usr/share/unit/welcome/welcome.md"
            }
        }
    ]
}

Signed-off-by: Ava Hahn <a.hahn@f5.com>
Signed-off-by: Gabor Javorszky <g.javorszky@f5.com>

2 files changed, 151 insertions, 0 deletions

diff --git a/src/nxt_conf_validation.c b/src/nxt_conf_validation.c
index 5d7f7c52..0dde39ff 100644
--- a/src/nxt_conf_validation.c
+++ b/src/nxt_conf_validation.c
@@ -241,9 +241,21 @@ static nxt_int_t nxt_conf_vldt_js_module_element(nxt_conf_validation_t *vldt,
     nxt_conf_value_t *value);
 #endif
 
+#if (NXT_HAVE_OTEL)
+static nxt_int_t nxt_otel_validate_batch_size(nxt_conf_validation_t *vldt,
+    nxt_conf_value_t *value, void *data);
+static nxt_int_t nxt_otel_validate_sample_ratio(nxt_conf_validation_t *vldt,
+    nxt_conf_value_t *value, void *data);
+static nxt_int_t nxt_otel_validate_protocol(nxt_conf_validation_t *vldt,
+    nxt_conf_value_t *value, void *data);
+#endif
+
 
 static nxt_conf_vldt_object_t  nxt_conf_vldt_setting_members[];
 static nxt_conf_vldt_object_t  nxt_conf_vldt_http_members[];
+#if (NXT_HAVE_OTEL)
+static nxt_conf_vldt_object_t  nxt_conf_vldt_otel_members[];
+#endif
 static nxt_conf_vldt_object_t  nxt_conf_vldt_websocket_members[];
 static nxt_conf_vldt_object_t  nxt_conf_vldt_static_members[];
 static nxt_conf_vldt_object_t  nxt_conf_vldt_forwarded_members[];
@@ -317,6 +329,13 @@ static nxt_conf_vldt_object_t  nxt_conf_vldt_setting_members[] = {
         .type       = NXT_CONF_VLDT_OBJECT,
         .validator  = nxt_conf_vldt_object,
         .u.members  = nxt_conf_vldt_http_members,
+#if (NXT_HAVE_OTEL)
+    }, {
+        .name       = nxt_string("telemetry"),
+        .type       = NXT_CONF_VLDT_OBJECT,
+        .validator  = nxt_conf_vldt_object,
+        .u.members  = nxt_conf_vldt_otel_members,
+#endif
 #if (NXT_HAVE_NJS)
     }, {
         .name       = nxt_string("js_module"),
@@ -385,6 +404,34 @@ static nxt_conf_vldt_object_t  nxt_conf_vldt_http_members[] = {
 };
 
 
+#if (NXT_HAVE_OTEL)
+
+static nxt_conf_vldt_object_t  nxt_conf_vldt_otel_members[] = {
+    {
+        .name       = nxt_string("endpoint"),
+        .type       = NXT_CONF_VLDT_STRING,
+        .flags      = NXT_CONF_VLDT_REQUIRED
+    }, {
+        .name       = nxt_string("batch_size"),
+        .type       = NXT_CONF_VLDT_INTEGER,
+        .validator  = nxt_otel_validate_batch_size,
+    }, {
+        .name       = nxt_string("protocol"),
+        .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
+};
+
+#endif
+
+
 static nxt_conf_vldt_object_t  nxt_conf_vldt_websocket_members[] = {
     {
         .name       = nxt_string("read_timeout"),
@@ -1465,6 +1512,60 @@ nxt_conf_validate(nxt_conf_validation_t *vldt)
     "a number, a string, an array, or an object"
 
 
+#if (NXT_HAVE_OTEL)
+
+nxt_int_t
+nxt_otel_validate_batch_size(nxt_conf_validation_t *vldt,
+    nxt_conf_value_t *value, void *data)
+{
+    double  batch_size;
+
+    batch_size = nxt_conf_get_number(value);
+    if (batch_size <= 0) {
+        return NXT_ERROR;
+    }
+
+    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,
+    nxt_conf_value_t *value, void *data)
+{
+    nxt_str_t  proto;
+
+    nxt_conf_get_string(value, &proto);
+
+    if (nxt_str_eq(&proto, "HTTP", 4) || nxt_str_eq(&proto, "http", 4)) {
+        return NXT_OK;
+    }
+
+    if (nxt_str_eq(&proto, "GRPC", 4) || nxt_str_eq(&proto, "grpc", 4)) {
+        return NXT_OK;
+    }
+
+    return NXT_ERROR;
+}
+
+#endif
+
+
 static nxt_int_t
 nxt_conf_vldt_type(nxt_conf_validation_t *vldt, const nxt_str_t *name,
     nxt_conf_value_t *value, nxt_conf_vldt_type_t type)
diff --git a/src/nxt_router.c b/src/nxt_router.c
index 076cd134..44ea823b 100644
--- a/src/nxt_router.c
+++ b/src/nxt_router.c
@@ -24,6 +24,11 @@
 
 #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 {
     nxt_str_t         type;
     uint32_t          processes;
@@ -1636,6 +1641,12 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
 #if (NXT_HAVE_NJS)
     nxt_conf_value_t            *js_module;
 #endif
+#if (NXT_HAVE_OTEL)
+    double                      telemetry_sample_fraction, telemetry_batching;
+    nxt_str_t                   telemetry_endpoint, telemetry_proto;
+    nxt_conf_value_t            *otel, *otel_endpoint, *otel_sampling,
+                                *otel_batching, *otel_proto;
+#endif
     nxt_conf_value_t            *root, *conf, *http, *value, *websocket;
     nxt_conf_value_t            *applications, *application, *settings;
     nxt_conf_value_t            *listeners, *listener;
@@ -1671,6 +1682,17 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
                                 nxt_string("/settings/http/websocket");
     static const nxt_str_t  forwarded_path = nxt_string("/forwarded");
     static const nxt_str_t  client_ip_path = nxt_string("/client_ip");
+#if (NXT_HAVE_OTEL)
+    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
 
     root = nxt_conf_json_parse(tmcf->mem_pool, start, end, NULL);
     if (root == NULL) {
@@ -2172,6 +2194,34 @@ nxt_router_conf_create(nxt_task_t *task, nxt_router_temp_conf_t *tmcf,
 
 #endif
 
+#if (NXT_HAVE_OTEL)
+    otel = nxt_conf_get_path(root, &telemetry_path);
+
+    if (otel) {
+        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);
+
+        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();
+    }
+#endif
+
     nxt_queue_add(&deleting_sockets, &router->sockets);
     nxt_queue_init(&router->sockets);