Measure average batch byte size and event count

config/logstash.yml

@@ -49,6 +49,12 @@
 #
 # 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 "disabled".
+#
+# pipeline.batch.metrics.sampling_mode: "disabled"
+#
 # 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"].round
+                }
+              },
+              :byte_size => {
+                :average => {
+                  :lifetime => stats[:batch][:byte_size][:average].value["lifetime"].round
+                }
+              }
+            }
+          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::PasswordSetting.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", "disabled", 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

@@ -16,6 +16,7 @@
 # under the License.
 
 require "logstash/instrument/namespaced_metric"
+# require "logstash/instrument/collector"
 
 describe LogStash::WrappedAckedQueue, :stress_test => true do
   let(:path) { Stud::Temporary.directory }
@@ -29,11 +30,13 @@
     let(:reject_memo_keys) { [:reject_memo_keys, :path, :queue, :writer_threads, :collector, :metric, :reader_threads, :output_strings] }
 
     let(:queue_settings) do
+      java_import org.logstash.ackedqueue.QueueFactoryExt::BatchMetricMode
       LogStash::AckedQueue.file_settings_builder(path)
                           .capacity(page_capacity)
                           .checkpointMaxAcks(queue_checkpoint_acks)
                           .checkpointMaxWrites(queue_checkpoint_writes)
                           .queueMaxBytes(queue_capacity)
+                          .batchMetricMode(BatchMetricMode::DISABLED)
                           .build
     end
 

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, "disabled") }
 
     before do
       read_client.set_events_metric(metric.namespace([:stats, :events]))
@@ -127,12 +127,14 @@ def threaded_read_client
     let(:path) { Stud::Temporary.directory }
 
     let(:queue_settings) do
+      java_import org.logstash.ackedqueue.QueueFactoryExt::BatchMetricMode
       LogStash::AckedQueue.file_settings_builder(path)
          .capacity(1024)
          .maxUnread(10)
          .checkpointMaxAcks(1024)
          .checkpointMaxWrites(1024)
          .queueMaxBytes(4096)
+         .batchMetricMode(BatchMetricMode::DISABLED)
          .build
     end
 

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", "disabled", 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

@@ -55,12 +55,14 @@
     let(:path) { Stud::Temporary.directory }
 
     let(:queue_settings) do
+      java_import org.logstash.ackedqueue.QueueFactoryExt::BatchMetricMode
       LogStash::AckedQueue.file_settings_builder(path)
                           .capacity(page_capacity)
                           .maxUnread(max_events)
                           .checkpointMaxAcks(checkpoint_acks)
                           .checkpointMaxWrites(checkpoint_writes)
                           .queueMaxBytes(max_bytes)
+                          .batchMetricMode(BatchMetricMode::DISABLED)
                           .build
     end
 

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

@@ -19,7 +19,7 @@
 require "logstash/instrument/collector"
 
 describe LogStash::WrappedSynchronousQueue do
-  subject {LogStash::WrappedSynchronousQueue.new(5)}
+  subject {LogStash::WrappedSynchronousQueue.new(5, "disabled")}
 
   describe "queue clients" do
     context "when requesting a write client" do

logstash-core/src/main/java/org/logstash/ConvertedMap.java

@@ -21,14 +21,26 @@
 package org.logstash;
 
 import java.io.Serializable;
+import java.math.BigDecimal;
+import java.math.BigInteger;
+import java.util.Collection;
 import java.util.HashMap;
 import java.util.IdentityHashMap;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+
+import org.jruby.RubyBignum;
+import org.jruby.RubyBoolean;
+import org.jruby.RubyFixnum;
+import org.jruby.RubyFloat;
 import org.jruby.RubyHash;
+import org.jruby.RubyNil;
 import org.jruby.RubyString;
+import org.jruby.RubySymbol;
+import org.jruby.ext.bigdecimal.RubyBigDecimal;
 import org.jruby.runtime.ThreadContext;
 import org.jruby.runtime.builtin.IRubyObject;
+import org.logstash.ext.JrubyTimestampExtLibrary;
 
 /**
  * <p>This class is an internal API and behaves very different from a standard {@link Map}.</p>
@@ -41,7 +53,7 @@
  * intern pool to ensure identity match of equivalent strings.
  * For performance, we keep a global cache of strings that have been interned for use with {@link ConvertedMap},
  * and encourage interning through {@link ConvertedMap#internStringForUseAsKey(String)} to avoid
- * the performance pentalty of the global string intern pool.
+ * the performance penalty of the global string intern pool.
  */
 public final class ConvertedMap extends IdentityHashMap<String, Object> {
 
@@ -157,4 +169,112 @@ public Object unconvert() {
     private static String convertKey(final RubyString key) {
         return internStringForUseAsKey(key.asJavaString());
     }
+
+    public long estimateMemory() {
+        return values().stream()
+                .map(this::estimateMemory)
+                .mapToLong(Long::longValue)
+                .sum();
+    }
+
+    @SuppressWarnings({"rawtypes", "unchecked"})
+    private long estimateMemory(Object o) {
+        if (o instanceof Boolean) {
+            return Byte.BYTES;
+        }
+        if (o instanceof Byte) {
+            return Byte.BYTES;
+        }
+        if (o instanceof Short) {
+            return Short.BYTES;
+        }
+        if (o instanceof Integer) {
+            return Integer.BYTES;
+        }
+        if (o instanceof Long) {
+            return Long.BYTES;
+        }
+        if (o instanceof Float) {
+            return Float.BYTES;
+        }
+        if (o instanceof Double) {
+            return Double.BYTES;
+        }
+        if (o instanceof Character) {
+            return Character.BYTES;
+        }
+        if (o instanceof String) {
+            return ((String) o).getBytes().length;
+        }
+        if (o instanceof RubyString) {
+            return ((RubyString) o).getBytes().length;
+        }
+
+        if (o instanceof Collection) {
+            Collection c = (Collection) o;
+            long memory = 0L;
+            for (Object v : c) {
+                memory += estimateMemory(v);
+            }
+            return memory;
+        }
+
+        if (o instanceof ConvertedMap) {
+            ConvertedMap c = (ConvertedMap) o;
+            return c.estimateMemory();
+        }
+
+        if (o instanceof Map) {
+            // this case shouldn't happen because all Map are converted to ConvertedMap
+            Map<String, Object> m = (Map<String, Object>) o;
+            long memory = 0L;
+            for (Map.Entry e : m.entrySet()) {
+                memory += estimateMemory(e.getKey());
+                memory += estimateMemory(e.getValue());
+            }
+            return memory;
+        }
+        if (o instanceof JrubyTimestampExtLibrary.RubyTimestamp) {
+            // wraps an java.time.Instant which is made of long and int
+            return Long.BYTES + Integer.BYTES;
+        }
+        if (o instanceof BigInteger) {
+            return ((BigInteger) o).toByteArray().length;
+        }
+        if (o instanceof BigDecimal) {
+            // BigInteger has 4 fields, one reference 2 ints (scale and precision) and a long.
+            return 8 + 2 * Integer.BYTES + Long.BYTES;
+        }
+        if (o instanceof RubyBignum) {
+            RubyBignum rbn = (RubyBignum) o;
+            return ((RubyFixnum) rbn.size()).getLongValue();
+        }
+        if (o instanceof RubyBigDecimal) {
+            RubyBigDecimal rbd = (RubyBigDecimal) o;
+            // wraps a Java BigDecimal so we can return the size of that:
+            return estimateMemory(rbd.getValue());
+        }
+        if (o instanceof RubyFixnum) {
+            // like an int value
+            return Integer.BYTES;
+        }
+        if (o instanceof RubyBoolean) {
+            return Byte.BYTES;
+        }
+        if (o instanceof RubyNil) {
+            return 8 + Integer.BYTES; // object reference, one int
+        }
+        if (o instanceof RubySymbol) {
+            return estimateMemory(((RubySymbol) o).asJavaString());
+        }
+        if (o instanceof RubyFloat) {
+            return Double.BYTES;
+        }
+
+        throw new IllegalArgumentException(
+                "Unsupported type encountered in estimateMemory: " + o.getClass().getName() +
+                        ". Please ensure all objects passed to estimateMemory are of supported types. " +
+                        "Refer to the ConvertedMap.estimateMemory method for the list of supported types."
+        );
+    }
 }

logstash-core/src/main/java/org/logstash/Event.java

@@ -529,6 +529,7 @@ private void initFailTag(final Object tag) {
      * and needs to be converted to a list before appending to it.
      * @param existing Existing Tag
      * @param tag Tag to add
+     *
      */
     private void scalarTagFallback(final String existing, final String tag) {
         final List<String> tags = new ArrayList<>(2);
@@ -567,4 +568,16 @@ private static String getCanonicalFieldReference(final FieldReference field) {
             return path.stream().collect(Collectors.joining("][", "[", "]"));
         }
     }
+
+    /**
+     * @return a byte size estimation of the event, based on the payloads carried by nested data structures,
+     * without considering the space needed by the JVM to represent the object itself.
+     *
+     * */
+    public long estimateMemory() {
+        long total = 0;
+        total += data.estimateMemory();
+        total += metadata.estimateMemory();
+        return total;
+    }
 }