Added setting 'pipeline.batch.metrics' to enable/disable the collection of batch size related metrics into histrograms

Author: andsel

logstash-core/lib/logstash/environment.rb

@@ -82,6 +82,7 @@ module Environment
   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::SettingString.new("pipeline.batch.metrics", "false", true, ["false", "true"]),
             Setting::SettingString.new("queue.type", "memory", true, ["persisted", "memory"]),
            Setting::Boolean.new("queue.drain", false),
              Setting::Bytes.new("queue.page_capacity", "64mb"),

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",
       "pipeline.batch.size",
       "pipeline.id",
       "pipeline.reloadable",

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, "false") }
 
     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::SettingNumeric.new("queue.checkpoint.writes", 1024),
       LogStash::Setting::Boolean.new("queue.checkpoint.retry", false),
       LogStash::Setting::SettingString.new("pipeline.id", pipeline_id),
+      LogStash::Setting::SettingString.new("pipeline.batch.metrics", "false", true, ["false", "true"]),
       LogStash::Setting::PositiveInteger.new("pipeline.batch.size", 125),
       LogStash::Setting::PositiveInteger.new("pipeline.workers", LogStash::Config::CpuCoreStrategy.maximum)
     ]

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, "false")}
 
   describe "queue clients" do
     context "when requesting a write client" do

logstash-core/src/main/java/org/logstash/ackedqueue/QueueFactoryExt.java

@@ -68,6 +68,10 @@ public QueueFactoryExt(final Ruby runtime, final RubyClass metaClass) {
     public static AbstractWrappedQueueExt create(final ThreadContext context, final IRubyObject recv,
         final IRubyObject settings) throws IOException {
         final String type = getSetting(context, settings, QUEUE_TYPE_CONTEXT_NAME).asJavaString();
+
+        final String histogramFlag = getSetting(context, settings, SettingKeyDefinitions.PIPELINE_BATCH_METRICS)
+                .asJavaString();
+
         if (PERSISTED_TYPE.equals(type)) {
             final Path queuePath = Paths.get(
                 getSetting(context, settings, SettingKeyDefinitions.PATH_QUEUE).asJavaString(),
@@ -93,15 +97,14 @@ public static AbstractWrappedQueueExt create(final ThreadContext context, final
                     }
                 );
         } else if (MEMORY_TYPE.equals(type)) {
+            final int batchSize = getSetting(context, settings, SettingKeyDefinitions.PIPELINE_BATCH_SIZE)
+                    .convertToInteger().getIntValue();
+            final int workers = getSetting(context, settings, SettingKeyDefinitions.PIPELINE_WORKERS)
+                    .convertToInteger().getIntValue();
             return new JrubyWrappedSynchronousQueueExt(
                 context.runtime, RubyUtil.WRAPPED_SYNCHRONOUS_QUEUE_CLASS
             ).initialize(
-                context, context.runtime.newFixnum(
-                    getSetting(context, settings, SettingKeyDefinitions.PIPELINE_BATCH_SIZE)
-                        .convertToInteger().getIntValue()
-                        * getSetting(context, settings, SettingKeyDefinitions.PIPELINE_WORKERS)
-                        .convertToInteger().getIntValue()
-                )
+                context, context.runtime.newFixnum(batchSize * workers), context.runtime.newString(histogramFlag)
             );
         } else {
             throw context.runtime.newRaiseException(

logstash-core/src/main/java/org/logstash/common/SettingKeyDefinitions.java

@@ -30,6 +30,8 @@ public class SettingKeyDefinitions {
 
     public static final String PIPELINE_BATCH_SIZE = "pipeline.batch.size";
 
+    public static final String PIPELINE_BATCH_METRICS = "pipeline.batch.metrics";
+
     public static final String PATH_QUEUE = "path.queue";
 
     public static final String QUEUE_PAGE_CAPACITY = "queue.page_capacity";

logstash-core/src/main/java/org/logstash/execution/QueueReadClient.java

@@ -35,9 +35,9 @@ public interface QueueReadClient {
     void addFilteredMetrics(int filteredSize);
     void closeBatch(QueueBatch batch) throws IOException;
 
-    public <V, E extends Exception> V executeWithTimers(final TimerMetric.ExceptionalSupplier<V,E> supplier) throws E;
+    <V, E extends Exception> V executeWithTimers(final TimerMetric.ExceptionalSupplier<V,E> supplier) throws E;
 
-    public <E extends Exception> void executeWithTimers(final TimerMetric.ExceptionalRunnable<E> runnable) throws E;
+    <E extends Exception> void executeWithTimers(final TimerMetric.ExceptionalRunnable<E> runnable) throws E;
 
     boolean isEmpty();
 }

logstash-core/src/main/java/org/logstash/ext/JrubyWrappedSynchronousQueueExt.java

@@ -23,9 +23,7 @@
 import java.util.concurrent.ArrayBlockingQueue;
 import java.util.concurrent.BlockingQueue;
 
-import org.jruby.Ruby;
-import org.jruby.RubyClass;
-import org.jruby.RubyNumeric;
+import org.jruby.*;
 import org.jruby.anno.JRubyClass;
 import org.jruby.anno.JRubyMethod;
 import org.jruby.runtime.ThreadContext;
@@ -50,8 +48,10 @@ public JrubyWrappedSynchronousQueueExt(final Ruby runtime, final RubyClass metaC
     @JRubyMethod
     @SuppressWarnings("unchecked")
     public JrubyWrappedSynchronousQueueExt initialize(final ThreadContext context,
-        IRubyObject size) {
+                                                      IRubyObject size,
+                                                      IRubyObject batchMetricsSampling) {
         int typedSize = ((RubyNumeric)size).getIntValue();
+        String batchMetricsSamplingType = ((RubyString) batchMetricsSampling).asJavaString();
         this.queue = new ArrayBlockingQueue<>(typedSize);
         return this;
     }