Merge pull request #2486 from beyonnex-io/feature/serializer-message-categories-and-kamon-metric-cache

Categorize cluster (de)serialization metrics and drop per-op allocation in metric wrappers

Author: thjaeckle

internal/utils/cluster/src/main/java/org/eclipse/ditto/internal/utils/cluster/AbstractJsonifiableWithDittoHeadersSerializer.java

@@ -21,7 +21,11 @@
 import java.nio.charset.Charset;
 import java.nio.charset.StandardCharsets;
 import java.text.MessageFormat;
+import java.util.EnumMap;
+import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
 import java.util.function.Function;
 
 import javax.annotation.Nullable;
@@ -40,8 +44,14 @@
 import org.eclipse.ditto.base.model.json.JsonSchemaVersion;
 import org.eclipse.ditto.base.model.json.Jsonifiable;
 import org.eclipse.ditto.base.model.signals.JsonParsable;
+import org.eclipse.ditto.base.model.signals.WithResource;
 import org.eclipse.ditto.base.model.signals.WithType;
+import org.eclipse.ditto.base.model.signals.acks.Acknowledgement;
+import org.eclipse.ditto.base.model.signals.acks.Acknowledgements;
+import org.eclipse.ditto.base.model.signals.announcements.Announcement;
 import org.eclipse.ditto.base.model.signals.commands.Command;
+import org.eclipse.ditto.base.model.signals.commands.CommandResponse;
+import org.eclipse.ditto.base.model.signals.events.Event;
 import org.eclipse.ditto.internal.utils.metrics.DittoMetrics;
 import org.eclipse.ditto.internal.utils.metrics.instruments.counter.Counter;
 import org.eclipse.ditto.internal.utils.metrics.instruments.tag.Tag;
@@ -90,14 +100,19 @@ public abstract class AbstractJsonifiableWithDittoHeadersSerializer extends Seri
 
     private static final String METRIC_NAME_SUFFIX = "_serializer_messages";
     private static final String METRIC_DIRECTION = "direction";
+    private static final String METRIC_CATEGORY = "category";
+    private static final String METRIC_RESOURCE_TYPE = "resource_type";
+    private static final String DIRECTION_IN = "in";
+    private static final String DIRECTION_OUT = "out";
+    private static final String RESOURCE_TYPE_OTHER = "other";
 
     private final int identifier;
     private final MappingStrategies mappingStrategies;
     private final Function<Object, String> manifestProvider;
     private final BufferPool byteBufferPool;
     private final Long defaultBufferSize;
-    private final Counter inCounter;
-    private final Counter outCounter;
+    private final Map<Category, ConcurrentMap<String, Counter>> inCounters;
+    private final Map<Category, ConcurrentMap<String, Counter>> outCounters;
     private final String serializerName;
 
     /**
@@ -127,10 +142,54 @@ protected AbstractJsonifiableWithDittoHeadersSerializer(
         final var maxPoolEntries = config.withFallback(FALLBACK_CONF).getInt(CONFIG_DIRECT_BUFFER_POOL_LIMIT);
         byteBufferPool = new DirectByteBufferPool(defaultBufferSize.intValue(), maxPoolEntries);
 
-        inCounter = DittoMetrics.counter(serializerName.toLowerCase() + METRIC_NAME_SUFFIX)
-                .tag(METRIC_DIRECTION, "in");
-        outCounter = DittoMetrics.counter(serializerName.toLowerCase() + METRIC_NAME_SUFFIX)
-                .tag(METRIC_DIRECTION, "out");
+        inCounters = newCategoryCounterCache();
+        outCounters = newCategoryCounterCache();
+    }
+
+    private static Map<Category, ConcurrentMap<String, Counter>> newCategoryCounterCache() {
+        final Map<Category, ConcurrentMap<String, Counter>> cache = new EnumMap<>(Category.class);
+        for (final Category category : Category.values()) {
+            cache.put(category, new ConcurrentHashMap<>());
+        }
+        return cache;
+    }
+
+    /**
+     * Increments the {@code <serializer>_serializer_messages} counter tagged with the {@code direction}, the coarse
+     * signal {@code category} and the {@code resource_type} of the passed {@code object}. The fully tagged counters
+     * are cached per {@code (category, resource_type)} combination so that on the hot (de)serialization path only a map
+     * lookup and the increment are performed; the tagged counter is built at most once per combination and serializer.
+     *
+     * @param countersByCategory the direction-specific counter cache ({@link #inCounters} or {@link #outCounters}).
+     * @param direction the {@code direction} tag value ({@value #DIRECTION_IN} or {@value #DIRECTION_OUT}).
+     * @param object the (de)serialized object to classify.
+     */
+    private void incrementCounter(final Map<Category, ConcurrentMap<String, Counter>> countersByCategory,
+            final String direction, final Object object) {
+
+        final var category = Category.of(object);
+        final var resourceType = resourceTypeOf(object);
+        final var countersByResourceType = countersByCategory.get(category);
+        var counter = countersByResourceType.get(resourceType);
+        if (null == counter) {
+            // slow path, hit at most once per (category, resource_type) combination for a serializer's lifetime:
+            counter = countersByResourceType.computeIfAbsent(resourceType,
+                    rt -> DittoMetrics.counter(serializerName.toLowerCase() + METRIC_NAME_SUFFIX)
+                            .tag(METRIC_DIRECTION, direction)
+                            .tag(METRIC_CATEGORY, category.getTag())
+                            .tag(METRIC_RESOURCE_TYPE, rt));
+        }
+        counter.increment();
+    }
+
+    private static String resourceTypeOf(final Object object) {
+        final String result;
+        if (object instanceof WithResource withResource) {
+            result = withResource.getResourceType();
+        } else {
+            result = RESOURCE_TYPE_OTHER;
+        }
+        return result;
     }
 
     @Override
@@ -155,7 +214,7 @@ public void toBinary(final Object object, final ByteBuffer buf) {
             try {
                 serializeIntoByteBuffer(jsonObject, buf);
                 LOG.trace("toBinary jsonStr about to send 'out': {}", jsonObject);
-                outCounter.increment();
+                incrementCounter(outCounters, DIRECTION_OUT, object);
             } catch (final BufferOverflowException e) {
                 final var errorMessage = MessageFormat.format(
                         "Could not put bytes of JSON string <{0}> into ByteBuffer due to BufferOverflow",
@@ -282,7 +341,7 @@ public Object fromBinary(final ByteBuffer buf, final String manifest) {
                         serializerName,
                         BinaryToHexConverter.createDebugMessageByTryingToConvertToHexString(buf));
             }
-            inCounter.increment();
+            incrementCounter(inCounters, DIRECTION_IN, jsonifiable);
             return jsonifiable;
         } catch (final NotSerializableException e) {
             return e;
@@ -448,4 +507,51 @@ private static String getDefaultManifestOrThrow(final JsonObject jsonObject) thr
                 .orElseThrow(() -> new NotSerializableException("No type found for inner JSON!"));
     }
 
+    /**
+     * Coarse, low-cardinality classification of a (de)serialized message, used as the {@code category} metric tag to
+     * understand the composition of cluster (de)serialization traffic - e.g. distinguishing event fan-out from command
+     * load. It is combined with the {@code resource_type} tag (derived from {@link WithResource#getResourceType()}).
+     */
+    private enum Category {
+
+        EVENT("event"),
+        COMMAND("command"),
+        RESPONSE("response"),
+        ACKNOWLEDGEMENT("acknowledgement"),
+        ANNOUNCEMENT("announcement"),
+        ERROR("error"),
+        OTHER("other");
+
+        private final String tag;
+
+        Category(final String tag) {
+            this.tag = tag;
+        }
+
+        private String getTag() {
+            return tag;
+        }
+
+        private static Category of(final Object object) {
+            final Category result;
+            // order matters: more specific types must be checked first (e.g. an Acknowledgement is a CommandResponse):
+            if (object instanceof Event) {
+                result = EVENT;
+            } else if (object instanceof Announcement) {
+                result = ANNOUNCEMENT;
+            } else if (object instanceof Acknowledgement || object instanceof Acknowledgements) {
+                result = ACKNOWLEDGEMENT;
+            } else if (object instanceof CommandResponse) {
+                result = RESPONSE;
+            } else if (object instanceof Command) {
+                result = COMMAND;
+            } else if (object instanceof DittoRuntimeException) {
+                result = ERROR;
+            } else {
+                result = OTHER;
+            }
+            return result;
+        }
+    }
+
 }

internal/utils/cluster/src/test/java/org/eclipse/ditto/internal/utils/cluster/SharedJsonifiableSerializerTest.java

@@ -14,6 +14,7 @@
 
 import static org.assertj.core.api.Assertions.assertThat;
 
+import java.util.List;
 import java.util.Map;
 
 import org.assertj.core.api.AutoCloseableSoftAssertions;
@@ -27,6 +28,10 @@
 import org.eclipse.ditto.base.model.signals.ShardedMessageEnvelope;
 import org.eclipse.ditto.base.model.signals.commands.GlobalCommandRegistry;
 import org.eclipse.ditto.base.model.signals.commands.GlobalCommandResponseRegistry;
+import org.eclipse.ditto.base.model.signals.events.GlobalEventRegistry;
+import org.eclipse.ditto.internal.utils.metrics.DittoMetrics;
+import org.eclipse.ditto.internal.utils.metrics.instruments.tag.Tag;
+import org.eclipse.ditto.internal.utils.metrics.instruments.tag.TagSet;
 import org.eclipse.ditto.internal.utils.tracing.DittoTracingInitResource;
 import org.eclipse.ditto.json.JsonObject;
 import org.eclipse.ditto.things.model.Thing;
@@ -35,6 +40,7 @@
 import org.eclipse.ditto.things.model.signals.commands.modify.CreateThing;
 import org.eclipse.ditto.things.model.signals.commands.modify.CreateThingResponse;
 import org.eclipse.ditto.things.model.signals.commands.query.RetrieveThings;
+import org.eclipse.ditto.things.model.signals.events.ThingCreated;
 import org.junit.AfterClass;
 import org.junit.Before;
 import org.junit.BeforeClass;
@@ -58,19 +64,25 @@ public final class SharedJsonifiableSerializerTest {
 
     private static enum SerializerImplementation {
 
-        JSONIFIABLE_SERIALIZER {
+        JSONIFIABLE_SERIALIZER("json_serializer_messages") {
             @Override
             public AbstractJsonifiableWithDittoHeadersSerializer getInstance(final ExtendedActorSystem actorSystem) {
                 return new JsonJsonifiableSerializer(actorSystem);
             }
         },
-        CBOR_JSONIFIABLE_SERIALIZER {
+        CBOR_JSONIFIABLE_SERIALIZER("cbor_serializer_messages") {
             @Override
             public AbstractJsonifiableWithDittoHeadersSerializer getInstance(final ExtendedActorSystem actorSystem) {
                 return new CborJsonifiableSerializer(actorSystem);
             }
         };
 
+        private final String metricName;
+
+        SerializerImplementation(final String metricName) {
+            this.metricName = metricName;
+        }
+
         abstract AbstractJsonifiableWithDittoHeadersSerializer getInstance(ExtendedActorSystem actorSystem);
 
     }
@@ -251,4 +263,109 @@ private static final class DittoHeadersStrategy extends MappingStrategies {
 
     }
 
+    /**
+     * Verifies that the {@code <serializer>_serializer_messages} counter is tagged with the expected
+     * {@code direction}, coarse signal {@code category} and {@code resource_type} for the (de)serialized signal.
+     */
+    @RunWith(Parameterized.class)
+    public static final class CategoryAndResourceTypeTagTest {
+
+        private static final String TAG_DIRECTION = "direction";
+        private static final String TAG_CATEGORY = "category";
+        private static final String TAG_RESOURCE_TYPE = "resource_type";
+
+        @ClassRule
+        public static final DittoTracingInitResource DITTO_TRACING_INIT_RESOURCE =
+                DittoTracingInitResource.disableDittoTracing();
+
+        private static Thing thing;
+        private static ExtendedActorSystem actorSystem;
+
+        @Parameterized.Parameter
+        public SerializerImplementation serializerImplementation;
+
+        private AbstractJsonifiableWithDittoHeadersSerializer underTest;
+
+        @Parameterized.Parameters(name = "{0}")
+        public static SerializerImplementation[] getSerializers() {
+            return SerializerImplementation.values();
+        }
+
+        @BeforeClass
+        public static void setUpClass() {
+            thing = Thing.newBuilder().setId(ThingId.generateRandom()).build();
+            actorSystem = getActorSystem(ThingSignalsStrategy.class);
+        }
+
+        @AfterClass
+        public static void tearDownClass() {
+            TestKit.shutdownActorSystem(actorSystem);
+        }
+
+        @Before
+        public void setUp() {
+            underTest = serializerImplementation.getInstance(actorSystem);
+        }
+
+        @Test
+        public void thingCommandIsTaggedAsCommandOnThingResource() {
+            assertRoundTripTaggedWith(CreateThing.of(thing, null, DITTO_HEADERS), "command", "thing");
+        }
+
+        @Test
+        public void thingCommandResponseIsTaggedAsResponseOnThingResource() {
+            assertRoundTripTaggedWith(CreateThingResponse.of(thing, DITTO_HEADERS), "response", "thing");
+        }
+
+        @Test
+        public void thingEventIsTaggedAsEventOnThingResource() {
+            assertRoundTripTaggedWith(ThingCreated.of(thing, 1L, null, DITTO_HEADERS, null), "event", "thing");
+        }
+
+        private void assertRoundTripTaggedWith(final Object signal, final String expectedCategory,
+                final String expectedResourceType) {
+
+            final long outBefore = counterCount("out", expectedCategory, expectedResourceType);
+            final long inBefore = counterCount("in", expectedCategory, expectedResourceType);
+
+            final byte[] serialized = underTest.toBinary(signal);
+            final Object deserialized = underTest.fromBinary(serialized, underTest.manifest(signal));
+
+            try (final AutoCloseableSoftAssertions softly = new AutoCloseableSoftAssertions()) {
+                softly.assertThat(deserialized)
+                        .as("round-trip result")
+                        .isEqualTo(signal);
+                softly.assertThat(counterCount("out", expectedCategory, expectedResourceType))
+                        .as("'out' count for category=<%s>, resource_type=<%s>", expectedCategory, expectedResourceType)
+                        .isEqualTo(outBefore + 1);
+                softly.assertThat(counterCount("in", expectedCategory, expectedResourceType))
+                        .as("'in' count for category=<%s>, resource_type=<%s>", expectedCategory, expectedResourceType)
+                        .isEqualTo(inBefore + 1);
+            }
+        }
+
+        private long counterCount(final String direction, final String category, final String resourceType) {
+            final TagSet tags = TagSet.ofTagCollection(List.of(
+                    Tag.of(TAG_DIRECTION, direction),
+                    Tag.of(TAG_CATEGORY, category),
+                    Tag.of(TAG_RESOURCE_TYPE, resourceType)));
+            return DittoMetrics.counter(serializerImplementation.metricName, tags).getCount();
+        }
+
+        private static final class ThingSignalsStrategy extends MappingStrategies {
+
+            ThingSignalsStrategy() {
+                super(MappingStrategiesBuilder.newInstance()
+                        .add(GlobalErrorRegistry.getInstance())
+                        .add(GlobalCommandRegistry.getInstance())
+                        .add(GlobalCommandResponseRegistry.getInstance())
+                        .add(GlobalEventRegistry.getInstance())
+                        .add(Thing.class, ThingsModelFactory::newThing)
+                        .build());
+            }
+
+        }
+
+    }
+
 }

internal/utils/metrics/src/main/java/org/eclipse/ditto/internal/utils/metrics/instruments/counter/KamonCounter.java

@@ -37,6 +37,14 @@ public final class KamonCounter implements Counter {
     private final String name;
     private final TagSet tags;
 
+    /**
+     * Memoized resolved Kamon counter. {@code name} and {@code tags} are immutable (and {@link #tag(Tag)} /
+     * {@link #tags(TagSet)} return new instances), so the resolved instrument is invariant and can be cached to avoid
+     * re-resolving it (registry lookup + tag-set conversion) on every {@link #increment()}.
+     */
+    @SuppressWarnings("squid:S3077")
+    private transient volatile kamon.metric.Counter kamonInternalCounter;
+
     private KamonCounter(final String name, final TagSet tags) {
         this.name = argumentNotEmpty(name, "name");
         this.tags = checkNotNull(tags, "tags");
@@ -102,7 +110,13 @@ private long getSnapshot() {
     }
 
     private kamon.metric.Counter getKamonInternalCounter() {
-        return Kamon.counter(name).withTags(KamonTagSetConverter.getKamonTagSet(tags));
+        kamon.metric.Counter result = kamonInternalCounter;
+        if (null == result) {
+            // benign race: for a given (name, tags) Kamon returns the same instrument, so a redundant resolve is safe:
+            result = Kamon.counter(name).withTags(KamonTagSetConverter.getKamonTagSet(tags));
+            kamonInternalCounter = result;
+        }
+        return result;
     }
 
     @Override