diff --git a/mantis-publish/mantis-publish-core/src/main/java/io/mantisrx/publish/EventProcessor.java b/mantis-publish/mantis-publish-core/src/main/java/io/mantisrx/publish/EventProcessor.java index ff104e08c..53d9bafd5 100644 --- a/mantis-publish/mantis-publish-core/src/main/java/io/mantisrx/publish/EventProcessor.java +++ b/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 matchingSubscriptions = new ArrayList<>(); if (streamManager.hasSubscriptions(stream)) { final Set streamSubscriptions = streamManager.getStreamSubscriptions(stream); + final Optional 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. diff --git a/mantis-publish/mantis-publish-core/src/main/java/io/mantisrx/publish/internal/metrics/StreamMetrics.java b/mantis-publish/mantis-publish-core/src/main/java/io/mantisrx/publish/internal/metrics/StreamMetrics.java index bdcc8e28d..b07b52fca 100644 --- a/mantis-publish/mantis-publish-core/src/main/java/io/mantisrx/publish/internal/metrics/StreamMetrics.java +++ b/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 subscriptionIdLimiter = + CardinalityLimiters.mostFrequent(SUBSCRIPTION_ID_CARDINALITY_LIMIT); + private final ConcurrentHashMap 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=, subscriptionId= + */ + 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; } diff --git a/mantis-publish/mantis-publish-core/src/test/java/io/mantisrx/publish/EventProcessorTest.java b/mantis-publish/mantis-publish-core/src/test/java/io/mantisrx/publish/EventProcessorTest.java index 38becee6a..784b64f97 100644 --- a/mantis-publish/mantis-publish-core/src/test/java/io/mantisrx/publish/EventProcessorTest.java +++ b/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 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 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 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 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 data = new HashMap<>(); diff --git a/mantis-publish/mantis-publish-core/src/test/java/io/mantisrx/publish/internal/metrics/StreamMetricsTest.java b/mantis-publish/mantis-publish-core/src/test/java/io/mantisrx/publish/internal/metrics/StreamMetricsTest.java new file mode 100644 index 000000000..fc9ea901c --- /dev/null +++ b/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"); + } +}