diff --git a/VERSION.in b/VERSION.in index 9de53f193..57432cbd9 100644 --- a/VERSION.in +++ b/VERSION.in @@ -1 +1 @@ -1.29 +1.30 diff --git a/cuebot/src/main/java/com/imageworks/spcue/LayerDetail.java b/cuebot/src/main/java/com/imageworks/spcue/LayerDetail.java index 572139039..47f3ad59c 100644 --- a/cuebot/src/main/java/com/imageworks/spcue/LayerDetail.java +++ b/cuebot/src/main/java/com/imageworks/spcue/LayerDetail.java @@ -15,6 +15,7 @@ package com.imageworks.spcue; +import java.sql.Timestamp; import java.util.ArrayList; import java.util.LinkedHashSet; import java.util.List; @@ -39,6 +40,10 @@ public class LayerDetail extends LayerEntity implements LayerInterface { public int dispatchOrder; public int totalFrameCount; + /** Do not book frames of this layer before this time. Null means no restriction. */ + public Timestamp startAfter; + public String startAfterReason; + public Set tags = new LinkedHashSet(); public Set services = new LinkedHashSet(); public Set limits = new LinkedHashSet(); diff --git a/cuebot/src/main/java/com/imageworks/spcue/PrometheusMetricsCollector.java b/cuebot/src/main/java/com/imageworks/spcue/PrometheusMetricsCollector.java index 3138acbe9..ae83d3f44 100644 --- a/cuebot/src/main/java/com/imageworks/spcue/PrometheusMetricsCollector.java +++ b/cuebot/src/main/java/com/imageworks/spcue/PrometheusMetricsCollector.java @@ -21,6 +21,7 @@ import org.springframework.core.env.Environment; import org.springframework.stereotype.Component; +import com.imageworks.spcue.dao.LayerDao; import com.imageworks.spcue.dispatcher.BookingQueue; import com.imageworks.spcue.dispatcher.DispatchQueue; import com.imageworks.spcue.dispatcher.HostReportHandler; @@ -169,6 +170,18 @@ public class PrometheusMetricsCollector { .name("cue_host_reports_received_total").help("Total number of host reports received") .labelNames("env", "cuebot_host", "facility").register(); + // Layer start-after backoff (dispatcher.layer_delay.rules). The counter ticks once per real + // delay write (concurrent reports that no-op on the conditional monotonic write do not count); + // the gauge is the number of layers currently gated, served by the i_layer_start_after partial + // index. A layer stuck re-delaying shows as a flat non-zero gauge with a climbing counter. + private static final Counter layerDelaysTotal = + Counter.build().name("cuebot_layer_delays_total") + .help("Number of automatic layer booking delays written, by exit status") + .labelNames("env", "cuebot_hosts", "exit_status").register(); + private static final Gauge layersDelayed = Gauge.build().name("cuebot_layers_delayed") + .help("Number of layers whose ts_start_after gate is currently in the future") + .labelNames("env", "cuebot_hosts").register(); + // Memory-stranded cores: idle cores that cannot be booked because their host is out of memory. // Reported per allocation. private static final Gauge coresTotal = @@ -185,6 +198,8 @@ public class PrometheusMetricsCollector { private HostManager hostManager; + private LayerDao layerDao; + private String deployment_environment; private String cuebot_host; @@ -310,6 +325,16 @@ public void collectPrometheusMetrics() { logger.error("Failed to collect memory-stranded core metrics", e); } } + + // Delayed-layer gauge, wrapped separately for the same reason as above. + if (layerDao != null) { + try { + layersDelayed.labels(this.deployment_environment, this.cuebot_host) + .set(layerDao.getDelayedLayerCount()); + } catch (Exception e) { + logger.error("Failed to collect delayed-layer metric", e); + } + } } } @@ -426,6 +451,17 @@ public void recordLayerMaxMemory(double memoryBytes, String show, String shot, .observe(memoryBytes); } + /** + * Record an automatic layer booking delay (a real ts_start_after write, not a no-op) + * + * @param exitStatus the configured exit status that triggered the delay + */ + public void recordLayerDelay(int exitStatus) { + layerDelaysTotal + .labels(this.deployment_environment, this.cuebot_host, String.valueOf(exitStatus)) + .inc(); + } + /** * Record a host report received * @@ -456,4 +492,8 @@ public void setReportQueue(HostReportQueue reportQueue) { public void setHostManager(HostManager hostManager) { this.hostManager = hostManager; } + + public void setLayerDao(LayerDao layerDao) { + this.layerDao = layerDao; + } } diff --git a/cuebot/src/main/java/com/imageworks/spcue/dao/LayerDao.java b/cuebot/src/main/java/com/imageworks/spcue/dao/LayerDao.java index 6c2470991..59f09d0c0 100644 --- a/cuebot/src/main/java/com/imageworks/spcue/dao/LayerDao.java +++ b/cuebot/src/main/java/com/imageworks/spcue/dao/LayerDao.java @@ -15,6 +15,8 @@ package com.imageworks.spcue.dao; +import java.sql.Timestamp; +import java.time.Duration; import java.util.List; import java.util.Map; import java.util.Set; @@ -330,6 +332,35 @@ public interface LayerDao { */ void updateTimeoutLLU(LayerInterface layer, int timeout_llu); + /** + * Authoritatively set or clear the layer's start-after gate. No frame of the layer may start + * before the given time. A null timestamp clears the gate. + * + * @param layer + * @param startAfter time before which no frame of the layer may start, or null to clear + * @param reason free-text provenance, displayed verbatim; ignored when startAfter is null + */ + void updateStartAfter(LayerInterface layer, Timestamp startAfter, String reason); + + /** + * Push the layer's start-after gate into the future by the given backoff duration, but only if + * that moves the gate later than it currently is (conditional monotonic). Used by the automatic + * exit-status backoff so an operator-set later time survives and concurrent reports collapse + * into a single write. + * + * @param layer + * @param backoff how far past now to delay the layer + * @param reason free-text provenance, displayed verbatim + * @return true if a row was written, i.e. this call actually delayed the layer + */ + boolean delayLayerForBackoff(LayerInterface layer, Duration backoff, String reason); + + /** + * Count layers whose start-after gate is currently in the future. Feeds the + * cuebot_layers_delayed gauge. + */ + int getDelayedLayerCount(); + /** * Lowers the minimum memory on a layer if the layer is using less memory and the currnet min * memory is the dispatcher default. diff --git a/cuebot/src/main/java/com/imageworks/spcue/dao/postgres/DispatchQuery.java b/cuebot/src/main/java/com/imageworks/spcue/dao/postgres/DispatchQuery.java index b8e9d6382..34d04bac1 100644 --- a/cuebot/src/main/java/com/imageworks/spcue/dao/postgres/DispatchQuery.java +++ b/cuebot/src/main/java/com/imageworks/spcue/dao/postgres/DispatchQuery.java @@ -76,6 +76,7 @@ public class DispatchQuery { + "job.str_os IN ? " + ") " + "AND (CASE WHEN layer_stat.int_waiting_count > 0 THEN 1 ELSE NULL END) = 1 " + + "AND (layer.ts_start_after IS NULL OR layer.ts_start_after <= current_timestamp) " + "AND layer.int_cores_min <= ? " + "AND layer.int_mem_min <= ? " + "AND (CASE WHEN layer.b_threadable = true THEN 1 ELSE 0 END) >= ? " @@ -139,6 +140,7 @@ public class DispatchQuery { + "job.str_os IN ? " + ") " + "AND (CASE WHEN layer_stat.int_waiting_count > 0 THEN 1 ELSE NULL END) = 1 " + + "AND (layer.ts_start_after IS NULL OR layer.ts_start_after <= current_timestamp) " + "AND layer.int_cores_min <= ? " + "AND layer.int_mem_min <= ? " + "AND (CASE WHEN layer.b_threadable = true THEN 1 ELSE 0 END) >= ? " @@ -264,6 +266,8 @@ private static final String replaceQueryForFifo(String query) { + "(CASE WHEN lst.int_waiting_count > 0 THEN lst.pk_layer ELSE NULL END) = l.pk_layer " + "AND " + "(CASE WHEN lst.int_waiting_count > 0 THEN 1 ELSE NULL END) = 1 " + + "AND " + + "(l.ts_start_after IS NULL OR l.ts_start_after <= current_timestamp) " + "AND " + "l.int_mem_min <= host_local.int_mem_idle " + "AND " @@ -363,6 +367,8 @@ private static final String replaceQueryForFifo(String query) { + "(CASE WHEN lst.int_waiting_count > 0 THEN lst.pk_layer ELSE NULL END) = l.pk_layer " + "AND " + "(CASE WHEN lst.int_waiting_count > 0 THEN 1 ELSE NULL END) = 1 " + + "AND " + + "(l.ts_start_after IS NULL OR l.ts_start_after <= current_timestamp) " + "AND " + "l.int_cores_min <= ? " + "AND " @@ -467,6 +473,8 @@ private static final String replaceQueryForFifo(String query) { + "(CASE WHEN lst.int_waiting_count > 0 THEN lst.pk_layer ELSE NULL END) = l.pk_layer " + "AND " + "(CASE WHEN lst.int_waiting_count > 0 THEN 1 ELSE NULL END) = 1 " + + "AND " + + "(l.ts_start_after IS NULL OR l.ts_start_after <= current_timestamp) " + "AND " + "l.int_cores_min <= ? " + "AND " @@ -601,6 +609,8 @@ private static final String replaceQueryForFifo(String query) { + "layer.int_gpu_mem_min BETWEEN ? AND ? " + "AND " + "frame.str_state='WAITING' " + + "AND " + + "(layer.ts_start_after IS NULL OR layer.ts_start_after <= current_timestamp) " + "AND " + "job.pk_job=? " + "AND layer.pk_layer IN ( " @@ -693,6 +703,8 @@ private static final String replaceQueryForFifo(String query) { + "layer.int_gpu_mem_min BETWEEN ? AND ? " + "AND " + "frame.str_state='WAITING' " + + "AND " + + "(layer.ts_start_after IS NULL OR layer.ts_start_after <= current_timestamp) " + "AND " + "job.pk_job=? " + "AND " @@ -778,6 +790,8 @@ private static final String replaceQueryForFifo(String query) { + "layer.int_gpu_mem_min <= ? " + "AND " + "frame.str_state='WAITING' " + + "AND " + + "(layer.ts_start_after IS NULL OR layer.ts_start_after <= current_timestamp) " + "AND " + "job.pk_job=? " + "AND " @@ -862,6 +876,8 @@ private static final String replaceQueryForFifo(String query) { + "layer.int_gpu_mem_min <= ? " + "AND " + "frame.str_state='WAITING' " + + "AND " + + "(layer.ts_start_after IS NULL OR layer.ts_start_after <= current_timestamp) " + "AND " + "job.pk_job=? " + "AND " @@ -953,6 +969,8 @@ private static final String replaceQueryForFifo(String query) { + "layer.int_gpu_mem_min <= ? " + "AND " + "frame.str_state='WAITING' " + + "AND " + + "(layer.ts_start_after IS NULL OR layer.ts_start_after <= current_timestamp) " + "AND " + "job.pk_layer=? " + "AND layer.pk_layer IN ( " @@ -1045,6 +1063,8 @@ private static final String replaceQueryForFifo(String query) { + "layer.int_gpu_mem_min <= ? " + "AND " + "frame.str_state='WAITING' " + + "AND " + + "(layer.ts_start_after IS NULL OR layer.ts_start_after <= current_timestamp) " + "AND " + "layer.pk_layer=? " + "AND " @@ -1130,6 +1150,8 @@ private static final String replaceQueryForFifo(String query) { + "layer.int_gpu_mem_min <= ? " + "AND " + "frame.str_state='WAITING' " + + "AND " + + "(layer.ts_start_after IS NULL OR layer.ts_start_after <= current_timestamp) " + "AND " + "layer.pk_layer = ? " + "AND " @@ -1214,6 +1236,8 @@ private static final String replaceQueryForFifo(String query) { + "layer.int_gpu_mem_min <= ? " + "AND " + "frame.str_state='WAITING' " + + "AND " + + "(layer.ts_start_after IS NULL OR layer.ts_start_after <= current_timestamp) " + "AND " + "layer.pk_layer= ? " + "AND " diff --git a/cuebot/src/main/java/com/imageworks/spcue/dao/postgres/FrameDaoJdbc.java b/cuebot/src/main/java/com/imageworks/spcue/dao/postgres/FrameDaoJdbc.java index bfdca2fa1..fd568b9d1 100644 --- a/cuebot/src/main/java/com/imageworks/spcue/dao/postgres/FrameDaoJdbc.java +++ b/cuebot/src/main/java/com/imageworks/spcue/dao/postgres/FrameDaoJdbc.java @@ -18,11 +18,14 @@ import java.sql.ResultSet; import java.sql.SQLException; import java.util.ArrayList; +import java.util.Arrays; import java.util.EnumSet; import java.util.List; import java.sql.Timestamp; import java.util.Optional; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.core.env.Environment; import org.springframework.dao.DataAccessException; import org.springframework.jdbc.core.RowMapper; import org.springframework.jdbc.core.support.JdbcDaoSupport; @@ -41,6 +44,7 @@ import com.imageworks.spcue.dao.criteria.FrameSearchInterface; import com.imageworks.spcue.dispatcher.Dispatcher; import com.imageworks.spcue.dispatcher.FrameReservationException; +import com.imageworks.spcue.dispatcher.LayerDelayRules; import com.imageworks.spcue.grpc.depend.DependType; import com.imageworks.spcue.grpc.job.CheckpointState; import com.imageworks.spcue.grpc.job.FrameExitStatus; @@ -55,6 +59,30 @@ public class FrameDaoJdbc extends JdbcDaoSupport implements FrameDao { + /** + * Exit statuses that do not consume a retry when the frame is next started: hardware or + * dispatch conditions not attributable to the frame itself, plus any statuses configured for + * automatic layer backoff (dispatcher.layer_delay.rules). + * + * The layer_delay half of this list is deliberately coupled to + * {@link com.imageworks.spcue.dispatcher.FrameCompleteHandler}, which parses the same property + * to decide which statuses send a frame back to WAITING and delay its layer. Both sides read + * dispatcher.layer_delay.rules through {@link LayerDelayRules#parse}; keep them reading the + * same property, or a delayed frame will silently burn its retries here. + */ + private final Integer[] retryExclusions; + + @Autowired + public FrameDaoJdbc(Environment env) { + List exclusions = new ArrayList(Arrays.asList(-1, + FrameExitStatus.SKIP_RETRY_VALUE, FrameExitStatus.FAILED_LAUNCH_VALUE, + Dispatcher.EXIT_STATUS_FRAME_CLEARED, Dispatcher.EXIT_STATUS_FRAME_ORPHAN, + Dispatcher.EXIT_STATUS_FAILED_KILL, Dispatcher.EXIT_STATUS_DOWN_HOST)); + exclusions.addAll(LayerDelayRules.parse(env.getProperty("dispatcher.layer_delay.rules", "")) + .keySet()); + this.retryExclusions = exclusions.toArray(new Integer[0]); + } + // spotless:off private static final String UPDATE_FRAME_STOPPED_NORSS = "UPDATE frame " @@ -186,8 +214,14 @@ public boolean updateFrameMemoryError(FrameInterface frame) { + "LEFT JOIN layer_stat ON layer_stat.pk_layer = layer_limit.pk_layer " + "GROUP BY limit_record.pk_limit_record) AS sum_running " + "ON limit_record.pk_limit_record = sum_running.pk_limit_record " - + "WHERE sum_running.int_sum_running < limit_record.int_max_value " - + "OR sum_running.int_sum_running IS NULL " + + "WHERE (" + + "sum_running.int_sum_running < limit_record.int_max_value " + + "OR sum_running.int_sum_running IS NULL" + + ") " + + "AND (" + + "layer.ts_start_after IS NULL " + + "OR layer.ts_start_after <= current_timestamp" + + ") " + ")"; // spotless:on @@ -196,7 +230,7 @@ public boolean updateFrameMemoryError(FrameInterface frame) { "UPDATE frame " + "SET int_retries = int_retries + 1 " + "WHERE pk_frame = ? " - + "AND int_exit_status NOT IN (?,?,?,?,?,?,?) "; + + "AND int_exit_status <> ALL(?) "; // spotless:on @Override @@ -210,6 +244,11 @@ public void updateFrameStarted(VirtualProc proc, FrameInterface frame) { proc.memoryReserved, proc.gpusReserved, proc.gpuMemoryReserved, frame.getFrameId(), FrameState.WAITING.toString(), frame.getVersion()); if (result == 0) { + // Zero rows matched is also the normal outcome for a layer held back by its + // limit or its start-after gate, not only a version race. + logger.debug("Reservation refused for frame " + frame + + "; frame was taken by another thread, or its layer is at a limit" + + " or delayed by ts_start_after."); String error_msg = "the frame " + frame + " was updated by another thread."; throw new FrameReservationException(error_msg); } @@ -223,13 +262,15 @@ public void updateFrameStarted(VirtualProc proc, FrameInterface frame) { /* * Frames that were killed via nimby or hardware errors not attributed to the software do * not increment the retry counter. Like failed launch, orphaned frame, failed kill or down - * host. + * host. Exit statuses configured for automatic layer backoff (dispatcher.layer_delay.rules) + * are excluded the same way, so delayed frames keep their full retry budget for genuine + * failures. */ try { - getJdbcTemplate().update(UPDATE_FRAME_RETRIES, frame.getFrameId(), -1, - FrameExitStatus.SKIP_RETRY_VALUE, FrameExitStatus.FAILED_LAUNCH_VALUE, - Dispatcher.EXIT_STATUS_FRAME_CLEARED, Dispatcher.EXIT_STATUS_FRAME_ORPHAN, - Dispatcher.EXIT_STATUS_FAILED_KILL, Dispatcher.EXIT_STATUS_DOWN_HOST); + getJdbcTemplate().update(UPDATE_FRAME_RETRIES, ps -> { + ps.setString(1, frame.getFrameId()); + ps.setArray(2, ps.getConnection().createArrayOf("integer", retryExclusions)); + }); } catch (DataAccessException e) { throw new FrameReservationException(e.getCause()); } diff --git a/cuebot/src/main/java/com/imageworks/spcue/dao/postgres/LayerDaoJdbc.java b/cuebot/src/main/java/com/imageworks/spcue/dao/postgres/LayerDaoJdbc.java index cd9047baa..0556a03a5 100644 --- a/cuebot/src/main/java/com/imageworks/spcue/dao/postgres/LayerDaoJdbc.java +++ b/cuebot/src/main/java/com/imageworks/spcue/dao/postgres/LayerDaoJdbc.java @@ -17,6 +17,8 @@ import java.sql.ResultSet; import java.sql.SQLException; +import java.sql.Timestamp; +import java.time.Duration; import java.util.Arrays; import java.util.HashMap; import java.util.List; @@ -220,6 +222,8 @@ public LayerDetail mapRow(ResultSet rs, int rowNum) throws SQLException { layer.services.addAll(Lists.newArrayList(rs.getString("str_services").split(","))); layer.timeout = rs.getInt("int_timeout"); layer.timeout_llu = rs.getInt("int_timeout_llu"); + layer.startAfter = rs.getTimestamp("ts_start_after"); + layer.startAfterReason = rs.getString("str_start_after_reason"); return layer; } }; @@ -702,6 +706,53 @@ public void updateTimeoutLLU(LayerInterface layer, int timeout_llu) { layer.getLayerId()); } + @Override + public void updateStartAfter(LayerInterface layer, Timestamp startAfter, String reason) { + if (startAfter == null) { + getJdbcTemplate() + .update("UPDATE layer SET ts_start_after=NULL, str_start_after_reason=NULL " + + "WHERE pk_layer=?", layer.getLayerId()); + } else { + getJdbcTemplate().update( + "UPDATE layer SET ts_start_after=?, str_start_after_reason=? WHERE pk_layer=?", + startAfter, reason, layer.getLayerId()); + } + } + + // spotless:off + /** + * Conditional monotonic: writes only when it moves the gate later, so an operator-set later + * time survives and a longer rule can extend a shorter active delay. Reports arriving in + * separate transactions each re-arm the backoff from their own transaction start, which is + * the intended semantic (wait N minutes after the most recent failure); only reports sharing + * a transaction collapse into one write, since current_timestamp is fixed per transaction. + */ + private static final String DELAY_LAYER_FOR_BACKOFF = + "UPDATE layer " + + "SET " + + "ts_start_after = current_timestamp + ?::interval, " + + "str_start_after_reason = ? " + + "WHERE pk_layer = ? " + + "AND (" + + "ts_start_after IS NULL " + + "OR ts_start_after < current_timestamp + ?::interval" + + ")"; + // spotless:on + + @Override + public boolean delayLayerForBackoff(LayerInterface layer, Duration backoff, String reason) { + String interval = backoff.getSeconds() + " seconds"; + return getJdbcTemplate().update(DELAY_LAYER_FOR_BACKOFF, interval, reason, + layer.getLayerId(), interval) > 0; + } + + @Override + public int getDelayedLayerCount() { + return getJdbcTemplate().queryForObject( + "SELECT COUNT(1) FROM layer WHERE ts_start_after > current_timestamp", + Integer.class); + } + @Override public void enableMemoryOptimizer(LayerInterface layer, boolean value) { getJdbcTemplate().update("UPDATE layer SET b_optimize=? WHERE pk_layer=?", value, diff --git a/cuebot/src/main/java/com/imageworks/spcue/dao/postgres/WhiteboardDaoJdbc.java b/cuebot/src/main/java/com/imageworks/spcue/dao/postgres/WhiteboardDaoJdbc.java index ba8aa773c..bdd8b7f98 100644 --- a/cuebot/src/main/java/com/imageworks/spcue/dao/postgres/WhiteboardDaoJdbc.java +++ b/cuebot/src/main/java/com/imageworks/spcue/dao/postgres/WhiteboardDaoJdbc.java @@ -1226,6 +1226,13 @@ public Layer mapRow(ResultSet rs, int rowNum) throws SQLException { } } + // Deferred-booking gate; 0 in the proto means not set. + Timestamp startAfter = rs.getTimestamp("ts_start_after"); + if (startAfter != null) { + builder.setStartAfter(startAfter.getTime() / 1000) + .setStartAfterReason(SqlUtil.getString(rs, "str_start_after_reason")); + } + LayerStats.Builder statsBuilder = LayerStats.newBuilder() .setReservedCores(Convert.coreUnitsToCores(rs.getInt("int_cores"))) .setReservedGpus(rs.getInt("int_gpus")).setMaxRss(rs.getLong("int_max_rss")) diff --git a/cuebot/src/main/java/com/imageworks/spcue/dispatcher/FrameCompleteHandler.java b/cuebot/src/main/java/com/imageworks/spcue/dispatcher/FrameCompleteHandler.java index 6f459c19a..0d29579c9 100644 --- a/cuebot/src/main/java/com/imageworks/spcue/dispatcher/FrameCompleteHandler.java +++ b/cuebot/src/main/java/com/imageworks/spcue/dispatcher/FrameCompleteHandler.java @@ -15,7 +15,9 @@ package com.imageworks.spcue.dispatcher; +import java.time.Duration; import java.util.EnumSet; +import java.util.Map; import java.util.Random; import java.util.concurrent.atomic.AtomicLong; @@ -51,6 +53,7 @@ import com.imageworks.spcue.util.CueExceptionUtil; import com.imageworks.spcue.util.CueUtil; +import com.imageworks.spcue.dao.LayerDao; import com.imageworks.spcue.dao.WhiteboardDao; import com.imageworks.spcue.dao.ShowDao; import com.imageworks.spcue.dao.ServiceDao; @@ -103,6 +106,7 @@ private enum ProcHealth { private WhiteboardDao whiteboardDao; private ServiceDao serviceDao; private ShowDao showDao; + private LayerDao layerDao; private Environment env; private KafkaEventPublisher kafkaEventPublisher; private MonitoringEventBuilder monitoringEventBuilder; @@ -131,6 +135,13 @@ private enum ProcHealth { */ private boolean satisfyDependOnlyOnFrameSuccess; + /** + * Exit statuses that defer the whole layer's booking instead of consuming a retry or killing + * the frame, mapped to how long the layer is deferred. Parsed at startup from + * dispatcher.layer_delay.rules; empty (the default) disables the automatic backoff. + */ + private volatile Map delayRules; + public boolean getSatisfyDependOnlyOnFrameSuccess() { return satisfyDependOnlyOnFrameSuccess; } @@ -139,11 +150,25 @@ public void setSatisfyDependOnlyOnFrameSuccess(boolean satisfyDependOnlyOnFrameS this.satisfyDependOnlyOnFrameSuccess = satisfyDependOnlyOnFrameSuccess; } + public Map getDelayRules() { + return delayRules; + } + + /** + * Replaces the parsed dispatcher.layer_delay.rules. Cuebot sets these once from configuration + * at startup; this exists so tests can exercise a rule set without standing up a second + * application context. + */ + public void setDelayRules(Map delayRules) { + this.delayRules = delayRules; + } + @Autowired public FrameCompleteHandler(Environment env) { this.env = env; satisfyDependOnlyOnFrameSuccess = env.getProperty("depend.satisfy_only_on_frame_success", Boolean.class, true); + delayRules = LayerDelayRules.parse(env.getProperty("dispatcher.layer_delay.rules", "")); } /** @@ -176,7 +201,7 @@ public void handleFrameCompleteReport(final FrameCompleteReport report) { jobManager.getFrameDetail(report.getFrame().getFrameId()); final DispatchFrame frame = jobManager.getDispatchFrame(report.getFrame().getFrameId()); final FrameState newFrameState = - determineFrameState(job, layer, frame, report, frameDetail); + determineFrameState(job, layer, frame, report, frameDetail, delayRules); final String key = proc.getJobId() + "_" + report.getFrame().getLayerId() + "_" + report.getFrame().getFrameId(); @@ -288,6 +313,8 @@ public void handlePostFrameCompleteOperations(VirtualProc proc, FrameCompleteRep dispatchSupport.updateUsageCounters(frame, report.getExitStatus()); + applyLayerDelayRule(frame, resolveExitStatus(report, frameDetail), newFrameState); + if (satisfyDependsAndCompleteLayerAndJob(job, frame, report, newFrameState)) { publishLayerCompletedTelemetry(frame); } @@ -633,9 +660,9 @@ private void finalizeOrphanedFrameComplete(FrameCompleteReport report) { // No proc owns the frame: it is genuinely orphaned, proceed with finalizing it. } - int exitStatus = resolveExitStatus(report, frameDetail); + final int exitStatus = resolveExitStatus(report, frameDetail); final FrameState newFrameState = - determineFrameState(job, layer, frame, report, frameDetail); + determineFrameState(job, layer, frame, report, frameDetail, delayRules); if (!dispatchSupport.stopFrame(frame, newFrameState, exitStatus, report.getFrame().getMaxRss())) { @@ -650,10 +677,11 @@ private void finalizeOrphanedFrameComplete(FrameCompleteReport report) { final String key = job.getJobId() + "_" + report.getFrame().getLayerId() + "_" + report.getFrame().getFrameId(); if (dispatcher.isTestMode()) { - handleOrphanedPostFrameComplete(report, job, frame, newFrameState); + handleOrphanedPostFrameComplete(report, job, frame, newFrameState, exitStatus); } else { queueDispatchTask(key, "handleOrphanedPostFrameComplete", - () -> handleOrphanedPostFrameComplete(report, job, frame, newFrameState)); + () -> handleOrphanedPostFrameComplete(report, job, frame, newFrameState, + exitStatus)); } } @@ -668,16 +696,53 @@ private void finalizeOrphanedFrameComplete(FrameCompleteReport report) { * reservation. */ private void handleOrphanedPostFrameComplete(FrameCompleteReport report, DispatchJob job, - DispatchFrame frame, FrameState newFrameState) { + DispatchFrame frame, FrameState newFrameState, int exitStatus) { if (prometheusMetrics != null) { prometheusMetrics.recordFrameCompleted(newFrameState.name(), frame.show, frame.shot); } dispatchSupport.updateUsageCounters(frame, report.getExitStatus()); + applyLayerDelayRule(frame, exitStatus, newFrameState); + satisfyDependsAndCompleteLayerAndJob(job, frame, report, newFrameState); } + /** + * Writes the layer-level backoff for an exit status configured in dispatcher.layer_delay.rules: + * pushes the layer's start-after gate a configured duration into the future so no frame of the + * layer re-books while the underlying condition (typically a license shortage) persists. The + * write is conditional monotonic, so an operator-set later time survives and the burst of + * reports from a layer's in-flight frames collapses into a single write. + * + * Skipped when the frame was EATEN: auto-eat wins over the delay rule, and nothing is going to + * retry an eaten frame, so a delay would only stretch the eating out and keep the job from + * finishing. + * + * Runs on the dispatch threadpool, so a queue rejection can drop the write. That is acceptable + * and self-healing: the condition persists, the layer re-books, and the next matching report + * writes the delay. + */ + private void applyLayerDelayRule(DispatchFrame frame, int exitStatus, + FrameState newFrameState) { + if (newFrameState.equals(FrameState.EATEN)) { + return; + } + Duration backoff = delayRules.get(exitStatus); + if (backoff == null) { + return; + } + boolean delayed = layerDao.delayLayerForBackoff((LayerInterface) frame, backoff, + "Automatic backoff: exit status " + exitStatus); + if (delayed) { + logger.info("Delayed layer " + frame.getLayerId() + " for " + backoff.toMinutes() + + " minutes: exit status " + exitStatus + " on frame " + frame.getName()); + if (prometheusMetrics != null) { + prometheusMetrics.recordLayerDelay(exitStatus); + } + } + } + /** * Proc-independent frame-completion bookkeeping shared by the normal and orphaned paths: * satisfies frame- and layer-level dependencies, optimizes the layer's resource requirements on @@ -772,15 +837,23 @@ public static int resolveExitStatus(FrameCompleteReport report, FrameDetail fram * (reported by rqd or stored on the frame by a Cuebot-initiated memory kill, see * {@link #resolveExitStatus}) are retried even when the retry count is exhausted. * + * An exit status configured in delayRules (dispatcher.layer_delay.rules) sends the frame back + * to Waiting unconditionally: the layer-level start-after backoff written by the caller is what + * prevents an immediate re-book, and the status is excluded from retry counting. Auto-eat still + * wins over a delay rule, and a delay-rule frame is deliberately immune to the timeout checks + * below (a matching failure exits in seconds, so timeouts are moot). + * * @param job * @param layer * @param frame * @param report * @param frameDetail + * @param delayRules exit statuses that defer the layer instead of failing the frame * @return */ public static final FrameState determineFrameState(DispatchJob job, LayerDetail layer, - DispatchFrame frame, FrameCompleteReport report, FrameDetail frameDetail) { + DispatchFrame frame, FrameCompleteReport report, FrameDetail frameDetail, + Map delayRules) { if (EnumSet.of(FrameState.WAITING, FrameState.EATEN).contains(frame.state)) { return frame.state; } @@ -807,6 +880,9 @@ public static final FrameState determineFrameState(DispatchJob job, LayerDetail if (job.autoEat) { return FrameState.EATEN; } + if (delayRules.containsKey(resolveExitStatus(report, frameDetail))) { + return FrameState.WAITING; + } // Log update (LLU) and run time timeouts. long minutesSinceLogUpdate = @@ -1018,6 +1094,14 @@ public void setShowDao(ShowDao showDao) { this.showDao = showDao; } + public LayerDao getLayerDao() { + return layerDao; + } + + public void setLayerDao(LayerDao layerDao) { + this.layerDao = layerDao; + } + public KafkaEventPublisher getKafkaEventPublisher() { return kafkaEventPublisher; } diff --git a/cuebot/src/main/java/com/imageworks/spcue/dispatcher/LayerDelayRules.java b/cuebot/src/main/java/com/imageworks/spcue/dispatcher/LayerDelayRules.java new file mode 100644 index 000000000..0815ba971 --- /dev/null +++ b/cuebot/src/main/java/com/imageworks/spcue/dispatcher/LayerDelayRules.java @@ -0,0 +1,79 @@ + +/* + * Copyright Contributors to the OpenCue Project + * + * 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.imageworks.spcue.dispatcher; + +import java.time.Duration; +import java.util.Collections; +import java.util.LinkedHashMap; +import java.util.Map; + +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.Logger; + +/** + * Parser for the dispatcher.layer_delay.rules property: comma-separated exit_status:minutes pairs + * (e.g. "330:5,332:60") mapping a frame exit status to how long its layer's booking should be + * deferred when a frame reports that status. An empty value (the default) disables the automatic + * backoff entirely. The exit statuses must agree with the substitute statuses RQD is configured to + * emit via rqd.yaml runner.log_exit_status_rules. + */ +public final class LayerDelayRules { + + private static final Logger logger = LogManager.getLogger(LayerDelayRules.class); + + private LayerDelayRules() {} + + /** + * Parse the rules property into a status-to-backoff map. Malformed entries are skipped with a + * WARN rather than failing startup, mirroring RQD's tolerance for invalid log rules. + */ + public static Map parse(String rules) { + if (rules == null || rules.trim().isEmpty()) { + return Collections.emptyMap(); + } + Map parsed = new LinkedHashMap(); + for (String entry : rules.split(",")) { + entry = entry.trim(); + if (entry.isEmpty()) { + continue; + } + String[] parts = entry.split(":"); + try { + if (parts.length != 2) { + throw new NumberFormatException("expected exit_status:minutes"); + } + int exitStatus = Integer.parseInt(parts[0].trim()); + long minutes = Long.parseLong(parts[1].trim()); + if (exitStatus == 0) { + // Exit status 0 is success: a rule on it would delay a layer on every + // frame that completes normally, which is never what an operator means. + throw new NumberFormatException( + "exit status 0 is success and cannot delay a layer"); + } + if (minutes <= 0) { + throw new NumberFormatException("minutes must be positive"); + } + parsed.put(exitStatus, Duration.ofMinutes(minutes)); + } catch (NumberFormatException | ArithmeticException e) { + // ArithmeticException: a minute count that parses as a long but overflows + // when Duration converts it to seconds is malformed like any other bad entry. + logger.warn("Skipping malformed dispatcher.layer_delay.rules entry \"" + entry + + "\": " + e.getMessage()); + } + } + return Collections.unmodifiableMap(parsed); + } +} diff --git a/cuebot/src/main/java/com/imageworks/spcue/servant/ManageLayer.java b/cuebot/src/main/java/com/imageworks/spcue/servant/ManageLayer.java index d8eb96f18..5d7b50718 100644 --- a/cuebot/src/main/java/com/imageworks/spcue/servant/ManageLayer.java +++ b/cuebot/src/main/java/com/imageworks/spcue/servant/ManageLayer.java @@ -15,6 +15,7 @@ package com.imageworks.spcue.servant; +import java.sql.Timestamp; import java.util.HashSet; import com.google.protobuf.Descriptors; @@ -104,6 +105,8 @@ import com.imageworks.spcue.grpc.job.LayerSetMinGpuMemoryResponse; import com.imageworks.spcue.grpc.job.LayerSetMinMemoryRequest; import com.imageworks.spcue.grpc.job.LayerSetMinMemoryResponse; +import com.imageworks.spcue.grpc.job.LayerSetStartAfterRequest; +import com.imageworks.spcue.grpc.job.LayerSetStartAfterResponse; import com.imageworks.spcue.grpc.job.LayerSetTagsRequest; import com.imageworks.spcue.grpc.job.LayerSetTagsResponse; import com.imageworks.spcue.grpc.job.LayerSetThreadableRequest; @@ -140,6 +143,22 @@ public class ManageLayer extends LayerInterfaceGrpc.LayerInterfaceImplBase { private LocalBookingSupport localBookingSupport; private FrameSearchFactory frameSearchFactory; private final String property = "layer.finished_jobs_readonly"; + + /** + * Width of layer.str_start_after_reason (see V47__Add_layer_start_after.sql). The reason embeds + * a client-supplied username, so it is truncated here rather than letting an oversized value + * fail the UPDATE with an opaque DataAccessException. + */ + private static final int START_AFTER_REASON_MAX_LENGTH = 255; + + /** + * Upper bound on how far in the future a start_after gate may be set. A value in milliseconds + * passed where seconds are expected lands around the year 58,000 - Postgres stores it and + * CueGUI renders it without a year, so the layer is silently gated forever. Anything beyond + * this bound (or negative) is rejected as INVALID_ARGUMENT instead. + */ + private static final long START_AFTER_MAX_FUTURE_YEARS = 5; + @Autowired private Environment env; @@ -445,6 +464,40 @@ public void setTimeoutLLU(LayerSetTimeoutLLURequest request, } } + @Override + public void setStartAfter(LayerSetStartAfterRequest request, + StreamObserver responseObserver) { + long startAfter = request.getStartAfter(); + long maxStartAfter = System.currentTimeMillis() / 1000L + + START_AFTER_MAX_FUTURE_YEARS * 365L * 24 * 3600; + if (startAfter < 0 || startAfter > maxStartAfter) { + responseObserver.onError(Status.INVALID_ARGUMENT + .withDescription("start_after must be a Unix timestamp in seconds no more " + + "than " + START_AFTER_MAX_FUTURE_YEARS + " years in the future, " + + "or 0 to clear the gate; got " + startAfter + + ". Was a milliseconds value passed where seconds are expected?") + .asRuntimeException()); + return; + } + updateLayer(request.getLayer()); + if (attemptChange(env, property, jobManager, layer, responseObserver)) { + if (request.getStartAfter() == 0) { + layerDao.updateStartAfter(layer, null, null); + } else { + String username = + request.getUsername().isEmpty() ? "unknown" : request.getUsername(); + String reason = "Set by " + username; + if (reason.length() > START_AFTER_REASON_MAX_LENGTH) { + reason = reason.substring(0, START_AFTER_REASON_MAX_LENGTH); + } + layerDao.updateStartAfter(layer, new Timestamp(request.getStartAfter() * 1000L), + reason); + } + responseObserver.onNext(LayerSetStartAfterResponse.newBuilder().build()); + responseObserver.onCompleted(); + } + } + @Override public void addLimit(LayerAddLimitRequest request, StreamObserver responseObserver) { diff --git a/cuebot/src/main/resources/conf/ddl/postgres/migrations/V47__Add_layer_start_after.sql b/cuebot/src/main/resources/conf/ddl/postgres/migrations/V47__Add_layer_start_after.sql new file mode 100644 index 000000000..6dde5a766 --- /dev/null +++ b/cuebot/src/main/resources/conf/ddl/postgres/migrations/V47__Add_layer_start_after.sql @@ -0,0 +1,18 @@ +-- Deferred layer booking: no frame of a layer may start before ts_start_after. +-- Written by the automatic license-shortage backoff and by the SetStartAfter RPC. +-- str_start_after_reason is free text displayed verbatim in tooling. + +ALTER TABLE layer + ADD COLUMN ts_start_after TIMESTAMP (6) WITH TIME ZONE DEFAULT NULL, + ADD COLUMN str_start_after_reason VARCHAR(255) DEFAULT NULL; + +-- Almost every row is NULL, so the partial index is tiny. It serves the +-- cuebot_layers_delayed gauge and any "which layers are delayed" query. +-- +-- Plain (non-CONCURRENTLY) form, as in V44: Flyway 5.2.0 (cuebot's test setup) +-- wraps each migration in a transaction and PostgreSQL rejects CREATE INDEX +-- CONCURRENTLY inside one. Every row is NULL at this point so the build is a +-- single scan of layer; on a deployment where even that lock window matters, +-- create it manually with CONCURRENTLY before running Flyway. +CREATE INDEX i_layer_start_after ON layer (ts_start_after) + WHERE ts_start_after IS NOT NULL; diff --git a/cuebot/src/main/resources/conf/spring/applicationContext-service.xml b/cuebot/src/main/resources/conf/spring/applicationContext-service.xml index cf783f129..6d35e00ff 100644 --- a/cuebot/src/main/resources/conf/spring/applicationContext-service.xml +++ b/cuebot/src/main/resources/conf/spring/applicationContext-service.xml @@ -306,6 +306,7 @@ + @@ -393,6 +394,7 @@ + diff --git a/cuebot/src/main/resources/opencue.properties b/cuebot/src/main/resources/opencue.properties index f7e3ce3db..a73604193 100644 --- a/cuebot/src/main/resources/opencue.properties +++ b/cuebot/src/main/resources/opencue.properties @@ -171,6 +171,16 @@ dispatcher.frame_verification_strict_fencing_enabled=true # kills, covering the book->first-report propagation window for a just-booked proc. Default = 120. dispatcher.frame_verification_grace_period_seconds=120 +# Automatic layer booking backoff: comma-separated exit_status:minutes pairs +# (e.g. "330:5,332:60"). When a frame reports one of these exit statuses, the whole layer's +# booking is deferred that many minutes (layer.ts_start_after) instead of the frame consuming a +# retry or dying — intended for operator-configured statuses like a license shortage, where every +# frame of the layer needs the same scarce resource. Frames retry indefinitely while the condition +# persists. The statuses must agree with the substitute exit statuses RQD is configured to emit +# via rqd.yaml runner.log_exit_status_rules; 330 is the conventional license-shortage code. +# Malformed entries are skipped with a warning. Empty (the default) disables the feature. +dispatcher.layer_delay.rules= + # Whether to turn off booking for all allocations. # On this mode, host will report their status but will never get booked. # Default = false diff --git a/cuebot/src/test/java/com/imageworks/spcue/test/dao/postgres/DispatcherDaoTests.java b/cuebot/src/test/java/com/imageworks/spcue/test/dao/postgres/DispatcherDaoTests.java index 0735123a4..1f5d23d38 100644 --- a/cuebot/src/test/java/com/imageworks/spcue/test/dao/postgres/DispatcherDaoTests.java +++ b/cuebot/src/test/java/com/imageworks/spcue/test/dao/postgres/DispatcherDaoTests.java @@ -382,6 +382,37 @@ public void testFindDispatchJobsByLocal() { assertTrue(jobs.size() > 0); } + /** + * Push every layer of a job past its start-after gate, the way the automatic backoff or an + * operator would. + */ + private void delayLayers(JobDetail job) { + jdbcTemplate.update("UPDATE layer SET ts_start_after = current_timestamp " + + "+ interval '5 minutes' WHERE pk_job=?", job.getJobId()); + } + + @Test + @Transactional + @Rollback(true) + public void testFindDispatchJobsByLocalIgnoresDelayedLayers() { + DispatchHost host = getHost(); + final JobDetail job = getJob1(); + assertNotNull(job); + + LocalHostAssignment lja = new LocalHostAssignment(); + lja.setThreads(1); + lja.setMaxMemory(CueUtil.GB16); + lja.setMaxCoreUnits(200); + lja.setMaxGpuMemory(1); + bookingDao.insertLocalHostAssignment(host, job, lja); + + assertTrue(dispatcherDao.findLocalDispatchJobs(host).size() > 0); + + // The job's only waiting layers are delayed, so it is not a local booking candidate. + delayLayers(job); + assertEquals(0, dispatcherDao.findLocalDispatchJobs(host).size()); + } + @Test @Transactional @Rollback(true) @@ -408,6 +439,11 @@ public void testfindUnderProcedJob() { boolean under = dispatcherDao.findUnderProcedJob(job1, proc); assertTrue(under); + + // Delaying the under-proced job's layers removes it as a preemption target: there is + // nothing bookable there, so the proc should not be unbooked for it. + delayLayers(job2); + assertFalse(dispatcherDao.findUnderProcedJob(job1, proc)); } @Test @@ -440,6 +476,10 @@ public void testHigherPriorityJobExistsTrue() { boolean isHigher = dispatcherDao.higherPriorityJobExists(job1, proc); assertTrue(isHigher); + + // Same for the higher-priority job: a delayed layer is not a reason to unbook. + delayLayers(job2); + assertFalse(dispatcherDao.higherPriorityJobExists(job1, proc)); } @Test diff --git a/cuebot/src/test/java/com/imageworks/spcue/test/dao/postgres/FrameDaoTests.java b/cuebot/src/test/java/com/imageworks/spcue/test/dao/postgres/FrameDaoTests.java index a801d888b..69099a601 100644 --- a/cuebot/src/test/java/com/imageworks/spcue/test/dao/postgres/FrameDaoTests.java +++ b/cuebot/src/test/java/com/imageworks/spcue/test/dao/postgres/FrameDaoTests.java @@ -16,6 +16,7 @@ package com.imageworks.spcue.test.dao.postgres; import java.io.File; +import java.util.Collections; import java.util.List; import java.util.Map; import javax.annotation.Resource; @@ -24,6 +25,9 @@ import org.junit.Rule; import org.junit.Test; import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.core.env.Environment; +import org.springframework.core.env.MapPropertySource; +import org.springframework.core.env.StandardEnvironment; import org.springframework.test.annotation.Rollback; import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.junit4.AbstractTransactionalJUnit4SpringContextTests; @@ -45,8 +49,11 @@ import com.imageworks.spcue.dao.ProcDao; import com.imageworks.spcue.dao.criteria.FrameSearchFactory; import com.imageworks.spcue.dao.criteria.FrameSearchInterface; +import com.imageworks.spcue.dao.postgres.FrameDaoJdbc; import com.imageworks.spcue.depend.FrameOnFrame; import com.imageworks.spcue.dispatcher.DispatchSupport; +import com.imageworks.spcue.dispatcher.Dispatcher; +import com.imageworks.spcue.dispatcher.FrameReservationException; import com.imageworks.spcue.grpc.host.HardwareState; import com.imageworks.spcue.grpc.job.CheckpointState; import com.imageworks.spcue.grpc.job.FrameSearchCriteria; @@ -64,6 +71,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; @Transactional @ContextConfiguration(classes = TestAppConfig.class, loader = AnnotationConfigContextLoader.class) @@ -287,6 +295,152 @@ public void testUpdateFrameStarted() { frameDao.updateFrameStarted(proc, fd); } + @Test + @Transactional + @Rollback(true) + public void testUpdateFrameStartedRefusesDelayedLayer() { + + DispatchHost host = createHost(); + JobDetail job = launchJob(); + FrameDetail frame = frameDao.findFrameDetail(job, "0001-pass_1_preprocess"); + DispatchFrame fd = frameDao.getDispatchFrame(frame.getId()); + VirtualProc proc = new VirtualProc(); + proc.allocationId = host.allocationId; + proc.coresReserved = 100; + proc.hostId = host.id; + proc.hostName = host.name; + proc.jobId = job.id; + proc.frameId = frame.id; + proc.layerId = frame.layerId; + proc.showId = frame.showId; + + procDao.insertVirtualProc(proc); + procDao.verifyRunningProc(proc.getId(), frame.getId()); + + // The reservation update is the authoritative start-after gate: a layer delayed into the + // future must refuse to start frames even when reached outside the dispatch queries. + jdbcTemplate.update( + "UPDATE layer SET ts_start_after = current_timestamp + interval '5 minutes' " + + "WHERE pk_layer=?", + frame.getLayerId()); + + try { + frameDao.updateFrameStarted(proc, fd); + fail("Expected FrameReservationException for a delayed layer"); + } catch (FrameReservationException e) { + // Expected: the layer's start-after gate is in the future. + } + + // Once the gate is in the past the same frame starts normally. + jdbcTemplate + .update("UPDATE layer SET ts_start_after = current_timestamp - interval '1 minute' " + + "WHERE pk_layer=?", frame.getLayerId()); + frameDao.updateFrameStarted(proc, fd); + } + + /** + * Exit status conventionally emitted by RQD for a license shortage and used here as the status + * an operator would configure in dispatcher.layer_delay.rules. + */ + private static final int LICENSE_EXIT_STATUS = 330; + + private VirtualProc buildProc(DispatchHost host, JobDetail job, FrameDetail frame) { + VirtualProc proc = new VirtualProc(); + proc.allocationId = host.allocationId; + proc.coresReserved = 100; + proc.hostId = host.id; + proc.hostName = host.name; + proc.jobId = job.id; + proc.frameId = frame.id; + proc.layerId = frame.layerId; + proc.showId = frame.showId; + return proc; + } + + /** + * Puts the frame back in WAITING carrying the given stored exit status, so the next + * updateFrameStarted exercises the retry-exclusion list against that status. + */ + private void rewindFrameWithExitStatus(FrameInterface frame, int exitStatus) { + jdbcTemplate.update("UPDATE frame SET str_state=?, int_exit_status=? WHERE pk_frame=?", + FrameState.WAITING.toString(), exitStatus, frame.getFrameId()); + } + + private int getRetries(FrameInterface frame) { + return jdbcTemplate.queryForObject("SELECT int_retries FROM frame WHERE pk_frame=?", + Integer.class, frame.getFrameId()); + } + + private Environment environmentWithDelayRules(String rules) { + StandardEnvironment testEnv = new StandardEnvironment(); + testEnv.getPropertySources().addFirst(new MapPropertySource("layerDelayRules", + Collections.singletonMap("dispatcher.layer_delay.rules", rules))); + return testEnv; + } + + @Test + @Transactional + @Rollback(true) + public void testUpdateFrameStartedRetryExclusions() { + + DispatchHost host = createHost(); + JobDetail job = launchJob(); + FrameDetail frame = frameDao.findFrameDetail(job, "0001-pass_1_preprocess"); + VirtualProc proc = buildProc(host, job, frame); + + procDao.insertVirtualProc(proc); + procDao.verifyRunningProc(proc.getId(), frame.getId()); + + // An ordinary failure exit status consumes a retry. + rewindFrameWithExitStatus(frame, 1); + frameDao.updateFrameStarted(proc, frameDao.getDispatchFrame(frame.getId())); + assertEquals(1, getRetries(frame)); + + // A hardware/dispatch status on the exclusion list does not: the frame keeps its retry + // budget for genuine failures. Also covers the Integer[] <> ALL(?) array binding. + rewindFrameWithExitStatus(frame, Dispatcher.EXIT_STATUS_DOWN_HOST); + frameDao.updateFrameStarted(proc, frameDao.getDispatchFrame(frame.getId())); + assertEquals(1, getRetries(frame)); + + rewindFrameWithExitStatus(frame, Dispatcher.EXIT_STATUS_FRAME_ORPHAN); + frameDao.updateFrameStarted(proc, frameDao.getDispatchFrame(frame.getId())); + assertEquals(1, getRetries(frame)); + } + + @Test + @Transactional + @Rollback(true) + public void testUpdateFrameStartedRetryExcludesLayerDelayStatuses() { + + DispatchHost host = createHost(); + JobDetail job = launchJob(); + FrameDetail frame = frameDao.findFrameDetail(job, "0001-pass_1_preprocess"); + VirtualProc proc = buildProc(host, job, frame); + + procDao.insertVirtualProc(proc); + procDao.verifyRunningProc(proc.getId(), frame.getId()); + + // dispatcher.layer_delay.rules is empty by default, so the status is an ordinary failure + // and consumes a retry. + rewindFrameWithExitStatus(frame, LICENSE_EXIT_STATUS); + frameDao.updateFrameStarted(proc, frameDao.getDispatchFrame(frame.getId())); + assertEquals(1, getRetries(frame)); + + // Configuring the same status for automatic layer backoff adds it to the exclusion list: + // a delayed frame retries indefinitely instead of burning its retry budget. + FrameDaoJdbc delayAwareDao = new FrameDaoJdbc(environmentWithDelayRules("330:5")); + delayAwareDao.setJdbcTemplate(jdbcTemplate); + + rewindFrameWithExitStatus(frame, LICENSE_EXIT_STATUS); + delayAwareDao.updateFrameStarted(proc, frameDao.getDispatchFrame(frame.getId())); + assertEquals(1, getRetries(frame)); + + // The configured status is the only addition; everything else still counts. + rewindFrameWithExitStatus(frame, 1); + delayAwareDao.updateFrameStarted(proc, frameDao.getDispatchFrame(frame.getId())); + assertEquals(2, getRetries(frame)); + } + @Test @Transactional @Rollback(true) diff --git a/cuebot/src/test/java/com/imageworks/spcue/test/dao/postgres/LayerDaoTests.java b/cuebot/src/test/java/com/imageworks/spcue/test/dao/postgres/LayerDaoTests.java index 5963df1d3..986423929 100644 --- a/cuebot/src/test/java/com/imageworks/spcue/test/dao/postgres/LayerDaoTests.java +++ b/cuebot/src/test/java/com/imageworks/spcue/test/dao/postgres/LayerDaoTests.java @@ -16,6 +16,8 @@ package com.imageworks.spcue.test.dao.postgres; import java.io.File; +import java.sql.Timestamp; +import java.time.Duration; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -63,6 +65,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; import static org.junit.Assert.assertThat; import static org.junit.Assert.assertTrue; @@ -780,4 +783,78 @@ public void testDropLimit() { List limitsB = layerDao.getLimits(layerResultB); assertEquals(limitsB.size(), 0); } + + @Test + @Transactional + @Rollback(true) + public void testUpdateStartAfter() { + LayerDetail layer = getLayer(); + assertNull(layerDao.getLayerDetail(layer.getLayerId()).startAfter); + + Timestamp startAfter = new Timestamp(System.currentTimeMillis() + 3600 * 1000L); + layerDao.updateStartAfter(layer, startAfter, "Set by testuser"); + + LayerDetail updated = layerDao.getLayerDetail(layer.getLayerId()); + assertEquals(startAfter.getTime() / 1000, updated.startAfter.getTime() / 1000); + assertEquals("Set by testuser", updated.startAfterReason); + + layerDao.updateStartAfter(layer, null, null); + updated = layerDao.getLayerDetail(layer.getLayerId()); + assertNull(updated.startAfter); + assertNull(updated.startAfterReason); + } + + @Test + @Transactional + @Rollback(true) + public void testDelayLayerForBackoffIsConditionalMonotonic() { + LayerDetail layer = getLayer(); + + // First write from NULL delays the layer. + assertTrue(layerDao.delayLayerForBackoff(layer, Duration.ofMinutes(5), + "Automatic backoff: exit status 330")); + LayerDetail updated = layerDao.getLayerDetail(layer.getLayerId()); + assertNotNull(updated.startAfter); + assertEquals("Automatic backoff: exit status 330", updated.startAfterReason); + + // A second, equal-or-shorter delay is a no-op (the in-flight report burst). + assertFalse(layerDao.delayLayerForBackoff(layer, Duration.ofMinutes(5), + "Automatic backoff: exit status 330")); + + // A longer rule extends the active delay. + assertTrue(layerDao.delayLayerForBackoff(layer, Duration.ofMinutes(60), + "Automatic backoff: exit status 332")); + assertEquals("Automatic backoff: exit status 332", + layerDao.getLayerDetail(layer.getLayerId()).startAfterReason); + + // An operator-set later time survives a backoff write... + Timestamp tonight = new Timestamp(System.currentTimeMillis() + 8 * 3600 * 1000L); + layerDao.updateStartAfter(layer, tonight, "Set by testuser"); + assertFalse(layerDao.delayLayerForBackoff(layer, Duration.ofMinutes(5), + "Automatic backoff: exit status 330")); + updated = layerDao.getLayerDetail(layer.getLayerId()); + assertEquals(tonight.getTime() / 1000, updated.startAfter.getTime() / 1000); + assertEquals("Set by testuser", updated.startAfterReason); + + // ...but the authoritative write may move the gate earlier or clear it. + layerDao.updateStartAfter(layer, null, null); + assertNull(layerDao.getLayerDetail(layer.getLayerId()).startAfter); + } + + @Test + @Transactional + @Rollback(true) + public void testGetDelayedLayerCount() { + LayerDetail layer = getLayer(); + int baseline = layerDao.getDelayedLayerCount(); + + layerDao.delayLayerForBackoff(layer, Duration.ofMinutes(5), + "Automatic backoff: exit status 330"); + assertEquals(baseline + 1, layerDao.getDelayedLayerCount()); + + // A gate in the past does not count as delayed. + layerDao.updateStartAfter(layer, new Timestamp(System.currentTimeMillis() - 60 * 1000L), + "Set by testuser"); + assertEquals(baseline, layerDao.getDelayedLayerCount()); + } } diff --git a/cuebot/src/test/java/com/imageworks/spcue/test/dispatcher/FrameCompleteHandlerFrameStateTests.java b/cuebot/src/test/java/com/imageworks/spcue/test/dispatcher/FrameCompleteHandlerFrameStateTests.java index e6d483dbf..7ab529c69 100644 --- a/cuebot/src/test/java/com/imageworks/spcue/test/dispatcher/FrameCompleteHandlerFrameStateTests.java +++ b/cuebot/src/test/java/com/imageworks/spcue/test/dispatcher/FrameCompleteHandlerFrameStateTests.java @@ -15,6 +15,10 @@ package com.imageworks.spcue.test.dispatcher; +import java.time.Duration; +import java.util.Collections; +import java.util.Map; + import org.junit.Test; import com.imageworks.spcue.DispatchFrame; @@ -44,6 +48,7 @@ public class FrameCompleteHandlerFrameStateTests { private LayerDetail layer; private DispatchFrame frame; private FrameDetail frameDetail; + private Map delayRules = Collections.emptyMap(); public FrameCompleteHandlerFrameStateTests() { job = new DispatchJob(); @@ -70,7 +75,8 @@ private FrameCompleteReport report(int exitStatus, int exitSignal) { } private FrameState determine(FrameCompleteReport report) { - return FrameCompleteHandler.determineFrameState(job, layer, frame, report, frameDetail); + return FrameCompleteHandler.determineFrameState(job, layer, frame, report, frameDetail, + delayRules); } @Test @@ -222,6 +228,57 @@ public void testStoredMemoryFailureWaitsEvenWhenRetriesExhausted() { assertEquals(FrameState.WAITING, determine(report(1, EXIT_SIGNAL_SIGTERM))); } + private static final int LICENSE_EXIT_STATUS = 330; + + @Test + public void testDelayRuleStatusWaits() { + delayRules = Collections.singletonMap(LICENSE_EXIT_STATUS, Duration.ofMinutes(5)); + assertEquals(FrameState.WAITING, determine(report(LICENSE_EXIT_STATUS, 0))); + } + + @Test + public void testDelayRuleStatusWaitsEvenWhenRetriesExhausted() { + delayRules = Collections.singletonMap(LICENSE_EXIT_STATUS, Duration.ofMinutes(5)); + frame.retries = job.maxRetries; + assertEquals(FrameState.WAITING, determine(report(LICENSE_EXIT_STATUS, 0))); + } + + @Test + public void testAutoEatWinsOverDelayRule() { + delayRules = Collections.singletonMap(LICENSE_EXIT_STATUS, Duration.ofMinutes(5)); + job.autoEat = true; + assertEquals(FrameState.EATEN, determine(report(LICENSE_EXIT_STATUS, 0))); + } + + @Test + public void testDelayRuleStatusImmuneToTimeouts() { + delayRules = Collections.singletonMap(LICENSE_EXIT_STATUS, Duration.ofMinutes(5)); + layer.timeout = 10; + FrameCompleteReport report = FrameCompleteReport.newBuilder(report(LICENSE_EXIT_STATUS, 0)) + .setRunTime(11 * 60).build(); + assertEquals(FrameState.WAITING, determine(report)); + } + + @Test + public void testDelayRuleUsesResolvedExitStatus() { + // A stored memory-failure status wins over the reported status (resolveExitStatus), so a + // delay rule keyed on the reported status must not match. With the layer timeout exceeded + // the frame goes DEAD, proving the delay branch did not fire on the raw reported status. + delayRules = Collections.singletonMap(LICENSE_EXIT_STATUS, Duration.ofMinutes(5)); + frameDetail.exitStatus = Dispatcher.EXIT_STATUS_MEMORY_FAILURE; + layer.timeout = 10; + FrameCompleteReport report = FrameCompleteReport.newBuilder(report(LICENSE_EXIT_STATUS, 0)) + .setRunTime(11 * 60).build(); + assertEquals(FrameState.DEAD, determine(report)); + } + + @Test + public void testUnconfiguredDelayStatusFollowsNormalPath() { + assertEquals(FrameState.WAITING, determine(report(LICENSE_EXIT_STATUS, 0))); + frame.retries = job.maxRetries; + assertEquals(FrameState.DEAD, determine(report(LICENSE_EXIT_STATUS, 0))); + } + @Test public void testResolveExitStatusPrefersStoredMemoryFailure() { FrameDetail frameDetail = new FrameDetail(); diff --git a/cuebot/src/test/java/com/imageworks/spcue/test/dispatcher/FrameCompleteHandlerLayerDelayTests.java b/cuebot/src/test/java/com/imageworks/spcue/test/dispatcher/FrameCompleteHandlerLayerDelayTests.java new file mode 100644 index 000000000..5ead83356 --- /dev/null +++ b/cuebot/src/test/java/com/imageworks/spcue/test/dispatcher/FrameCompleteHandlerLayerDelayTests.java @@ -0,0 +1,245 @@ + +/* + * Copyright Contributors to the OpenCue Project + * + * 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.imageworks.spcue.test.dispatcher; + +import java.io.File; +import java.time.Duration; +import java.util.List; +import java.util.Map; +import javax.annotation.Resource; + +import org.junit.After; +import org.junit.Before; +import org.junit.Test; +import org.springframework.test.annotation.Rollback; +import org.springframework.transaction.annotation.Transactional; + +import com.imageworks.spcue.DispatchHost; +import com.imageworks.spcue.JobDetail; +import com.imageworks.spcue.LayerDetail; +import com.imageworks.spcue.VirtualProc; +import com.imageworks.spcue.dao.FrameDao; +import com.imageworks.spcue.dao.JobDao; +import com.imageworks.spcue.dao.LayerDao; +import com.imageworks.spcue.dispatcher.Dispatcher; +import com.imageworks.spcue.dispatcher.FrameCompleteHandler; +import com.imageworks.spcue.dispatcher.LayerDelayRules; +import com.imageworks.spcue.grpc.host.HardwareState; +import com.imageworks.spcue.grpc.job.FrameState; +import com.imageworks.spcue.grpc.report.FrameCompleteReport; +import com.imageworks.spcue.grpc.report.RenderHost; +import com.imageworks.spcue.grpc.report.RunningFrameInfo; +import com.imageworks.spcue.service.AdminManager; +import com.imageworks.spcue.service.HostManager; +import com.imageworks.spcue.service.JobLauncher; +import com.imageworks.spcue.service.JobManager; +import com.imageworks.spcue.test.TransactionalTest; +import com.imageworks.spcue.util.CueUtil; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNotNull; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +/** + * End-to-end coverage of the automatic layer backoff: a FrameCompleteReport carrying an exit status + * configured in dispatcher.layer_delay.rules must write the reporting layer's start-after gate, and + * nothing else must. + * + * The rules property is empty by default. Rather than start a second application context to + * override it -- the test harness supports only one, since the embedded database and the gRPC + * server's fixed port are both per-process -- these tests install the rule set directly on the + * shared FrameCompleteHandler and restore it afterwards. LayerDelayRulesTests covers the parsing of + * the property itself. + */ +public class FrameCompleteHandlerLayerDelayTests extends TransactionalTest { + + @Resource + AdminManager adminManager; + + @Resource + FrameCompleteHandler frameCompleteHandler; + + @Resource + HostManager hostManager; + + @Resource + JobLauncher jobLauncher; + + @Resource + JobManager jobManager; + + @Resource + JobDao jobDao; + + @Resource + FrameDao frameDao; + + @Resource + LayerDao layerDao; + + @Resource + Dispatcher dispatcher; + + private static final String HOSTNAME = "beta"; + + /** The exit status these tests configure a delay rule for. */ + private static final int LICENSE_EXIT_STATUS = 330; + + private static final int CONFIGURED_DELAY_MINUTES = 5; + + private Map originalDelayRules; + + @Before + public void setTestMode() { + dispatcher.setTestMode(true); + } + + /** + * FrameCompleteHandler is a context-wide singleton, so the rules are restored in + * {@link #tearDown()} to keep the other dispatcher tests running against the configured default + * (no rules). + */ + @Before + public void installDelayRules() { + originalDelayRules = frameCompleteHandler.getDelayRules(); + frameCompleteHandler.setDelayRules( + LayerDelayRules.parse(LICENSE_EXIT_STATUS + ":" + CONFIGURED_DELAY_MINUTES)); + } + + @After + public void tearDown() { + frameCompleteHandler.setDelayRules(originalDelayRules); + } + + @Before + public void launchJob() { + jobLauncher.testMode = true; + jobLauncher.launch(new File("src/test/resources/conf/jobspec/jobspec_gpus_test.xml")); + } + + @Before + public void createHost() { + RenderHost host = RenderHost.newBuilder().setName(HOSTNAME).setBootTime(1192369572) + // The minimum amount of free space in the temporary directory to book a host. + .setFreeMcp(CueUtil.GB).setFreeMem((int) CueUtil.GB8).setFreeSwap(20760).setLoad(0) + .setTotalMcp(CueUtil.GB4).setTotalMem(CueUtil.GB8).setTotalSwap(CueUtil.GB2) + .setNimbyEnabled(false).setNumProcs(40).setCoresPerProc(100) + .setState(HardwareState.UP).setFacility("spi").putAttributes("SP_OS", "Linux") + .setNumGpus(8).setFreeGpuMem(CueUtil.GB16 * 8).setTotalGpuMem(CueUtil.GB16 * 8) + .build(); + + hostManager.createHost(host, adminManager.findAllocationDetail("spi", "general")); + } + + /** + * Books the single frame of pipe-default-testuser_test0/layer0 and returns its proc. + */ + private VirtualProc bookOneFrame(JobDetail job) { + jobManager.setJobPaused(job, false); + DispatchHost host = hostManager.findDispatchHost(HOSTNAME); + List procs = dispatcher.dispatchHost(host); + assertEquals(1, procs.size()); + return procs.get(0); + } + + private void reportFrameComplete(VirtualProc proc, int exitStatus) { + RunningFrameInfo info = RunningFrameInfo.newBuilder().setJobId(proc.getJobId()) + .setLayerId(proc.getLayerId()).setFrameId(proc.getFrameId()) + .setResourceId(proc.getProcId()).build(); + frameCompleteHandler.handleFrameCompleteReport( + FrameCompleteReport.newBuilder().setFrame(info).setExitStatus(exitStatus).build()); + } + + @Test + @Transactional + @Rollback(true) + public void testConfiguredExitStatusDelaysLayer() { + JobDetail job = jobManager.findJobDetail("pipe-default-testuser_test0"); + LayerDetail layer = layerDao.findLayerDetail(job, "layer0"); + assertNull(layerDao.getLayerDetail(layer.getLayerId()).startAfter); + + VirtualProc proc = bookOneFrame(job); + reportFrameComplete(proc, LICENSE_EXIT_STATUS); + + LayerDetail delayed = layerDao.getLayerDetail(layer.getLayerId()); + assertNotNull("A configured exit status must write the layer's start-after gate", + delayed.startAfter); + assertEquals("Automatic backoff: exit status " + LICENSE_EXIT_STATUS, + delayed.startAfterReason); + + // The gate is the configured backoff into the future. Postgres current_timestamp is the + // transaction's start time, which is somewhat before now, so only bound the window + // loosely: what matters is that the configured duration was used and not some default. + long millisOut = delayed.startAfter.getTime() - System.currentTimeMillis(); + assertTrue( + "Expected a gate up to " + CONFIGURED_DELAY_MINUTES + " minutes out, got " + + millisOut + "ms", + millisOut > 0 && millisOut <= CONFIGURED_DELAY_MINUTES * 60 * 1000L); + assertTrue("Gate is implausibly close, the configured backoff was probably not used", + millisOut > (CONFIGURED_DELAY_MINUTES - 2) * 60 * 1000L); + + // The frame itself is left retriable rather than dead: the layer gate is what holds it. + assertEquals(FrameState.WAITING, frameDao.getFrameDetail(proc.getFrameId()).state); + } + + @Test + @Transactional + @Rollback(true) + public void testEatenFrameDoesNotDelayLayer() { + JobDetail job = jobManager.findJobDetail("pipe-default-testuser_test0"); + LayerDetail layer = layerDao.findLayerDetail(job, "layer0"); + jobDao.updateAutoEat(job, true); + + VirtualProc proc = bookOneFrame(job); + reportFrameComplete(proc, LICENSE_EXIT_STATUS); + + // Auto-eat wins over the delay rule: nothing is going to retry an eaten frame, so + // delaying the layer would only keep the job from finishing. + assertEquals(FrameState.EATEN, frameDao.getFrameDetail(proc.getFrameId()).state); + assertNull("An eaten frame must not delay its layer", + layerDao.getLayerDetail(layer.getLayerId()).startAfter); + } + + @Test + @Transactional + @Rollback(true) + public void testUnconfiguredExitStatusDoesNotDelayLayer() { + JobDetail job = jobManager.findJobDetail("pipe-default-testuser_test0"); + LayerDetail layer = layerDao.findLayerDetail(job, "layer0"); + + VirtualProc proc = bookOneFrame(job); + reportFrameComplete(proc, 1); + + assertNull("An exit status with no rule must not delay its layer", + layerDao.getLayerDetail(layer.getLayerId()).startAfter); + assertNull(layerDao.getLayerDetail(layer.getLayerId()).startAfterReason); + } + + @Test + @Transactional + @Rollback(true) + public void testSuccessfulFrameDoesNotDelayLayer() { + JobDetail job = jobManager.findJobDetail("pipe-default-testuser_test0"); + LayerDetail layer = layerDao.findLayerDetail(job, "layer0"); + + VirtualProc proc = bookOneFrame(job); + reportFrameComplete(proc, 0); + + assertNull("A successful frame must not delay its layer", + layerDao.getLayerDetail(layer.getLayerId()).startAfter); + } +} diff --git a/cuebot/src/test/java/com/imageworks/spcue/test/dispatcher/LayerDelayRulesTests.java b/cuebot/src/test/java/com/imageworks/spcue/test/dispatcher/LayerDelayRulesTests.java new file mode 100644 index 000000000..eafd96511 --- /dev/null +++ b/cuebot/src/test/java/com/imageworks/spcue/test/dispatcher/LayerDelayRulesTests.java @@ -0,0 +1,104 @@ + +/* + * Copyright Contributors to the OpenCue Project + * + * 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.imageworks.spcue.test.dispatcher; + +import java.time.Duration; +import java.util.Map; + +import org.junit.Test; + +import com.imageworks.spcue.dispatcher.LayerDelayRules; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertNull; +import static org.junit.Assert.assertTrue; + +/** + * Pure unit tests for the dispatcher.layer_delay.rules parser. No Spring context or database + * required. + */ +public class LayerDelayRulesTests { + + @Test + public void testEmptyDisablesFeature() { + assertTrue(LayerDelayRules.parse("").isEmpty()); + assertTrue(LayerDelayRules.parse(" ").isEmpty()); + assertTrue(LayerDelayRules.parse(null).isEmpty()); + } + + @Test + public void testSingleRule() { + Map rules = LayerDelayRules.parse("330:5"); + assertEquals(1, rules.size()); + assertEquals(Duration.ofMinutes(5), rules.get(330)); + } + + @Test + public void testMultipleRules() { + Map rules = LayerDelayRules.parse("330:5,332:60"); + assertEquals(2, rules.size()); + assertEquals(Duration.ofMinutes(5), rules.get(330)); + assertEquals(Duration.ofMinutes(60), rules.get(332)); + } + + @Test + public void testWhitespaceTolerated() { + Map rules = LayerDelayRules.parse(" 330 : 5 , 332 : 60 "); + assertEquals(2, rules.size()); + assertEquals(Duration.ofMinutes(5), rules.get(330)); + } + + @Test + public void testMalformedEntriesSkippedNotFatal() { + // A bad entry is dropped with a warning; valid entries around it survive. + Map rules = LayerDelayRules.parse("330:5,bogus,332:sixty,333"); + assertEquals(1, rules.size()); + assertEquals(Duration.ofMinutes(5), rules.get(330)); + } + + @Test + public void testNonPositiveMinutesSkipped() { + Map rules = LayerDelayRules.parse("330:0,332:-5,334:10"); + assertEquals(1, rules.size()); + assertEquals(Duration.ofMinutes(10), rules.get(334)); + } + + @Test + public void testExitStatusZeroSkipped() { + // Exit status 0 is success: delaying a layer on every successful frame is nonsense, so + // the entry is dropped with a warning like any other malformed one. + Map rules = LayerDelayRules.parse("0:5,330:5"); + assertEquals(1, rules.size()); + assertNull(rules.get(0)); + assertEquals(Duration.ofMinutes(5), rules.get(330)); + } + + @Test + public void testOverflowingMinutesSkipped() { + // Long.MAX_VALUE parses as a long but overflows when Duration converts minutes to + // seconds. The entry must be dropped like any other malformed one rather than throwing + // out of parse() and failing dispatcher startup. + Map rules = LayerDelayRules.parse("330:" + Long.MAX_VALUE + ",332:5"); + assertEquals(1, rules.size()); + assertNull(rules.get(330)); + assertEquals(Duration.ofMinutes(5), rules.get(332)); + } + + @Test + public void testTrailingCommaTolerated() { + assertEquals(1, LayerDelayRules.parse("330:5,").size()); + } +} diff --git a/cuebot/src/test/java/com/imageworks/spcue/test/servant/ManageLayerStartAfterValidationTests.java b/cuebot/src/test/java/com/imageworks/spcue/test/servant/ManageLayerStartAfterValidationTests.java new file mode 100644 index 000000000..892ec07a2 --- /dev/null +++ b/cuebot/src/test/java/com/imageworks/spcue/test/servant/ManageLayerStartAfterValidationTests.java @@ -0,0 +1,87 @@ + +/* + * Copyright Contributors to the OpenCue Project + * + * 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.imageworks.spcue.test.servant; + +import io.grpc.Status; +import io.grpc.stub.StreamObserver; + +import com.imageworks.spcue.grpc.job.LayerSetStartAfterRequest; +import com.imageworks.spcue.grpc.job.LayerSetStartAfterResponse; +import com.imageworks.spcue.servant.ManageLayer; + +import org.junit.Test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotNull; + +/** + * setStartAfter input validation rejects out-of-range timestamps before touching the layer, so + * these tests run against an unwired servant with no database. + */ +public class ManageLayerStartAfterValidationTests { + + private static class CapturingObserver implements StreamObserver { + Throwable error; + boolean completed; + + @Override + public void onNext(T value) {} + + @Override + public void onError(Throwable t) { + error = t; + } + + @Override + public void onCompleted() { + completed = true; + } + } + + private CapturingObserver setStartAfter(long startAfter) { + CapturingObserver observer = new CapturingObserver<>(); + new ManageLayer().setStartAfter( + LayerSetStartAfterRequest.newBuilder().setStartAfter(startAfter).build(), observer); + return observer; + } + + @Test + public void testRejectsMillisecondTimestamp() { + // The realistic client mistake: milliseconds where seconds are expected. + CapturingObserver observer = + setStartAfter(System.currentTimeMillis()); + assertNotNull(observer.error); + assertEquals(Status.Code.INVALID_ARGUMENT, Status.fromThrowable(observer.error).getCode()); + assertFalse(observer.completed); + } + + @Test + public void testRejectsNegativeTimestamp() { + CapturingObserver observer = setStartAfter(-1L); + assertNotNull(observer.error); + assertEquals(Status.Code.INVALID_ARGUMENT, Status.fromThrowable(observer.error).getCode()); + assertFalse(observer.completed); + } + + @Test + public void testRejectsValueThatWouldOverflowMultiplication() { + CapturingObserver observer = setStartAfter(Long.MAX_VALUE); + assertNotNull(observer.error); + assertEquals(Status.Code.INVALID_ARGUMENT, Status.fromThrowable(observer.error).getCode()); + assertFalse(observer.completed); + } +} diff --git a/cuegui/cuegui/DarkPalette.py b/cuegui/cuegui/DarkPalette.py index 9a9daf272..03144e7fa 100644 --- a/cuegui/cuegui/DarkPalette.py +++ b/cuegui/cuegui/DarkPalette.py @@ -121,6 +121,7 @@ def ColorF(r, g, b): COLOR_JOB_WITHOUT_PROCS = QtGui.QColor(68, 172, 65, 100) COLOR_JOB_DEPENDED = QtGui.QColor(238, 130, 238, 100) COLOR_JOB_HIGH_MEMORY = QtGui.QColor(132, 132, 26) +COLOR_LAYER_DELAYED_BACKGROUND = QtGui.QColor(108, 79, 24) COLOR_GROUP_BACKGROUND = GreyF(0.18) COLOR_GROUP_FOREGROUND = GreyF(0.79) diff --git a/cuegui/cuegui/LayerDialog.py b/cuegui/cuegui/LayerDialog.py index ef4e0551a..3f54ae266 100644 --- a/cuegui/cuegui/LayerDialog.py +++ b/cuegui/cuegui/LayerDialog.py @@ -513,6 +513,101 @@ def apply(self): layer.dropLimit(limit.id()) +class LayerStartAfterDialog(QtWidgets.QDialog): + """Dialog for deferring booking of the selected layers until a chosen time. + + Displays local time and sends UTC epoch seconds. The same field is written + automatically by Cuebot's exit-status backoff (e.g. a license shortage), so + a cleared layer may be delayed again automatically while the underlying + condition persists; a value set here replaces any automatic delay.""" + + def __init__(self, layers, parent=None): + QtWidgets.QDialog.__init__(self, parent) + self.__layers = layers + self.setWindowTitle("Set Start After") + self.setModal(True) + + layout = QtWidgets.QVBoxLayout(self) + + if len(layers) > 1: + layout.addWidget(QtWidgets.QLabel('%d layers selected' % len(layers), self)) + + current = layers[0].data.start_after + if current: + seed = QtCore.QDateTime.fromSecsSinceEpoch(current) + else: + seed = QtCore.QDateTime.currentDateTime() + self.__dateTimeEdit = QtWidgets.QDateTimeEdit(seed, self) + self.__dateTimeEdit.setDisplayFormat('yyyy-MM-dd HH:mm') + self.__dateTimeEdit.setCalendarPopup(True) + pickerLayout = QtWidgets.QHBoxLayout() + pickerLayout.addWidget(self.__dateTimeEdit) + pickerLayout.addWidget(QtWidgets.QLabel('(local time)', self)) + pickerLayout.addStretch() + layout.addLayout(pickerLayout) + + # Presets fill the picker; they are not a separate input mode. + presetLayout = QtWidgets.QHBoxLayout() + for label, minutes in (('+15m', 15), ('+1h', 60), ('+4h', 240)): + button = QtWidgets.QPushButton(label, self) + button.clicked.connect( + lambda checked=False, m=minutes: self.__dateTimeEdit.setDateTime( + QtCore.QDateTime.currentDateTime().addSecs(m * 60))) + presetLayout.addWidget(button) + tonightButton = QtWidgets.QPushButton('Tonight 18:00', self) + tonightButton.clicked.connect(self.__presetTonight) + presetLayout.addWidget(tonightButton) + presetLayout.addStretch() + layout.addLayout(presetLayout) + + if current: + layout.addWidget(QtWidgets.QLabel( + 'current: %s' % cuegui.Utils.dateToMMDDHHMM(current), self)) + if layers[0].data.start_after_reason: + reasonLabel = QtWidgets.QLabel(layers[0].data.start_after_reason, self) + # The reason is free text that embeds a client-supplied username and is + # promised to be displayed verbatim; Qt's default AutoText format would + # otherwise render anything markup-shaped as HTML. + reasonLabel.setTextFormat(QtCore.Qt.PlainText) + layout.addWidget(reasonLabel) + + note = QtWidgets.QLabel( + 'A layer may be delayed again automatically while the underlying ' + 'failure condition (e.g. a license shortage) persists.', self) + note.setWordWrap(True) + layout.addWidget(note) + + buttons = QtWidgets.QDialogButtonBox(QtWidgets.QDialogButtonBox.Cancel, + QtCore.Qt.Horizontal, self) + clearButton = buttons.addButton('Clear', QtWidgets.QDialogButtonBox.ResetRole) + setButton = buttons.addButton('Set', QtWidgets.QDialogButtonBox.AcceptRole) + # pylint: disable=no-member + clearButton.clicked.connect(self.__clear) + setButton.clicked.connect(self.accept) + buttons.rejected.connect(self.reject) + # pylint: enable=no-member + layout.addWidget(buttons) + + def __presetTonight(self): + tonight = QtCore.QDateTime.currentDateTime() + tonight.setTime(QtCore.QTime(18, 0)) + if tonight <= QtCore.QDateTime.currentDateTime(): + tonight = tonight.addDays(1) + self.__dateTimeEdit.setDateTime(tonight) + + def __clear(self): + for layer in self.__layers: + layer.clearStartAfter() + self.close() + + def accept(self): + """Sends the picked time (as UTC epoch seconds) to every selected layer.""" + epoch = self.__dateTimeEdit.dateTime().toSecsSinceEpoch() + for layer in self.__layers: + layer.setStartAfter(epoch) + self.close() + + class LayerTagsDialog(QtWidgets.QDialog): """Dialog for displaying a layer's tags.""" diff --git a/cuegui/cuegui/LayerMonitorTree.py b/cuegui/cuegui/LayerMonitorTree.py index 54f622fc7..8da1a993f 100644 --- a/cuegui/cuegui/LayerMonitorTree.py +++ b/cuegui/cuegui/LayerMonitorTree.py @@ -21,6 +21,8 @@ from __future__ import division import functools +import html +import time from qtpy import QtCore from qtpy import QtGui @@ -36,11 +38,11 @@ import cuegui.Constants import cuegui.Logger import cuegui.MenuActions +import cuegui.Style import cuegui.Utils logger = cuegui.Logger.getLogger(__file__) - def displayRange(layer): """Returns a string representation of a layer's frame range.""" if layer.data.chunk_size != 1: @@ -180,6 +182,15 @@ def __init__(self, parent): tip="The time the last frame of the layer finished. Blank while\n" "any frame is still pending, running, or in DEPEND - mirroring\n" "the job's Stop Time semantics.") + self.addColumn("Start After", 100, id=27, + data=lambda layer: ( + cuegui.Utils.dateToMMDDHHMM(layer.data.start_after) + if layer.data.start_after else ""), + sort=lambda layer: layer.data.start_after, + tip="The time before which no frame of this layer may start.\n" + "Set by an operator (Set Start After...) or written\n" + "automatically by Cuebot's exit-status backoff, e.g. a\n" + "license shortage. Hover a delayed row for the reason.") cuegui.AbstractTreeWidget.AbstractTreeWidget.__init__(self, parent) # pylint: disable=no-member @@ -332,6 +343,7 @@ def contextMenuEvent(self, e): menu.addSeparator() self.__menuActions.layers().addAction(menu, "setProperties").setEnabled(not readonly) + self.__menuActions.layers().addAction(menu, "setStartAfter").setEnabled(not readonly) menu.addSeparator() self.__menuActions.layers().addAction(menu, "kill").setEnabled(not readonly) self.__menuActions.layers().addAction(menu, "eat").setEnabled(not readonly) @@ -390,6 +402,37 @@ def __handle_select_layers(self, layerRpcObjects): class LayerWidgetItem(cuegui.AbstractWidgetItem.AbstractWidgetItem): """Widget item for displaying a single layer.""" + # Index of the "Start After" column, resolved from the column definitions on + # first use so it survives columns being added or reordered above it. + __startAfterColumn = None + def __init__(self, rpcObject, parent): cuegui.AbstractWidgetItem.AbstractWidgetItem.__init__( self, cuegui.Constants.TYPE_LAYER, rpcObject, parent) + if LayerWidgetItem.__startAfterColumn is None: + LayerWidgetItem.__startAfterColumn = next( + (col for col, info in enumerate(self.column_info) + if info[cuegui.AbstractWidgetItem.NAME] == "Start After"), -1) + + def data(self, col, role): + """Extends the base data with the delayed-layer treatment: a tinted + row while the layer's start-after gate is in the future (self-clearing + once the deadline passes) and the gate's reason as the Start After + column's tooltip.""" + if role == QtCore.Qt.BackgroundRole and \ + self.rpcObject.data.start_after > time.time(): + if cuegui.Style.ColorTheme is None: + cuegui.Style.init() + return cuegui.Style.ColorTheme.COLOR_LAYER_DELAYED_BACKGROUND + + if role == QtCore.Qt.ToolTipRole and col == LayerWidgetItem.__startAfterColumn: + reason = self.rpcObject.data.start_after_reason + if not reason: + return reason + # The reason is free text embedding a client-supplied username and is + # promised to be displayed verbatim. Tooltips have no setTextFormat, so + # escape the text and force the rich-text path with an wrapper; + # escaping alone would leave Qt in plain-text mode showing "<". + return '%s' % html.escape(reason) + + return cuegui.AbstractWidgetItem.AbstractWidgetItem.data(self, col, role) diff --git a/cuegui/cuegui/MenuActions.py b/cuegui/cuegui/MenuActions.py index 34fc063c0..54dc5f20c 100644 --- a/cuegui/cuegui/MenuActions.py +++ b/cuegui/cuegui/MenuActions.py @@ -982,6 +982,17 @@ def setTags(self, rpcObjects=None): dialog.exec_() self._update() + setStartAfter_info = [ + "Set Start After...", + "Defer booking of the selected layers until a chosen time", "configure"] + + def setStartAfter(self, rpcObjects=None): + layers = self._getOnlyLayerObjects(rpcObjects) + if layers: + dialog = cuegui.LayerDialog.LayerStartAfterDialog(layers, self._caller) + dialog.exec_() + self._update() + kill_info = ["&Kill", None, "kill"] def kill(self, rpcObjects=None): diff --git a/docs/_docs/reference/rust-rqd.md b/docs/_docs/reference/rust-rqd.md index 1df1ac972..4e50cda30 100644 --- a/docs/_docs/reference/rust-rqd.md +++ b/docs/_docs/reference/rust-rqd.md @@ -269,6 +269,26 @@ Behavior notes: - **Invalid regex is skipped** (with a warning) rather than disabling the whole rule set. - **Efficient tail read**: the log tail is read backward in fixed-size chunks and stops once enough lines are collected — typically only a few kilobytes are read even for large logs, with a hard 1 MiB cap so a pathologically large log is never read in full. +#### Cuebot-side handling: automatic layer backoff + +A substitute exit status is most useful when Cuebot is told what to do with it. Cuebot's +`dispatcher.layer_delay.rules` property (in `opencue.properties`) maps exit statuses to a number of +minutes; when a frame reports a configured status, Cuebot defers booking of the frame's whole layer +for that long (`layer.ts_start_after`) instead of consuming a retry or killing the frame — the +right behavior for a shared-resource shortage like a license pool, where every frame of the layer +would hit the same wall: + +```properties +# Comma-separated exit_status:minutes pairs. Empty (default) disables the feature. +# Must agree with the exit statuses configured in rqd.yaml log_exit_status_rules. +dispatcher.layer_delay.rules=330:5 +``` + +The exit status is an arbitrary number chosen in `rqd.yaml` and repeated in `opencue.properties`; +`330` is the conventional license-shortage code. Delayed layers are visible in CueGUI (tinted row +plus a *Start After* column) and in the `cuebot_layers_delayed` / `cuebot_layer_delays_total` +Prometheus metrics. + ## Testing ### Unit Tests diff --git a/docs/news/2019-04-18-season-of-docs-2019.md b/docs/news/2019-04-18-season-of-docs-2019.md index 2ee695f91..469eadee1 100644 --- a/docs/news/2019-04-18-season-of-docs-2019.md +++ b/docs/news/2019-04-18-season-of-docs-2019.md @@ -2,7 +2,7 @@ layout: default title: "Apr 18, 2019: Season of Docs 2019" parent: News -nav_order: 13 +nav_order: 14 --- # Season of Docs 2019 diff --git a/docs/news/2019-07-08-opencue-birds-of-a-feather-at-siggraph.md b/docs/news/2019-07-08-opencue-birds-of-a-feather-at-siggraph.md index a9533f57f..2cbc16014 100644 --- a/docs/news/2019-07-08-opencue-birds-of-a-feather-at-siggraph.md +++ b/docs/news/2019-07-08-opencue-birds-of-a-feather-at-siggraph.md @@ -2,7 +2,7 @@ layout: default title: "Jul 8, 2019: OpenCue Birds of a Feather at SIGGRAPH" parent: News -nav_order: 12 +nav_order: 13 --- # OpenCue Birds of a Feather at SIGGRAPH diff --git a/docs/news/2019-07-22-opencue-steering-committee-at-siggraph.md b/docs/news/2019-07-22-opencue-steering-committee-at-siggraph.md index 67dc45f0f..9c90557a3 100644 --- a/docs/news/2019-07-22-opencue-steering-committee-at-siggraph.md +++ b/docs/news/2019-07-22-opencue-steering-committee-at-siggraph.md @@ -2,7 +2,7 @@ layout: default title: "Jul 22, 2019: OpenCue Steering Committee at SIGGRAPH" parent: News -nav_order: 11 +nav_order: 12 --- # OpenCue Steering Committee at SIGGRAPH diff --git a/docs/news/2019-09-20-opencue-at-siggraph-recording.md b/docs/news/2019-09-20-opencue-at-siggraph-recording.md index 3c2ca4d09..bfb32c632 100644 --- a/docs/news/2019-09-20-opencue-at-siggraph-recording.md +++ b/docs/news/2019-09-20-opencue-at-siggraph-recording.md @@ -2,7 +2,7 @@ layout: default title: "Sept 20, 2019: OpenCue at SIGGRAPH recording" parent: News -nav_order: 10 +nav_order: 11 --- # OpenCue at SIGGRAPH recording diff --git a/docs/news/2019-12-05-la-pipeline-developers-meetup.md b/docs/news/2019-12-05-la-pipeline-developers-meetup.md index 40b3b9cdd..2a6f1dec6 100644 --- a/docs/news/2019-12-05-la-pipeline-developers-meetup.md +++ b/docs/news/2019-12-05-la-pipeline-developers-meetup.md @@ -2,7 +2,7 @@ layout: default title: "Dec 5, 2019: LA Pipeline Developers Meetup" parent: News -nav_order: 9 +nav_order: 10 --- # JLA Pipeline Developers Meetup diff --git a/docs/news/2019-12-18-sony-pictures-imageworks-case-study.md b/docs/news/2019-12-18-sony-pictures-imageworks-case-study.md index c38d8c29f..c3a1a9736 100644 --- a/docs/news/2019-12-18-sony-pictures-imageworks-case-study.md +++ b/docs/news/2019-12-18-sony-pictures-imageworks-case-study.md @@ -2,7 +2,7 @@ layout: default title: "Dec 18, 2019: Sony Pictures Imageworks case study" parent: News -nav_order: 8 +nav_order: 9 --- # Sony Pictures Imageworks case study diff --git a/docs/news/2020-08-27-google-summer-of-code-20-cloud-plugin.md b/docs/news/2020-08-27-google-summer-of-code-20-cloud-plugin.md index 520f9c60b..63a2abe4d 100644 --- a/docs/news/2020-08-27-google-summer-of-code-20-cloud-plugin.md +++ b/docs/news/2020-08-27-google-summer-of-code-20-cloud-plugin.md @@ -2,7 +2,7 @@ layout: default title: "Aug 27, 2020: Google Summer of Code '20 - Cloud Plugin" parent: News -nav_order: 7 +nav_order: 8 --- # Google Summer of Code '20 - Cloud Plugin diff --git a/docs/news/2021-08-04-open-source-days-2021.md b/docs/news/2021-08-04-open-source-days-2021.md index adc427f32..bddd9a036 100644 --- a/docs/news/2021-08-04-open-source-days-2021.md +++ b/docs/news/2021-08-04-open-source-days-2021.md @@ -2,7 +2,7 @@ layout: default title: "Aug 4, 2021: Open Source Days 2021" parent: News -nav_order: 6 +nav_order: 7 --- # Open Source Days 2021 diff --git a/docs/news/2024-05-24-opencue-project-review-2024.md b/docs/news/2024-05-24-opencue-project-review-2024.md index dedbcf71b..21d1f277e 100644 --- a/docs/news/2024-05-24-opencue-project-review-2024.md +++ b/docs/news/2024-05-24-opencue-project-review-2024.md @@ -2,7 +2,7 @@ layout: default title: "May 24, 2024: OpenCue Project Review 2024" parent: News -nav_order: 5 +nav_order: 6 --- # OpenCue Project Review 2024 diff --git a/docs/news/2025-08-10-opencue-project-review-2025.md b/docs/news/2025-08-10-opencue-project-review-2025.md index 35b883581..11f51adc2 100644 --- a/docs/news/2025-08-10-opencue-project-review-2025.md +++ b/docs/news/2025-08-10-opencue-project-review-2025.md @@ -2,7 +2,7 @@ layout: default title: "August 10, 2025: OpenCue Project Review 2025" parent: News -nav_order: 4 +nav_order: 5 --- # OpenCue Project Review 2025 diff --git a/docs/news/2025-12-12-distributed-scheduler-release.md b/docs/news/2025-12-12-distributed-scheduler-release.md index 29d12770e..221079b35 100644 --- a/docs/news/2025-12-12-distributed-scheduler-release.md +++ b/docs/news/2025-12-12-distributed-scheduler-release.md @@ -2,7 +2,7 @@ layout: default title: "December 12, 2025: Distributed Scheduler Release" parent: News -nav_order: 3 +nav_order: 4 --- # Distributed Scheduler Release diff --git a/docs/news/2026-01-21-opencue-major-releases-2026-roadmap.md b/docs/news/2026-01-21-opencue-major-releases-2026-roadmap.md index f915266e2..e5d27a398 100644 --- a/docs/news/2026-01-21-opencue-major-releases-2026-roadmap.md +++ b/docs/news/2026-01-21-opencue-major-releases-2026-roadmap.md @@ -2,7 +2,7 @@ layout: default title: "January 21, 2026: OpenCue Project Update: Major Releases and 2026 Roadmap" parent: News -nav_order: 2 +nav_order: 3 --- # OpenCue Project Update: Major Releases and 2026 Roadmap diff --git a/docs/news/2026-07-07-cueweb-full-cuegui-parity-release.md b/docs/news/2026-07-07-cueweb-full-cuegui-parity-release.md index 898527ebd..8a8812a1d 100644 --- a/docs/news/2026-07-07-cueweb-full-cuegui-parity-release.md +++ b/docs/news/2026-07-07-cueweb-full-cuegui-parity-release.md @@ -2,7 +2,7 @@ layout: default title: "July 7, 2026: Announcing OpenCueWeb: The Complete Web-Based OpenCue GUI" parent: News -nav_order: 1 +nav_order: 2 --- # Announcing OpenCueWeb: The Complete Web-Based OpenCue GUI diff --git a/docs/news/2026-08-06-rqd-log-exit-status-rules.md b/docs/news/2026-08-06-rqd-log-exit-status-rules.md index e71ddab63..036fa85f7 100644 --- a/docs/news/2026-08-06-rqd-log-exit-status-rules.md +++ b/docs/news/2026-08-06-rqd-log-exit-status-rules.md @@ -2,7 +2,7 @@ layout: default title: "August 6, 2026: Log-Based Exit-Status Rules in Rust RQD" parent: News -nav_order: 0 +nav_order: 1 --- # Log-Based Exit-Status Rules in Rust RQD diff --git a/docs/news/2026-08-07-layer-start-after-deferred-booking.md b/docs/news/2026-08-07-layer-start-after-deferred-booking.md new file mode 100644 index 000000000..bee9528d7 --- /dev/null +++ b/docs/news/2026-08-07-layer-start-after-deferred-booking.md @@ -0,0 +1,118 @@ +--- +layout: default +title: "August 7, 2026: Deferred Layer Booking with Start After" +parent: News +nav_order: 0 +--- + +# Deferred Layer Booking with Start After + +### Automatic license-shortage backoff and operator-scheduled layer starts + +#### August 7, 2026 + +--- + +Layers can now carry a **start-after time**: a timestamp meaning "do not book frames of this layer +before this time." Two writers use it: + +1. **Automatic backoff.** When a frame reports an operator-configured exit status — such as the + license shortage that RQD's [log-based exit-status rules](/news/2026-08-06-rqd-log-exit-status-rules/) + can detect — Cuebot pushes the whole layer's start-after time a few minutes into the future + instead of letting the frames retry immediately or die. Frames retry indefinitely without ever + consuming a retry. +2. **Operators and tools**, via a new `SetStartAfter` RPC, a CueGUI *Set Start After…* right-click + action, and pycue's `Layer.setStartAfter()` — which also delivers a general-purpose "start this + layer at 18:00" capability. + +## The Challenge + +A license shortage is not a property of the frame that hit it. Previously Cuebot could not tell one +from a generic crash, so a farm at license cap produced a wave of dead frames that someone had to +clean up by hand, every single time. Retrying immediately is pointless — the license is still gone — +and `maxRetries` is typically low enough that the frames died within seconds of each other. + +RQD's log-based exit-status rules gave RQD the ability to recognise the failure from the frame log +and report a substitute exit status. This feature is the Cuebot half: react to that status by +pausing the *layer* for a few minutes rather than killing its frames. Pausing the layer is correct +because every frame in a layer depends on the same license — one failure is enough to establish +that the pool is exhausted. + +## The Solution + +### Automatic backoff configuration + +Add matching configuration to both sides: + +`rqd.yaml` (see the [Rust RQD reference](/docs/reference/rust-rqd/)): + +```yaml +runner: + log_exit_status_rules: + - name: "HOUDINI_LICENSE_ERROR" + regex: "A usable license to run the application is installed but they are all in use" + exit_status: 330 +``` + +`opencue.properties`: + +```properties +# Comma-separated exit_status:minutes pairs. Empty (default) disables the feature. +dispatcher.layer_delay.rules=330:5 +``` + +When a frame exits `330`, Cuebot marks the frame `WAITING` (no retry consumed), records +`Automatic backoff: exit status 330` on the layer, and defers the layer's booking for 5 minutes. +In-flight frames on the same layer reporting the same status collapse into that one write. When the +delay expires the layer books again; if the license is still gone, the next report re-delays it. +`330` is the conventional license-shortage code — it sits safely outside the exit statuses Cuebot +reserves internally. + +Key behaviors: + +- **Off by default** — an empty rule list changes nothing on upgrade. +- **No retries consumed** — configured statuses are excluded from retry counting, so frames keep + their full retry budget for genuine failures. +- **Auto-eat wins** — on a job with auto-eat enabled a matching failure is still eaten, so the job + finishes promptly. +- **Operator intent survives** — the automatic write only ever moves the time later; an + operator-set 18:00 start cannot be pulled earlier by a backoff, and a longer rule can extend a + shorter active delay. +- **Both dispatchers honour the gate** — the Cuebot dispatcher (including local dispatch) and the + Rust scheduler both enforce it at the frame-reservation update, the single authoritative choke + point. + +### Operator scheduling + +In CueGUI's layer view, right-click → *Set Start After…* opens a picker with quick presets +(+15m, +1h, +4h, Tonight 18:00). Delayed layers show a tinted row and a *Start After* column whose +tooltip explains why the layer is delayed — `Automatic backoff: exit status 330` or +`Set by `. *Clear* makes the layer bookable immediately (though it may be re-delayed +automatically while the underlying condition persists). + +From python: + +```python +layer.setStartAfter(epoch_seconds) # defer booking +layer.startAfter() # read it back (0 = not set) +layer.startAfterReason() # provenance, displayed verbatim +layer.clearStartAfter() # bookable immediately +``` + +### Monitoring + +Two Prometheus metrics make a broken license server loud instead of silent: + +| Metric | Meaning | +| --- | --- | +| `cuebot_layer_delays_total{exit_status}` | Automatic delays written, by exit status | +| `cuebot_layers_delayed` | Layers currently gated | + +A layer stuck re-delaying for hours shows as a flat non-zero gauge with a climbing counter — alert +on `cuebot_layers_delayed > 0` sustained. + +## Availability + +The layer start-after gate is available now in Cuebot, the Rust scheduler, pycue, and CueGUI. +CueWeb parity (a read-only column, then set/clear) and `rest_gateway` registration of +`SetStartAfter` are tracked as follow-up work. diff --git a/proto/src/job.proto b/proto/src/job.proto index 3692b33ae..581a33946 100644 --- a/proto/src/job.proto +++ b/proto/src/job.proto @@ -383,6 +383,11 @@ service LayerInterface { // Set the Min Memory value for the layer rpc SetMinMemory(LayerSetMinMemoryRequest) returns (LayerSetMinMemoryResponse); + // Do not book frames of this layer before the given time. Also written automatically by + // Cuebot's exit-status backoff (dispatcher.layer_delay.rules); an operator-set time is + // authoritative and replaces any automatic delay. + rpc SetStartAfter(LayerSetStartAfterRequest) returns (LayerSetStartAfterResponse); + // Set tags on the layer rpc SetTags(LayerSetTagsRequest) returns (LayerSetTagsResponse); @@ -727,6 +732,11 @@ message Layer { int32 eligible_time = 23; int32 start_time = 24; int32 stop_time = 25; + // Do not book frames of this layer before this time. Epoch seconds; 0 = not set. + // int64 because this is a future timestamp an operator may set past the int32 epoch limit. + int64 start_after = 26; + // Free-text provenance for start_after, displayed verbatim. + string start_after_reason = 27; } message LayerSeq { @@ -1785,6 +1795,17 @@ message LayerSetMinMemoryRequest { message LayerSetMinMemoryResponse {} // Empty +// SetStartAfter +message LayerSetStartAfterRequest { + Layer layer = 1; + // Epoch seconds; 0 clears the delay. See Layer.start_after for why this is int64. + int64 start_after = 2; + // Recorded in the layer's start_after_reason for provenance. + string username = 3; +} + +message LayerSetStartAfterResponse {} // Empty + // SetTags message LayerSetTagsRequest { Layer layer = 1; diff --git a/pycue/opencue/wrappers/layer.py b/pycue/opencue/wrappers/layer.py index 30e487410..351aa2fe0 100644 --- a/pycue/opencue/wrappers/layer.py +++ b/pycue/opencue/wrappers/layer.py @@ -223,6 +223,38 @@ def setTimeoutLLU(self, timeout_llu): layer=self.data, timeout_llu=timeout_llu), timeout=Cuebot.Timeout) + def setStartAfter(self, epoch_seconds, username=None): + """Defers booking of this layer: no frame of the layer will start before + the given time. + + The same field is written automatically by Cuebot's exit-status backoff + (e.g. a license shortage); a value set here is authoritative and replaces + any automatic delay, and a later automatic delay can only move the time + further into the future. + + :type epoch_seconds: int + :param epoch_seconds: UTC epoch time before which no frame may start; + 0 clears the delay + :type username: str + :param username: recorded in the start-after reason for provenance; + defaults to the current user + """ + username = username if username else getpass.getuser() + return self.stub.SetStartAfter(job_pb2.LayerSetStartAfterRequest( + layer=self.data, start_after=epoch_seconds, username=username), + timeout=Cuebot.Timeout) + + def clearStartAfter(self, username=None): + """Clears the layer's start-after delay, making it bookable immediately. + + Note the layer may be delayed again automatically while the condition + that triggered an automatic delay (e.g. a license shortage) persists. + + :type username: str + :param username: recorded for provenance; defaults to the current user + """ + return self.setStartAfter(0, username=username) + def addRenderPartition(self, hostname, threads, max_cores, max_mem, max_gpu_memory, max_gpus): """Adds a render partition to the layer. @@ -696,3 +728,40 @@ def stopTime(self, format=None): if not format: return self.data.stop_time return time.strftime(format, time.localtime(self.data.stop_time)) + + # pylint: disable=redefined-builtin + def startAfter(self, format=None): + """Returns the layer's start-after time in the desired format. + + No frame of the layer may start before this time. Written by operators + (setStartAfter) or automatically by Cuebot's exit-status backoff. + Returns 0 when no delay is set. + + Examples: + None + "%m/%d %H:%M" => 05/17 18:00 + "%a %b %d %H:%M:%S %Y" => Sun May 17 18:00:00 2026 + + See the format table at: + https://docs.python.org/3/library/time.html + + :type format: str + :param format: desired time format + :rtype: int/str + :return: layer start-after time in epoch, or string version of that + timestamp if format given; an empty string when no delay is + set and a format was given""" + if not format: + return self.data.start_after + if not self.data.start_after: + # 0 means "no delay", not the epoch; formatting it would show 1970. + return '' + return time.strftime(format, time.localtime(self.data.start_after)) + + def startAfterReason(self): + """Returns the free-text provenance of the layer's start-after time, + e.g. "Automatic backoff: exit status 330" or "Set by ". + + :rtype: str + :return: reason the start-after time was set, or an empty string""" + return self.data.start_after_reason diff --git a/pycue/tests/wrappers/test_layer.py b/pycue/tests/wrappers/test_layer.py index 08f99d01a..3573f2cff 100644 --- a/pycue/tests/wrappers/test_layer.py +++ b/pycue/tests/wrappers/test_layer.py @@ -198,6 +198,55 @@ def testSetTags(self, getStubMock): stubMock.SetTags.assert_called_with( job_pb2.LayerSetTagsRequest(layer=layer.data, tags=tags), timeout=mock.ANY) + def testSetStartAfter(self, getStubMock): + stubMock = mock.Mock() + stubMock.SetStartAfter.return_value = job_pb2.LayerSetStartAfterResponse() + getStubMock.return_value = stubMock + + testEpoch = 1800000000 + layer = opencue.wrappers.layer.Layer( + job_pb2.Layer(name=TEST_LAYER_NAME)) + layer.setStartAfter(testEpoch, username='testuser') + + stubMock.SetStartAfter.assert_called_with( + job_pb2.LayerSetStartAfterRequest( + layer=layer.data, start_after=testEpoch, username='testuser'), + timeout=mock.ANY) + + def testClearStartAfter(self, getStubMock): + stubMock = mock.Mock() + stubMock.SetStartAfter.return_value = job_pb2.LayerSetStartAfterResponse() + getStubMock.return_value = stubMock + + layer = opencue.wrappers.layer.Layer( + job_pb2.Layer(name=TEST_LAYER_NAME)) + layer.clearStartAfter(username='testuser') + + stubMock.SetStartAfter.assert_called_with( + job_pb2.LayerSetStartAfterRequest( + layer=layer.data, start_after=0, username='testuser'), + timeout=mock.ANY) + + def testStartAfter(self, getStubMock): + testEpoch = 1800000000 + layer = opencue.wrappers.layer.Layer( + job_pb2.Layer(name=TEST_LAYER_NAME, start_after=testEpoch, + start_after_reason='Automatic backoff: exit status 330')) + + self.assertEqual(layer.startAfter(), testEpoch) + self.assertEqual(layer.startAfter('%Y'), + time.strftime('%Y', time.localtime(testEpoch))) + self.assertEqual(layer.startAfterReason(), 'Automatic backoff: exit status 330') + + def testStartAfterUnset(self, getStubMock): + # 0 means "no delay": raw callers still see 0, but a formatted call must not + # render the epoch as 1970. + layer = opencue.wrappers.layer.Layer(job_pb2.Layer(name=TEST_LAYER_NAME)) + + self.assertEqual(layer.startAfter(), 0) + self.assertEqual(layer.startAfter('%m/%d %H:%M'), '') + self.assertEqual(layer.startAfterReason(), '') + def testSetMaxCores(self, getStubMock): stubMock = mock.Mock() stubMock.SetMaxCores.return_value = job_pb2.LayerSetMaxCoresResponse() diff --git a/rust/config/rqd.yaml b/rust/config/rqd.yaml index 9adbb9cc8..5c55e8367 100644 --- a/rust/config/rqd.yaml +++ b/rust/config/rqd.yaml @@ -239,9 +239,18 @@ runner: # retry differently (e.g. license shortages). Rules are evaluated top-to-bottom; first match # wins. Leave `log_exit_status_rules` empty to disable. Not applied to Loki-only frames. # + # `log_scan_last_lines` and `log_exit_status_rules` are reloaded live: RQD re-reads this + # file every `log_exit_status_rules_reload_interval` and applies changes to these two keys + # without a restart, including to frames that were already running. Changes to any other + # key in this file still require a restart. + # # Number of trailing log lines to scan on failure (Default: 50) # log_scan_last_lines: 50 # + # How often to re-read this file for rule changes. 0 disables live reloading. + # (Default: 5m) + # log_exit_status_rules_reload_interval: 5m + # # log_exit_status_rules: # - name: "HOUDINI_LICENSE_ERROR" # regex: "A usable license to run the application is installed but they are all in use" diff --git a/rust/crates/rqd/src/config/mod.rs b/rust/crates/rqd/src/config/mod.rs index 7d0da8957..11550bfd3 100644 --- a/rust/crates/rqd/src/config/mod.rs +++ b/rust/crates/rqd/src/config/mod.rs @@ -22,10 +22,10 @@ use std::{ collections::HashMap, env, fs, path::Path, - sync::{Arc, OnceLock}, + sync::{Arc, RwLock}, time::Duration, }; -use tracing::warn; +use tracing::{info, warn}; static DEFAULT_CONFIG_FILE: &str = "~/.local/share/rqd.yaml"; @@ -207,6 +207,73 @@ mod tests { assert!(config.machine.nimby_mode); assert!(config.machine.nimby_lock_by_default); } + + #[test] + fn compiled_exit_status_rules_seed_lazily_from_raw_fields() { + let mut config = super::RunnerConfig::default(); + config.log_scan_last_lines = 25; + config.log_exit_status_rules = vec![super::LogExitStatusRule { + name: "LICENSE".to_string(), + regex: "all in use".to_string(), + exit_status: 330, + }]; + + let rule_set = config.compiled_exit_status_rules(); + assert_eq!(rule_set.scan_last_lines, 25); + assert_eq!(rule_set.rules.len(), 1); + assert_eq!(rule_set.rules[0].name, "LICENSE"); + } + + #[test] + fn reload_exit_status_rules_reaches_previously_made_clones() { + // A clone taken before the reload (as every RunningFrame holds) must observe the new + // rules — the compiled-rules cell is shared across clones, not copied. + let config = super::RunnerConfig::default(); + let clone_before_reload = config.clone(); + // Seed the cell from the (empty) raw fields first, as startup does. + assert!(clone_before_reload + .compiled_exit_status_rules() + .rules + .is_empty()); + + config.reload_exit_status_rules( + 10, + &[super::LogExitStatusRule { + name: "NEW_RULE".to_string(), + regex: "added later".to_string(), + exit_status: 331, + }], + ); + + let rule_set = clone_before_reload.compiled_exit_status_rules(); + assert_eq!(rule_set.scan_last_lines, 10); + assert_eq!(rule_set.rules.len(), 1); + assert_eq!(rule_set.rules[0].name, "NEW_RULE"); + } + + #[test] + fn reload_exit_status_rules_skips_invalid_regex() { + let config = super::RunnerConfig::default(); + config.reload_exit_status_rules( + 50, + &[ + super::LogExitStatusRule { + name: "BAD".to_string(), + regex: "(unclosed".to_string(), + exit_status: 1, + }, + super::LogExitStatusRule { + name: "GOOD".to_string(), + regex: "valid".to_string(), + exit_status: 2, + }, + ], + ); + + let rule_set = config.compiled_exit_status_rules(); + assert_eq!(rule_set.rules.len(), 1); + assert_eq!(rule_set.rules[0].name, "GOOD"); + } } /// A rule that reclassifies a failed frame's exit status based on its log output. @@ -217,7 +284,7 @@ mod tests { /// operators single out failures that deserve special dispatcher handling, e.g. a Houdini /// license shortage that should be retried differently without the render wrapper needing to /// translate the error into an exit code itself. -#[derive(Debug, Serialize, Deserialize, Clone)] +#[derive(Debug, Serialize, Deserialize, Clone, PartialEq, Eq)] pub struct LogExitStatusRule { /// Human-readable identifier, used only in log messages (e.g. "HOUDINI_LICENSE_ERROR"). #[serde(default)] @@ -237,6 +304,16 @@ pub struct CompiledExitStatusRule { pub exit_status: i32, } +/// A compiled snapshot of the exit-status scanning knobs: the rules together with the scan +/// depth they were configured with. Handed out as one `Arc` so a frame completing mid-reload +/// sees a consistent pair instead of new rules with an old scan depth. +#[derive(Debug)] +pub struct ExitStatusRuleSet { + /// Number of trailing log lines to scan (`log_scan_last_lines`); 0 disables scanning. + pub scan_last_lines: usize, + pub rules: Vec, +} + /// Compiles the configured rules, skipping (with a warning) any whose regex is invalid so a /// single bad pattern can neither disable the whole feature nor fail frame completion. pub(crate) fn compile_exit_status_rules( @@ -288,11 +365,22 @@ pub struct RunnerConfig { /// Ordered list of regex→exit-status rules applied to failed frames' logs. The first /// matching rule wins. Empty by default, which disables the feature. pub log_exit_status_rules: Vec, - /// Compiled form of `log_exit_status_rules`, populated lazily on first access and cached - /// so the regexes are compiled once (at startup via [`RunnerConfig::compiled_exit_status_rules`]) - /// rather than on every failed frame. Not part of the serialized config. + /// How often the watcher re-reads the config file to pick up changes to + /// `log_exit_status_rules`/`log_scan_last_lines` without a restart (restarting RQD kills + /// running frames on Linux, where recover mode is not available). Set to 0 to disable + /// live reloading. Only these two keys are live-reloaded; every other config change still + /// requires a restart. + #[serde(with = "humantime_serde")] + pub log_exit_status_rules_reload_interval: Duration, + /// Compiled form of `log_exit_status_rules`, seeded lazily on first access (forced at + /// startup, see `async_main`) and replaced live by the config watcher on reload. + /// + /// The outer `Arc` is shared by every clone of this config — each `RunningFrame` holds a + /// clone frozen at frame creation, so this cell is what lets a reload reach frames that + /// were already running. Not part of the serialized config: a frame recovered from a + /// snapshot gets the live cell back when its config is replaced in `from_snapshot`. #[serde(skip)] - compiled_exit_status_rules: OnceLock>>, + compiled_exit_status_rules: Arc>>>, } #[derive(Debug, Serialize, Deserialize, Clone)] @@ -342,21 +430,58 @@ impl Default for RunnerConfig { docker_images: HashMap::new(), log_scan_last_lines: 50, log_exit_status_rules: Vec::new(), - compiled_exit_status_rules: OnceLock::new(), + log_exit_status_rules_reload_interval: Duration::from_secs(300), // 5 min + compiled_exit_status_rules: Arc::new(RwLock::new(None)), } } } impl RunnerConfig { - /// Returns the compiled `log_exit_status_rules`, compiling and caching them on first call. + /// Returns the current compiled `log_exit_status_rules`, seeding the shared cell from this + /// config's raw fields on first call. /// - /// Because compilation happens once (forced at startup, see `async_main`), the warning for - /// an invalid pattern is emitted a single time rather than repeating on every failed frame, - /// and no frame pays the cost of recompiling every regex when it fails. - pub fn compiled_exit_status_rules(&self) -> &[CompiledExitStatusRule] { - self.compiled_exit_status_rules - .get_or_init(|| Arc::new(compile_exit_status_rules(&self.log_exit_status_rules))) - .as_slice() + /// Because compilation happens once per rule set (forced at startup, see `async_main`, and + /// again only when the watcher applies a reload), the warning for an invalid pattern is + /// emitted once per load rather than repeating on every failed frame, and no frame pays the + /// cost of recompiling every regex when it fails. + pub fn compiled_exit_status_rules(&self) -> Arc { + let guard = self + .compiled_exit_status_rules + .read() + .unwrap_or_else(|err| err.into_inner()); + if let Some(rule_set) = guard.as_ref() { + return Arc::clone(rule_set); + } + drop(guard); + + let mut guard = self + .compiled_exit_status_rules + .write() + .unwrap_or_else(|err| err.into_inner()); + // Another thread may have seeded the cell between the read and write locks. + if let Some(rule_set) = guard.as_ref() { + return Arc::clone(rule_set); + } + let rule_set = Arc::new(ExitStatusRuleSet { + scan_last_lines: self.log_scan_last_lines, + rules: compile_exit_status_rules(&self.log_exit_status_rules), + }); + *guard = Some(Arc::clone(&rule_set)); + rule_set + } + + /// Compiles `rules` and swaps them into the shared cell, making them the set every frame — + /// including frames launched before this call — scans on failure from now on. Rules with + /// invalid regexes are skipped with a warning, same as at startup. + pub fn reload_exit_status_rules(&self, scan_last_lines: usize, rules: &[LogExitStatusRule]) { + let rule_set = Arc::new(ExitStatusRuleSet { + scan_last_lines, + rules: compile_exit_status_rules(rules), + }); + *self + .compiled_exit_status_rules + .write() + .unwrap_or_else(|err| err.into_inner()) = Some(rule_set); } } @@ -394,18 +519,20 @@ pub struct Config { } impl Config { - // load the current config from the system config and environment variables - fn load() -> Result { - let mut required = false; - let config_file = match env::var("OPENCUE_RQD_CONFIG") { - Ok(v) => { - required = true; - v - } - Err(_) => DEFAULT_CONFIG_FILE.to_string(), - }; + /// Returns the config file path and whether its presence is required (it is when the + /// operator pointed at it explicitly via `OPENCUE_RQD_CONFIG`). + fn config_file_source() -> (String, bool) { + match env::var("OPENCUE_RQD_CONFIG") { + Ok(v) => (v, true), + Err(_) => (DEFAULT_CONFIG_FILE.to_string(), false), + } + } - println!(" INFO Config::load: using config file: {:?}", config_file); + /// Reads and deserializes the config from its sources (config file + `OPENRQD` environment + /// variables) without performing any filesystem setup. Used both by the initial [`load`] + /// and by the watcher re-reading the file at runtime. + fn read_sources() -> Result { + let (config_file, required) = Self::config_file_source(); let config = ConfigBase::builder() .add_source(File::with_name(&config_file).required(required)) @@ -422,12 +549,20 @@ impl Config { )) })?; - let deserialized_config = Config::deserialize(config).map_err(|err| { + Config::deserialize(config).map_err(|err| { RqdConfigError::LoadConfigError(format!( "{:?} config could not be deserialized. {}", &config_file, err )) - })?; + }) + } + + // load the current config from the system config and environment variables + fn load() -> Result { + let (config_file, _) = Self::config_file_source(); + println!(" INFO Config::load: using config file: {:?}", config_file); + + let deserialized_config = Self::read_sources()?; Self::setup(&deserialized_config)?; @@ -503,3 +638,68 @@ impl Config { Ok(()) } } + +/// Periodically re-reads the config sources and applies changes to `log_exit_status_rules` and +/// `log_scan_last_lines` to the live rule set, so operators can register new license-error +/// patterns without restarting RQD. +/// +/// Only those two keys are live-reloaded; changes to anything else in the file are ignored +/// until the next restart. A file that is missing, unreadable, or fails to parse leaves the +/// current rules untouched (with a warning), so a half-written edit can never wipe the rules +/// out from under running frames. +/// +/// Runs forever; spawn it as a background task. Returns immediately when +/// `log_exit_status_rules_reload_interval` is 0. +pub async fn watch_exit_status_rules() { + let interval = CONFIG.runner.log_exit_status_rules_reload_interval; + if interval.is_zero() { + info!("log_exit_status_rules live reload is disabled (reload interval = 0)"); + return; + } + + let mut last_applied = ( + CONFIG.runner.log_scan_last_lines, + CONFIG.runner.log_exit_status_rules.clone(), + ); + let mut ticker = tokio::time::interval(interval); + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + // The first tick of a tokio interval fires immediately; skip it, startup already + // compiled the initial rules. + ticker.tick().await; + + loop { + ticker.tick().await; + + let new_config = match Config::read_sources() { + Ok(config) => config, + Err(err) => { + warn!("Skipping log_exit_status_rules reload, config re-read failed: {err}"); + continue; + } + }; + + let candidate = ( + new_config.runner.log_scan_last_lines, + new_config.runner.log_exit_status_rules, + ); + if candidate == last_applied { + continue; + } + + CONFIG + .runner + .reload_exit_status_rules(candidate.0, &candidate.1); + info!( + "Reloaded log_exit_status_rules: {} rule(s) [{}], log_scan_last_lines={}", + candidate.1.len(), + candidate + .1 + .iter() + .map(|rule| rule.name.as_str()) + .collect::>() + .join(", "), + candidate.0, + ); + last_applied = candidate; + } +} diff --git a/rust/crates/rqd/src/frame/running_frame.rs b/rust/crates/rqd/src/frame/running_frame.rs index b7b3a1fa9..8475269c0 100644 --- a/rust/crates/rqd/src/frame/running_frame.rs +++ b/rust/crates/rqd/src/frame/running_frame.rs @@ -630,7 +630,11 @@ impl RunningFrame { if exit_code == 0 { return None; } - if self.config.log_exit_status_rules.is_empty() || self.config.log_scan_last_lines == 0 { + // Read the rules through the live cell rather than this frame's frozen raw config, so + // rules added by a config reload apply to frames that were already running when the + // reload happened. + let rule_set = self.config.compiled_exit_status_rules(); + if rule_set.rules.is_empty() || rule_set.scan_last_lines == 0 { return None; } // If RQD itself killed this frame (OOM, NIMBY, timeout, manual kill), that reason is @@ -643,12 +647,8 @@ impl RunningFrame { if !self.request.loki_url.is_empty() { return None; } - let compiled = self.config.compiled_exit_status_rules(); - if compiled.is_empty() { - return None; - } - let lines = match read_last_lines(&self.log_path, self.config.log_scan_last_lines).await { + let lines = match read_last_lines(&self.log_path, rule_set.scan_last_lines).await { Ok(lines) => lines, Err(err) => { warn!( @@ -660,7 +660,7 @@ impl RunningFrame { }; let log_tail = lines.join("\n"); - match match_exit_status_rules(&log_tail, compiled) { + match match_exit_status_rules(&log_tail, &rule_set.rules) { Some((name, exit_status)) => { info!( "Frame {}: log matched rule '{}'; overriding exit status {} -> {}", @@ -2111,6 +2111,39 @@ mod tests { let _ = std::fs::remove_file(&frame.log_path); } + #[tokio::test] + async fn test_scan_uses_rules_added_after_frame_creation() { + // The live-reload guarantee: a frame launched with NO rules configured must pick up + // rules swapped in later (as the config watcher does on file change), because every + // RunnerConfig clone shares the compiled-rules cell. + let mut config_handle: Option = None; + let mut frame = create_running_frame_cfg("false", 1, 1, HashMap::new(), "", |cfg| { + config_handle = Some(cfg.clone()); + }); + let log_file = std::env::temp_dir().join(format!("rqd_scan_test_{}.rqlog", Uuid::new_v4())); + std::fs::write(&log_file, LICENSE_LOG).unwrap(); + frame.log_path = log_file.to_string_lossy().to_string(); + + // No rules yet: the failure keeps its exit status (this also seeds the shared cell, + // proving a reload replaces an already-seeded set). + assert_eq!(frame.scan_log_for_exit_status_override(3).await, None); + + config_handle.unwrap().reload_exit_status_rules( + 50, + &[rule( + "HOUDINI_LICENSE_ERROR", + "A usable license to run the application is installed but they are all in use", + 330, + )], + ); + + assert_eq!( + frame.scan_log_for_exit_status_override(3).await, + Some(("HOUDINI_LICENSE_ERROR".to_string(), 330)) + ); + let _ = std::fs::remove_file(&frame.log_path); + } + #[tokio::test] async fn test_scan_skips_killed_frames() { // A frame RQD killed keeps its kill-driven classification, even if the log matches. diff --git a/rust/crates/rqd/src/main.rs b/rust/crates/rqd/src/main.rs index 8a821b625..d8528a983 100644 --- a/rust/crates/rqd/src/main.rs +++ b/rust/crates/rqd/src/main.rs @@ -69,6 +69,11 @@ async fn async_main() -> miette::Result<()> { // frame (and no frame later pays to recompile them). let _ = CONFIG.runner.compiled_exit_status_rules(); + // Keep the exit-status rules editable without a restart (which would kill running frames + // on Linux): a watcher re-reads the config file periodically and swaps changed rules into + // the live set that all frames — including already-running ones — scan against. + tokio::spawn(config::watch_exit_status_rules()); + // Fail fast if the config requires elevated privileges the process does not hold, // instead of letting every frame fail later with an opaque error. capabilities::preflight(&CONFIG.runner)?; diff --git a/rust/crates/scheduler/src/dao/frame_dao.rs b/rust/crates/scheduler/src/dao/frame_dao.rs index 79eb7a5aa..6a9ee6e8b 100644 --- a/rust/crates/scheduler/src/dao/frame_dao.rs +++ b/rust/crates/scheduler/src/dao/frame_dao.rs @@ -171,6 +171,13 @@ UPDATE frame SET WHERE pk_frame = $6 AND str_state = 'WAITING' AND int_version = $7 + -- Deferred layer booking: refuse to start a frame whose layer's start-after + -- gate is in the future. Authoritative counterpart of the advisory predicate + -- in the pending-frames query (layer_dao.rs). + AND EXISTS ( + SELECT 1 FROM layer + WHERE layer.pk_layer = frame.pk_layer + AND (layer.ts_start_after IS NULL OR layer.ts_start_after <= current_timestamp)) RETURNING int_version "#; diff --git a/rust/crates/scheduler/src/dao/layer_dao.rs b/rust/crates/scheduler/src/dao/layer_dao.rs index 4e9989df6..dac7c8589 100644 --- a/rust/crates/scheduler/src/dao/layer_dao.rs +++ b/rust/crates/scheduler/src/dao/layer_dao.rs @@ -219,6 +219,8 @@ WITH dispatch_frames AS ( AND ls.int_waiting_count > 0 AND string_to_array(REPLACE($2, ' ', ''), '|') && string_to_array(REPLACE(l.str_tags, ' ', ''), '|') AND f.str_state = 'WAITING' + -- Deferred layer booking: skip layers whose start-after gate is in the future + AND (l.ts_start_after IS NULL OR l.ts_start_after <= current_timestamp) ), limited_frames AS ( SELECT * FROM dispatch_frames diff --git a/rust/crates/scheduler/tests/layer_start_after.rs b/rust/crates/scheduler/tests/layer_start_after.rs new file mode 100644 index 000000000..fb7e085e5 --- /dev/null +++ b/rust/crates/scheduler/tests/layer_start_after.rs @@ -0,0 +1,399 @@ +// Copyright Contributors to the OpenCue Project +// +// 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. + +//! Standalone integration test for the layer start-after gate (`layer.ts_start_after`). +//! +//! The reservation update (`UPDATE_FRAME_STARTED`) is the authoritative gate: it must refuse +//! to start a frame whose layer's start-after time is in the future, and allow it once the +//! gate is NULL or in the past. Self-contained on purpose - it seeds its own minimal +//! facility -> ... -> frame chain and does NOT depend on the cluster/Tag fixtures in +//! `smoke_tests.rs`. Gated behind the `smoke-tests` feature; requires the repo-root +//! `docker compose up -d flyway` Postgres (host `localhost:5432`, db/user `cuebot`). +//! +//! Run with: `cargo test -p scheduler --features smoke-tests --test layer_start_after`. + +#[cfg(feature = "smoke-tests")] +mod util; + +#[cfg(feature = "smoke-tests")] +mod layer_start_after { + use std::collections::HashMap; + use std::sync::Arc; + use std::time::SystemTime; + + use bytesize::ByteSize; + use scheduler::dao::{FrameDao, FrameDaoError, LayerDao}; + use scheduler::models::{ + CoreSize, CoreSizeWithMultiplier, DispatchFrame, DispatchLayer, VirtualProc, + }; + use serial_test::serial; + use sqlx::{Pool, Postgres}; + use uuid::Uuid; + + use crate::util::test_connection_pool; + + /// Seeds a minimal facility -> dept -> show -> folder -> job -> layer -> frame chain with + /// a single WAITING frame. Column lists and values mirror the `frame_dao_compensation` + /// harness. Returns the frame id, layer id, job id and the unique row-name prefix used for + /// cleanup. + async fn seed_one_waiting_frame( + pool: &Pool, + ) -> Result<(Uuid, Uuid, Uuid, String), sqlx::Error> { + let suffix = Uuid::new_v4().to_string()[..8].to_string(); + let prefix = format!("integ_test_startafter_{}", suffix); + + let facility_id = Uuid::new_v4(); + let dept_id = Uuid::new_v4(); + let show_id = Uuid::new_v4(); + let folder_id = Uuid::new_v4(); + let job_id = Uuid::new_v4(); + let layer_id = Uuid::new_v4(); + let frame_id = Uuid::new_v4(); + + let mut tx = pool.begin().await?; + + sqlx::query("INSERT INTO facility (pk_facility, str_name) VALUES ($1, $2)") + .bind(facility_id.to_string()) + .bind(format!("{}_facility", prefix)) + .execute(&mut *tx) + .await?; + + sqlx::query("INSERT INTO dept (pk_dept, str_name) VALUES ($1, $2)") + .bind(dept_id.to_string()) + .bind(format!("{}_dept", prefix)) + .execute(&mut *tx) + .await?; + + sqlx::query("INSERT INTO show (pk_show, str_name) VALUES ($1, $2)") + .bind(show_id.to_string()) + .bind(format!("{}_show", prefix)) + .execute(&mut *tx) + .await?; + + sqlx::query( + "INSERT INTO folder (pk_folder, pk_show, pk_dept, str_name) VALUES ($1, $2, $3, $4)", + ) + .bind(folder_id.to_string()) + .bind(show_id.to_string()) + .bind(dept_id.to_string()) + .bind(format!("{}_folder", prefix)) + .execute(&mut *tx) + .await?; + + sqlx::query( + "INSERT INTO job (pk_job, pk_folder, pk_show, pk_facility, pk_dept, str_name, \ + str_visible_name, str_shot, str_user, str_state) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)", + ) + .bind(job_id.to_string()) + .bind(folder_id.to_string()) + .bind(show_id.to_string()) + .bind(facility_id.to_string()) + .bind(dept_id.to_string()) + .bind(format!("{}_job", prefix)) + .bind(format!("{}_job", prefix)) + .bind(format!("{}_shot", prefix)) + .bind(format!("{}_user", prefix)) + .bind("PENDING") + .execute(&mut *tx) + .await?; + + sqlx::query( + "INSERT INTO layer (pk_layer, pk_job, str_name, str_cmd, str_range, str_tags, \ + str_type, int_cores_min, int_mem_min, int_gpus_min, int_gpu_mem_min) \ + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)", + ) + .bind(layer_id.to_string()) + .bind(job_id.to_string()) + .bind(format!("{}_layer", prefix)) + .bind("echo 'start-after test frame'") + .bind("1-1") + .bind("general") + .bind("PRE") + .bind(100_i64) + .bind(1_000_000_i64) + .bind(0_i64) + .bind(0_i64) + .execute(&mut *tx) + .await?; + + // The `after_insert_layer` trigger already created this row with zero counts, so the + // conflict branch must overwrite them: the pending-frames query gates on + // `layer_stat.int_waiting_count > 0` and would otherwise find nothing at all. + sqlx::query( + "INSERT INTO layer_stat (pk_layer_stat, pk_layer, pk_job, int_waiting_count, \ + int_total_count) VALUES ($1, $2, $3, $4, $5) \ + ON CONFLICT (pk_layer) DO UPDATE SET \ + int_waiting_count = EXCLUDED.int_waiting_count, \ + int_total_count = EXCLUDED.int_total_count", + ) + .bind(Uuid::new_v4().to_string()) + .bind(layer_id.to_string()) + .bind(job_id.to_string()) + .bind(1_i64) + .bind(1_i64) + .execute(&mut *tx) + .await?; + + sqlx::query( + "INSERT INTO frame (pk_frame, pk_layer, pk_job, str_name, str_state, int_number, \ + int_layer_order, int_dispatch_order) VALUES ($1, $2, $3, $4, $5, $6, $7, $8)", + ) + .bind(frame_id.to_string()) + .bind(layer_id.to_string()) + .bind(job_id.to_string()) + .bind(format!("{}_frame", prefix)) + .bind("WAITING") + .bind(1_i32) + .bind(1_i32) + .bind(1_i32) + .execute(&mut *tx) + .await?; + + tx.commit().await?; + Ok((frame_id, layer_id, job_id, prefix)) + } + + /// Total number of frames the pending-frames query surfaced across all layers. + fn count_dispatch_frames(layers: &[DispatchLayer]) -> usize { + layers.iter().map(|l| l.frames.len()).sum() + } + + /// Best-effort teardown of everything seeded under `prefix`, triggers disabled so stat + /// bookkeeping doesn't interfere with the deletes. + async fn cleanup(pool: &Pool, prefix: &str) { + let like = format!("{}%", prefix); + let mut tx = match pool.begin().await { + Ok(tx) => tx, + Err(_) => return, + }; + let _ = sqlx::query("SET session_replication_role = 'replica'") + .execute(&mut *tx) + .await; + for stmt in [ + "DELETE FROM frame WHERE str_name LIKE $1", + "DELETE FROM layer_stat WHERE pk_layer IN (SELECT pk_layer FROM layer WHERE str_name LIKE $1)", + "DELETE FROM layer_resource WHERE pk_layer IN (SELECT pk_layer FROM layer WHERE str_name LIKE $1)", + "DELETE FROM layer WHERE str_name LIKE $1", + "DELETE FROM job_stat WHERE pk_job IN (SELECT pk_job FROM job WHERE str_name LIKE $1)", + "DELETE FROM job_resource WHERE pk_job IN (SELECT pk_job FROM job WHERE str_name LIKE $1)", + "DELETE FROM job WHERE str_name LIKE $1", + "DELETE FROM folder WHERE str_name LIKE $1", + "DELETE FROM show WHERE str_name LIKE $1", + "DELETE FROM facility WHERE str_name LIKE $1", + "DELETE FROM dept WHERE str_name LIKE $1", + ] { + let _ = sqlx::query(stmt).bind(&like).execute(&mut *tx).await; + } + let _ = sqlx::query("SET session_replication_role = 'origin'") + .execute(&mut *tx) + .await; + let _ = tx.commit().await; + } + + /// Builds the minimal VirtualProc update_frame_started needs (host name, reserved + /// resources, and the frame id/version driving the optimistic-lock guard). + fn virtual_proc( + frame_id: Uuid, + layer_id: Uuid, + frame_version: u32, + prefix: &str, + ) -> VirtualProc { + VirtualProc { + proc_id: Uuid::new_v4(), + host_id: Uuid::new_v4(), + show_id: Uuid::new_v4(), + folder_id: Uuid::new_v4(), + dept_id: Uuid::new_v4(), + layer_id, + job_id: Uuid::new_v4(), + frame_id, + alloc_id: Uuid::new_v4(), + host_name: format!("{}_host", prefix), + cores_reserved: CoreSizeWithMultiplier(100), + memory_reserved: ByteSize::gb(1), + gpus_reserved: 0, + gpu_memory_reserved: ByteSize::b(0), + os: "linux".to_string(), + is_local_dispatch: false, + frame: DispatchFrame { + id: frame_id, + frame_name: format!("{}_frame", prefix), + show_id: Uuid::new_v4(), + facility_id: Uuid::new_v4().to_string(), + job_id: Uuid::new_v4(), + layer_id, + command: "echo 'start-after test frame'".to_string(), + range: "1-1".to_string(), + chunk_size: 1, + show_name: format!("{}_show", prefix), + shot: format!("{}_shot", prefix), + user: format!("{}_user", prefix), + uid: None, + log_dir: "/tmp".to_string(), + layer_name: format!("{}_layer", prefix), + job_name: format!("{}_job", prefix), + min_cores: CoreSize(1), + layer_cores_limit: None, + threadable: false, + has_selfish_service: false, + min_gpus: 0, + min_gpu_memory: ByteSize::b(0), + min_memory: ByteSize::gb(1), + services: None, + os: Some("linux".to_string()), + loki_url: None, + version: frame_version, + updated_at: SystemTime::now(), + env: HashMap::new(), + }, + } + } + + /// The reservation update must refuse a frame whose layer is delayed into the future and + /// accept it once the gate has passed (or was cleared). + #[tokio::test] + #[serial] + async fn update_frame_started_honours_layer_start_after() { + let pool: Arc> = test_connection_pool().await.expect("connection pool"); + let (frame_id, layer_id, _job_id, prefix) = + seed_one_waiting_frame(&pool).await.expect("seed frame"); + + let version: i32 = sqlx::query_scalar("SELECT int_version FROM frame WHERE pk_frame = $1") + .bind(frame_id.to_string()) + .fetch_one(&*pool) + .await + .expect("frame version"); + + let frame_dao = FrameDao::new().await.expect("frame dao"); + let proc = virtual_proc(frame_id, layer_id, version as u32, &prefix); + + // Delay the layer: the reservation must be refused and the frame left WAITING. + sqlx::query( + "UPDATE layer SET ts_start_after = current_timestamp + interval '5 minutes', \ + str_start_after_reason = 'Automatic backoff: exit status 330' WHERE pk_layer = $1", + ) + .bind(layer_id.to_string()) + .execute(&*pool) + .await + .expect("delay layer"); + + let mut tx = pool.begin().await.expect("tx"); + let refused = frame_dao.update_frame_started(&mut tx, &proc).await; + tx.commit().await.expect("commit"); + assert!( + matches!(refused, Err(FrameDaoError::FrameCouldNotBeUpdated)), + "update_frame_started must refuse a frame on a delayed layer, got {:?}", + refused.map(|_| ()), + ); + let state: String = sqlx::query_scalar("SELECT str_state FROM frame WHERE pk_frame = $1") + .bind(frame_id.to_string()) + .fetch_one(&*pool) + .await + .expect("frame state"); + assert_eq!(state, "WAITING", "refused frame must stay WAITING"); + + // Move the gate into the past: the same reservation now succeeds. + sqlx::query( + "UPDATE layer SET ts_start_after = current_timestamp - interval '1 minute' \ + WHERE pk_layer = $1", + ) + .bind(layer_id.to_string()) + .execute(&*pool) + .await + .expect("expire layer delay"); + + let mut tx = pool.begin().await.expect("tx"); + let started = frame_dao + .update_frame_started(&mut tx, &proc) + .await + .expect("update_frame_started after the gate passed"); + tx.commit().await.expect("commit"); + assert_eq!(started, (version + 1) as u32, "version bumps on start"); + let state: String = sqlx::query_scalar("SELECT str_state FROM frame WHERE pk_frame = $1") + .bind(frame_id.to_string()) + .fetch_one(&*pool) + .await + .expect("frame state"); + assert_eq!(state, "RUNNING", "frame must start once the gate is past"); + + cleanup(&pool, &prefix).await; + } + + /// The pending-frames query (`QUERY_LAYERS_WITH_FRAMES` in `layer_dao.rs`) is the advisory + /// half of the gate: a layer delayed into the future must surface no frames, and must + /// surface them again once the gate has passed. Cheap to get wrong silently, since the + /// authoritative reservation gate would still hold the line. + #[tokio::test] + #[serial] + async fn pending_frames_query_skips_delayed_layer() { + let pool: Arc> = test_connection_pool().await.expect("connection pool"); + let (_frame_id, layer_id, job_id, prefix) = + seed_one_waiting_frame(&pool).await.expect("seed frame"); + + let layer_dao = LayerDao::new().await.expect("layer dao"); + // Matches `str_tags` on the seeded layer; the query intersects this list with the + // layer's tags. + let tags = vec!["general".to_string()]; + + // Bookable while the gate is unset. + let layers = layer_dao + .query_layers(job_id, tags.clone()) + .await + .expect("query_layers failed"); + assert!( + count_dispatch_frames(&layers) > 0, + "Expected frames while the layer has no start-after gate" + ); + + // Delay the layer: the query must return nothing for it. + sqlx::query( + "UPDATE layer SET ts_start_after = current_timestamp + interval '5 minutes', \ + str_start_after_reason = 'Automatic backoff: exit status 330' WHERE pk_layer = $1", + ) + .bind(layer_id.to_string()) + .execute(&*pool) + .await + .expect("delay layer"); + + let layers = layer_dao + .query_layers(job_id, tags.clone()) + .await + .expect("query_layers failed"); + assert_eq!( + count_dispatch_frames(&layers), + 0, + "Expected no frames while the layer's start-after gate is in the future" + ); + + // Expire the gate: frames surface again. + sqlx::query( + "UPDATE layer SET ts_start_after = current_timestamp - interval '1 minute' \ + WHERE pk_layer = $1", + ) + .bind(layer_id.to_string()) + .execute(&*pool) + .await + .expect("expire layer delay"); + + let layers = layer_dao + .query_layers(job_id, tags) + .await + .expect("query_layers failed"); + assert!( + count_dispatch_frames(&layers) > 0, + "Expected frames again once the start-after gate is in the past" + ); + + cleanup(&pool, &prefix).await; + } +}