NMS-19278 Expose batch-size of sink modules

core/ipc/sink/api/src/main/java/org/opennms/core/ipc/sink/api/AggregationPolicy.java

@@ -96,4 +96,6 @@ public interface AggregationPolicy<S, T, U> {
      */
      T build(U accumulator);
 
+
+     int getBatchSize();
 }

core/ipc/sink/common/src/main/java/org/opennms/core/ipc/sink/aggregation/ArrayListAggregationPolicy.java

@@ -63,6 +63,11 @@ public List<S> build(List<S> accumulator) {
         return accumulator;
     }
 
+    @Override
+    public int getBatchSize() {
+        return m_completionSize;
+    }
+
     @Override
     public int getCompletionSize() {
         return m_completionSize;

core/ipc/sink/common/src/main/java/org/opennms/core/ipc/sink/aggregation/IdentityAggregationPolicy.java

@@ -47,6 +47,11 @@ public T build(T accumulator) {
         return accumulator;
     }
 
+    @Override
+    public int getBatchSize() {
+        return 1;
+    }
+
     @Override
     public int getCompletionSize() {
         return 1;

core/ipc/sink/common/src/main/java/org/opennms/core/ipc/sink/aggregation/MappingAggregationPolicy.java

@@ -59,4 +59,8 @@ public T build(T accumulator) {
         return accumulator;
     }
 
+    @Override
+    public int getBatchSize() {
+        return 1;
+    }
 }

core/ipc/sink/common/src/main/java/org/opennms/core/ipc/sink/common/AsyncDispatcherImpl.java

@@ -103,6 +103,12 @@ public AsyncDispatcherImpl(DispatcherState<W, S, T> state, AsyncPolicy asyncPoli
 
         state.getMetrics().register(queueSizeMetricName(), (Gauge<Integer>) activeDispatchers::get);
 
+        if (state.getModule() != null) {
+            state.getMetrics().register(
+                    batchSizeMetricName(),
+                    (Gauge<Integer>) () -> state.getModule().getAggregationPolicy().getBatchSize()
+            );
+        }
         droppedCounter = state.getMetrics().counter(MetricRegistry.name(state.getModule().getId(), "dropped"));
 
         executor = Executors.newFixedThreadPool(asyncPolicy.getNumThreads(),
@@ -115,6 +121,11 @@ private String queueSizeMetricName() {
         return MetricRegistry.name(state.getModule().getId(), "queue-size");
     }
 
+    private String batchSizeMetricName() {
+        return MetricRegistry.name(state.getModule().getId(), "batch-size");
+    }
+
+
     private void dispatchFromQueue() {
         while (true) {
             try {

core/ipc/sink/common/src/test/java/org/opennms/core/ipc/sink/aggregation/AggregationTest.java

@@ -321,6 +321,12 @@ public UDPPacketLog aggregate(UDPPacketLog oldPacketLog, UDPPacket newPacket) {
                 public UDPPacketLog build(UDPPacketLog accumulator) {
                     return accumulator;
                 }
+
+                @Override
+                public int getBatchSize() {
+                    return COMPLETION_SIZE;
+                }
+
             };
         }
     }
@@ -359,6 +365,13 @@ public UDPPacketLog aggregate(UDPPacketLog oldPacketLog, UDPPacket newPacket) {
                 public UDPPacketLog build(UDPPacketLog accumulator) {
                     return accumulator;
                 }
+
+                @Override
+                public int getBatchSize() {
+                    return COMPLETION_SIZE;
+                }
+
+
             };
         }
     }
@@ -398,6 +411,10 @@ public UDPPacketLog aggregate(UDPPacketLog oldPacketLog, UDPPacket newPacket) {
                 public UDPPacketLog build(UDPPacketLog accumulator) {
                     return accumulator;
                 }
+                @Override
+                public int getBatchSize() {
+                    return 1;
+                }
             };
         }
     }

features/events/sink/dispatcher/src/main/java/org/opennms/features/events/sink/module/EventSinkModule.java

@@ -90,6 +90,12 @@ public Log aggregate(Log eventLog, Event event) {
             public Log build(Log accumulator) {
                 return accumulator;
             }
+
+            @Override
+            public int getBatchSize() {
+                return m_config.getBatchSize();
+            }
+
         };
     }
 

features/events/syslog/src/main/java/org/opennms/netmgt/syslogd/SyslogSinkModule.java

@@ -90,6 +90,10 @@ public SyslogMessageLogDTO aggregate(SyslogMessageLogDTO accumulator, SyslogConn
             public SyslogMessageLogDTO build(SyslogMessageLogDTO accumulator) {
                 return accumulator;
             }
+            @Override
+            public int getBatchSize() {
+                return config.getBatchSize();
+            }
         };
     }
 

features/events/traps/src/main/java/org/opennms/netmgt/trapd/TrapSinkModule.java

@@ -105,6 +105,12 @@ public TrapLogDTO aggregate(TrapLogDTO accumulator, TrapInformationWrapper newMe
             public TrapLogDTO build(TrapLogDTO accumulator) {
                 return accumulator;
             }
+
+            @Override
+            public int getBatchSize() {
+                return config.getBatchSize();
+            }
+
         };
     }
 

features/telemetry/common/src/main/java/org/opennms/netmgt/telemetry/common/ipc/TelemetrySinkModule.java

@@ -138,6 +138,10 @@ public TelemetryProtos.TelemetryMessageLog.Builder aggregate(TelemetryProtos.Tel
             public TelemetryProtos.TelemetryMessageLog build(TelemetryProtos.TelemetryMessageLog.Builder accumulator) {
                 return accumulator.build();
             }
+            @Override
+            public int getBatchSize() {
+                return TelemetrySinkModule.this.queueConfig.getBatchSize().orElse(DEFAULT_BATCH_SIZE);
+            }
         };
     }