#2115: fix exceptions during outbound mapping not being recorded via "metrics"

Author: thjaeckle

connectivity/model/src/main/java/org/eclipse/ditto/connectivity/model/MetricType.java

@@ -43,6 +43,11 @@ public enum MetricType {
      */
     MAPPED("mapped", MetricDirection.INBOUND, MetricDirection.OUTBOUND),
 
+    /**
+     * Counts others/uncategorized for messages.
+     */
+    OTHER("other", MetricDirection.INBOUND, MetricDirection.OUTBOUND),
+
     /**
      * Counts messages that were dropped (not published by intention e.g. because no reply-to address was given).
      */

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/OutboundMappingProcessorActor.java

@@ -15,6 +15,7 @@
 import static org.eclipse.ditto.base.model.common.ConditionChecker.checkNotEmpty;
 import static org.eclipse.ditto.connectivity.model.MetricType.DROPPED;
 import static org.eclipse.ditto.connectivity.model.MetricType.MAPPED;
+import static org.eclipse.ditto.connectivity.model.MetricType.OTHER;
 
 import java.io.PrintWriter;
 import java.io.StringWriter;
@@ -157,6 +158,7 @@ public final class OutboundMappingProcessorActor
     private final ConnectionMonitor responseDispatchedMonitor;
     private final ConnectionMonitor responseDroppedMonitor;
     private final ConnectionMonitor responseMappedMonitor;
+    private final ConnectionMonitor responseOtherMonitor;
     private final SignalEnrichmentFacade signalEnrichmentFacade;
     private final int processorPoolSize;
     private final DittoRuntimeExceptionToErrorResponseFunction toErrorResponseFunction;
@@ -187,6 +189,7 @@ private OutboundMappingProcessorActor(final ActorRef clientActor,
         responseDispatchedMonitor = connectionMonitorRegistry.forResponseDispatched(this.connection);
         responseDroppedMonitor = connectionMonitorRegistry.forResponseDropped(this.connection);
         responseMappedMonitor = connectionMonitorRegistry.forResponseMapped(this.connection);
+        responseOtherMonitor = connectionMonitorRegistry.forResponseOther(this.connection);
         signalEnrichmentFacade = ConnectivitySignalEnrichmentProvider.get(system, dittoExtensionConfig).getFacade(this.connection.getId());
         this.processorPoolSize = determinePoolSize(processorPoolSize, mappingConfig.getMaxPoolSize());
         toErrorResponseFunction = DittoRuntimeExceptionToErrorResponseFunction.of(DittoHeadersValidator.get(system, dittoExtensionConfig));
@@ -661,15 +664,13 @@ private Object handleSignal(final Signal<?> signal, final ActorRef sender) {
                         })
                         .onError((mapperId, exception, topicPath, unused) -> {
                             if (exception instanceof DittoRuntimeException e) {
-                                monitorsForOther.forEach(monitor ->
-                                        monitor.getLogger().failure(infoProvider, e));
+                                monitorsForOther.forEach(monitor -> monitor.failure(infoProvider, e));
                                 logger.withCorrelationId(e)
                                         .info("Got DittoRuntimeException during processing Signal: {} - {}",
                                                 e.getMessage(),
                                                 e.getDescription().orElse(""));
                             } else {
-                                monitorsForOther.forEach(monitor ->
-                                        monitor.getLogger().exception(infoProvider, exception));
+                                monitorsForOther.forEach(monitor -> monitor.exception(infoProvider, exception));
                                 logger.withCorrelationId(outbound.getSource())
                                         .warning("Got unexpected exception during processing Signal <{}>.",
                                                 exception.getMessage());
@@ -696,7 +697,7 @@ private Set<ConnectionMonitor> getMonitorsForMappedSignal(final OutboundSignal o
 
     private Set<ConnectionMonitor> getMonitorsForOther(final OutboundSignal outbound) {
 
-        return getMonitorsForOutboundSignal(outbound, MAPPED, LogType.OTHER, responseMappedMonitor);
+        return getMonitorsForOutboundSignal(outbound, OTHER, LogType.OTHER, responseOtherMonitor);
     }
 
     private Set<ConnectionMonitor> getMonitorsForOutboundSignal(final OutboundSignal outbound,

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/monitoring/ConnectionMonitorRegistry.java

@@ -170,6 +170,15 @@ public interface ConnectionMonitorRegistry<T> {
      */
     T forResponseMapped(Connection connection);
 
+    /**
+     * Gets counter for {@link org.eclipse.ditto.connectivity.model.MetricDirection#OUTBOUND}/{@link
+     * org.eclipse.ditto.connectivity.model.MetricType#OTHER} messages for responses.
+     *
+     * @param connection connection
+     * @return the response mapped counter
+     */
+    T forResponseOther(Connection connection);
+
     /**
      * Gets counter for {@link org.eclipse.ditto.connectivity.model.MetricDirection#OUTBOUND}/{@link
      * org.eclipse.ditto.connectivity.model.MetricType#PUBLISHED} messages for responses.

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/monitoring/DefaultConnectionMonitorRegistry.java

@@ -183,6 +183,14 @@ public ConnectionMonitor forResponseMapped(final Connection connection) {
                 .build();
     }
 
+    @Override
+    public ConnectionMonitor forResponseOther(final Connection connection) {
+        return DefaultConnectionMonitor.builder(
+                        connectionCounterRegistry.forResponseOther(connection),
+                        connectionLoggerRegistry.forResponseOther(connection))
+                .build();
+    }
+
     @Override
     public ConnectionMonitor forResponsePublished(final Connection connection) {
         return DefaultConnectionMonitor.builder(

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/monitoring/logs/ConnectionLoggerRegistry.java

@@ -464,6 +464,11 @@ public ConnectionLogger forResponseMapped(final Connection connection) {
         return getLogger(connection.getId(), LogCategory.RESPONSE, LogType.MAPPED, RESPONSES_ADDRESS);
     }
 
+    @Override
+    public ConnectionLogger forResponseOther(final Connection connection) {
+        return getLogger(connection.getId(), LogCategory.RESPONSE, LogType.OTHER, RESPONSES_ADDRESS);
+    }
+
     @Override
     public ConnectionLogger forResponsePublished(final Connection connection) {
         return getLogger(connection.getId(), LogCategory.RESPONSE, LogType.PUBLISHED, RESPONSES_ADDRESS);

connectivity/service/src/main/java/org/eclipse/ditto/connectivity/service/messaging/monitoring/metrics/ConnectivityCounterRegistry.java

@@ -246,6 +246,11 @@ public ConnectionMetricsCounter forResponseMapped(final Connection connection) {
         return getCounter(connection, MetricType.MAPPED, MetricDirection.OUTBOUND, RESPONSES_ADDRESS);
     }
 
+    @Override
+    public ConnectionMetricsCounter forResponseOther(final Connection connection) {
+        return getCounter(connection, MetricType.OTHER, MetricDirection.OUTBOUND, RESPONSES_ADDRESS);
+    }
+
     @Override
     public ConnectionMetricsCounter forResponsePublished(final Connection connection) {
         return getCounter(connection, MetricType.PUBLISHED, MetricDirection.OUTBOUND, RESPONSES_ADDRESS);

connectivity/service/src/test/java/org/eclipse/ditto/connectivity/service/messaging/monitoring/metrics/ConnectivityCounterRegistryTest.java

@@ -23,6 +23,7 @@
 import static org.eclipse.ditto.connectivity.model.MetricType.ENFORCED;
 import static org.eclipse.ditto.connectivity.model.MetricType.FILTERED;
 import static org.eclipse.ditto.connectivity.model.MetricType.MAPPED;
+import static org.eclipse.ditto.connectivity.model.MetricType.OTHER;
 import static org.eclipse.ditto.connectivity.model.MetricType.PUBLISHED;
 import static org.eclipse.ditto.connectivity.model.MetricType.THROTTLED;
 import static org.eclipse.ditto.connectivity.service.messaging.monitoring.metrics.MeasurementWindow.ONE_MINUTE_WITH_ONE_MINUTE_RESOLUTION;
@@ -208,7 +209,7 @@ public void testThrottlingCounterRegistered() {
 
         final SourceMetrics sourceMetrics = counterRegistry.aggregateSourceMetrics(connection.getId());
 
-        final MetricType[] expectedTypes = {CONSUMED, MAPPED, DROPPED, ACKNOWLEDGED, ENFORCED, THROTTLED};
+        final MetricType[] expectedTypes = {OTHER, CONSUMED, MAPPED, DROPPED, ACKNOWLEDGED, ENFORCED, THROTTLED};
         connection.getSources()
                 .stream()
                 .flatMap(source -> source.getAddresses().stream())