ipc: add option to disable inflight (#1150)

The inflight metrics are poorly modeled and can often be
confusing. They do not aggregate as expected so aren't
really meaningful unless you understand the full set of
dimensions used. Add option to disable them. Cannot
remove for now due to backwards compatibility.

spectator-ext-ipc/src/main/java/com/netflix/spectator/ipc/IpcLogEntry.java

@@ -156,7 +156,7 @@ public IpcLogEntry withLatency(long latency, TimeUnit unit) {
    * the request completes it is recommended to call {@link #markEnd()}.
    */
   public IpcLogEntry markStart() {
-    if (registry != null && !disableMetrics) {
+    if (registry != null && !disableMetrics && logger.inflightEnabled()) {
       inflightId = getInflightId();
       int n = logger.inflightRequests(inflightId).incrementAndGet();
       registry.distributionSummary(inflightId).record(n);

spectator-ext-ipc/src/main/java/com/netflix/spectator/ipc/IpcLogger.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2019 Netflix, Inc.
+ * Copyright 2014-2024 Netflix, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -47,32 +47,40 @@ public class IpcLogger {
   private final Registry registry;
   private final Clock clock;
   private final Logger logger;
+  private final IpcLoggerConfig config;
 
+  private final boolean inflightEnabled;
   private final ConcurrentHashMap<Id, AtomicInteger> inflightRequests;
   private final ConcurrentHashMap<String, Function<String, String>> limiters;
 
   private final LinkedBlockingQueue<IpcLogEntry> entries;
 
   /**
-   * Create a new instance. Allows the clock to be explicitly set for unit tests.
+   * Create a new instance.
    */
-  IpcLogger(Registry registry, Clock clock, Logger logger) {
+  IpcLogger(Registry registry, Logger logger, IpcLoggerConfig config) {
     this.registry = registry;
-    this.clock = clock;
+    this.clock = registry.clock();
     this.logger = logger;
+    this.config = config;
+    this.inflightEnabled = config.inflightMetricsEnabled();
     this.inflightRequests = new ConcurrentHashMap<>();
     this.limiters = new ConcurrentHashMap<>();
-    this.entries = new LinkedBlockingQueue<>(1000);
+    this.entries = new LinkedBlockingQueue<>(config.entryQueueSize());
   }
 
   /** Create a new instance. */
   public IpcLogger(Registry registry, Logger logger) {
-    this(registry, registry.clock(), logger);
+    this(registry, logger, k -> null);
   }
 
   /** Create a new instance. */
   public IpcLogger(Registry registry) {
-    this(registry, registry.clock(), LoggerFactory.getLogger(IpcLogger.class));
+    this(registry, LoggerFactory.getLogger(IpcLogger.class), k -> null);
+  }
+
+  boolean inflightEnabled() {
+    return inflightEnabled;
   }
 
   /** Return the number of inflight requests associated with the given id. */
@@ -85,7 +93,10 @@ AtomicInteger inflightRequests(Id id) {
    * backend from a metrics explosion if some dimensions have a high cardinality.
    */
   Function<String, String> limiterForKey(String key) {
-    return Utils.computeIfAbsent(limiters, key, k -> CardinalityLimiters.rollup(25));
+    return Utils.computeIfAbsent(limiters, key, k -> {
+      final int n = config.cardinalityLimit(k);
+      return CardinalityLimiters.rollup(n);
+    });
   }
 
   private IpcLogEntry newEntry() {

spectator-ext-ipc/src/main/java/com/netflix/spectator/ipc/IpcLoggerConfig.java

@@ -0,0 +1,76 @@
+/*
+ * Copyright 2014-2024 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 com.netflix.spectator.ipc;
+
+/**
+ * Configuration settings for IpcLogger.
+ */
+public interface IpcLoggerConfig {
+
+  /**
+   * Get the value for a given key or return null if it is not present.
+   */
+  String get(String key);
+
+  /**
+   * Get the value for a given key or return the specified default if it is not present.
+   */
+  default String get(String key, String dflt) {
+    String v = get(key);
+    return v == null ? dflt : v;
+  }
+
+  /**
+   * Get the value for a given key or return the specified default if it is not present.
+   */
+  default int getInt(String key, int dflt) {
+    int n = dflt;
+    String v = get(key, Integer.toString(n));
+    try {
+      n = Integer.parseInt(v);
+    } catch (NumberFormatException ignored) {
+    }
+    return n;
+  }
+
+  /**
+   * Determines whether the metrics for number of in flight requests will be generated.
+   * These metrics are poorly modeled and often lead to confusion. To estimate number
+   * of in flight requests for a given scope, Little's law can be used along with the
+   * latency metrics. Defaults to true for backwards compatibility.
+   */
+  default boolean inflightMetricsEnabled() {
+    String v = get("spectator.ipc.inflight-metrics-enabled", "true");
+    return "true".equals(v);
+  }
+
+  /**
+   * Limit to use for the cardinality limiter applied on a given tag key. IPC metrics
+   * have many dimensions and a high volume, be careful with increasing the limit as
+   * it can easily cause a large increase.
+   */
+  default int cardinalityLimit(String tagKey) {
+    String propKey = "spectator.ipc.cardinality-limit." + tagKey;
+    return getInt(propKey, 25);
+  }
+
+  /**
+   * Size to use for the queue of reusable logger entries.
+   */
+  default int entryQueueSize() {
+    return getInt("spectator.ipc.entry-queue-size", 1000);
+  }
+}

spectator-ext-ipc/src/test/java/com/netflix/spectator/ipc/IpcLogEntryTest.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2014-2023 Netflix, Inc.
+ * Copyright 2014-2024 Netflix, Inc.
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -22,16 +22,19 @@
 import com.netflix.spectator.api.ManualClock;
 import com.netflix.spectator.api.Registry;
 import com.netflix.spectator.api.Utils;
+import com.netflix.spectator.api.patterns.CardinalityLimiters;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
 import java.net.URI;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 
 public class IpcLogEntryTest {
 
@@ -660,11 +663,67 @@ public void responseContentLength() {
     Assertions.assertEquals(expected, actual.longValue());
   }
 
+  @Test
+  public void keyLimiterDefault() {
+    Registry registry = new DefaultRegistry(clock);
+    IpcLogger logger = new IpcLogger(registry, LoggerFactory.getLogger(getClass()));
+    for (int i = 0; i < 200; ++i) {
+      logger.createClientEntry()
+          .markStart()
+          .addTag("id", Integer.toString(i))
+          .markEnd()
+          .log();
+    }
+    AtomicInteger count = new AtomicInteger();
+    registry
+        .counters()
+        .filter(c -> "ipc.client.call".equals(c.id().name()))
+        .forEach(c -> {
+          count.incrementAndGet();
+          String id = Utils.getTagValue(c.id(), "id");
+          if (CardinalityLimiters.AUTO_ROLLUP.equals(id)) {
+            Assertions.assertEquals(175, c.count());
+          } else {
+            Assertions.assertEquals(1, c.count());
+          }
+        });
+    Assertions.assertEquals(26, count.get());
+  }
+
+  @Test
+  public void keyLimiterCustom() {
+    Map<String, String> props = Collections
+        .singletonMap("spectator.ipc.cardinality-limit.id", "100");
+    Registry registry = new DefaultRegistry(clock);
+    IpcLogger logger = new IpcLogger(registry, LoggerFactory.getLogger(getClass()), props::get);
+    for (int i = 0; i < 200; ++i) {
+      logger.createClientEntry()
+          .markStart()
+          .addTag("id", Integer.toString(i))
+          .markEnd()
+          .log();
+    }
+    AtomicInteger count = new AtomicInteger();
+    registry
+        .counters()
+        .filter(c -> "ipc.client.call".equals(c.id().name()))
+        .forEach(c -> {
+          count.incrementAndGet();
+          String id = Utils.getTagValue(c.id(), "id");
+          if (CardinalityLimiters.AUTO_ROLLUP.equals(id)) {
+            Assertions.assertEquals(100, c.count());
+          } else {
+            Assertions.assertEquals(1, c.count());
+          }
+        });
+    Assertions.assertEquals(101, count.get());
+  }
+
   @Test
   public void inflightRequests() {
-    Registry registry = new DefaultRegistry();
+    Registry registry = new DefaultRegistry(clock);
     DistributionSummary summary = registry.distributionSummary("ipc.client.inflight");
-    IpcLogger logger = new IpcLogger(registry, clock, LoggerFactory.getLogger(getClass()));
+    IpcLogger logger = new IpcLogger(registry, LoggerFactory.getLogger(getClass()));
     IpcLogEntry logEntry = logger.createClientEntry();
 
     Assertions.assertEquals(0L, summary.totalAmount());
@@ -674,11 +733,27 @@ public void inflightRequests() {
     Assertions.assertEquals(1L, summary.totalAmount());
   }
 
+  @Test
+  public void inflightRequestsDisabled() {
+    Map<String, String> props = Collections
+        .singletonMap("spectator.ipc.inflight-metrics-enabled", "false");
+    Registry registry = new DefaultRegistry(clock);
+    DistributionSummary summary = registry.distributionSummary("ipc.client.inflight");
+    IpcLogger logger = new IpcLogger(registry, LoggerFactory.getLogger(getClass()), props::get);
+    IpcLogEntry logEntry = logger.createClientEntry();
+
+    Assertions.assertEquals(0L, summary.totalAmount());
+    logEntry.markStart();
+    Assertions.assertEquals(0L, summary.totalAmount());
+    logEntry.markEnd();
+    Assertions.assertEquals(0L, summary.totalAmount());
+  }
+
   @Test
   public void inflightRequestsMany() {
-    Registry registry = new DefaultRegistry();
+    Registry registry = new DefaultRegistry(clock);
     DistributionSummary summary = registry.distributionSummary("ipc.client.inflight");
-    IpcLogger logger = new IpcLogger(registry, clock, LoggerFactory.getLogger(getClass()));
+    IpcLogger logger = new IpcLogger(registry, LoggerFactory.getLogger(getClass()));
 
     for (int i = 0; i < 10; ++i) {
       logger.createClientEntry().markStart().markEnd();
@@ -689,8 +764,8 @@ public void inflightRequestsMany() {
 
   @Test
   public void clientMetricsValidate() {
-    Registry registry = new DefaultRegistry();
-    IpcLogger logger = new IpcLogger(registry, clock, LoggerFactory.getLogger(getClass()));
+    Registry registry = new DefaultRegistry(clock);
+    IpcLogger logger = new IpcLogger(registry, LoggerFactory.getLogger(getClass()));
 
     logger.createClientEntry()
         .withOwner("test")
@@ -703,8 +778,8 @@ public void clientMetricsValidate() {
 
   @Test
   public void clientMetricsValidateHttpSuccess() {
-    Registry registry = new DefaultRegistry();
-    IpcLogger logger = new IpcLogger(registry, clock, LoggerFactory.getLogger(getClass()));
+    Registry registry = new DefaultRegistry(clock);
+    IpcLogger logger = new IpcLogger(registry, LoggerFactory.getLogger(getClass()));
 
     logger.createClientEntry()
         .withOwner("test")
@@ -720,8 +795,8 @@ public void clientMetricsValidateHttpSuccess() {
 
   @Test
   public void clientMetricsDisbled() {
-    Registry registry = new DefaultRegistry();
-    IpcLogger logger = new IpcLogger(registry, clock, LoggerFactory.getLogger(getClass()));
+    Registry registry = new DefaultRegistry(clock);
+    IpcLogger logger = new IpcLogger(registry, LoggerFactory.getLogger(getClass()));
 
     logger.createClientEntry()
         .withOwner("test")
@@ -735,8 +810,8 @@ public void clientMetricsDisbled() {
 
   @Test
   public void serverMetricsValidate() {
-    Registry registry = new DefaultRegistry();
-    IpcLogger logger = new IpcLogger(registry, clock, LoggerFactory.getLogger(getClass()));
+    Registry registry = new DefaultRegistry(clock);
+    IpcLogger logger = new IpcLogger(registry, LoggerFactory.getLogger(getClass()));
 
     logger.createServerEntry()
         .withOwner("test")
@@ -749,8 +824,8 @@ public void serverMetricsValidate() {
 
   @Test
   public void serverMetricsDisabled() {
-    Registry registry = new DefaultRegistry();
-    IpcLogger logger = new IpcLogger(registry, clock, LoggerFactory.getLogger(getClass()));
+    Registry registry = new DefaultRegistry(clock);
+    IpcLogger logger = new IpcLogger(registry, LoggerFactory.getLogger(getClass()));
 
     logger.createServerEntry()
         .withOwner("test")
@@ -764,8 +839,8 @@ public void serverMetricsDisabled() {
 
   @Test
   public void serverMetricsDisabledViaHeader() {
-    Registry registry = new DefaultRegistry();
-    IpcLogger logger = new IpcLogger(registry, clock, LoggerFactory.getLogger(getClass()));
+    Registry registry = new DefaultRegistry(clock);
+    IpcLogger logger = new IpcLogger(registry, LoggerFactory.getLogger(getClass()));
 
     logger.createServerEntry()
         .withOwner("test")
@@ -779,8 +854,8 @@ public void serverMetricsDisabledViaHeader() {
 
   @Test
   public void serverMetricsDisabledReuseEntry() {
-    Registry registry = new DefaultRegistry();
-    IpcLogger logger = new IpcLogger(registry, clock, LoggerFactory.getLogger(getClass()));
+    Registry registry = new DefaultRegistry(clock);
+    IpcLogger logger = new IpcLogger(registry, LoggerFactory.getLogger(getClass()));
 
     logger.createServerEntry()
         .withOwner("test")
@@ -801,8 +876,8 @@ public void serverMetricsDisabledReuseEntry() {
 
   @Test
   public void endpointUnknownIfNotSet() {
-    Registry registry = new DefaultRegistry();
-    IpcLogger logger = new IpcLogger(registry, clock, LoggerFactory.getLogger(getClass()));
+    Registry registry = new DefaultRegistry(clock);
+    IpcLogger logger = new IpcLogger(registry, LoggerFactory.getLogger(getClass()));
 
     logger.createServerEntry()
         .withOwner("test")