threaded-sources: move initial window size calculation to base class

This also fixes opentelemetry() and axosyslog-otlp() when workers() > 1, because originally init_window_size was always set to 0 (-1 / num_workers). Signed-off-by: László Várady <[email protected]>
axoflow · Sep 25, 2024 · 87f77fa · 87f77fa
1 parent 288c432
commit 87f77fa
Show file tree

Hide file tree

Showing 3 changed files with 10 additions and 5 deletions.
diff --git a/lib/logthrsource/logthrsourcedrv.c b/lib/logthrsource/logthrsourcedrv.c
@@ -108,8 +108,15 @@ log_threaded_source_worker_options_defaults(LogThreadedSourceWorkerOptions *opti
 
 void
 log_threaded_source_worker_options_init(LogThreadedSourceWorkerOptions *options, GlobalConfig *cfg,
-                                        const gchar *group_name)
+                                        const gchar *group_name, gint num_workers)
 {
+  if (options->super.init_window_size == -1)
+    {
+      options->super.init_window_size = 100 * num_workers;
+    }
+
+  options->super.init_window_size /= num_workers;
+
   log_source_options_init(&options->super, cfg, group_name);
   msg_format_options_init(&options->parse_options, cfg);
 }
@@ -285,7 +292,7 @@ _init_workers(LogThreadedSourceDriver *self)
 
   GlobalConfig *cfg = log_pipe_get_config(&self->super.super.super);
 
-  log_threaded_source_worker_options_init(&self->worker_options, cfg, self->super.super.group);
+  log_threaded_source_worker_options_init(&self->worker_options, cfg, self->super.super.group, self->num_workers);
 
   for (size_t i = 0; i < self->num_workers; i++)
     {

diff --git a/lib/logthrsource/logthrsourcedrv.h b/lib/logthrsource/logthrsourcedrv.h
@@ -84,7 +84,7 @@ struct _LogThreadedSourceDriver
 
 void log_threaded_source_worker_options_defaults(LogThreadedSourceWorkerOptions *options);
 void log_threaded_source_worker_options_init(LogThreadedSourceWorkerOptions *options, GlobalConfig *cfg,
-                                             const gchar *group_name);
+                                             const gchar *group_name, gint num_workers);
 void log_threaded_source_worker_options_destroy(LogThreadedSourceWorkerOptions *options);
 
 void log_threaded_source_driver_set_transport_name(LogThreadedSourceDriver *self, const gchar *transport_name);

diff --git a/modules/grpc/otel/otel-source.cpp b/modules/grpc/otel/otel-source.cpp
@@ -116,8 +116,6 @@ syslogng::grpc::otel::SourceDriver::init()
 
   msg_info("OpenTelemetry server accepting connections", evt_tag_int("port", port));
 
-  super->super.worker_options.super.init_window_size /= super->super.num_workers;
-
   if (fetch_limit == -1)
     fetch_limit = super->super.worker_options.super.init_window_size;