kfaraz commented on code in PR #19651:
URL: https://github.com/apache/druid/pull/19651#discussion_r3556921411
##########
docs/operations/metrics.md:
##########
@@ -317,6 +317,12 @@ batch ingestion emit the following metrics. These metrics
are deltas for each em
|`ingest/notices/time`|Milliseconds taken to process a notice by the
supervisor.|`supervisorId`, `dataSource`, `tags`| < 1s |
|`ingest/pause/time`|Milliseconds spent by a task in a paused state without
ingesting.|`dataSource`, `taskId`, `tags`| < 10 seconds|
|`ingest/handoff/time`|Total number of milliseconds taken to handoff a set of
segments.|`dataSource`, `taskId`, `taskType`, `groupId`, `tags`|Depends on the
coordinator cycle time.|
+|`ingest/realtime/segmentUpgrade/count`|Number of pending segments that a
concurrent replace (for example, compaction) upgraded to a new version and
asked the supervisor to have running tasks announce under the new version.
Emitted by the replace task only when streaming ingestion is running
concurrently with replace on the same interval.|`dataSource`, `taskId`,
`taskType`, `groupId`, `interval`, `version`, `tags`|0 unless [concurrent
append and replace](../ingestion/concurrent-append-replace.md) is in use.|
Review Comment:
We should probably rename this metric from `count` to `persisted` to signify
that this is the count that was actually persisted to the metadata store.
Also, rephrased the text a bit.
```suggestion
|`ingest/realtime/segmentUpgrade/persisted`|Number of pending segments that
were upgraded in the metadata store while publishing segments to an interval.
Emitted by the Overlord only when a REPLACE task (e.g., compaction task with
`useConcurrentLocks` set to `true`) commits segments to an interval already
containing pending segments allocated to an APPEND task (e.g. streaming task
with `useConcurrentLocks` set to `true`). |`dataSource`, `taskId`, `taskType`,
`groupId`, `interval`, `version`, `tags`|0 unless [concurrent append and
replace](../ingestion/concurrent-append-replace.md) is in use.|
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -1415,26 +1417,102 @@ public void resetOffsets(@Nonnull DataSourceMetadata
resetDataSourceMetadata)
addNotice(new ResetOffsetsNotice(resetDataSourceMetadata));
}
- public void registerNewVersionOfPendingSegment(
+ /**
+ * Notifies every running task in the matching task group(s) of an upgraded
pending segment.
+ *
+ * @return the number of tasks notified, which the caller aggregates into a
per-batch summary
Review Comment:
```suggestion
* @return the number of tasks notified
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceAction.java:
##########
@@ -169,19 +173,45 @@ private void registerUpgradedPendingSegmentsOnSupervisor(
List<PendingSegmentRecord> upgradedPendingSegments
)
{
+ // Emit one count per upgraded segment (rather than a single aggregate)
regardless of whether a supervisor exists
+ // to receive them, so the total can be compared against the count
actually announced by tasks and so each event
+ // carries the segment's interval and version.
+ for (PendingSegmentRecord upgradedPendingSegment :
upgradedPendingSegments) {
+ final ServiceMetricEvent.Builder metricBuilder = new
ServiceMetricEvent.Builder();
+ IndexTaskUtils.setTaskDimensions(metricBuilder, task);
+ SegmentUpgradeMetrics.setSegmentDimensions(metricBuilder,
upgradedPendingSegment);
+
toolbox.getEmitter().emit(metricBuilder.setMetric(SegmentUpgradeMetrics.COUNT,
1));
+ }
+
final SupervisorManager supervisorManager = toolbox.getSupervisorManager();
final Optional<String> activeSupervisorIdWithAppendLock =
supervisorManager.getActiveSupervisorIdForDatasourceWithAppendLock(task.getDataSource());
if (!activeSupervisorIdWithAppendLock.isPresent()) {
+ log.info("No active streaming supervisor for datasource[%s]; the [%d]
upgraded pending segment(s) from task[%s]"
+ + " will become queryable when their tasks hand off.",
+ task.getDataSource(),
+ upgradedPendingSegments.size(),
+ task.getId()
+ );
Review Comment:
minor rephrase
```suggestion
log.info(
"Could not find any active concurrent streaming supervisor for
datasource[%s]."
+ " Ignoring registry of [%d] pending segment(s) upgraded by
task[%s]. These will become queryable only after handoff."
task.getDataSource(),
upgradedPendingSegments.size(),
task.getId()
);
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -1415,26 +1417,102 @@ public void resetOffsets(@Nonnull DataSourceMetadata
resetDataSourceMetadata)
addNotice(new ResetOffsetsNotice(resetDataSourceMetadata));
}
- public void registerNewVersionOfPendingSegment(
+ /**
+ * Notifies every running task in the matching task group(s) of an upgraded
pending segment.
+ *
+ * @return the number of tasks notified, which the caller aggregates into a
per-batch summary
+ */
+ public int registerNewVersionOfPendingSegment(
PendingSegmentRecord pendingSegmentRecord
)
{
+ final String taskAllocatorId = pendingSegmentRecord.getTaskAllocatorId();
+ int notifiedTasks = 0;
+
for (TaskGroup taskGroup : activelyReadingTaskGroups.values()) {
- if
(taskGroup.baseSequenceName.equals(pendingSegmentRecord.getTaskAllocatorId())) {
+ if (taskGroup.baseSequenceName.equals(taskAllocatorId)) {
for (String taskId : taskGroup.taskIds()) {
- taskClient.registerNewVersionOfPendingSegmentAsync(taskId,
pendingSegmentRecord);
+ notifyTaskOfUpgradedPendingSegment(taskId, pendingSegmentRecord);
+ notifiedTasks++;
}
}
}
for (List<TaskGroup> taskGroupList : pendingCompletionTaskGroups.values())
{
for (TaskGroup taskGroup : taskGroupList) {
- if
(taskGroup.baseSequenceName.equals(pendingSegmentRecord.getTaskAllocatorId())) {
+ if (taskGroup.baseSequenceName.equals(taskAllocatorId)) {
for (String taskId : taskGroup.taskIds()) {
- taskClient.registerNewVersionOfPendingSegmentAsync(taskId,
pendingSegmentRecord);
+ notifyTaskOfUpgradedPendingSegment(taskId, pendingSegmentRecord);
+ notifiedTasks++;
}
}
}
}
+
+ // Only the exceptional no-match case is logged per segment; the notified
count is returned to the task action,
+ // which summarizes the whole batch in one log line, and each notification
is captured by the per-task metric.
+ if (notifiedTasks == 0) {
+ // No running task matched: the segment will not be re-announced until
handoff. This is a potential silent-loss
+ // window where data will not be queryable until handoff.
+ log.warn(
+ "Could not find any task matching taskAllocatorId[%s] in
supervisor[%s] for upgraded pending segment[%s]"
+ + " (upgradedFrom[%s]); it will not be re-announced until handoff.
Currently tracking [%d] activelyReading"
+ + " and [%d] pendingCompletion task group(s).",
Review Comment:
Do you think logging the number of actively reading / pending completion
groups would be useful for debugging? I feel we may omit these for now.
##########
server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java:
##########
@@ -1181,12 +1181,53 @@ private void unannounceSegment(DataSegment segment)
}
}
- public void registerUpgradedPendingSegment(PendingSegmentRecord
pendingSegmentRecord) throws IOException
+ /**
+ * Result of a {@link #registerUpgradedPendingSegment} call. Each skipped
outcome carries the human-readable
+ * {@link #getReason() reason} emitted as the {@code reason} metric
dimension.
+ */
+ public enum PendingSegmentUpgradeResult
+ {
+ ANNOUNCED(null),
+ /** The task holds no pending segment matching upgradedFromSegmentId
(request targeted the wrong task). */
+ SKIPPED_UNKNOWN_BASE("unknown base"),
Review Comment:
```suggestion
SKIPPED_UNKNOWN_BASE("unknown base sink"),
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/common/SegmentUpgradeMetrics.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you 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 org.apache.druid.indexing.common;
+
+import org.apache.druid.java.util.emitter.service.ServiceMetricEvent;
+import org.apache.druid.metadata.PendingSegmentRecord;
+import org.apache.druid.query.DruidMetrics;
+import org.apache.druid.segment.realtime.appenderator.SegmentIdWithShardSpec;
+
+/**
+ * Metric names and dimension values for the re-announcement of pending
segments upgraded by a concurrent REPLACE.
+ * <ul>
+ * <li>the task action emits {@link #COUNT} (how many upgrades a commit
produced),</li>
+ * <li>the supervisor emits {@link #NOTIFIED}, {@link #UNMATCHED} and {@link
#SEND_FAILED} as it fans requests out,</li>
+ * <li>the streaming task emits {@link #ANNOUNCED} and {@link #SKIPPED}
(with a {@code reason}) as it applies them.</li>
+ * </ul>
+ * Two reconciliations bound the visibility gap:
+ * <ul>
+ * <li>per segment, {@link #UNMATCHED} counts upgrades that reached no
running task (delayed until handoff);</li>
+ * <li>per task, {@link #NOTIFIED} should equal {@link #ANNOUNCED} + {@link
#SKIPPED} + {@link #SEND_FAILED},
+ * since every notified task either announces, skips, or fails to
receive the request. A shortfall means a
+ * notification was silently dropped.</li>
+ * </ul>
+ */
+public class SegmentUpgradeMetrics
+{
+ /** Number of upgraded pending segments a REPLACE commit created and handed
to the supervisor. Task-action dims. */
+ public static final String COUNT = "ingest/realtime/segmentUpgrade/count";
+
+ /** A notification was sent to a running task (once per task). Supervisor
dims plus {@code taskId}. */
+ public static final String NOTIFIED =
"ingest/realtime/segmentUpgrade/notified";
+
+ /** A record matched no running task and will not be re-announced until
handoff. Supervisor dims. */
+ public static final String UNMATCHED =
"ingest/realtime/segmentUpgrade/unmatched";
+
+ /** An upgrade request failed to reach a task over the wire. Supervisor dims
plus {@code taskId}. */
+ public static final String SEND_FAILED =
"ingest/realtime/segmentUpgrade/sendFailed";
+
+ /** A task announced an upgraded segment under the new version. Task dims. */
+ public static final String ANNOUNCED =
"ingest/realtime/segmentUpgrade/announced";
+
+ /**
+ * A task received an upgrade request but did not announce it. The {@code
reason} dimension carries
+ * {@link
org.apache.druid.segment.realtime.appenderator.StreamAppenderator.PendingSegmentUpgradeResult#getReason()}.
+ * Task dims.
+ */
+ public static final String SKIPPED =
"ingest/realtime/segmentUpgrade/skipped";
+
+ /**
+ * Adds the upgraded pending segment's {@code interval} and {@code version}
to a metric builder so that every
+ * segment-upgrade metric can be sliced by the specific segment being
re-announced. Mirrors
+ * {@code IndexTaskUtils.setSegmentDimensions}, which serves the same
purpose for {@code DataSegment}s.
+ */
+ public static ServiceMetricEvent.Builder setSegmentDimensions(
Review Comment:
Nit: Maybe move this method to `IndexTaskUtils` itself, as it feels like a
more generic place (rather than upgrade specific) allowing for reuse for this
method.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java:
##########
@@ -1415,26 +1417,102 @@ public void resetOffsets(@Nonnull DataSourceMetadata
resetDataSourceMetadata)
addNotice(new ResetOffsetsNotice(resetDataSourceMetadata));
}
- public void registerNewVersionOfPendingSegment(
+ /**
+ * Notifies every running task in the matching task group(s) of an upgraded
pending segment.
+ *
+ * @return the number of tasks notified, which the caller aggregates into a
per-batch summary
+ */
+ public int registerNewVersionOfPendingSegment(
PendingSegmentRecord pendingSegmentRecord
)
{
+ final String taskAllocatorId = pendingSegmentRecord.getTaskAllocatorId();
+ int notifiedTasks = 0;
+
for (TaskGroup taskGroup : activelyReadingTaskGroups.values()) {
- if
(taskGroup.baseSequenceName.equals(pendingSegmentRecord.getTaskAllocatorId())) {
+ if (taskGroup.baseSequenceName.equals(taskAllocatorId)) {
for (String taskId : taskGroup.taskIds()) {
- taskClient.registerNewVersionOfPendingSegmentAsync(taskId,
pendingSegmentRecord);
+ notifyTaskOfUpgradedPendingSegment(taskId, pendingSegmentRecord);
+ notifiedTasks++;
}
}
}
for (List<TaskGroup> taskGroupList : pendingCompletionTaskGroups.values())
{
for (TaskGroup taskGroup : taskGroupList) {
- if
(taskGroup.baseSequenceName.equals(pendingSegmentRecord.getTaskAllocatorId())) {
+ if (taskGroup.baseSequenceName.equals(taskAllocatorId)) {
for (String taskId : taskGroup.taskIds()) {
- taskClient.registerNewVersionOfPendingSegmentAsync(taskId,
pendingSegmentRecord);
+ notifyTaskOfUpgradedPendingSegment(taskId, pendingSegmentRecord);
+ notifiedTasks++;
}
}
}
}
+
+ // Only the exceptional no-match case is logged per segment; the notified
count is returned to the task action,
+ // which summarizes the whole batch in one log line, and each notification
is captured by the per-task metric.
Review Comment:
We need not call this out. This method should be agnostic of the behaviour
of its caller.
We choose not to log the successful case here to reduce verbosity,
irrespective of the behaviour of the caller.
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java:
##########
@@ -1865,6 +1869,61 @@ public Response registerUpgradedPendingSegment(
}
}
+ /**
+ * Logs and emits a metric for the outcome of a pending-segment upgrade
request, keyed off the {@code outcome}
+ * returned by {@link StreamAppenderator#registerUpgradedPendingSegment}.
+ */
+ private void recordUpgradeOutcome(
+ PendingSegmentRecord upgradedPendingSegment,
+ StreamAppenderator.PendingSegmentUpgradeResult outcome
Review Comment:
```suggestion
StreamAppenderator.PendingSegmentUpgradeResult result
```
##########
indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java:
##########
@@ -1865,6 +1869,61 @@ public Response registerUpgradedPendingSegment(
}
}
+ /**
+ * Logs and emits a metric for the outcome of a pending-segment upgrade
request, keyed off the {@code outcome}
+ * returned by {@link StreamAppenderator#registerUpgradedPendingSegment}.
+ */
+ private void recordUpgradeOutcome(
Review Comment:
```suggestion
private void recordUpgradeResult(
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
