diff --git a/docs/operations/metrics.md b/docs/operations/metrics.md index f4757760b719..58c9d276d206 100644 --- a/docs/operations/metrics.md +++ b/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.| +|`ingest/realtime/segmentUpgrade/notified`|Number of notifications the supervisor sent to running tasks, emitted once per task so it scales with the replica count. Reconcile against the per-task outcomes: `notified` should equal `announced` + `skipped` + `sendFailed` over the same period and `dataSource`, since every notified task either announces, skips, or fails to receive the request. A shortfall indicates a notification was silently dropped.|`supervisorId`, `dataSource`, `stream`, `taskId`, `interval`, `version`, `tags`|Equal to `announced` + `skipped` + `sendFailed`.| +|`ingest/realtime/segmentUpgrade/unmatched`|Number of upgraded pending segments the supervisor could not route to any running task. These are not announced under the new version until handoff, so the corresponding data may be briefly missing from queries.|`supervisorId`, `dataSource`, `stream`, `interval`, `version`, `tags`|0. A non-zero value indicates a lost upgrade announcement.| +|`ingest/realtime/segmentUpgrade/sendFailed`|Number of upgrade requests that matched a running task but failed to reach it over the wire after retries.|`supervisorId`, `dataSource`, `stream`, `taskId`, `interval`, `version`, `tags`|0| +|`ingest/realtime/segmentUpgrade/announced`|Number of upgraded segments a task announced under the new version. Emitted once per task, so it scales with the replica count. Reconcile against `notified` (also per task), not `ingest/realtime/segmentUpgrade/count`, which is per segment rather than per replica.|`dataSource`, `taskId`, `taskType`, `groupId`, `interval`, `version`, `tags`|Greater than 0 while concurrent replace occurs.| +|`ingest/realtime/segmentUpgrade/skipped`|Number of upgrade requests a task received but did not announce. The `reason` dimension is one of `unknown base` (the request reached the wrong task), `base sink already dropped` (the base sink is no longer present), or `dropping base sink` (the base sink is handing off, which is benign and covered by the durable publish path).|`dataSource`, `taskId`, `taskType`, `groupId`, `interval`, `version`, `tags`, `reason`|0, excluding `reason=dropping base sink`.| |`task/autoScaler/requiredCount`|Count of required tasks based on the calculations of the auto scaler.|`supervisorId`, `dataSource`, `stream`, `scalingSkipReason`|Depends on auto scaler config.| |`task/autoScaler/scaleActionTime`|Time taken in milliseconds to complete the scale action.|`supervisorId`, `dataSource`, `stream`, `tags`|Depends on auto scaler config.| |`task/autoScaler/costBased/optimalTaskCount`|Optimal task count computed by the cost-based auto scaler.|`supervisorId`, `dataSource`, `stream`|Depends on auto scaler config.| diff --git a/extensions-contrib/dropwizard-emitter/src/main/resources/defaultMetricDimensions.json b/extensions-contrib/dropwizard-emitter/src/main/resources/defaultMetricDimensions.json index 3a681d25cd9c..e0c69c529f80 100644 --- a/extensions-contrib/dropwizard-emitter/src/main/resources/defaultMetricDimensions.json +++ b/extensions-contrib/dropwizard-emitter/src/main/resources/defaultMetricDimensions.json @@ -214,6 +214,60 @@ "type": "timer", "timeUnit": "NANOSECONDS" }, + "ingest/realtime/segmentUpgrade/count": { + "dimensions": [ + "dataSource", + "interval", + "version" + ], + "type": "counter" + }, + "ingest/realtime/segmentUpgrade/notified": { + "dimensions": [ + "dataSource", + "stream", + "taskId", + "interval", + "version" + ], + "type": "counter" + }, + "ingest/realtime/segmentUpgrade/unmatched": { + "dimensions": [ + "dataSource", + "stream", + "interval", + "version" + ], + "type": "counter" + }, + "ingest/realtime/segmentUpgrade/sendFailed": { + "dimensions": [ + "dataSource", + "stream", + "taskId", + "interval", + "version" + ], + "type": "counter" + }, + "ingest/realtime/segmentUpgrade/announced": { + "dimensions": [ + "dataSource", + "interval", + "version" + ], + "type": "counter" + }, + "ingest/realtime/segmentUpgrade/skipped": { + "dimensions": [ + "dataSource", + "reason", + "interval", + "version" + ], + "type": "counter" + }, "task/run/time": { "dimensions": [ "dataSource", diff --git a/extensions-contrib/opentsdb-emitter/src/main/resources/defaultMetrics.json b/extensions-contrib/opentsdb-emitter/src/main/resources/defaultMetrics.json index f38f115348f8..e04e0a5a9423 100644 --- a/extensions-contrib/opentsdb-emitter/src/main/resources/defaultMetrics.json +++ b/extensions-contrib/opentsdb-emitter/src/main/resources/defaultMetrics.json @@ -105,6 +105,42 @@ "ingest/events/messageGap": [ "dataSource" ], + "ingest/realtime/segmentUpgrade/count": [ + "dataSource", + "interval", + "version" + ], + "ingest/realtime/segmentUpgrade/notified": [ + "dataSource", + "stream", + "taskId", + "interval", + "version" + ], + "ingest/realtime/segmentUpgrade/unmatched": [ + "dataSource", + "stream", + "interval", + "version" + ], + "ingest/realtime/segmentUpgrade/sendFailed": [ + "dataSource", + "stream", + "taskId", + "interval", + "version" + ], + "ingest/realtime/segmentUpgrade/announced": [ + "dataSource", + "interval", + "version" + ], + "ingest/realtime/segmentUpgrade/skipped": [ + "dataSource", + "reason", + "interval", + "version" + ], "ingest/kafka/lag": [ "dataSource", "stream" diff --git a/extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json b/extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json index e582958e7683..0336806e4185 100644 --- a/extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json +++ b/extensions-contrib/prometheus-emitter/src/main/resources/defaultMetrics.json @@ -134,6 +134,12 @@ "ingest/notices/time" : { "dimensions" : ["dataSource"], "type" : "timer", "conversionFactor": 1000.0, "help": "Seconds taken to process a notice by the supervisor." }, "ingest/pause/time" : { "dimensions" : ["dataSource"], "type" : "timer", "conversionFactor": 1000.0, "help": "Seconds spent by a task in a paused state without ingesting." }, "ingest/handoff/time" : { "dimensions" : ["dataSource"], "type" : "timer", "conversionFactor": 1000.0, "help": "Total number of seconds taken to handoff a set of segments." }, + "ingest/realtime/segmentUpgrade/count" : { "dimensions" : ["dataSource", "interval", "version"], "type" : "count", "help": "Number of pending segments upgraded by a concurrent replace." }, + "ingest/realtime/segmentUpgrade/notified" : { "dimensions" : ["dataSource", "stream", "taskId", "interval", "version"], "type" : "count", "help": "Number of notifications the supervisor sent to running tasks, one per task." }, + "ingest/realtime/segmentUpgrade/unmatched" : { "dimensions" : ["dataSource", "stream", "interval", "version"], "type" : "count", "help": "Number of upgraded pending segments that matched no running task." }, + "ingest/realtime/segmentUpgrade/sendFailed" : { "dimensions" : ["dataSource", "stream", "taskId", "interval", "version"], "type" : "count", "help": "Number of upgrade requests that failed to reach a task." }, + "ingest/realtime/segmentUpgrade/announced" : { "dimensions" : ["dataSource", "interval", "version"], "type" : "count", "help": "Number of upgraded segments a task announced under the new version." }, + "ingest/realtime/segmentUpgrade/skipped" : { "dimensions" : ["dataSource", "reason", "interval", "version"], "type" : "count", "help": "Number of upgrade requests a task received but did not announce." }, "task/autoScaler/requiredCount" : { "dimensions" : ["dataSource"], "type" : "count", "help": "Count of required tasks based on the calculations of lagBased auto scaler." }, "task/run/time" : { "dimensions" : ["dataSource", "taskType"], "type" : "timer", "conversionFactor": 1000.0, "help": "Seconds taken to run a task."}, diff --git a/extensions-contrib/statsd-emitter/src/main/resources/defaultMetricDimensions.json b/extensions-contrib/statsd-emitter/src/main/resources/defaultMetricDimensions.json index 9eae1dd5a62e..282c7f8fc312 100644 --- a/extensions-contrib/statsd-emitter/src/main/resources/defaultMetricDimensions.json +++ b/extensions-contrib/statsd-emitter/src/main/resources/defaultMetricDimensions.json @@ -57,6 +57,13 @@ "ingest/segments/count" : { "dimensions" : ["dataSource"], "type" : "count" }, "ingest/rows/published": { "dimensions" : ["dataSource"], "type" : "count" }, + "ingest/realtime/segmentUpgrade/count" : { "dimensions" : ["dataSource", "interval", "version"], "type" : "count" }, + "ingest/realtime/segmentUpgrade/notified" : { "dimensions" : ["dataSource", "stream", "taskId", "interval", "version"], "type" : "count" }, + "ingest/realtime/segmentUpgrade/unmatched" : { "dimensions" : ["dataSource", "stream", "interval", "version"], "type" : "count" }, + "ingest/realtime/segmentUpgrade/sendFailed" : { "dimensions" : ["dataSource", "stream", "taskId", "interval", "version"], "type" : "count" }, + "ingest/realtime/segmentUpgrade/announced" : { "dimensions" : ["dataSource", "interval", "version"], "type" : "count" }, + "ingest/realtime/segmentUpgrade/skipped" : { "dimensions" : ["dataSource", "reason", "interval", "version"], "type" : "count" }, + "ingest/kafka/lag" : { "dimensions" : ["dataSource", "stream"], "type" : "gauge" }, "ingest/kafka/maxLag" : { "dimensions" : ["dataSource", "stream"], "type" : "gauge" }, "ingest/kafka/avgLag" : { "dimensions" : ["dataSource", "stream"], "type" : "gauge" }, diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/SegmentUpgradeMetrics.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/SegmentUpgradeMetrics.java new file mode 100644 index 000000000000..af67fa90165a --- /dev/null +++ b/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. + * + * Two reconciliations bound the visibility gap: + * + */ +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( + ServiceMetricEvent.Builder metricBuilder, + PendingSegmentRecord pendingSegmentRecord + ) + { + final SegmentIdWithShardSpec id = pendingSegmentRecord.getId(); + return metricBuilder + .setDimension(DruidMetrics.INTERVAL, id.getInterval().toString()) + .setDimension(DruidMetrics.VERSION, id.getVersion()); + } + + private SegmentUpgradeMetrics() + { + } +} diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceAction.java b/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceAction.java index 6546ed80d9d9..541939d45093 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceAction.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/common/actions/SegmentTransactionalReplaceAction.java @@ -24,12 +24,14 @@ import com.fasterxml.jackson.core.type.TypeReference; import com.google.common.base.Optional; import com.google.common.collect.ImmutableSet; +import org.apache.druid.indexing.common.SegmentUpgradeMetrics; import org.apache.druid.indexing.common.task.IndexTaskUtils; import org.apache.druid.indexing.common.task.Task; import org.apache.druid.indexing.overlord.CriticalAction; import org.apache.druid.indexing.overlord.SegmentPublishResult; import org.apache.druid.indexing.overlord.supervisor.SupervisorManager; import org.apache.druid.java.util.common.logger.Logger; +import org.apache.druid.java.util.emitter.service.ServiceMetricEvent; import org.apache.druid.metadata.PendingSegmentRecord; import org.apache.druid.metadata.ReplaceTaskLock; import org.apache.druid.segment.SegmentSchemaMapping; @@ -37,7 +39,9 @@ import org.apache.druid.timeline.DataSegment; import javax.annotation.Nullable; +import java.util.LinkedHashMap; import java.util.List; +import java.util.Map; import java.util.Set; import java.util.stream.Collectors; @@ -169,19 +173,45 @@ private void registerUpgradedPendingSegmentsOnSupervisor( List 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 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() + ); return; } - upgradedPendingSegments.forEach( - upgradedPendingSegment -> supervisorManager.registerUpgradedPendingSegmentOnSupervisor( - activeSupervisorIdWithAppendLock.get(), - upgradedPendingSegment - ) + // Register each upgraded pending segment on the supervisor and summarize the whole batch in one log line, + // mapping each segment to the number of running tasks the supervisor notified. + final Map notifiedTasksBySegment = new LinkedHashMap<>(); + for (PendingSegmentRecord upgradedPendingSegment : upgradedPendingSegments) { + supervisorManager + .registerUpgradedPendingSegmentOnSupervisor(activeSupervisorIdWithAppendLock.get(), upgradedPendingSegment) + .ifPresent(notified -> notifiedTasksBySegment.put(upgradedPendingSegment.getId().toString(), notified)); + } + log.info( + "Registered [%d] upgraded pending segment(s) created by task[%s] on supervisor[%s]; tasks notified per" + + " segment[%s].", + notifiedTasksBySegment.size(), + task.getId(), + activeSupervisorIdWithAppendLock.get(), + notifiedTasksBySegment ); } diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java index fa7d96634ae6..929f34bf89eb 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManager.java @@ -60,6 +60,7 @@ import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.OptionalInt; import java.util.Set; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.Future; @@ -511,8 +512,11 @@ public boolean checkPointDataSourceMetadata( * Registers a new version of the given pending segment on a supervisor. This * allows the supervisor to include the pending segment in queries fired against * that segment version. + * + * @return the number of tasks notified if the segment was registered on a seekable stream supervisor, or + * {@link OptionalInt#empty()} if no such supervisor was found or the registration failed */ - public boolean registerUpgradedPendingSegmentOnSupervisor( + public OptionalInt registerUpgradedPendingSegmentOnSupervisor( String supervisorId, PendingSegmentRecord upgradedPendingSegment ) @@ -529,12 +533,11 @@ public boolean registerUpgradedPendingSegmentOnSupervisor( Pair supervisor = supervisors.get(supervisorId); Preconditions.checkNotNull(supervisor, "supervisor could not be found"); if (!(supervisor.lhs instanceof SeekableStreamSupervisor)) { - return false; + return OptionalInt.empty(); } SeekableStreamSupervisor seekableStreamSupervisor = (SeekableStreamSupervisor) supervisor.lhs; - seekableStreamSupervisor.registerNewVersionOfPendingSegment(upgradedPendingSegment); - return true; + return OptionalInt.of(seekableStreamSupervisor.registerNewVersionOfPendingSegment(upgradedPendingSegment)); } catch (Exception e) { log.error( @@ -545,7 +548,7 @@ public boolean registerUpgradedPendingSegmentOnSupervisor( supervisorId ); } - return false; + return OptionalInt.empty(); } /** diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java index 2390792fd4d2..fd40a8800fb6 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/SeekableStreamIndexTaskRunner.java @@ -58,6 +58,7 @@ import org.apache.druid.indexer.report.TaskContextReport; import org.apache.druid.indexer.report.TaskReport; import org.apache.druid.indexing.common.LockGranularity; +import org.apache.druid.indexing.common.SegmentUpgradeMetrics; import org.apache.druid.indexing.common.TaskLock; import org.apache.druid.indexing.common.TaskLockType; import org.apache.druid.indexing.common.TaskToolbox; @@ -80,6 +81,7 @@ import org.apache.druid.java.util.common.concurrent.Execs; import org.apache.druid.java.util.emitter.EmittingLogger; import org.apache.druid.metadata.PendingSegmentRecord; +import org.apache.druid.query.DruidMetrics; import org.apache.druid.segment.incremental.InputRowFilterResult; import org.apache.druid.segment.incremental.ParseExceptionHandler; import org.apache.druid.segment.incremental.ParseExceptionReport; @@ -1846,7 +1848,9 @@ public Response registerUpgradedPendingSegment( { authorizationCheck(req); try { - ((StreamAppenderator) appenderator).registerUpgradedPendingSegment(upgradedPendingSegment); + final StreamAppenderator.PendingSegmentUpgradeResult outcome = + ((StreamAppenderator) appenderator).registerUpgradedPendingSegment(upgradedPendingSegment); + recordUpgradeOutcome(upgradedPendingSegment, outcome); return Response.ok().build(); } catch (DruidException e) { @@ -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 + ) + { + final SegmentIdWithShardSpec upgradedId = upgradedPendingSegment.getId(); + final String upgradedFromSegmentId = upgradedPendingSegment.getUpgradedFromSegmentId(); + + if (outcome == StreamAppenderator.PendingSegmentUpgradeResult.ANNOUNCED) { + toolbox.getEmitter().emit( + SegmentUpgradeMetrics.setSegmentDimensions(task.getMetricBuilder(), upgradedPendingSegment) + .setMetric(SegmentUpgradeMetrics.ANNOUNCED, 1) + ); + return; + } + // Log at the level appropriate to the outcome; the reason string itself is carried by the enum. + switch (outcome) { + case SKIPPED_UNKNOWN_BASE: + // The request targeted the wrong task: this task never held a base sink matching upgradedFromSegmentId. + log.warn( + "Not announcing upgraded pending segment[%s] on task[%s] because it has no base sink matching" + + " upgradedFromSegmentId[%s]; the upgrade request likely targeted the wrong task.", + upgradedId, task.getId(), upgradedFromSegmentId + ); + break; + case SKIPPED_NO_SINK: + // Unexpected: the base sink is gone even though this task once held it. + log.info( + "Not announcing upgraded pending segment[%s] (upgradedFrom[%s]) on task[%s] because the base sink is no" + + " longer present.", + upgradedId, upgradedFromSegmentId, task.getId() + ); + break; + case SKIPPED_DROPPING: + // Expected during handoff: the base sink is being dropped and the durable path re-announces at the new version. + log.debug( + "Not announcing upgraded pending segment[%s] (upgradedFrom[%s]) on task[%s] because the base sink is being" + + " dropped.", + upgradedId, upgradedFromSegmentId, task.getId() + ); + break; + default: + return; + } + toolbox.getEmitter().emit( + SegmentUpgradeMetrics.setSegmentDimensions(task.getMetricBuilder(), upgradedPendingSegment) + .setDimension(DruidMetrics.REASON, outcome.getReason()) + .setMetric(SegmentUpgradeMetrics.SKIPPED, 1) + ); + } + public Map doGetRowStats() { Map returnMap = new HashMap<>(); diff --git a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java index 74329c68e1d2..e55551da5513 100644 --- a/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java +++ b/indexing-service/src/main/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisor.java @@ -32,6 +32,7 @@ import com.google.common.collect.ImmutableSet; import com.google.common.collect.Sets; import com.google.common.util.concurrent.AsyncFunction; +import com.google.common.util.concurrent.FutureCallback; import com.google.common.util.concurrent.Futures; import com.google.common.util.concurrent.ListenableFuture; import com.google.common.util.concurrent.ListeningExecutorService; @@ -48,6 +49,7 @@ import org.apache.druid.indexer.TaskLocation; import org.apache.druid.indexer.TaskState; import org.apache.druid.indexer.TaskStatus; +import org.apache.druid.indexing.common.SegmentUpgradeMetrics; import org.apache.druid.indexing.common.TaskInfoProvider; import org.apache.druid.indexing.common.task.Task; import org.apache.druid.indexing.overlord.DataSourceMetadata; @@ -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 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).", + taskAllocatorId, + supervisorId, + pendingSegmentRecord.getId(), + pendingSegmentRecord.getUpgradedFromSegmentId(), + activelyReadingTaskGroups.size(), + pendingCompletionTaskGroups.size() + ); + emitter.emit( + SegmentUpgradeMetrics.setSegmentDimensions(getMetricBuilder(), pendingSegmentRecord) + .setMetric(SegmentUpgradeMetrics.UNMATCHED, 1) + ); + } + return notifiedTasks; + } + + /** + * Sends an upgraded pending segment to a single task, emitting {@link SegmentUpgradeMetrics#NOTIFIED} for the + * notification attempt and {@link SegmentUpgradeMetrics#SEND_FAILED} if the request fails to reach the task. Both + * are per-task (keyed by {@code taskId}), so {@code notified} can be reconciled against the per-task + * {@code announced}, {@code skipped} and {@code sendFailed} outcomes. + */ + private void notifyTaskOfUpgradedPendingSegment(String taskId, PendingSegmentRecord pendingSegmentRecord) + { + emitter.emit( + SegmentUpgradeMetrics.setSegmentDimensions(getMetricBuilder(), pendingSegmentRecord) + .setDimension(DruidMetrics.TASK_ID, taskId) + .setMetric(SegmentUpgradeMetrics.NOTIFIED, 1) + ); + Futures.addCallback( + taskClient.registerNewVersionOfPendingSegmentAsync(taskId, pendingSegmentRecord), + new FutureCallback<>() + { + @Override + public void onSuccess(Boolean result) + { + // The request reached the task; what the task does with it (announce or skip) is reported by the task's + // own announced/skipped metrics, so there is nothing to record here. + } + + @Override + public void onFailure(Throwable t) + { + log.warn( + t, + "Failed to register upgraded pending segment[%s] on task[%s] of supervisor[%s].", + pendingSegmentRecord.getId(), taskId, supervisorId + ); + emitter.emit( + SegmentUpgradeMetrics.setSegmentDimensions(getMetricBuilder(), pendingSegmentRecord) + .setDimension(DruidMetrics.TASK_ID, taskId) + .setMetric(SegmentUpgradeMetrics.SEND_FAILED, 1) + ); + } + }, + MoreExecutors.directExecutor() + ); } public ReentrantLock getRecordSupplierLock() diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/concurrent/ConcurrentReplaceAndStreamingAppendTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/concurrent/ConcurrentReplaceAndStreamingAppendTest.java index 868d8c15568e..35382746ae13 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/common/task/concurrent/ConcurrentReplaceAndStreamingAppendTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/common/task/concurrent/ConcurrentReplaceAndStreamingAppendTest.java @@ -84,6 +84,7 @@ import java.util.HashSet; import java.util.List; import java.util.Map; +import java.util.OptionalInt; import java.util.Set; import java.util.concurrent.atomic.AtomicInteger; @@ -170,7 +171,7 @@ public void setUpIngestionTestBase() throws IOException EasyMock.expect(supervisorManager.registerUpgradedPendingSegmentOnSupervisor( EasyMock.capture(supervisorId), EasyMock.capture(pendingSegment) - )).andReturn(true).anyTimes(); + )).andReturn(OptionalInt.of(1)).anyTimes(); replaceTask = createAndStartTask(); EasyMock.replay(supervisorManager); versionToIntervalToLoadSpecs = new HashMap<>(); diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java index 36e87b604d3e..7bbedd87fb84 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/overlord/supervisor/SupervisorManagerTest.java @@ -75,6 +75,7 @@ import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.OptionalInt; import java.util.concurrent.ConcurrentHashMap; @RunWith(EasyMockRunner.class) @@ -894,8 +895,7 @@ public void testRegisterUpgradedPendingSegmentOnSupervisor() SeekableStreamSupervisorSpec streamingSpec = EasyMock.createNiceMock(SeekableStreamSupervisorSpec.class); SeekableStreamSupervisor streamSupervisor = EasyMock.createNiceMock(SeekableStreamSupervisor.class); - streamSupervisor.registerNewVersionOfPendingSegment(EasyMock.anyObject()); - EasyMock.expectLastCall().once(); + EasyMock.expect(streamSupervisor.registerNewVersionOfPendingSegment(EasyMock.anyObject())).andReturn(1).once(); EasyMock.expect(streamingSpec.getId()).andReturn("sss").anyTimes(); EasyMock.expect(streamingSpec.isSuspended()).andReturn(false).anyTimes(); EasyMock.expect(streamingSpec.getDataSources()).andReturn(ImmutableList.of("DS")).anyTimes(); @@ -923,10 +923,10 @@ public void testRegisterUpgradedPendingSegmentOnSupervisor() manager.start(); manager.createOrUpdateAndStartSupervisor(noopSpec); - Assert.assertFalse(manager.registerUpgradedPendingSegmentOnSupervisor("noop", pendingSegment)); + Assert.assertFalse(manager.registerUpgradedPendingSegmentOnSupervisor("noop", pendingSegment).isPresent()); manager.createOrUpdateAndStartSupervisor(streamingSpec); - Assert.assertTrue(manager.registerUpgradedPendingSegmentOnSupervisor("sss", pendingSegment)); + Assert.assertEquals(OptionalInt.of(1), manager.registerUpgradedPendingSegmentOnSupervisor("sss", pendingSegment)); verifyAll(); } diff --git a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java index 9e45920ad719..6c12c8b38855 100644 --- a/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java +++ b/indexing-service/src/test/java/org/apache/druid/indexing/seekablestream/supervisor/SeekableStreamSupervisorStateTest.java @@ -39,6 +39,7 @@ import org.apache.druid.indexer.TaskLocation; import org.apache.druid.indexer.TaskStatus; import org.apache.druid.indexer.granularity.UniformGranularitySpec; +import org.apache.druid.indexing.common.SegmentUpgradeMetrics; import org.apache.druid.indexing.common.TestUtils; import org.apache.druid.indexing.common.task.Task; import org.apache.druid.indexing.overlord.DataSourceMetadata; @@ -2116,6 +2117,85 @@ public void testRegisterNewVersionOfPendingSegment() Assert.assertEquals(pendingSegmentRecord0, captured0.getValue()); Assert.assertEquals(pendingSegmentRecord1, captured1.getValue()); + + // Both records matched a running task, so each emits a notified metric and none is unmatched. + Assert.assertEquals(2, emitter.getMetricEventCount(SegmentUpgradeMetrics.NOTIFIED)); + Assert.assertEquals(0, emitter.getMetricEventCount(SegmentUpgradeMetrics.UNMATCHED)); + verifyAll(); + } + + @Test + public void testRegisterNewVersionOfPendingSegmentUnmatchedEmitsMetric() + { + EasyMock.expect(spec.isSuspended()).andReturn(false); + replayAll(); + + final TestSeekableStreamSupervisor supervisor = new TestSeekableStreamSupervisor(); + supervisor.getIoConfig().setTaskCount(3); + supervisor.start(); + + supervisor.addTaskGroupToActivelyReadingTaskGroup( + supervisor.getTaskGroupIdForPartition("0"), + ImmutableMap.of("0", "5"), + null, + null, + Set.of("task0"), + Set.of(), + null + ); + + // A taskAllocatorId that matches no running task group + final PendingSegmentRecord record = PendingSegmentRecord.create( + new SegmentIdWithShardSpec("DS", Intervals.of("2024/2025"), "2024", new NumberedShardSpec(1, 0)), + "no-such-allocator", + "prevId", + "someAppendedSegment", + "no-such-allocator" + ); + + supervisor.registerNewVersionOfPendingSegment(record); + + Assert.assertEquals(1, emitter.getMetricEventCount(SegmentUpgradeMetrics.UNMATCHED)); + Assert.assertEquals(0, emitter.getMetricEventCount(SegmentUpgradeMetrics.NOTIFIED)); + verifyAll(); + } + + @Test + public void testRegisterNewVersionOfPendingSegmentSendFailureEmitsMetric() + { + EasyMock.expect(spec.isSuspended()).andReturn(false); + EasyMock.expect( + indexTaskClient.registerNewVersionOfPendingSegmentAsync(EasyMock.eq("task0"), EasyMock.anyObject()) + ).andReturn(Futures.immediateFailedFuture(new RuntimeException("boom"))); + replayAll(); + + final TestSeekableStreamSupervisor supervisor = new TestSeekableStreamSupervisor(); + supervisor.getIoConfig().setTaskCount(3); + supervisor.start(); + + final SeekableStreamSupervisor.TaskGroup taskGroup0 = supervisor.addTaskGroupToActivelyReadingTaskGroup( + supervisor.getTaskGroupIdForPartition("0"), + ImmutableMap.of("0", "5"), + null, + null, + Set.of("task0"), + Set.of(), + null + ); + + final PendingSegmentRecord record = PendingSegmentRecord.create( + new SegmentIdWithShardSpec("DS", Intervals.of("2024/2025"), "2024", new NumberedShardSpec(1, 0)), + taskGroup0.getBaseSequenceName(), + "prevId", + "someAppendedSegment", + taskGroup0.getBaseSequenceName() + ); + + supervisor.registerNewVersionOfPendingSegment(record); + + // The record matched task0 (so notified fires) but delivery failed over the wire (so sendFailed fires). + Assert.assertEquals(1, emitter.getMetricEventCount(SegmentUpgradeMetrics.NOTIFIED)); + Assert.assertEquals(1, emitter.getMetricEventCount(SegmentUpgradeMetrics.SEND_FAILED)); verifyAll(); } diff --git a/processing/src/main/java/org/apache/druid/query/DruidMetrics.java b/processing/src/main/java/org/apache/druid/query/DruidMetrics.java index 532553aef6ef..713ee017c003 100644 --- a/processing/src/main/java/org/apache/druid/query/DruidMetrics.java +++ b/processing/src/main/java/org/apache/druid/query/DruidMetrics.java @@ -58,6 +58,7 @@ public class DruidMetrics public static final String PARTITION = "partition"; public static final String SUPERVISOR_ID = "supervisorId"; public static final String REASON = "reason"; + public static final String VERSION = "version"; public static final String TAGS = "tags"; diff --git a/server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java b/server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java index 94e7a56679b1..da9125327c7f 100644 --- a/server/src/main/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderator.java +++ b/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"), + /** The base sink is gone even though this task once held it. */ + SKIPPED_NO_SINK("base sink already dropped"), + /** The base sink is being dropped (handoff in progress); the durable path re-announces at the new version. */ + SKIPPED_DROPPING("dropping base sink"); + + @Nullable + private final String reason; + + PendingSegmentUpgradeResult(@Nullable String reason) + { + this.reason = reason; + } + + /** + * Reason a request was skipped, for the {@code reason} metric dimension; null for {@link #ANNOUNCED}. + */ + @Nullable + public String getReason() + { + return reason; + } + } + + public PendingSegmentUpgradeResult registerUpgradedPendingSegment(PendingSegmentRecord pendingSegmentRecord) throws IOException { SegmentIdWithShardSpec basePendingSegment = idToPendingSegment.get(pendingSegmentRecord.getUpgradedFromSegmentId()); SegmentIdWithShardSpec upgradedPendingSegment = pendingSegmentRecord.getId(); - if (!sinks.containsKey(basePendingSegment) || droppingSinks.contains(basePendingSegment)) { - return; + if (basePendingSegment == null || droppingSinks.contains(basePendingSegment) || !sinks.containsKey(basePendingSegment)) { + if (basePendingSegment == null) { + // This task never allocated a segment matching upgradedFromSegmentId, i.e. the request targeted the wrong task. + return PendingSegmentUpgradeResult.SKIPPED_UNKNOWN_BASE; + } else if (droppingSinks.contains(basePendingSegment)) { + // Expected during handoff: the base sink is being dropped. + return PendingSegmentUpgradeResult.SKIPPED_DROPPING; + } else { + // Unexpected: the base sink is gone even though this task once held it. + return PendingSegmentUpgradeResult.SKIPPED_NO_SINK; + } } final Sink sink = sinks.get(basePendingSegment); @@ -1201,6 +1242,8 @@ public void registerUpgradedPendingSegment(PendingSegmentRecord pendingSegmentRe segmentAnnouncer.announceSegment(newSegment); baseSegmentToUpgradedSegments.get(basePendingSegment).add(upgradedPendingSegment); upgradedSegmentToBaseSegment.put(upgradedPendingSegment, basePendingSegment); + log.info("Announced upgraded segment[%s] for base segment[%s] on task[%s]", upgradedPendingSegment, basePendingSegment, myId); + return PendingSegmentUpgradeResult.ANNOUNCED; } private DataSegment getUpgradedSegment(DataSegment baseSegment, SegmentIdWithShardSpec upgradedVersion) diff --git a/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTest.java b/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTest.java index bc5b3c6de360..4f4a0d390467 100644 --- a/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTest.java +++ b/server/src/test/java/org/apache/druid/segment/realtime/appenderator/StreamAppenderatorTest.java @@ -1160,6 +1160,60 @@ ScheduledFuture getLastScheduledFuture() } } + @Test + public void testRegisterUpgradedPendingSegmentReturnsAnnouncedWhenBaseSinkExists() throws Exception + { + try ( + final StreamAppenderatorTester tester = + new StreamAppenderatorTester.Builder().maxRowsInMemory(2) + .basePersistDirectory(temporaryFolder.newFolder()) + .build()) { + final StreamAppenderator appenderator = (StreamAppenderator) tester.getAppenderator(); + appenderator.startJob(); + // Create the base sink for IDENTIFIERS.get(0) so the upgrade can be announced against it. + appenderator.add(IDENTIFIERS.get(0), ir("2000", "foo", 1), Suppliers.ofInstance(Committers.nil())); + + final StreamAppenderator.PendingSegmentUpgradeResult outcome = appenderator.registerUpgradedPendingSegment( + PendingSegmentRecord.create( + si("2000/2001", "B", 1), + si("2000/2001", "B", 1).asSegmentId().toString(), + IDENTIFIERS.get(0).asSegmentId().toString(), + IDENTIFIERS.get(0).asSegmentId().toString(), + StreamAppenderatorTester.DATASOURCE + ) + ); + + Assert.assertEquals(StreamAppenderator.PendingSegmentUpgradeResult.ANNOUNCED, outcome); + } + } + + @Test + public void testRegisterUpgradedPendingSegmentReturnsSkippedUnknownBaseWhenBaseNotHeld() throws Exception + { + try ( + final StreamAppenderatorTester tester = + new StreamAppenderatorTester.Builder().maxRowsInMemory(2) + .basePersistDirectory(temporaryFolder.newFolder()) + .build()) { + final StreamAppenderator appenderator = (StreamAppenderator) tester.getAppenderator(); + appenderator.startJob(); + + // No sink has ever been created for the upgradedFromSegmentId below, so this task cannot announce it. + // This is the case where the upgrade request reached the wrong task. + final StreamAppenderator.PendingSegmentUpgradeResult outcome = appenderator.registerUpgradedPendingSegment( + PendingSegmentRecord.create( + si("2050/2051", "Z", 1), + si("2050/2051", "Z", 1).asSegmentId().toString(), + si("2050/2051", "Y", 0).asSegmentId().toString(), + si("2050/2051", "Y", 0).asSegmentId().toString(), + StreamAppenderatorTester.DATASOURCE + ) + ); + + Assert.assertEquals(StreamAppenderator.PendingSegmentUpgradeResult.SKIPPED_UNKNOWN_BASE, outcome); + } + } + @Test public void testQueryBySegments_withSegmentVersionUpgrades() throws Exception {