From 2d1db7717a75c61136b469e7b8f6048d1e8a3ec7 Mon Sep 17 00:00:00 2001
From: Attila Szakacs <attila.szakacs@axoflow.com>
Date: Fri, 11 Oct 2024 14:45:52 +0200
Subject: [PATCH 01/13] logthrdest: fix leaking worker_partition_key template

Signed-off-by: Attila Szakacs <attila.szakacs@axoflow.com>
---
 lib/logthrdest/logthrdestdrv.c | 1 +
 1 file changed, 1 insertion(+)

diff --git a/lib/logthrdest/logthrdestdrv.c b/lib/logthrdest/logthrdestdrv.c
index 134358422..83a0d15c3 100644
--- a/lib/logthrdest/logthrdestdrv.c
+++ b/lib/logthrdest/logthrdestdrv.c
@@ -1482,6 +1482,7 @@ log_threaded_dest_driver_free(LogPipe *s)
   LogThreadedDestDriver *self = (LogThreadedDestDriver *)s;
 
   g_free(self->workers);
+  log_template_unref(self->worker_partition_key);
   log_dest_driver_free((LogPipe *)self);
 }
 

From 695ef4aab093d10de64bdae948417d66fa2db0e5 Mon Sep 17 00:00:00 2001
From: Attila Szakacs <attila.szakacs@axoflow.com>
Date: Fri, 11 Oct 2024 14:45:13 +0200
Subject: [PATCH 02/13] grpc/metrics: fix leaking stats cluster key

Signed-off-by: Attila Szakacs <attila.szakacs@axoflow.com>
---
 modules/grpc/common/metrics/grpc-metrics.cpp | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/modules/grpc/common/metrics/grpc-metrics.cpp b/modules/grpc/common/metrics/grpc-metrics.cpp
index 4085032f4..2220a20c4 100644
--- a/modules/grpc/common/metrics/grpc-metrics.cpp
+++ b/modules/grpc/common/metrics/grpc-metrics.cpp
@@ -99,6 +99,8 @@ DestDriverMetrics::create_grpc_request_cluster(::grpc::StatusCode response_code)
 
     StatsCounterItem *counter;
     cluster = stats_register_counter(stats_level, sc_key, SC_TYPE_SINGLE_VALUE, &counter);
+
+    stats_cluster_key_free(sc_key);
   }
   stats_cluster_key_builder_pop(kb);
 

From 61d21347560087b9522a7f9898732870371ac9d6 Mon Sep 17 00:00:00 2001
From: Attila Szakacs <attila.szakacs@axoflow.com>
Date: Fri, 11 Oct 2024 14:44:35 +0200
Subject: [PATCH 03/13] grpc: add missing worker super deinit()

Signed-off-by: Attila Szakacs <attila.szakacs@axoflow.com>
---
 modules/grpc/common/grpc-dest-worker.cpp | 1 +
 1 file changed, 1 insertion(+)

diff --git a/modules/grpc/common/grpc-dest-worker.cpp b/modules/grpc/common/grpc-dest-worker.cpp
index d11652b97..52189d94b 100644
--- a/modules/grpc/common/grpc-dest-worker.cpp
+++ b/modules/grpc/common/grpc-dest-worker.cpp
@@ -78,6 +78,7 @@ DestWorker::init()
 void
 DestWorker::deinit()
 {
+  log_threaded_dest_worker_deinit_method(&super->super);
 }
 
 bool

From e246b5f84ca522d5c8cc92fd0f4ab19ae018ff06 Mon Sep 17 00:00:00 2001
From: Attila Szakacs <attila.szakacs@axoflow.com>
Date: Thu, 10 Oct 2024 22:23:48 +0200
Subject: [PATCH 04/13] grpc/loki: flush on worker-partition-key change

Out intention is to batch messages that correspond
to the same label set. If between two messages the
label set changes, but they arrive to the same
worker, we should flush the previous batch.

Signed-off-by: Attila Szakacs <attila.szakacs@axoflow.com>
---
 modules/grpc/loki/loki-dest.cpp | 9 +++++++--
 1 file changed, 7 insertions(+), 2 deletions(-)

diff --git a/modules/grpc/loki/loki-dest.cpp b/modules/grpc/loki/loki-dest.cpp
index ac6dff79a..38fa21433 100644
--- a/modules/grpc/loki/loki-dest.cpp
+++ b/modules/grpc/loki/loki-dest.cpp
@@ -93,9 +93,14 @@ DestinationDriver::init()
     }
 
   if (log_template_is_literal_string(worker_partition_key))
-    log_template_unref(worker_partition_key);
+    {
+      log_template_unref(worker_partition_key);
+    }
   else
-    log_threaded_dest_driver_set_worker_partition_key_ref(&this->super->super.super.super, worker_partition_key);
+    {
+      log_threaded_dest_driver_set_worker_partition_key_ref(&this->super->super.super.super, worker_partition_key);
+      log_threaded_dest_driver_set_flush_on_worker_key_change(&this->super->super.super.super, TRUE);
+    }
 
   return syslogng::grpc::DestDriver::init();
 }

From fc7423de1a45a2fe7837df37c0230ec3eb5086c1 Mon Sep 17 00:00:00 2001
From: Attila Szakacs <attila.szakacs@axoflow.com>
Date: Thu, 10 Oct 2024 22:52:05 +0200
Subject: [PATCH 05/13] grpc: add worker_partition_key related functions to
 base class

Signed-off-by: Attila Szakacs <attila.szakacs@axoflow.com>
---
 modules/grpc/common/grpc-dest.cpp | 33 ++++++++++++++++++++++++++++++-
 modules/grpc/common/grpc-dest.hpp | 14 +++++++++++++
 modules/grpc/loki/loki-dest.cpp   | 31 ++---------------------------
 3 files changed, 48 insertions(+), 30 deletions(-)

diff --git a/modules/grpc/common/grpc-dest.cpp b/modules/grpc/common/grpc-dest.cpp
index 587b9f9a5..560500f2b 100644
--- a/modules/grpc/common/grpc-dest.cpp
+++ b/modules/grpc/common/grpc-dest.cpp
@@ -31,11 +31,39 @@ using namespace syslogng::grpc;
 
 DestDriver::DestDriver(GrpcDestDriver *s)
   : super(s), compression(false), batch_bytes(4 * 1000 * 1000),
-    keepalive_time(-1), keepalive_timeout(-1), keepalive_max_pings_without_data(-1)
+    keepalive_time(-1), keepalive_timeout(-1), keepalive_max_pings_without_data(-1),
+    flush_on_key_change(false)
 {
   credentials_builder_wrapper.self = &credentials_builder;
 }
 
+bool
+DestDriver::set_worker_partition_key()
+{
+  GlobalConfig *cfg = log_pipe_get_config(&this->super->super.super.super.super);
+
+  LogTemplate *worker_partition_key_tpl = log_template_new(cfg, NULL);
+  if (!log_template_compile(worker_partition_key_tpl, this->worker_partition_key.str().c_str(), NULL))
+    {
+      msg_error("Error compiling worker partition key template",
+                evt_tag_str("template", this->worker_partition_key.str().c_str()));
+      return false;
+    }
+
+  if (log_template_is_literal_string(worker_partition_key_tpl))
+    {
+      log_template_unref(worker_partition_key_tpl);
+    }
+  else
+    {
+      log_threaded_dest_driver_set_worker_partition_key_ref(&this->super->super.super.super, worker_partition_key_tpl);
+      log_threaded_dest_driver_set_flush_on_worker_key_change(&this->super->super.super.super,
+                                                              this->flush_on_key_change);
+    }
+
+  return true;
+}
+
 bool
 DestDriver::init()
 {
@@ -51,6 +79,9 @@ DestDriver::init()
       return false;
     }
 
+  if (this->worker_partition_key.rdbuf()->in_avail() && !this->set_worker_partition_key())
+    return false;
+
   if (!log_threaded_dest_driver_init_method(&this->super->super.super.super.super))
     return false;
 
diff --git a/modules/grpc/common/grpc-dest.hpp b/modules/grpc/common/grpc-dest.hpp
index 4667269aa..261ccf27c 100644
--- a/modules/grpc/common/grpc-dest.hpp
+++ b/modules/grpc/common/grpc-dest.hpp
@@ -37,6 +37,7 @@
 #include <grpcpp/server.h>
 
 #include <list>
+#include <sstream>
 
 namespace syslogng {
 namespace grpc {
@@ -118,11 +119,21 @@ class DestDriver
     this->headers.push_back(std::make_pair(name, value));
   }
 
+  void extend_worker_partition_key(const std::string &extension)
+  {
+    if (this->worker_partition_key.rdbuf()->in_avail())
+      this->worker_partition_key << ",";
+    this->worker_partition_key << extension;
+  }
+
   GrpcClientCredentialsBuilderW *get_credentials_builder_wrapper()
   {
     return &this->credentials_builder_wrapper;
   }
 
+private:
+  bool set_worker_partition_key();
+
 public:
   GrpcDestDriver *super;
   DestDriverMetrics metrics;
@@ -139,6 +150,9 @@ class DestDriver
   int keepalive_timeout;
   int keepalive_max_pings_without_data;
 
+  std::stringstream worker_partition_key;
+  bool flush_on_key_change;
+
   std::list<std::pair<std::string, long>> int_extra_channel_args;
   std::list<std::pair<std::string, std::string>> string_extra_channel_args;
 
diff --git a/modules/grpc/loki/loki-dest.cpp b/modules/grpc/loki/loki-dest.cpp
index 38fa21433..cb48086db 100644
--- a/modules/grpc/loki/loki-dest.cpp
+++ b/modules/grpc/loki/loki-dest.cpp
@@ -43,6 +43,7 @@ DestinationDriver::DestinationDriver(GrpcDestDriver *s)
   : syslogng::grpc::DestDriver(s), timestamp(LM_TS_PROCESSED)
 {
   this->url = "localhost:9095";
+  this->flush_on_key_change = true;
   log_template_options_defaults(&this->template_options);
 }
 
@@ -71,36 +72,8 @@ DestinationDriver::init()
 
   log_template_options_init(&this->template_options, cfg);
 
-  LogTemplate *worker_partition_key = log_template_new(cfg, NULL);
-
-  std::stringstream template_str;
-  bool comma_needed = false;
   for (const auto &label : this->labels)
-    {
-      if (comma_needed)
-        template_str << ",";
-      template_str << label.name << "=" << label.value->template_str;
-
-      comma_needed = true;
-    }
-
-  std::string worker_partition_key_str = template_str.str();
-  if (!log_template_compile(worker_partition_key, worker_partition_key_str.c_str(), NULL))
-    {
-      msg_error("Error compiling worker partition key template",
-                evt_tag_str("template", worker_partition_key_str.c_str()));
-      return false;
-    }
-
-  if (log_template_is_literal_string(worker_partition_key))
-    {
-      log_template_unref(worker_partition_key);
-    }
-  else
-    {
-      log_threaded_dest_driver_set_worker_partition_key_ref(&this->super->super.super.super, worker_partition_key);
-      log_threaded_dest_driver_set_flush_on_worker_key_change(&this->super->super.super.super, TRUE);
-    }
+    this->extend_worker_partition_key(label.name + "=" + label.value->template_str);
 
   return syslogng::grpc::DestDriver::init();
 }

From ce6b2d9ef87cf59e11554df460b314bb1d83e07a Mon Sep 17 00:00:00 2001
From: Attila Szakacs <attila.szakacs@axoflow.com>
Date: Fri, 11 Oct 2024 11:02:47 +0200
Subject: [PATCH 06/13] grpc: factor out loki::Label class as
 grpc::NameValueTemplatePair

Will be useful for templated headers as well.

Signed-off-by: Attila Szakacs <attila.szakacs@axoflow.com>
---
 modules/grpc/common/grpc-dest.hpp | 27 +++++++++++++++++++++++++++
 modules/grpc/loki/loki-dest.cpp   |  2 +-
 modules/grpc/loki/loki-dest.hpp   | 29 +----------------------------
 3 files changed, 29 insertions(+), 29 deletions(-)

diff --git a/modules/grpc/common/grpc-dest.hpp b/modules/grpc/common/grpc-dest.hpp
index 261ccf27c..7f42f788f 100644
--- a/modules/grpc/common/grpc-dest.hpp
+++ b/modules/grpc/common/grpc-dest.hpp
@@ -42,6 +42,33 @@
 namespace syslogng {
 namespace grpc {
 
+struct NameValueTemplatePair
+{
+  std::string name;
+  LogTemplate *value;
+
+  NameValueTemplatePair(std::string name_, LogTemplate *value_)
+    : name(name_), value(log_template_ref(value_)) {}
+
+  NameValueTemplatePair(const NameValueTemplatePair &a)
+    : name(a.name), value(log_template_ref(a.value)) {}
+
+  NameValueTemplatePair &operator=(const NameValueTemplatePair &a)
+  {
+    name = a.name;
+    log_template_unref(value);
+    value = log_template_ref(a.value);
+
+    return *this;
+  }
+
+  ~NameValueTemplatePair()
+  {
+    log_template_unref(value);
+  }
+
+};
+
 class DestDriver
 {
 public:
diff --git a/modules/grpc/loki/loki-dest.cpp b/modules/grpc/loki/loki-dest.cpp
index cb48086db..8fe0601f9 100644
--- a/modules/grpc/loki/loki-dest.cpp
+++ b/modules/grpc/loki/loki-dest.cpp
@@ -56,7 +56,7 @@ DestinationDriver::~DestinationDriver()
 void
 DestinationDriver::add_label(std::string name, LogTemplate *value)
 {
-  this->labels.push_back(Label{name, value});
+  this->labels.push_back(NameValueTemplatePair{name, value});
 }
 
 bool
diff --git a/modules/grpc/loki/loki-dest.hpp b/modules/grpc/loki/loki-dest.hpp
index 8b4f70d44..5b56ea6f5 100644
--- a/modules/grpc/loki/loki-dest.hpp
+++ b/modules/grpc/loki/loki-dest.hpp
@@ -43,33 +43,6 @@ namespace syslogng {
 namespace grpc {
 namespace loki {
 
-struct Label
-{
-  std::string name;
-  LogTemplate *value;
-
-  Label(std::string name_, LogTemplate *value_)
-    : name(name_), value(log_template_ref(value_)) {}
-
-  Label(const Label &a)
-    : name(a.name), value(log_template_ref(a.value)) {}
-
-  Label &operator=(const Label &a)
-  {
-    name = a.name;
-    log_template_unref(value);
-    value = log_template_ref(a.value);
-
-    return *this;
-  }
-
-  ~Label()
-  {
-    log_template_unref(value);
-  }
-
-};
-
 class DestinationDriver final : public syslogng::grpc::DestDriver
 {
 public:
@@ -118,7 +91,7 @@ class DestinationDriver final : public syslogng::grpc::DestDriver
   std::string tenant_id;
 
   LogTemplate *message = nullptr;
-  std::vector<Label> labels;
+  std::vector<NameValueTemplatePair> labels;
   LogMessageTimeStamp timestamp;
 };
 

From 9dff07f6a390d63119ef4745a9707551418a2dfd Mon Sep 17 00:00:00 2001
From: Attila Szakacs <attila.szakacs@axoflow.com>
Date: Fri, 11 Oct 2024 14:53:51 +0200
Subject: [PATCH 07/13] grpc/bigquery: use NameValueTemplatePair in Field

Signed-off-by: Attila Szakacs <attila.szakacs@axoflow.com>
---
 modules/grpc/bigquery/bigquery-dest.cpp   |  2 +-
 modules/grpc/bigquery/bigquery-dest.hpp   | 16 ++++------------
 modules/grpc/bigquery/bigquery-worker.cpp |  4 ++--
 3 files changed, 7 insertions(+), 15 deletions(-)

diff --git a/modules/grpc/bigquery/bigquery-dest.cpp b/modules/grpc/bigquery/bigquery-dest.cpp
index f8033ac0f..b4dae2e18 100644
--- a/modules/grpc/bigquery/bigquery-dest.cpp
+++ b/modules/grpc/bigquery/bigquery-dest.cpp
@@ -246,7 +246,7 @@ DestinationDriver::construct_schema_prototype()
   for (auto &field : this->fields)
     {
       google::protobuf::FieldDescriptorProto *field_desc_proto = descriptor_proto->add_field();
-      field_desc_proto->set_name(field.name);
+      field_desc_proto->set_name(field.nv.name);
       field_desc_proto->set_type(field.type);
       field_desc_proto->set_number(num++);
     }
diff --git a/modules/grpc/bigquery/bigquery-dest.hpp b/modules/grpc/bigquery/bigquery-dest.hpp
index a6e3f57c8..2286ca6fe 100644
--- a/modules/grpc/bigquery/bigquery-dest.hpp
+++ b/modules/grpc/bigquery/bigquery-dest.hpp
@@ -49,33 +49,25 @@ namespace bigquery {
 
 struct Field
 {
-  std::string name;
+  NameValueTemplatePair nv;
   google::protobuf::FieldDescriptorProto::Type type;
-  LogTemplate *value;
   const google::protobuf::FieldDescriptor *field_desc;
 
   Field(std::string name_, google::protobuf::FieldDescriptorProto::Type type_, LogTemplate *value_)
-    : name(name_), type(type_), value(log_template_ref(value_)), field_desc(nullptr) {}
+    : nv(name_, value_), type(type_), field_desc(nullptr) {}
 
   Field(const Field &a)
-    : name(a.name), type(a.type), value(log_template_ref(a.value)), field_desc(a.field_desc) {}
+    : nv(a.nv), type(a.type), field_desc(a.field_desc) {}
 
   Field &operator=(const Field &a)
   {
-    name = a.name;
+    nv = a.nv;
     type = a.type;
-    log_template_unref(value);
-    value = log_template_ref(a.value);
     field_desc = a.field_desc;
 
     return *this;
   }
 
-  ~Field()
-  {
-    log_template_unref(value);
-  }
-
 };
 
 class DestinationDriver final : public syslogng::grpc::DestDriver
diff --git a/modules/grpc/bigquery/bigquery-worker.cpp b/modules/grpc/bigquery/bigquery-worker.cpp
index bc908ccf2..c2703ca28 100644
--- a/modules/grpc/bigquery/bigquery-worker.cpp
+++ b/modules/grpc/bigquery/bigquery-worker.cpp
@@ -180,13 +180,13 @@ DestinationWorker::insert_field(const google::protobuf::Reflection *reflection,
 
   LogMessageValueType type;
 
-  Slice value = this->format_template(field.value, msg, buf, &type);
+  Slice value = this->format_template(field.nv.value, msg, buf, &type);
 
   if (type == LM_VT_NULL)
     {
       if (field.field_desc->is_required())
         {
-          msg_error("Missing required field", evt_tag_str("field", field.name.c_str()));
+          msg_error("Missing required field", evt_tag_str("field", field.nv.name.c_str()));
           goto error;
         }
 

From 090118e6899556e8181d6bce1efb3893e7949b96 Mon Sep 17 00:00:00 2001
From: Attila Szakacs <attila.szakacs@axoflow.com>
Date: Fri, 11 Oct 2024 11:26:44 +0200
Subject: [PATCH 08/13] grpc: support template_options for all destinations

This will be needed for the dynamic headers.

Signed-off-by: Attila Szakacs <attila.szakacs@axoflow.com>
---
 modules/grpc/bigquery/bigquery-dest.cpp   | 13 -------------
 modules/grpc/bigquery/bigquery-dest.h     |  2 --
 modules/grpc/bigquery/bigquery-dest.hpp   |  7 -------
 modules/grpc/bigquery/bigquery-grammar.ym |  1 -
 modules/grpc/common/grpc-dest.cpp         | 17 +++++++++++++++++
 modules/grpc/common/grpc-dest.h           |  1 +
 modules/grpc/common/grpc-dest.hpp         |  9 ++++++++-
 modules/grpc/common/grpc-grammar.ym       |  1 +
 modules/grpc/loki/loki-dest.cpp           | 12 ------------
 modules/grpc/loki/loki-dest.h             |  2 --
 modules/grpc/loki/loki-dest.hpp           |  6 ------
 modules/grpc/loki/loki-grammar.ym         |  1 -
 12 files changed, 27 insertions(+), 45 deletions(-)

diff --git a/modules/grpc/bigquery/bigquery-dest.cpp b/modules/grpc/bigquery/bigquery-dest.cpp
index b4dae2e18..e329449a3 100644
--- a/modules/grpc/bigquery/bigquery-dest.cpp
+++ b/modules/grpc/bigquery/bigquery-dest.cpp
@@ -92,13 +92,11 @@ DestinationDriver::DestinationDriver(GrpcDestDriver *s)
 {
   this->url = "bigquerystorage.googleapis.com";
   this->credentials_builder.set_mode(GCAM_ADC);
-  log_template_options_defaults(&this->template_options);
 }
 
 DestinationDriver::~DestinationDriver()
 {
   g_list_free_full(this->protobuf_schema.values, _template_unref);
-  log_template_options_destroy(&this->template_options);
 }
 
 bool
@@ -166,9 +164,6 @@ DestinationDriver::init()
       return false;
     }
 
-  GlobalConfig *cfg = log_pipe_get_config(&this->super->super.super.super.super);
-  log_template_options_init(&this->template_options, cfg);
-
   if (this->protobuf_schema.proto_path.empty())
     this->construct_schema_prototype();
   else
@@ -376,14 +371,6 @@ bigquery_dd_set_protobuf_schema(LogDriver *d, const gchar *proto_path, GList *va
   cpp->set_protobuf_schema(proto_path, values);
 }
 
-LogTemplateOptions *
-bigquery_dd_get_template_options(LogDriver *d)
-{
-  GrpcDestDriver *self = (GrpcDestDriver *) d;
-  DestinationDriver *cpp = bigquery_dd_get_cpp(self);
-  return &cpp->get_template_options();
-}
-
 LogDriver *
 bigquery_dd_new(GlobalConfig *cfg)
 {
diff --git a/modules/grpc/bigquery/bigquery-dest.h b/modules/grpc/bigquery/bigquery-dest.h
index a63490493..77af41f5b 100644
--- a/modules/grpc/bigquery/bigquery-dest.h
+++ b/modules/grpc/bigquery/bigquery-dest.h
@@ -40,8 +40,6 @@ void bigquery_dd_set_table(LogDriver *d, const gchar *table);
 gboolean bigquery_dd_add_field(LogDriver *d, const gchar *name, const gchar *type, LogTemplate *value);
 void bigquery_dd_set_protobuf_schema(LogDriver *d, const gchar *proto_path, GList *values);
 
-LogTemplateOptions *bigquery_dd_get_template_options(LogDriver *d);
-
 #include "compat/cpp-end.h"
 
 #endif
diff --git a/modules/grpc/bigquery/bigquery-dest.hpp b/modules/grpc/bigquery/bigquery-dest.hpp
index 2286ca6fe..d60fd4263 100644
--- a/modules/grpc/bigquery/bigquery-dest.hpp
+++ b/modules/grpc/bigquery/bigquery-dest.hpp
@@ -83,11 +83,6 @@ class DestinationDriver final : public syslogng::grpc::DestDriver
   bool add_field(std::string name, std::string type, LogTemplate *value);
   void set_protobuf_schema(std::string proto_path, GList *values);
 
-  LogTemplateOptions &get_template_options()
-  {
-    return this->template_options;
-  }
-
   void set_project(std::string p)
   {
     this->project = p;
@@ -124,8 +119,6 @@ class DestinationDriver final : public syslogng::grpc::DestDriver
   bool load_protobuf_schema();
 
 private:
-  LogTemplateOptions template_options;
-
   std::string project;
   std::string dataset;
   std::string table;
diff --git a/modules/grpc/bigquery/bigquery-grammar.ym b/modules/grpc/bigquery/bigquery-grammar.ym
index 826172142..cef67cc11 100644
--- a/modules/grpc/bigquery/bigquery-grammar.ym
+++ b/modules/grpc/bigquery/bigquery-grammar.ym
@@ -90,7 +90,6 @@ bigquery_dest_option
       free($3);
     }
   | grpc_dest_option
-  | { last_template_options = bigquery_dd_get_template_options(last_driver); } template_option
   ;
 
 bigquery_schema_fields
diff --git a/modules/grpc/common/grpc-dest.cpp b/modules/grpc/common/grpc-dest.cpp
index 560500f2b..413e088e1 100644
--- a/modules/grpc/common/grpc-dest.cpp
+++ b/modules/grpc/common/grpc-dest.cpp
@@ -34,9 +34,15 @@ DestDriver::DestDriver(GrpcDestDriver *s)
     keepalive_time(-1), keepalive_timeout(-1), keepalive_max_pings_without_data(-1),
     flush_on_key_change(false)
 {
+  log_template_options_defaults(&this->template_options);
   credentials_builder_wrapper.self = &credentials_builder;
 }
 
+DestDriver::~DestDriver()
+{
+  log_template_options_destroy(&this->template_options);
+}
+
 bool
 DestDriver::set_worker_partition_key()
 {
@@ -67,6 +73,8 @@ DestDriver::set_worker_partition_key()
 bool
 DestDriver::init()
 {
+  GlobalConfig *cfg = log_pipe_get_config(&this->super->super.super.super.super);
+
   if (url.length() == 0)
     {
       msg_error("url() option is mandatory",
@@ -82,6 +90,8 @@ DestDriver::init()
   if (this->worker_partition_key.rdbuf()->in_avail() && !this->set_worker_partition_key())
     return false;
 
+  log_template_options_init(&this->template_options, cfg);
+
   if (!log_threaded_dest_driver_init_method(&this->super->super.super.super.super))
     return false;
 
@@ -166,6 +176,13 @@ grpc_dd_add_header(LogDriver *s, const gchar *name, const gchar *value)
   self->cpp->add_header(name, value);
 }
 
+LogTemplateOptions *
+grpc_dd_get_template_options(LogDriver *d)
+{
+  GrpcDestDriver *self = (GrpcDestDriver *) d;
+  return &self->cpp->get_template_options();
+}
+
 GrpcClientCredentialsBuilderW *
 grpc_dd_get_credentials_builder(LogDriver *s)
 {
diff --git a/modules/grpc/common/grpc-dest.h b/modules/grpc/common/grpc-dest.h
index 9c051cb66..eac697454 100644
--- a/modules/grpc/common/grpc-dest.h
+++ b/modules/grpc/common/grpc-dest.h
@@ -43,6 +43,7 @@ void grpc_dd_set_keepalive_max_pings(LogDriver *s, gint p);
 void grpc_dd_add_int_channel_arg(LogDriver *s, const gchar *name, glong value);
 void grpc_dd_add_string_channel_arg(LogDriver *s, const gchar *name, const gchar *value);
 void grpc_dd_add_header(LogDriver *s, const gchar *name, const gchar *value);
+LogTemplateOptions *grpc_dd_get_template_options(LogDriver *d);
 GrpcClientCredentialsBuilderW *grpc_dd_get_credentials_builder(LogDriver *s);
 
 #include "compat/cpp-end.h"
diff --git a/modules/grpc/common/grpc-dest.hpp b/modules/grpc/common/grpc-dest.hpp
index 7f42f788f..7bee02047 100644
--- a/modules/grpc/common/grpc-dest.hpp
+++ b/modules/grpc/common/grpc-dest.hpp
@@ -73,7 +73,7 @@ class DestDriver
 {
 public:
   DestDriver(GrpcDestDriver *s);
-  virtual ~DestDriver() {};
+  virtual ~DestDriver();
 
   virtual bool init();
   virtual bool deinit();
@@ -153,6 +153,11 @@ class DestDriver
     this->worker_partition_key << extension;
   }
 
+  LogTemplateOptions &get_template_options()
+  {
+    return this->template_options;
+  }
+
   GrpcClientCredentialsBuilderW *get_credentials_builder_wrapper()
   {
     return &this->credentials_builder_wrapper;
@@ -185,6 +190,8 @@ class DestDriver
 
   std::list<std::pair<std::string, std::string>> headers;
 
+  LogTemplateOptions template_options;
+
   GrpcClientCredentialsBuilderW credentials_builder_wrapper;
 };
 
diff --git a/modules/grpc/common/grpc-grammar.ym b/modules/grpc/common/grpc-grammar.ym
index 8d2511ee7..1e50a610a 100644
--- a/modules/grpc/common/grpc-grammar.ym
+++ b/modules/grpc/common/grpc-grammar.ym
@@ -101,6 +101,7 @@ grpc_dest_option
   | threaded_dest_driver_general_option
   | threaded_dest_driver_batch_option
   | threaded_dest_driver_workers_option
+  | { last_template_options = grpc_dd_get_template_options(last_driver); } template_option
   ;
 
 grpc_keepalive_options
diff --git a/modules/grpc/loki/loki-dest.cpp b/modules/grpc/loki/loki-dest.cpp
index 8fe0601f9..4739ff44c 100644
--- a/modules/grpc/loki/loki-dest.cpp
+++ b/modules/grpc/loki/loki-dest.cpp
@@ -44,12 +44,10 @@ DestinationDriver::DestinationDriver(GrpcDestDriver *s)
 {
   this->url = "localhost:9095";
   this->flush_on_key_change = true;
-  log_template_options_defaults(&this->template_options);
 }
 
 DestinationDriver::~DestinationDriver()
 {
-  log_template_options_destroy(&this->template_options);
   log_template_unref(this->message);
 }
 
@@ -70,8 +68,6 @@ DestinationDriver::init()
       log_template_compile(this->message, DEFAULT_MESSAGE_TEMPLATE, NULL);
     }
 
-  log_template_options_init(&this->template_options, cfg);
-
   for (const auto &label : this->labels)
     this->extend_worker_partition_key(label.name + "=" + label.value->template_str);
 
@@ -149,14 +145,6 @@ loki_dd_set_tenant_id(LogDriver *d, const gchar *tid)
   return cpp->set_tenant_id(tid);
 }
 
-LogTemplateOptions *
-loki_dd_get_template_options(LogDriver *d)
-{
-  GrpcDestDriver *self = (GrpcDestDriver *) d;
-  DestinationDriver *cpp = loki_dd_get_cpp(self);
-  return &cpp->get_template_options();
-}
-
 LogDriver *
 loki_dd_new(GlobalConfig *cfg)
 {
diff --git a/modules/grpc/loki/loki-dest.h b/modules/grpc/loki/loki-dest.h
index 2c9773e09..af38b14b0 100644
--- a/modules/grpc/loki/loki-dest.h
+++ b/modules/grpc/loki/loki-dest.h
@@ -40,8 +40,6 @@ void loki_dd_add_label(LogDriver *d, const gchar *name, LogTemplate *value);
 gboolean loki_dd_set_timestamp(LogDriver *d, const gchar *t);
 void loki_dd_set_tenant_id(LogDriver *d, const gchar *tid);
 
-LogTemplateOptions *loki_dd_get_template_options(LogDriver *d);
-
 #include "compat/cpp-end.h"
 
 #endif
diff --git a/modules/grpc/loki/loki-dest.hpp b/modules/grpc/loki/loki-dest.hpp
index 5b56ea6f5..0d7b949d0 100644
--- a/modules/grpc/loki/loki-dest.hpp
+++ b/modules/grpc/loki/loki-dest.hpp
@@ -55,11 +55,6 @@ class DestinationDriver final : public syslogng::grpc::DestDriver
 
   void add_label(std::string name, LogTemplate *value);
 
-  LogTemplateOptions &get_template_options()
-  {
-    return this->template_options;
-  }
-
   void set_message_template_ref(LogTemplate *msg)
   {
     log_template_unref(this->message);
@@ -86,7 +81,6 @@ class DestinationDriver final : public syslogng::grpc::DestDriver
 
 private:
   friend class DestinationWorker;
-  LogTemplateOptions template_options;
 
   std::string tenant_id;
 
diff --git a/modules/grpc/loki/loki-grammar.ym b/modules/grpc/loki/loki-grammar.ym
index 85658ba2f..baa87a1f0 100644
--- a/modules/grpc/loki/loki-grammar.ym
+++ b/modules/grpc/loki/loki-grammar.ym
@@ -85,7 +85,6 @@ loki_dest_option
     }
   | KW_TEMPLATE '(' template_name_or_content ')' { loki_dd_set_message_template_ref(last_driver, $3); }
   | grpc_dest_option
-  | { last_template_options = loki_dd_get_template_options(last_driver); } template_option
   ;
 
 loki_labels

From 1237e1808040d3f22a492b445fd1f94598cedfb0 Mon Sep 17 00:00:00 2001
From: Attila Szakacs <attila.szakacs@axoflow.com>
Date: Fri, 11 Oct 2024 13:44:10 +0200
Subject: [PATCH 09/13] grpc: add support for templated header() values

It must be enabled in the derived class, and
prepare_context_dynamic() must be used if so.

This is not part of this commit, enabling it will be
done in subsequent commits.

Signed-off-by: Attila Szakacs <attila.szakacs@axoflow.com>
---
 modules/grpc/common/grpc-dest-worker.cpp | 35 +++++++++++++++++++++++-
 modules/grpc/common/grpc-dest-worker.hpp |  1 +
 modules/grpc/common/grpc-dest.cpp        |  8 +++---
 modules/grpc/common/grpc-dest.h          |  2 +-
 modules/grpc/common/grpc-dest.hpp        | 23 ++++++++++++++--
 modules/grpc/common/grpc-grammar.ym      |  7 ++++-
 6 files changed, 66 insertions(+), 10 deletions(-)

diff --git a/modules/grpc/common/grpc-dest-worker.cpp b/modules/grpc/common/grpc-dest-worker.cpp
index 52189d94b..d4f46d4a1 100644
--- a/modules/grpc/common/grpc-dest-worker.cpp
+++ b/modules/grpc/common/grpc-dest-worker.cpp
@@ -28,6 +28,10 @@
 
 #include "grpc-dest-worker.hpp"
 
+#include "compat/cpp-start.h"
+#include "scratch-buffers.h"
+#include "compat/cpp-end.h"
+
 using namespace syslogng::grpc;
 
 /* C++ Implementations */
@@ -95,8 +99,37 @@ DestWorker::disconnect()
 void
 DestWorker::prepare_context(::grpc::ClientContext &context)
 {
+  g_assert(!this->owner.dynamic_headers_enabled);
+
+  for (auto nv : owner.headers)
+    context.AddMetadata(nv.name, nv.value->template_str);
+}
+
+void
+DestWorker::prepare_context_dynamic(::grpc::ClientContext &context, LogMessage *msg)
+{
+  g_assert(this->owner.dynamic_headers_enabled);
+
+  LogTemplateEvalOptions options = {&this->owner.template_options, LTZ_SEND, this->super->super.seq_num, NULL,
+                                    LM_VT_STRING
+                                   };
+
+  ScratchBuffersMarker marker;
+  GString *buf = scratch_buffers_alloc_and_mark(&marker);
+
   for (auto nv : owner.headers)
-    context.AddMetadata(nv.first, nv.second);
+    {
+      if (log_template_is_literal_string(nv.value))
+        {
+          context.AddMetadata(nv.name, log_template_get_literal_value(nv.value, NULL));
+          continue;
+        }
+
+      log_template_format(nv.value, msg, &options, buf);
+      context.AddMetadata(nv.name, buf->str);
+    }
+
+  scratch_buffers_reclaim_marked(marker);
 }
 
 /* C Wrappers */
diff --git a/modules/grpc/common/grpc-dest-worker.hpp b/modules/grpc/common/grpc-dest-worker.hpp
index ea4ee43e6..169f72d59 100644
--- a/modules/grpc/common/grpc-dest-worker.hpp
+++ b/modules/grpc/common/grpc-dest-worker.hpp
@@ -51,6 +51,7 @@ class DestWorker
 
 protected:
   void prepare_context(::grpc::ClientContext &context);
+  void prepare_context_dynamic(::grpc::ClientContext &context, LogMessage *msg);
   std::shared_ptr<::grpc::ChannelCredentials> create_credentials();
   ::grpc::ChannelArguments create_channel_args();
 
diff --git a/modules/grpc/common/grpc-dest.cpp b/modules/grpc/common/grpc-dest.cpp
index 413e088e1..7887f25d9 100644
--- a/modules/grpc/common/grpc-dest.cpp
+++ b/modules/grpc/common/grpc-dest.cpp
@@ -32,7 +32,7 @@ using namespace syslogng::grpc;
 DestDriver::DestDriver(GrpcDestDriver *s)
   : super(s), compression(false), batch_bytes(4 * 1000 * 1000),
     keepalive_time(-1), keepalive_timeout(-1), keepalive_max_pings_without_data(-1),
-    flush_on_key_change(false)
+    flush_on_key_change(false), dynamic_headers_enabled(false)
 {
   log_template_options_defaults(&this->template_options);
   credentials_builder_wrapper.self = &credentials_builder;
@@ -169,11 +169,11 @@ grpc_dd_add_string_channel_arg(LogDriver *s, const gchar *name, const gchar *val
   self->cpp->add_extra_channel_arg(name, value);
 }
 
-void
-grpc_dd_add_header(LogDriver *s, const gchar *name, const gchar *value)
+gboolean
+grpc_dd_add_header(LogDriver *s, const gchar *name, LogTemplate *value)
 {
   GrpcDestDriver *self = (GrpcDestDriver *) s;
-  self->cpp->add_header(name, value);
+  return self->cpp->add_header(name, value);
 }
 
 LogTemplateOptions *
diff --git a/modules/grpc/common/grpc-dest.h b/modules/grpc/common/grpc-dest.h
index eac697454..a1aaf791e 100644
--- a/modules/grpc/common/grpc-dest.h
+++ b/modules/grpc/common/grpc-dest.h
@@ -42,7 +42,7 @@ void grpc_dd_set_keepalive_timeout(LogDriver *s, gint t);
 void grpc_dd_set_keepalive_max_pings(LogDriver *s, gint p);
 void grpc_dd_add_int_channel_arg(LogDriver *s, const gchar *name, glong value);
 void grpc_dd_add_string_channel_arg(LogDriver *s, const gchar *name, const gchar *value);
-void grpc_dd_add_header(LogDriver *s, const gchar *name, const gchar *value);
+gboolean grpc_dd_add_header(LogDriver *s, const gchar *name, LogTemplate *value);
 LogTemplateOptions *grpc_dd_get_template_options(LogDriver *d);
 GrpcClientCredentialsBuilderW *grpc_dd_get_credentials_builder(LogDriver *s);
 
diff --git a/modules/grpc/common/grpc-dest.hpp b/modules/grpc/common/grpc-dest.hpp
index 7bee02047..83242b62a 100644
--- a/modules/grpc/common/grpc-dest.hpp
+++ b/modules/grpc/common/grpc-dest.hpp
@@ -136,14 +136,24 @@ class DestDriver
     this->string_extra_channel_args.push_back(std::make_pair(name, value));
   }
 
-  void add_header(std::string name, std::string value)
+  bool add_header(std::string name, LogTemplate *value)
   {
+    bool is_literal_string = log_template_is_literal_string(value);
+
+    if (!this->dynamic_headers_enabled && !is_literal_string)
+      return false;
+
     std::transform(name.begin(), name.end(), name.begin(),
                    [](auto c)
     {
       return ::tolower(c);
     });
-    this->headers.push_back(std::make_pair(name, value));
+    this->headers.push_back(NameValueTemplatePair{name, value});
+
+    if (!is_literal_string)
+      this->extend_worker_partition_key(value->template_str);
+
+    return true;
   }
 
   void extend_worker_partition_key(const std::string &extension)
@@ -153,6 +163,12 @@ class DestDriver
     this->worker_partition_key << extension;
   }
 
+  void enable_dynamic_headers()
+  {
+    this->dynamic_headers_enabled = true;
+    this->flush_on_key_change = true;
+  }
+
   LogTemplateOptions &get_template_options()
   {
     return this->template_options;
@@ -188,7 +204,8 @@ class DestDriver
   std::list<std::pair<std::string, long>> int_extra_channel_args;
   std::list<std::pair<std::string, std::string>> string_extra_channel_args;
 
-  std::list<std::pair<std::string, std::string>> headers;
+  std::list<NameValueTemplatePair> headers;
+  bool dynamic_headers_enabled;
 
   LogTemplateOptions template_options;
 
diff --git a/modules/grpc/common/grpc-grammar.ym b/modules/grpc/common/grpc-grammar.ym
index 1e50a610a..3d6fa6ea6 100644
--- a/modules/grpc/common/grpc-grammar.ym
+++ b/modules/grpc/common/grpc-grammar.ym
@@ -132,7 +132,12 @@ grpc_dest_headers
   ;
 
 grpc_dest_header
-  : string LL_ARROW string { grpc_dd_add_header(last_driver, $1, $3); free($1); free($3); }
+  : string LL_ARROW template_name_or_content
+      {
+        CHECK_ERROR(grpc_dd_add_header(last_driver, $1, $3), @3, "Failed to set header: templating is not supported");
+        free($1);
+        log_template_unref($3);
+      }
   ;
 
 grpc_server_credentials_builder_option

From 885051d331f9ce52a1ac90dfd8e4f28f0d741f73 Mon Sep 17 00:00:00 2001
From: Attila Szakacs <attila.szakacs@axoflow.com>
Date: Fri, 11 Oct 2024 14:48:20 +0200
Subject: [PATCH 10/13] grpc/otel: support dynamic headers

Signed-off-by: Attila Szakacs <attila.szakacs@axoflow.com>
---
 modules/grpc/otel/otel-dest-worker.cpp | 24 +++++++++++-------------
 modules/grpc/otel/otel-dest-worker.hpp |  1 +
 modules/grpc/otel/otel-dest.cpp        |  6 ++++++
 modules/grpc/otel/otel-dest.hpp        |  2 +-
 4 files changed, 19 insertions(+), 14 deletions(-)

diff --git a/modules/grpc/otel/otel-dest-worker.cpp b/modules/grpc/otel/otel-dest-worker.cpp
index 28ffeb0dd..d25d0c3a8 100644
--- a/modules/grpc/otel/otel-dest-worker.cpp
+++ b/modules/grpc/otel/otel-dest-worker.cpp
@@ -344,6 +344,12 @@ DestWorker::insert(LogMessage *msg)
       g_assert_not_reached();
     }
 
+  if (!client_context.get())
+    {
+      client_context = std::make_unique<::grpc::ClientContext>();
+      prepare_context_dynamic(*client_context, msg);
+    }
+
   if (should_initiate_flush())
     return log_threaded_dest_worker_flush(&super->super, LTF_FLUSH_NORMAL);
 
@@ -408,11 +414,8 @@ _map_grpc_status_to_log_threaded_result(const ::grpc::Status &status)
 LogThreadedResult
 DestWorker::flush_log_records()
 {
-  ::grpc::ClientContext client_context;
-  prepare_context(client_context);
-
   logs_service_response.Clear();
-  ::grpc::Status status = logs_service_stub->Export(&client_context, logs_service_request,
+  ::grpc::Status status = logs_service_stub->Export(client_context.get(), logs_service_request,
                                                     &logs_service_response);
   owner.metrics.insert_grpc_request_stats(status);
   LogThreadedResult result = _map_grpc_status_to_log_threaded_result(status);
@@ -429,11 +432,8 @@ DestWorker::flush_log_records()
 LogThreadedResult
 DestWorker::flush_metrics()
 {
-  ::grpc::ClientContext client_context;
-  prepare_context(client_context);
-
   metrics_service_response.Clear();
-  ::grpc::Status status = metrics_service_stub->Export(&client_context, metrics_service_request,
+  ::grpc::Status status = metrics_service_stub->Export(client_context.get(), metrics_service_request,
                                                        &metrics_service_response);
   owner.metrics.insert_grpc_request_stats(status);
   LogThreadedResult result = _map_grpc_status_to_log_threaded_result(status);
@@ -450,11 +450,8 @@ DestWorker::flush_metrics()
 LogThreadedResult
 DestWorker::flush_spans()
 {
-  ::grpc::ClientContext client_context;
-  prepare_context(client_context);
-
   trace_service_response.Clear();
-  ::grpc::Status status = trace_service_stub->Export(&client_context, trace_service_request,
+  ::grpc::Status status = trace_service_stub->Export(client_context.get(), trace_service_request,
                                                      &trace_service_response);
   owner.metrics.insert_grpc_request_stats(status);
   LogThreadedResult result = _map_grpc_status_to_log_threaded_result(status);
@@ -471,7 +468,7 @@ DestWorker::flush_spans()
 LogThreadedResult
 DestWorker::flush(LogThreadedFlushMode mode)
 {
-  LogThreadedResult result;
+  LogThreadedResult result = LTR_SUCCESS;
 
   if (mode == LTF_FLUSH_EXPEDITE)
     return LTR_RETRY;
@@ -498,6 +495,7 @@ DestWorker::flush(LogThreadedFlushMode mode)
     }
 
 exit:
+  client_context.reset();
   logs_service_request.Clear();
   metrics_service_request.Clear();
   trace_service_request.Clear();
diff --git a/modules/grpc/otel/otel-dest-worker.hpp b/modules/grpc/otel/otel-dest-worker.hpp
index 62f171b06..d30a95c5d 100644
--- a/modules/grpc/otel/otel-dest-worker.hpp
+++ b/modules/grpc/otel/otel-dest-worker.hpp
@@ -82,6 +82,7 @@ class DestWorker : public syslogng::grpc::DestWorker
 
 protected:
   std::shared_ptr<::grpc::Channel> channel;
+  std::unique_ptr<::grpc::ClientContext> client_context;
   std::unique_ptr<LogsService::Stub> logs_service_stub;
   std::unique_ptr<MetricsService::Stub> metrics_service_stub;
   std::unique_ptr<TraceService::Stub> trace_service_stub;
diff --git a/modules/grpc/otel/otel-dest.cpp b/modules/grpc/otel/otel-dest.cpp
index 2e9f388c8..3f8422c13 100644
--- a/modules/grpc/otel/otel-dest.cpp
+++ b/modules/grpc/otel/otel-dest.cpp
@@ -28,6 +28,12 @@ using namespace syslogng::grpc::otel;
 
 /* C++ Implementations */
 
+DestDriver::DestDriver(GrpcDestDriver *s) :
+  syslogng::grpc::DestDriver(s)
+{
+  this->enable_dynamic_headers();
+}
+
 const char *
 DestDriver::generate_persist_name()
 {
diff --git a/modules/grpc/otel/otel-dest.hpp b/modules/grpc/otel/otel-dest.hpp
index 4c26ae9de..a53ac06e9 100644
--- a/modules/grpc/otel/otel-dest.hpp
+++ b/modules/grpc/otel/otel-dest.hpp
@@ -43,7 +43,7 @@ namespace otel {
 class DestDriver: public syslogng::grpc::DestDriver
 {
 public:
-  DestDriver(GrpcDestDriver *s) : syslogng::grpc::DestDriver(s) {};
+  DestDriver(GrpcDestDriver *s);
 
   const char *format_stats_key(StatsClusterKeyBuilder *kb);
   const char *generate_persist_name();

From 96ec7a2e8cca2f25efcc536d5fae936511bcba84 Mon Sep 17 00:00:00 2001
From: Attila Szakacs <attila.szakacs@axoflow.com>
Date: Fri, 11 Oct 2024 14:48:26 +0200
Subject: [PATCH 11/13] grpc/loki: support dynamic headers

Signed-off-by: Attila Szakacs <attila.szakacs@axoflow.com>
---
 modules/grpc/loki/loki-dest.cpp   |  2 +-
 modules/grpc/loki/loki-worker.cpp | 17 ++++++++++-------
 modules/grpc/loki/loki-worker.hpp |  1 +
 3 files changed, 12 insertions(+), 8 deletions(-)

diff --git a/modules/grpc/loki/loki-dest.cpp b/modules/grpc/loki/loki-dest.cpp
index 4739ff44c..292c99888 100644
--- a/modules/grpc/loki/loki-dest.cpp
+++ b/modules/grpc/loki/loki-dest.cpp
@@ -43,7 +43,7 @@ DestinationDriver::DestinationDriver(GrpcDestDriver *s)
   : syslogng::grpc::DestDriver(s), timestamp(LM_TS_PROCESSED)
 {
   this->url = "localhost:9095";
-  this->flush_on_key_change = true;
+  this->enable_dynamic_headers();
 }
 
 DestinationDriver::~DestinationDriver()
diff --git a/modules/grpc/loki/loki-worker.cpp b/modules/grpc/loki/loki-worker.cpp
index 89f2113ea..c0bd1e606 100644
--- a/modules/grpc/loki/loki-worker.cpp
+++ b/modules/grpc/loki/loki-worker.cpp
@@ -121,6 +121,7 @@ DestinationWorker::prepare_batch()
   this->current_batch = logproto::PushRequest{};
   this->current_batch.add_streams();
   this->current_batch_bytes = 0;
+  this->client_context.reset();
 }
 
 bool
@@ -205,6 +206,14 @@ DestinationWorker::insert(LogMessage *msg)
   this->current_batch_bytes += message->len;
   log_threaded_dest_driver_insert_msg_length_stats(super->super.owner, message->len);
 
+  if (!this->client_context.get())
+    {
+      this->client_context = std::make_unique<::grpc::ClientContext>();
+      this->prepare_context_dynamic(*this->client_context, msg);
+      if (!owner_->tenant_id.empty())
+        client_context->AddMetadata("x-scope-orgid", owner_->tenant_id);
+    }
+
   msg_trace("Message added to Loki batch", log_pipe_location_tag((LogPipe *) this->super->super.owner));
 
   if (this->should_initiate_flush())
@@ -224,13 +233,7 @@ DestinationWorker::flush(LogThreadedFlushMode mode)
   LogThreadedResult result;
   logproto::PushResponse response{};
 
-  ::grpc::ClientContext ctx;
-  this->prepare_context(ctx);
-
-  if (!owner_->tenant_id.empty())
-    ctx.AddMetadata("x-scope-orgid", owner_->tenant_id);
-
-  ::grpc::Status status = this->stub->Push(&ctx, this->current_batch, &response);
+  ::grpc::Status status = this->stub->Push(client_context.get(), this->current_batch, &response);
   this->get_owner()->metrics.insert_grpc_request_stats(status);
 
   if (!status.ok())
diff --git a/modules/grpc/loki/loki-worker.hpp b/modules/grpc/loki/loki-worker.hpp
index 743d745a1..2e3e5cfe2 100644
--- a/modules/grpc/loki/loki-worker.hpp
+++ b/modules/grpc/loki/loki-worker.hpp
@@ -66,6 +66,7 @@ class DestinationWorker final: public syslogng::grpc::DestWorker
   bool connected;
 
   std::shared_ptr<::grpc::Channel> channel;
+  std::unique_ptr<::grpc::ClientContext> client_context;
   std::unique_ptr<logproto::Pusher::Stub> stub;
   logproto::PushRequest current_batch;
   size_t current_batch_bytes = 0;

From c26fddc913de88de97b5bc3408ecb412af0b7f42 Mon Sep 17 00:00:00 2001
From: Attila Szakacs <attila.szakacs@axoflow.com>
Date: Fri, 11 Oct 2024 15:02:43 +0200
Subject: [PATCH 12/13] grpc: use local subchannel pool

Signed-off-by: Attila Szakacs <attila.szakacs@axoflow.com>
---
 modules/grpc/common/grpc-dest-worker.cpp | 1 +
 modules/grpc/common/grpc-source.cpp      | 2 ++
 2 files changed, 3 insertions(+)

diff --git a/modules/grpc/common/grpc-dest-worker.cpp b/modules/grpc/common/grpc-dest-worker.cpp
index d4f46d4a1..134b1cc21 100644
--- a/modules/grpc/common/grpc-dest-worker.cpp
+++ b/modules/grpc/common/grpc-dest-worker.cpp
@@ -64,6 +64,7 @@ DestWorker::create_channel_args()
     args.SetCompressionAlgorithm(GRPC_COMPRESS_GZIP);
 
   args.SetInt(GRPC_ARG_KEEPALIVE_PERMIT_WITHOUT_CALLS, 1);
+  args.SetInt(GRPC_ARG_USE_LOCAL_SUBCHANNEL_POOL, 1);
 
   for (auto nv : this->owner.int_extra_channel_args)
     args.SetInt(nv.first, nv.second);
diff --git a/modules/grpc/common/grpc-source.cpp b/modules/grpc/common/grpc-source.cpp
index 313f7e6a0..d4a3d4ed7 100644
--- a/modules/grpc/common/grpc-source.cpp
+++ b/modules/grpc/common/grpc-source.cpp
@@ -74,6 +74,8 @@ SourceDriver::prepare_server_builder(::grpc::ServerBuilder &builder)
 
   builder.AddListeningPort(address, credentials_builder.build());
 
+  builder.AddChannelArgument(GRPC_ARG_USE_LOCAL_SUBCHANNEL_POOL, 1);
+
   for (auto nv : int_extra_channel_args)
     builder.AddChannelArgument(nv.first, nv.second);
   for (auto nv : string_extra_channel_args)

From a12a4a84b61dc6ca28445166e8b4d4d25fb602b6 Mon Sep 17 00:00:00 2001
From: Attila Szakacs <attila.szakacs@axoflow.com>
Date: Fri, 11 Oct 2024 15:12:05 +0200
Subject: [PATCH 13/13] news: add entry for #334

Signed-off-by: Attila Szakacs <attila.szakacs@axoflow.com>
---
 news/feature-334.md | 1 +
 1 file changed, 1 insertion(+)
 create mode 100644 news/feature-334.md

diff --git a/news/feature-334.md b/news/feature-334.md
new file mode 100644
index 000000000..139fcb250
--- /dev/null
+++ b/news/feature-334.md
@@ -0,0 +1 @@
+`opentelemetry()`, `loki()` destination: Add support for templated `header()` values