Track events filtered out by MQL/subscription queries in mantis-publish-core

mantis-publish/mantis-publish-core/src/main/java/io/mantisrx/publish/EventProcessor.java

@@ -20,9 +20,11 @@
 import io.mantisrx.publish.api.StreamType;
 import io.mantisrx.publish.config.MrePublishConfiguration;
 import io.mantisrx.publish.core.Subscription;
+import io.mantisrx.publish.internal.metrics.StreamMetrics;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
+import java.util.Optional;
 import java.util.Random;
 import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -90,14 +92,18 @@ public Event process(String stream, Event event) {
         List<Subscription> matchingSubscriptions = new ArrayList<>();
         if (streamManager.hasSubscriptions(stream)) {
             final Set<Subscription> streamSubscriptions = streamManager.getStreamSubscriptions(stream);
+            final Optional<StreamMetrics> streamMetricsOpt = streamManager.getStreamMetrics(stream);
 
             for (Subscription s : streamSubscriptions) {
                 try {
                     if (s.matches(event)) {
                         matchingSubscriptions.add(s);
+                    } else {
+                        streamMetricsOpt
+                                .ifPresent(m -> m.getMantisEventsFilteredCounter(s.getSubscriptionId()).increment());
                     }
                 } catch (Exception e) {
-                    streamManager.getStreamMetrics(stream)
+                    streamMetricsOpt
                             .ifPresent(m -> m.getMantisQueryFailedCounter().increment());
 
                     // Send errors only for a sample of events.

mantis-publish/mantis-publish-core/src/main/java/io/mantisrx/publish/internal/metrics/StreamMetrics.java

@@ -19,13 +19,27 @@
 import com.netflix.spectator.api.Counter;
 import com.netflix.spectator.api.Registry;
 import com.netflix.spectator.api.Timer;
+import com.netflix.spectator.api.patterns.CardinalityLimiters;
 import com.netflix.spectator.impl.AtomicDouble;
+import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.Function;
 
 
 public class StreamMetrics {
 
+    /**
+     * Bounds the number of distinct {@code subscriptionId} tag values registered for the
+     * {@code mantisEventsFiltered} counter. Matches the limit used by {@code MqlEvalStage} on
+     * the Mantis source-job side (Netflix-internal PR #949).
+     */
+    private static final int SUBSCRIPTION_ID_CARDINALITY_LIMIT = 50;
+
     private final String streamName;
+    private final Registry registry;
+    private final Function<String, String> subscriptionIdLimiter =
+            CardinalityLimiters.mostFrequent(SUBSCRIPTION_ID_CARDINALITY_LIMIT);
+    private final ConcurrentHashMap<String, Counter> mantisEventsFilteredCounters = new ConcurrentHashMap<>();
 
     private final Counter mantisEventsDroppedCounter;
     private final Counter mantisEventsDroppedProcessingExceptionCounter;
@@ -42,6 +56,7 @@ public class StreamMetrics {
 
     public StreamMetrics(Registry registry, final String streamName) {
         this.streamName = streamName;
+        this.registry = registry;
 
         this.mantisEventsDroppedCounter = SpectatorUtils.buildAndRegisterCounter(
                 registry, "mantisEventsDropped", "stream", streamName, "reason", "publisherQueueFull");
@@ -99,6 +114,29 @@ public Counter getMantisQueryProjectionFailedCounter() {
         return mantisQueryProjectionFailedCounter;
     }
 
+    /**
+     * Returns the per-subscription "events filtered out" counter for the given
+     * subscription id. The subscriptionId tag value is cardinality-limited; ids
+     * beyond the limit collapse into a single "--others--" bucket. Counters are
+     * cached one per (limited) id for the lifetime of this StreamMetrics.
+     *
+     * @param subscriptionId the id of the subscription whose query did not match
+     * @return a Spectator Counter tagged stream=<streamName>, subscriptionId=<limited id>
+     */
+    public Counter getMantisEventsFilteredCounter(String subscriptionId) {
+        // CardinalityLimiters.mostFrequent(...) is backed by a non-thread-safe LinkedHashMap;
+        // this class is shared across concurrently-processed events, so apply() must be
+        // synchronized to avoid corrupting its internal LRU state.
+        String limitedId;
+        synchronized (subscriptionIdLimiter) {
+            limitedId = subscriptionIdLimiter.apply(subscriptionId);
+        }
+        return mantisEventsFilteredCounters.computeIfAbsent(
+                limitedId,
+                lid -> SpectatorUtils.buildAndRegisterCounter(
+                        registry, "mantisEventsFiltered", "stream", streamName, "subscriptionId", lid));
+    }
+
     public AtomicDouble getMantisEventsQueuedGauge() {
         return mantisEventsQueuedGauge;
     }

mantis-publish/mantis-publish-core/src/test/java/io/mantisrx/publish/EventProcessorTest.java

@@ -29,11 +29,13 @@
 import com.netflix.archaius.api.PropertyRepository;
 import com.netflix.archaius.api.config.SettableConfig;
 import com.netflix.archaius.config.DefaultSettableConfig;
+import com.netflix.spectator.api.DefaultRegistry;
 import io.mantisrx.publish.api.Event;
 import io.mantisrx.publish.api.StreamType;
 import io.mantisrx.publish.config.MrePublishConfiguration;
 import io.mantisrx.publish.config.SampleArchaiusMrePublishConfiguration;
 import io.mantisrx.publish.core.Subscription;
+import io.mantisrx.publish.internal.metrics.StreamMetrics;
 import io.mantisrx.publish.internal.mql.MQLSubscription;
 import java.time.Instant;
 import java.util.*;
@@ -77,11 +79,11 @@ void shouldReturnEnrichedEventForStream() throws Exception {
         event.set("k1", "v1");
         Event actual = eventProcessor.process(StreamType.DEFAULT_EVENT_STREAM, event);
         // Single event with a `select * where true` yields the single event.
-        assertEquals(actual.get("mantisStream"), StreamType.DEFAULT_EVENT_STREAM);
-        assertEquals(actual.get("type"), "EVENT");
-        assertEquals(actual.get("k1"), "v1");
-        assertEquals(((ArrayList)actual.get("matched-clients")).size(), 1);
-        assertEquals(((ArrayList)actual.get("matched-clients")).get(0), "id");
+        assertEquals(StreamType.DEFAULT_EVENT_STREAM, actual.get("mantisStream"));
+        assertEquals("EVENT", actual.get("type"));
+        assertEquals("v1", actual.get("k1"));
+        assertEquals(1, ((ArrayList)actual.get("matched-clients")).size());
+        assertEquals("id", ((ArrayList)actual.get("matched-clients")).get(0));
     }
 
     @Test
@@ -104,6 +106,89 @@ void shouldReturnEmptyEventForStream() throws Exception {
         assertNull(actual);
     }
 
+    @Test
+    void shouldIncrementMantisEventsFilteredForNonMatchingSubscription() throws Exception {
+        StreamMetrics metrics = new StreamMetrics(new DefaultRegistry(), StreamType.DEFAULT_EVENT_STREAM);
+        when(streamManager.getStreamMetrics(anyString())).thenReturn(Optional.of(metrics));
+        when(streamManager.hasSubscriptions(anyString())).thenReturn(true);
+
+        Subscription subscription = mock(MQLSubscription.class);
+        when(subscription.getSubscriptionId()).thenReturn("nonMatchingSub");
+        when(subscription.matches(any(Event.class))).thenReturn(false);
+        Set<Subscription> subscriptions = new ConcurrentSkipListSet<>();
+        subscriptions.add(subscription);
+        when(streamManager.getStreamSubscriptions(anyString())).thenReturn(subscriptions);
+
+        Event event = new Event();
+        event.set("k1", "v1");
+        Event actual = eventProcessor.process(StreamType.DEFAULT_EVENT_STREAM, event);
+
+        assertNull(actual);
+        assertEquals(1, metrics.getMantisEventsFilteredCounter("nonMatchingSub").count());
+    }
+
+    @Test
+    void shouldNotIncrementMantisEventsFilteredForMatchingSubscription() throws Exception {
+        StreamMetrics metrics = new StreamMetrics(new DefaultRegistry(), StreamType.DEFAULT_EVENT_STREAM);
+        when(streamManager.getStreamMetrics(anyString())).thenReturn(Optional.of(metrics));
+        when(streamManager.hasSubscriptions(anyString())).thenReturn(true);
+
+        Subscription subscription = new MQLSubscription("matchingSub", "select * where true");
+        Set<Subscription> subscriptions = new ConcurrentSkipListSet<>();
+        subscriptions.add(subscription);
+        when(streamManager.getStreamSubscriptions(anyString())).thenReturn(subscriptions);
+
+        Event event = new Event();
+        event.set("k1", "v1");
+        eventProcessor.process(StreamType.DEFAULT_EVENT_STREAM, event);
+
+        assertEquals(0, metrics.getMantisEventsFilteredCounter("matchingSub").count());
+    }
+
+    @Test
+    void shouldOnlyIncrementMantisEventsFilteredForNonMatchingSubscriptionInMixedSet() throws Exception {
+        StreamMetrics metrics = new StreamMetrics(new DefaultRegistry(), StreamType.DEFAULT_EVENT_STREAM);
+        when(streamManager.getStreamMetrics(anyString())).thenReturn(Optional.of(metrics));
+        when(streamManager.hasSubscriptions(anyString())).thenReturn(true);
+
+        Subscription matchingSubscription = new MQLSubscription("matchingSub", "select * where true");
+        Subscription nonMatchingSubscription = new MQLSubscription("nonMatchingSub", "select * where false");
+        Set<Subscription> subscriptions = new ConcurrentSkipListSet<>();
+        subscriptions.add(matchingSubscription);
+        subscriptions.add(nonMatchingSubscription);
+        when(streamManager.getStreamSubscriptions(anyString())).thenReturn(subscriptions);
+
+        Event event = new Event();
+        event.set("k1", "v1");
+        Event actual = eventProcessor.process(StreamType.DEFAULT_EVENT_STREAM, event);
+
+        assertEquals(0, metrics.getMantisEventsFilteredCounter("matchingSub").count());
+        assertEquals(1, metrics.getMantisEventsFilteredCounter("nonMatchingSub").count());
+        assertEquals(StreamType.DEFAULT_EVENT_STREAM, actual.get("mantisStream"));
+    }
+
+    @Test
+    void shouldNotIncrementMantisEventsFilteredWhenMatchesThrows() throws Exception {
+        StreamMetrics metrics = new StreamMetrics(new DefaultRegistry(), StreamType.DEFAULT_EVENT_STREAM);
+        when(streamManager.getStreamMetrics(anyString())).thenReturn(Optional.of(metrics));
+        when(streamManager.hasSubscriptions(anyString())).thenReturn(true);
+
+        Subscription subscription = mock(MQLSubscription.class);
+        when(subscription.getSubscriptionId()).thenReturn("throwingSub");
+        when(subscription.matches(any(Event.class))).thenThrow(new RuntimeException("query failed"));
+        Set<Subscription> subscriptions = new ConcurrentSkipListSet<>();
+        subscriptions.add(subscription);
+        when(streamManager.getStreamSubscriptions(anyString())).thenReturn(subscriptions);
+
+        Event event = new Event();
+        event.set("k1", "v1");
+        Event actual = eventProcessor.process(StreamType.DEFAULT_EVENT_STREAM, event);
+
+        assertNull(actual);
+        assertEquals(1, metrics.getMantisQueryFailedCounter().count());
+        assertEquals(0, metrics.getMantisEventsFilteredCounter("throwingSub").count());
+    }
+
     @Test
     void shouldMaskSensitiveFields() {
         Map<String, Object> data = new HashMap<>();

mantis-publish/mantis-publish-core/src/test/java/io/mantisrx/publish/internal/metrics/StreamMetricsTest.java

@@ -0,0 +1,115 @@
+/*
+ * Copyright 2019 Netflix, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.mantisrx.publish.internal.metrics;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertSame;
+
+import com.netflix.spectator.api.Counter;
+import com.netflix.spectator.api.DefaultRegistry;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+
+class StreamMetricsTest {
+
+    private static final String STREAM_NAME = "testStream";
+
+    private DefaultRegistry registry;
+    private StreamMetrics streamMetrics;
+
+    @BeforeEach
+    void setUp() {
+        registry = new DefaultRegistry();
+        streamMetrics = new StreamMetrics(registry, STREAM_NAME);
+    }
+
+    @Test
+    void testGetMantisEventsFilteredCounterIsRegisteredWithExpectedTags() {
+        Counter counter = streamMetrics.getMantisEventsFilteredCounter("sub1");
+        assertNotNull(counter);
+
+        counter.increment();
+
+        Counter registeredCounter = registry.counter(
+                registry.createId("mantisEventsFiltered")
+                        .withTag("stream", STREAM_NAME)
+                        .withTag("subscriptionId", "sub1"));
+        assertEquals(1, registeredCounter.count());
+    }
+
+    @Test
+    void testGetMantisEventsFilteredCounterCachesInstancePerSubscriptionId() {
+        Counter first = streamMetrics.getMantisEventsFilteredCounter("sub1");
+        Counter second = streamMetrics.getMantisEventsFilteredCounter("sub1");
+
+        assertSame(first, second);
+
+        first.increment();
+        second.increment();
+
+        assertEquals(2, first.count());
+    }
+
+    @Test
+    void testGetMantisEventsFilteredCounterReturnsDistinctCountersPerSubscriptionId() {
+        Counter subOneCounter = streamMetrics.getMantisEventsFilteredCounter("sub1");
+        Counter subTwoCounter = streamMetrics.getMantisEventsFilteredCounter("sub2");
+
+        assertNotSame(subOneCounter, subTwoCounter);
+
+        subOneCounter.increment();
+
+        assertEquals(1, subOneCounter.count());
+        assertEquals(0, subTwoCounter.count());
+    }
+
+    @Test
+    void testSubscriptionIdsAboveCardinalityLimitCollapseToOtherBucket() {
+        for (int i = 0; i < 50; i++) {
+            streamMetrics.getMantisEventsFilteredCounter("sub" + i).increment();
+        }
+
+        streamMetrics.getMantisEventsFilteredCounter("overflow-sub").increment();
+
+        Counter otherBucket = registry.counter(
+                registry.createId("mantisEventsFiltered")
+                        .withTag("stream", STREAM_NAME)
+                        .withTag("subscriptionId", "--others--"));
+        assertEquals(1, otherBucket.count(),
+                "subscriptionIds beyond the cardinality limit must be collapsed to '--others--'");
+    }
+
+    @Test
+    void testMultipleOverflowIdsAccumulateInSameBucket() {
+        for (int i = 0; i < 50; i++) {
+            streamMetrics.getMantisEventsFilteredCounter("sub" + i).increment();
+        }
+
+        streamMetrics.getMantisEventsFilteredCounter("overflow-1").increment();
+        streamMetrics.getMantisEventsFilteredCounter("overflow-2").increment();
+
+        Counter otherBucket = registry.counter(
+                registry.createId("mantisEventsFiltered")
+                        .withTag("stream", STREAM_NAME)
+                        .withTag("subscriptionId", "--others--"));
+        assertEquals(2, otherBucket.count(),
+                "all overflow subscriptionIds must accumulate in the single '--others--' bucket");
+    }
+}