[cuebot/cuegui/pycue/rqd] Allow delaying a layer start

VERSION.in

@@ -1 +1 @@
-1.29
+1.30

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<String> tags = new LinkedHashSet<String>();
     public Set<String> services = new LinkedHashSet<String>();
     public Set<String> limits = new LinkedHashSet<String>();

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;
+    }
 }

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.

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 "

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<Integer> exclusions = new ArrayList<Integer>(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());
         }