Extract stats via /rowStats call, remove /metrics call

Author: Fly-Style

COST_BASED_AUTOSCALER_ANALYSIS.md

@@ -42,7 +42,6 @@
 import org.joda.time.DateTime;
 import org.joda.time.DateTimeZone;
 import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.Timeout;
 
@@ -160,7 +159,7 @@ public void test_autoScaler_computesOptimalTaskCountAndProduceScaleDown()
   }
 
   @Test
-  @Disabled // Actually make it work, it does not ingest poverfully enough
+  // @Disabled // Actually make it work, it does not ingest poverfully enough
   @Timeout(60)
   public void test_autoScaler_computesOptimalTaskCountAndProducesScaleUp()
   {
@@ -182,14 +181,14 @@ public void test_autoScaler_computesOptimalTaskCountAndProducesScaleUp()
         .taskCountMin(1)
         .taskCountMax(50)
         .taskCountStart(lowInitialTaskCount)
-        .metricsCollectionIntervalMillis(200)
+        .metricsCollectionIntervalMillis(500)
         .scaleActionStartDelayMillis(500)
         .scaleActionPeriodMillis(500)
         .minTriggerScaleActionFrequencyMillis(1000)
         // Weight configuration: favor lag as the primary signal for scale-up scenarios
         // High lag with low idle (overloaded) will trigger scale-up
-        .lagWeight(0.7)
-        .idleWeight(0.3)
+        .lagWeight(0.3)
+        .idleWeight(0.7)
         .build();
 
     final KafkaSupervisorSpec kafkaSupervisorSpec = createKafkaSupervisorWithAutoScaler(

embedded-tests/src/test/java/org/apache/druid/testing/embedded/indexing/autoscaler/CostBasedAutoScalerIntegrationTest.java

@@ -4403,48 +4403,6 @@ public void testGetCurrentTotalStats()
     Assert.assertEquals(ImmutableMap.of("task2", ImmutableMap.of("prop2", "val2")), stats.get("1"));
   }
 
-  @Test
-  public void testGetTaskMetrics()
-  {
-    supervisor = getTestableSupervisor(1, 2, true, "PT1H", null, null, false, kafkaHost);
-    supervisor.addTaskGroupToActivelyReadingTaskGroup(
-        supervisor.getTaskGroupIdForPartition(new KafkaTopicPartition(false, topic, 0)),
-        singlePartitionMap(topic, 0, 0L),
-        null,
-        null,
-        ImmutableSet.of("task1"),
-        ImmutableSet.of()
-    );
-
-    supervisor.addTaskGroupToPendingCompletionTaskGroup(
-        supervisor.getTaskGroupIdForPartition(new KafkaTopicPartition(false, topic, 1)),
-        singlePartitionMap(topic, 0, 0L),
-        null,
-        null,
-        ImmutableSet.of("task2"),
-        ImmutableSet.of()
-    );
-
-    EasyMock.expect(taskClient.getMetrics("task1"))
-            .andReturn(Futures.immediateFuture(ImmutableMap.of("pollIdleRatio", 0.5)))
-            .times(1);
-
-    EasyMock.expect(taskClient.getMetrics("task2"))
-            .andReturn(Futures.immediateFuture(ImmutableMap.of("pollIdleRatio", 0.8)))
-            .times(1);
-
-    replayAll();
-
-    Map<String, Map<String, Object>> metrics = supervisor.getTaskMetrics();
-
-    verifyAll();
-
-    Assert.assertEquals(2, metrics.size());
-    Assert.assertEquals(ImmutableSet.of("0", "1"), metrics.keySet());
-    Assert.assertEquals(ImmutableMap.of("task1", ImmutableMap.of("pollIdleRatio", 0.5)), metrics.get("0"));
-    Assert.assertEquals(ImmutableMap.of("task2", ImmutableMap.of("pollIdleRatio", 0.8)), metrics.get("1"));
-  }
-
   @Test
   public void testGetCurrentParseErrors()
   {

extensions-core/kafka-indexing-service/src/test/java/org/apache/druid/indexing/kafka/supervisor/KafkaSupervisorTest.java

@@ -80,13 +80,6 @@ ListenableFuture<TreeMap<Integer, Map<PartitionIdType, SequenceOffsetType>>> get
    */
   ListenableFuture<DateTime> getStartTimeAsync(String id);
 
-  /**
-   * Gets all available metrics for a task.
-   *
-   * @param id task id
-   */
-  ListenableFuture<Map<String, Object>> getMetrics(String id);
-
   /**
    * Pause a task.
    *

indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskClient.java

@@ -257,22 +257,6 @@ public ListenableFuture<DateTime> getStartTimeAsync(String id)
         .go();
   }
 
-  @Override
-  public ListenableFuture<Map<String, Object>> getMetrics(String id)
-  {
-    return makeRequest(id, new RequestBuilder(HttpMethod.GET, "/metrics"))
-        .handler(new BytesFullResponseHandler())
-        .onSuccess(r -> {
-          if (isNullOrEmpty(r.getContent())) {
-            return null;
-          } else {
-            return JacksonUtils.readValue(jsonMapper, r.getContent(), JacksonUtils.TYPE_REFERENCE_MAP_STRING_OBJECT);
-          }
-        })
-        .onNotAvailable(e -> Either.value(null))
-        .go();
-  }
-
   @Override
   public ListenableFuture<Map<PartitionIdType, SequenceOffsetType>> pauseAsync(String id)
   {

indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskClientAsyncImpl.java

@@ -957,7 +957,6 @@ public void onFailure(Throwable t)
     }
     finally {
       try {
-        this.recordSupplier = null;
         if (driver != null) {
           driver.close();
         }
@@ -1665,6 +1664,9 @@ public Map<String, Object> doGetRowStats()
 
     returnMap.put("movingAverages", averagesMap);
     returnMap.put("totals", totalsMap);
+    if (this.recordSupplier != null) {
+      returnMap.put("pollIdleRatio", this.recordSupplier.getPollIdleRatioMetric());
+    }
     return returnMap;
   }
 
@@ -1989,39 +1991,6 @@ public DateTime getStartTime(@Context final HttpServletRequest req)
     return startTime;
   }
 
-  /**
-   * Returns the poll-idle-ratio-avg metric from the underlying record supplier (e.g., Kafka consumer).
-   * This metric indicates how much time the consumer spends idle vs polling:
-   * <ul>
-   *   <li>0 means the consumer is never idle (always consuming)</li>
-   *   <li>1 means the consumer is always idle (not receiving data)</li>
-   * </ul>
-   *
-   * @return HTTP 200 with JSON containing pollIdleRatio, HTTP 503 if supplier not initialized,
-   *         or HTTP 501 if the stream type doesn't support this metric
-   */
-  @GET
-  @Path("/metrics")
-  @Produces(MediaType.APPLICATION_JSON)
-  public Response getMetrics(@Context final HttpServletRequest req)
-  {
-    authorizationCheck(req);
-    final RecordSupplier<PartitionIdType, SequenceOffsetType, RecordType> supplier = recordSupplier;
-    if (supplier == null) {
-      return Response.status(Response.Status.SERVICE_UNAVAILABLE)
-                     .entity(ImmutableMap.of("error", "Record supplier isn't initialized"))
-                     .build();
-    }
-    try {
-      return Response.ok(ImmutableMap.of("pollIdleRatio", supplier.getPollIdleRatioMetric())).build();
-    }
-    catch (UnsupportedOperationException e) {
-      return Response.status(501)
-                     .entity(ImmutableMap.of("error", "Poll idle ratio metric isn't supported by this stream type"))
-                     .build();
-    }
-  }
-
   @VisibleForTesting
   public long getNextCheckpointTime()
   {

indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java

@@ -1481,22 +1481,6 @@ public Map<String, Map<String, Object>> getStats()
     }
   }
 
-  @Override
-  public Map<String, Map<String, Object>> getTaskMetrics()
-  {
-    try {
-      return getCurrentTaskMetrics();
-    }
-    catch (InterruptedException ie) {
-      Thread.currentThread().interrupt();
-      log.warn(ie, "getTaskMetrics() interrupted.");
-      throw new RuntimeException(ie);
-    }
-    catch (ExecutionException ee) {
-      throw new RuntimeException(ee);
-    }
-  }
-
   @Override
   public List<ParseExceptionReport> getParseErrors()
   {
@@ -1592,83 +1576,6 @@ private Map<String, Map<String, Object>> getCurrentTotalStats()
     return allStats;
   }
 
-  /**
-   * Collect metrics from all tasks managed by this supervisor.
-   * Note that {@code StatsFromTaskResult} are reused for metrics collection since
-   * both stats and metrics are represented as {@code Map<String, Object>}.
-   *
-   * @return A map of groupId->taskId->task metrics
-   * @throws InterruptedException
-   * @throws ExecutionException
-   */
-  private Map<String, Map<String, Object>> getCurrentTaskMetrics()
-      throws InterruptedException, ExecutionException
-  {
-    Map<String, Map<String, Object>> allMetrics = new HashMap<>();
-    final List<ListenableFuture<StatsFromTaskResult>> futures = new ArrayList<>();
-    final List<Pair<Integer, String>> groupAndTaskIds = new ArrayList<>();
-
-    for (int groupId : activelyReadingTaskGroups.keySet()) {
-      TaskGroup group = activelyReadingTaskGroups.get(groupId);
-      for (String taskId : group.taskIds()) {
-        futures.add(
-            Futures.transform(
-                taskClient.getMetrics(taskId),
-                (Function<Map<String, Object>, StatsFromTaskResult>) (currentStats) -> new StatsFromTaskResult(
-                    groupId,
-                    taskId,
-                    currentStats
-                ),
-                MoreExecutors.directExecutor()
-            )
-        );
-        groupAndTaskIds.add(new Pair<>(groupId, taskId));
-      }
-    }
-
-    for (int groupId : pendingCompletionTaskGroups.keySet()) {
-      List<TaskGroup> pendingGroups = pendingCompletionTaskGroups.get(groupId);
-      for (TaskGroup pendingGroup : pendingGroups) {
-        for (String taskId : pendingGroup.taskIds()) {
-          futures.add(
-              Futures.transform(
-                  taskClient.getMetrics(taskId),
-                  (Function<Map<String, Object>, StatsFromTaskResult>) (currentStats) -> new StatsFromTaskResult(
-                      groupId,
-                      taskId,
-                      currentStats
-                  ),
-                  MoreExecutors.directExecutor()
-              )
-          );
-          groupAndTaskIds.add(new Pair<>(groupId, taskId));
-        }
-      }
-    }
-
-    List<Either<Throwable, StatsFromTaskResult>> results = coalesceAndAwait(futures);
-    for (int i = 0; i < results.size(); i++) {
-      if (results.get(i).isValue()) {
-        StatsFromTaskResult result = results.get(i).valueOrThrow();
-
-        if (result != null) {
-          Map<String, Object> groupMap = allMetrics.computeIfAbsent(result.getGroupId(), k -> new HashMap<>());
-          groupMap.put(result.getTaskId(), result.getStats());
-        }
-      } else {
-        Pair<Integer, String> groupAndTaskId = groupAndTaskIds.get(i);
-        log.noStackTrace().warn(
-            results.get(i).error(),
-            "Failed to get metrics for group[%d]-task[%s]",
-            groupAndTaskId.lhs,
-            groupAndTaskId.rhs
-        );
-      }
-    }
-
-    return allMetrics;
-  }
-
   /**
    * Collect parse errors from all tasks managed by this supervisor.
    *
@@ -4472,7 +4379,7 @@ public ConcurrentHashMap<PartitionIdType, SequenceOffsetType> getPartitionOffset
    */
   public double getPollIdleRatioMetric()
   {
-    Map<String, Map<String, Object>> taskMetrics = getTaskMetrics();
+    Map<String, Map<String, Object>> taskMetrics = getStats();
     if (taskMetrics.isEmpty()) {
       return 1.;
     }

indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java

@@ -169,7 +169,7 @@ private void collectMetrics()
    * Returns -1 (no scaling needed) in the following cases:
    * <ul>
    *   <li>Metrics are not available</li>
-   *   <li>Current idle ratio is in the ideal range [0.2, 0.6] - optimal utilization achieved</li>
+   *   <li>The current idle ratio is in the ideal range [0.2, 0.6] - optimal utilization achieved</li>
    *   <li>Optimal task count equals current task count</li>
    * </ul>
    *

indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/autoscaler/CostBasedAutoScaler.java

@@ -342,45 +342,6 @@ public void test_getStartTimeAsync_notAvailable() throws Exception
     Assert.assertNull(client.getStartTimeAsync(TASK_ID).get());
   }
 
-  @Test
-  public void test_getMetricsAsync() throws Exception
-  {
-    final Map<String, Object> retVal = ImmutableMap.of("pollIdleRatio", 0.5);
-
-    serviceClient.expectAndRespond(
-        new RequestBuilder(HttpMethod.GET, "/metrics").timeout(httpTimeout),
-        HttpResponseStatus.OK,
-        Collections.emptyMap(),
-        jsonMapper.writeValueAsBytes(retVal)
-    );
-
-    Assert.assertEquals(retVal, client.getMetrics(TASK_ID).get());
-  }
-
-  @Test
-  public void test_getMetricsAsync_empty() throws Exception
-  {
-    serviceClient.expectAndRespond(
-        new RequestBuilder(HttpMethod.GET, "/metrics").timeout(httpTimeout),
-        HttpResponseStatus.OK,
-        Collections.emptyMap(),
-        ByteArrays.EMPTY_ARRAY
-    );
-
-    Assert.assertNull(client.getMetrics(TASK_ID).get());
-  }
-
-  @Test
-  public void test_getMetricsAsync_notAvailable() throws Exception
-  {
-    serviceClient.expectAndThrow(
-        new RequestBuilder(HttpMethod.GET, "/metrics").timeout(httpTimeout),
-        new ServiceNotAvailableException(TASK_ID)
-    );
-
-    Assert.assertNull(client.getMetrics(TASK_ID).get());
-  }
-
   @Test
   public void test_pauseAsync_immediateOk() throws Exception
   {

indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskClientAsyncImplTest.java

@@ -71,11 +71,6 @@ default Map<String, Map<String, Object>> getStats()
     return ImmutableMap.of();
   }
 
-  default Map<String, Map<String, Object>> getTaskMetrics()
-  {
-    return ImmutableMap.of();
-  }
-
   default List<ParseExceptionReport> getParseErrors()
   {
     return ImmutableList.of();