From ff12746ad7d815f6bf3ea3fde148e545378ffa5a Mon Sep 17 00:00:00 2001
From: Pavel Pautov <p.pautov@f5.com>
Date: Mon, 2 Dec 2024 21:07:38 -0800
Subject: [PATCH] Support sending custom headers to export endpoint (fix #62).

The headers are configured by "header" directive in "otel_exporter" block, e.g.
    otel_exporter {
        endpoint localhost:4317;
        header X-API-Token "token value";
    }
---
 src/http_module.cpp          | 36 +++++++++++++++++++++++++++++++++++-
 src/trace_service_client.hpp | 23 ++++++++++++++++++++++-
 2 files changed, 57 insertions(+), 2 deletions(-)

diff --git a/src/http_module.cpp b/src/http_module.cpp
index 93898b1c..ef77ffbd 100644
--- a/src/http_module.cpp
+++ b/src/http_module.cpp
@@ -30,6 +30,7 @@ struct MainConf : MainConfBase {
     std::map<StrView, StrView> resourceAttrs;
     bool ssl;
     std::string trustedCert;
+    Target::HeaderVec headers;
 };
 
 struct SpanAttr {
@@ -49,6 +50,7 @@ char* setExporter(ngx_conf_t* cf, ngx_command_t* cmd, void* conf);
 char* addResourceAttr(ngx_conf_t* cf, ngx_command_t* cmd, void* conf);
 char* addSpanAttr(ngx_conf_t* cf, ngx_command_t* cmd, void* conf);
 char* setTrustedCertificate(ngx_conf_t* cf, ngx_command_t* cmd, void* conf);
+char* addExporterHeader(ngx_conf_t* cf, ngx_command_t* cmd, void* conf);
 
 namespace Propagation {
 
@@ -120,6 +122,10 @@ ngx_command_t gExporterCommands[] = {
       NGX_CONF_TAKE1,
       setTrustedCertificate },
 
+    { ngx_string("header"),
+      NGX_CONF_TAKE2,
+      addExporterHeader },
+
     { ngx_string("interval"),
       NGX_CONF_TAKE1,
       ngx_conf_set_msec_slot,
@@ -580,6 +586,7 @@ ngx_int_t initWorkerProcess(ngx_cycle_t* cycle)
         target.endpoint = std::string(toStrView(mcf->endpoint));
         target.ssl = mcf->ssl;
         target.trustedCert = mcf->trustedCert;
+        target.headers = mcf->headers;
 
         gExporter.reset(new BatchExporter(
             target,
@@ -651,7 +658,7 @@ char* setExporter(ngx_conf_t* cf, ngx_command_t* cmd, void* conf)
                 continue;
             }
 
-            if (cf->args->nelts != 2) {
+            if (cf->args->nelts != static_cast<unsigned>(ffs(cmd->type))) {
                 ngx_conf_log_error(NGX_LOG_EMERG, cf, 0,
                     "invalid number of arguments in \"%V\" "
                     "directive of \"otel_exporter\"", name);
@@ -747,6 +754,33 @@ char* setTrustedCertificate(ngx_conf_t* cf, ngx_command_t* cmd, void* conf)
     return NGX_CONF_OK;
 }
 
+char* addExporterHeader(ngx_conf_t* cf, ngx_command_t* cmd, void* conf)
+{
+    auto args = (ngx_str_t*)cf->args->elts;
+
+    // don't force on users lower case name requirement of gRPC
+    ngx_strlow(args[1].data, args[1].data, args[1].len);
+
+    try {
+        // validate header here to avoid runtime assert failure in gRPC
+        auto name = toStrView(args[1]);
+        if (!Target::validateHeaderName(name)) {
+            return (char*)"has invalid header name";
+        }
+        auto value = toStrView(args[2]);
+        if (!Target::validateHeaderValue(value)) {
+            return (char*)"has invalid header value";
+        }
+
+        getMainConf(cf)->headers.emplace_back(name, value);
+    } catch (const std::exception& e) {
+        ngx_conf_log_error(NGX_LOG_EMERG, cf, 0, "OTel: %s", e.what());
+        return (char*)NGX_CONF_ERROR;
+    }
+
+    return NGX_CONF_OK;
+}
+
 void* createMainConf(ngx_conf_t* cf)
 {
     auto cln = ngx_pool_cleanup_add(cf->pool, sizeof(MainConf));
diff --git a/src/trace_service_client.hpp b/src/trace_service_client.hpp
index 485143c8..4ed92e79 100644
--- a/src/trace_service_client.hpp
+++ b/src/trace_service_client.hpp
@@ -9,9 +9,24 @@
 namespace otel_proto_trace = opentelemetry::proto::collector::trace::v1;
 
 struct Target {
+    typedef std::vector<std::pair<std::string, std::string>> HeaderVec;
+
     std::string endpoint;
     bool ssl;
     std::string trustedCert;
+    HeaderVec headers;
+
+    static bool validateHeaderName(StrView name)
+    {
+        return grpc_header_key_is_legal(
+            grpc_slice_from_static_buffer(name.data(), name.size()));
+    }
+
+    static bool validateHeaderValue(StrView value)
+    {
+        return grpc_header_nonbin_value_is_legal(
+            grpc_slice_from_static_buffer(value.data(), value.size()));
+    }
 };
 
 class TraceServiceClient {
@@ -23,7 +38,7 @@ class TraceServiceClient {
     typedef std::function<void (Request, Response, grpc::Status)>
         ResponseCb;
 
-    TraceServiceClient(const Target& target)
+    TraceServiceClient(const Target& target) : headers(target.headers)
     {
         std::shared_ptr<grpc::ChannelCredentials> creds;
         if (target.ssl) {
@@ -44,6 +59,10 @@ class TraceServiceClient {
     {
         std::unique_ptr<ActiveCall> call{new ActiveCall{}};
 
+        for (auto& header : headers) {
+            call->context.AddMetadata(header.first, header.second);
+        }
+
         call->request = std::move(req);
         call->cb = std::move(cb);
 
@@ -113,6 +132,8 @@ class TraceServiceClient {
         ResponseCb cb;
     };
 
+    Target::HeaderVec headers;
+
     std::unique_ptr<TraceService::Stub> stub;
     grpc::CompletionQueue queue;