jasonk000 commented on code in PR #14782:
URL: https://github.com/apache/druid/pull/14782#discussion_r1289028958
##########
server/src/main/java/org/apache/druid/server/coordinator/duty/KillUnusedSegments.java:
##########
@@ -112,69 +112,112 @@ public KillUnusedSegments(
@Override
public DruidCoordinatorRuntimeParams run(DruidCoordinatorRuntimeParams
params)
{
+
+ TaskStats taskStats = TaskStats.EMPTY;
Collection<String> dataSourcesToKill =
params.getCoordinatorDynamicConfig().getSpecificDataSourcesToKillUnusedSegmentsIn();
double killTaskSlotRatio =
params.getCoordinatorDynamicConfig().getKillTaskSlotRatio();
int maxKillTaskSlots =
params.getCoordinatorDynamicConfig().getMaxKillTaskSlots();
- int availableKillTaskSlots = getAvailableKillTaskSlots(killTaskSlotRatio,
maxKillTaskSlots);
- if (0 == availableKillTaskSlots) {
- log.debug("Not killing any unused segments because there are no
available kill task slots at this time.");
- return params;
- }
+ int killTaskCapacity = getKillTaskCapacity(
+ CoordinatorDutyUtils.getTotalWorkerCapacity(overlordClient),
+ killTaskSlotRatio,
+ maxKillTaskSlots
+ );
+ int availableKillTaskSlots = getAvailableKillTaskSlots(
+ killTaskCapacity,
+ CoordinatorDutyUtils.getNumActiveTaskSlots(overlordClient,
TASK_PREDICATE).size()
+ );
+ final CoordinatorRunStats stats = params.getCoordinatorStats();
- // If no datasource has been specified, all are eligible for killing
unused segments
- if (CollectionUtils.isNullOrEmpty(dataSourcesToKill)) {
- dataSourcesToKill = segmentsMetadataManager.retrieveAllDataSourceNames();
- }
+ taskStats.availableTaskSlots = availableKillTaskSlots;
+ taskStats.maxSlots = killTaskCapacity;
- final long currentTimeMillis = System.currentTimeMillis();
- if (CollectionUtils.isNullOrEmpty(dataSourcesToKill)) {
- log.debug("No eligible datasource to kill unused segments.");
- } else if (lastKillTime + period > currentTimeMillis) {
- log.debug("Skipping kill of unused segments as kill period has not
elapsed yet.");
- } else {
- log.debug("Killing unused segments in datasources: %s",
dataSourcesToKill);
- lastKillTime = currentTimeMillis;
- killUnusedSegments(dataSourcesToKill, availableKillTaskSlots);
+ if (0 < availableKillTaskSlots) {
+
+ // If no datasource has been specified, all are eligible for killing
unused segments
+ if (CollectionUtils.isNullOrEmpty(dataSourcesToKill)) {
+ dataSourcesToKill =
segmentsMetadataManager.retrieveAllDataSourceNames();
+ }
+
+ final long currentTimeMillis = System.currentTimeMillis();
+ if (CollectionUtils.isNullOrEmpty(dataSourcesToKill)) {
+ log.debug("No eligible datasource to kill unused segments.");
+ } else if (lastKillTime + period > currentTimeMillis) {
+ log.debug("Skipping kill of unused segments as kill period has not
elapsed yet.");
+ } else {
+ log.debug("Killing unused segments in datasources: %s",
dataSourcesToKill);
+ lastKillTime = currentTimeMillis;
+ taskStats.submittedTasks = killUnusedSegments(dataSourcesToKill,
availableKillTaskSlots);
+ }
}
+ addStats(taskStats, stats);
return params;
}
- private void killUnusedSegments(Collection<String> dataSourcesToKill, int
availableKillTaskSlots)
+ private void addStats(
+ TaskStats taskStats,
+ CoordinatorRunStats stats
+ )
+ {
+ stats.add(Stats.Kill.AVAILABLE_SLOTS, taskStats.availableTaskSlots);
+ stats.add(Stats.Kill.SUBMITTED_TASKS, taskStats.submittedTasks);
+ stats.add(Stats.Kill.MAX_SLOTS, taskStats.maxSlots);
+ }
+
+ private int killUnusedSegments(
+ Collection<String> dataSourcesToKill,
+ int availableKillTaskSlots
+ )
{
int submittedTasks = 0;
- for (String dataSource : dataSourcesToKill) {
- if (submittedTasks >= availableKillTaskSlots) {
- log.info(StringUtils.format(
- "Submitted [%d] kill tasks and reached kill task slot limit [%d].
Will resume "
- + "on the next coordinator cycle.", submittedTasks,
availableKillTaskSlots));
- break;
- }
- final Interval intervalToKill = findIntervalForKill(dataSource);
- if (intervalToKill == null) {
- continue;
- }
-
- try {
- FutureUtils.getUnchecked(overlordClient.runKillTask(
- TASK_ID_PREFIX,
- dataSource,
- intervalToKill,
- maxSegmentsToKill
- ), true);
- ++submittedTasks;
- }
- catch (Exception ex) {
- log.error(ex, "Failed to submit kill task for dataSource [%s]",
dataSource);
- if (Thread.currentThread().isInterrupted()) {
- log.warn("skipping kill task scheduling because thread is
interrupted.");
+ if (0 < availableKillTaskSlots &&
!CollectionUtils.isNullOrEmpty(dataSourcesToKill)) {
+ log.info("datasourcesToKill: %s", dataSourcesToKill);
+ for (String dataSource : dataSourcesToKill) {
+ if (submittedTasks >= availableKillTaskSlots) {
+ log.info(StringUtils.format(
+ "Submitted [%d] kill tasks and reached kill task slot limit
[%d]. Will resume "
+ + "on the next coordinator cycle.", submittedTasks,
availableKillTaskSlots));
break;
}
+ final Interval intervalToKill = findIntervalForKill(dataSource);
+ if (intervalToKill == null) {
+ continue;
+ }
+
+ try {
+ FutureUtils.getUnchecked(overlordClient.runKillTask(
+ TASK_ID_PREFIX,
+ dataSource,
+ intervalToKill,
+ maxSegmentsToKill
+ ), true);
+ ++submittedTasks;
+ }
+ catch (Exception ex) {
+ log.error(ex, "Failed to submit kill task for dataSource [%s]",
dataSource);
+ if (Thread.currentThread().isInterrupted()) {
+ log.warn("skipping kill task scheduling because thread is
interrupted.");
+ break;
+ }
+ }
}
}
- log.debug("Submitted [%d] kill tasks for [%d] datasources.",
submittedTasks, dataSourcesToKill.size());
+ log.debug(
+ "Submitted [%d] kill tasks for [%d] datasources.%s",
+ submittedTasks,
+ dataSourcesToKill.size(),
+ availableKillTaskSlots < dataSourcesToKill.size()
+ ? StringUtils.format(
+ " Datasources skipped: %s",
+ ImmutableList.copyOf(dataSourcesToKill).subList(submittedTasks,
dataSourcesToKill.size())
Review Comment:
fwiw, I think this might need to be wrapped in an `log.isDebug()`, since the
string format will likely be executed even if debug is not enabled.
--
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]
