Measure average batch byte size and event count (#18000)

Implements average batch event count and byte size metrics. The collection of such metric could be disabled, enabled for each batch or done on a sample of the total batches.

Exposing metric related to average batch byte size and event count let the user discover the average structure of their batches, understanding if the batches are fulfilled and eventually understand how to set `pipeline.batch.size` and `pipeline.batch.delay` so that goal is reached.

- Instantiate metric `pipelines.<pipeline id>.batch.count` to count number of matches to compute the average events and bytes per batch
- Instantiate metric `pipelines.<pipeline id>.batch.total_bytes` to sumup all the batches event's byte estimation. Exposed metric `pipelines.<pipeline id>.batch.byte_size.average.lifetime` containing the average byte size of each batch.
- create new setting `pipeline.batch.metrics.sampling_mode` which could have 3 values: `disabled`, `minimal` and `full`. In this case id `disable` no `batch` metric is exposed in the `_node/stats` API. `minimal` count batches and estimates the size only for 1% of the total while `full` is for every batch. This setting leverages existing Logstash setting infrastructure so that one defined at pipeline level (defined in `pipelines.yml`) takes precedence over the global one (defined in `logstash.yml`).

Author: andsel

config/logstash.yml

@@ -49,6 +49,13 @@
 #
 # pipeline.batch.delay: 50
 #
+# Set the pipeline's batch metrics reporting mode. It can be "disabled" to disable it.
+# "minimal" to collect only 1% of the batches metrics, "full" to collect all batches.
+# Default is "minimal".
+#
+# pipeline.batch.metrics.sampling_mode: "minimal"
+pipeline.batch.metrics.sampling_mode: minimal
+#
 # Force Logstash to exit during shutdown even if there are still inflight
 # events in memory. By default, logstash will refuse to quit until all
 # received events have been pushed to the outputs.

logstash-core/lib/logstash/api/commands/stats.rb

@@ -172,6 +172,23 @@ def plugin_stats(stats, plugin_type)
             end
           end
 
+          def refine_batch_metrics(stats)
+            {
+              :event_count => {
+                :average => {
+                  # average return a FlowMetric which and we need to invoke getValue to obtain the map with metric details.
+                  :lifetime => stats[:batch][:event_count][:average].value["lifetime"] ? stats[:batch][:event_count][:average].value["lifetime"].round : 0
+                }
+              },
+              :byte_size => {
+                :average => {
+                  :lifetime => stats[:batch][:byte_size][:average].value["lifetime"] ? stats[:batch][:byte_size][:average].value["lifetime"].round : 0
+                }
+              }
+            }
+          end
+          private :refine_batch_metrics
+
           def report(stats, extended_stats = nil, opts = {})
             ret = {
               :events => stats[:events],
@@ -190,6 +207,7 @@ def report(stats, extended_stats = nil, opts = {})
                 :batch_delay => stats.dig(:config, :batch_delay),
               }
             }
+            ret[:batch] = refine_batch_metrics(stats) if stats.include?(:batch)
             ret[:dead_letter_queue] = stats[:dlq] if stats.include?(:dlq)
 
             # if extended_stats were provided, enrich the return value

logstash-core/lib/logstash/environment.rb

@@ -86,6 +86,7 @@ def self.as_java_range(r)
   Setting::ExistingFilePath.new("api.ssl.keystore.path", nil, false).nullable,
           Setting::Password.new("api.ssl.keystore.password", nil, false).nullable,
        Setting::StringArray.new("api.ssl.supported_protocols", nil, true, %w[TLSv1 TLSv1.1 TLSv1.2 TLSv1.3]),
+           Setting::StringSetting.new("pipeline.batch.metrics.sampling_mode", "minimal", true, ["disabled", "minimal", "full"]),
             Setting::StringSetting.new("queue.type", "memory", true, ["persisted", "memory"]),
            Setting::BooleanSetting.new("queue.drain", false),
              Setting::Bytes.new("queue.page_capacity", "64mb"),

logstash-core/lib/logstash/java_pipeline.rb

@@ -267,6 +267,7 @@ def start_workers
       @preserve_event_order = preserve_event_order?(pipeline_workers)
       batch_size = settings.get("pipeline.batch.size")
       batch_delay = settings.get("pipeline.batch.delay")
+      batch_metric_sampling = settings.get("pipeline.batch.metrics.sampling_mode")
 
       max_inflight = batch_size * pipeline_workers
 
@@ -287,6 +288,7 @@ def start_workers
         "pipeline.batch.size" => batch_size,
         "pipeline.batch.delay" => batch_delay,
         "pipeline.max_inflight" => max_inflight,
+        "batch_metric_sampling" => batch_metric_sampling,
         "pipeline.sources" => pipeline_source_details)
       @logger.info("Starting pipeline", pipeline_log_params)
 

logstash-core/lib/logstash/settings.rb

@@ -58,6 +58,7 @@ def self.included(base)
       "path.dead_letter_queue",
       "path.queue",
       "pipeline.batch.delay",
+      "pipeline.batch.metrics.sampling_mode",
       "pipeline.batch.size",
       "pipeline.id",
       "pipeline.reloadable",

logstash-core/spec/logstash/acked_queue_concurrent_stress_spec.rb

@@ -38,7 +38,7 @@
     end
 
     let(:queue) do
-      described_class.new(queue_settings)
+      described_class.new(queue_settings, org.logstash.ackedqueue.QueueFactoryExt::BatchMetricMode::DISABLED)
     end
 
     let(:writer_threads) do

logstash-core/spec/logstash/api/modules/node_stats_spec.rb

@@ -21,8 +21,13 @@
 require "logstash/api/modules/node_stats"
 
 describe LogStash::Api::Modules::NodeStats do
-  # enable PQ to ensure PQ-related metrics are present
-  include_context "api setup", {"queue.type" => "persisted"}
+
+  include_context "api setup", {
+    # enable PQ to ensure PQ-related metrics are present
+    "queue.type" => "persisted",
+    #enable batch metrics
+    "pipeline.batch.metrics.sampling_mode" => "full"
+  }
   include_examples "not found"
 
   extend ResourceDSLMethods
@@ -142,6 +147,18 @@
            "path" => String,
            "free_space_in_bytes" => Numeric
          }
+       },
+       "batch" => {
+         "event_count" => {
+           "average" => {
+             "lifetime" => Numeric
+           }
+         },
+         "byte_size" => {
+           "average" => {
+             "lifetime" => Numeric
+           }
+         }
        }
      }
    },

logstash-core/spec/logstash/instrument/wrapped_write_client_spec.rb

@@ -113,7 +113,7 @@ def threaded_read_client
   end
 
   context "WrappedSynchronousQueue" do
-    let(:queue) { LogStash::WrappedSynchronousQueue.new(1024) }
+    let(:queue) { LogStash::WrappedSynchronousQueue.new(1024, org.logstash.ackedqueue.QueueFactoryExt::BatchMetricMode::DISABLED) }
 
     before do
       read_client.set_events_metric(metric.namespace([:stats, :events]))
@@ -136,7 +136,9 @@ def threaded_read_client
          .build
     end
 
-    let(:queue) { LogStash::WrappedAckedQueue.new(queue_settings) }
+    let(:queue) do
+      LogStash::WrappedAckedQueue.new(queue_settings, org.logstash.ackedqueue.QueueFactoryExt::BatchMetricMode::DISABLED)
+    end
 
     before do
       read_client.set_events_metric(metric.namespace([:stats, :events]))

logstash-core/spec/logstash/queue_factory_spec.rb

@@ -31,6 +31,7 @@
       LogStash::Setting::NumericSetting.new("queue.checkpoint.writes", 1024),
       LogStash::Setting::BooleanSetting.new("queue.checkpoint.retry", false),
       LogStash::Setting::StringSetting.new("pipeline.id", pipeline_id),
+      LogStash::Setting::StringSetting.new("pipeline.batch.metrics.sampling_mode", "minimal", true, ["disabled", "minimal", "full"]),
       LogStash::Setting::PositiveIntegerSetting.new("pipeline.batch.size", 125),
       LogStash::Setting::PositiveIntegerSetting.new("pipeline.workers", LogStash::Config::CpuCoreStrategy.maximum)
     ]

logstash-core/spec/logstash/util/wrapped_acked_queue_spec.rb

@@ -64,7 +64,7 @@
                           .build
     end
 
-    let(:queue) { LogStash::WrappedAckedQueue.new(queue_settings) }
+    let(:queue) { LogStash::WrappedAckedQueue.new(queue_settings, org.logstash.ackedqueue.QueueFactoryExt::BatchMetricMode::DISABLED) }
 
     after do
       queue.close