[cuebot] Reject job specs referencing undefined limits

cuebot/src/main/java/com/imageworks/spcue/dao/LimitDao.java

@@ -15,6 +15,9 @@
 
 package com.imageworks.spcue.dao;
 
+import java.util.Collection;
+import java.util.List;
+
 import com.imageworks.spcue.LimitEntity;
 import com.imageworks.spcue.LimitInterface;
 
@@ -44,6 +47,15 @@ public interface LimitDao {
      */
     public LimitEntity findLimit(String name);
 
+    /**
+     * Returns the subset of the given names that have no matching limit record. Names are returned
+     * in the order they were given, without duplicates.
+     *
+     * @param names
+     * @return names that do not exist
+     */
+    public List<String> findMissingLimitNames(Collection<String> names);
+
     /**
      * Gets a limit by Id
      *

cuebot/src/main/java/com/imageworks/spcue/dao/postgres/LimitDaoJdbc.java

@@ -17,6 +17,13 @@
 
 import java.sql.ResultSet;
 import java.sql.SQLException;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
 
 import org.springframework.jdbc.core.RowMapper;
 import org.springframework.jdbc.core.support.JdbcDaoSupport;
@@ -62,6 +69,23 @@ public LimitEntity findLimit(String name) {
         return getJdbcTemplate().queryForObject(findLimitQuery, LIMIT_MAPPER, name);
     }
 
+    @Override
+    public List<String> findMissingLimitNames(Collection<String> names) {
+        Set<String> uniqueNames = new LinkedHashSet<String>(names);
+        if (uniqueNames.isEmpty()) {
+            return Collections.emptyList();
+        }
+
+        String placeholders =
+                uniqueNames.stream().map(name -> "?").collect(Collectors.joining(","));
+        Set<String> existing = new HashSet<String>(getJdbcTemplate().queryForList(
+                "SELECT str_name FROM limit_record WHERE str_name IN (" + placeholders + ")",
+                String.class, uniqueNames.toArray()));
+
+        return uniqueNames.stream().filter(name -> !existing.contains(name))
+                .collect(Collectors.toList());
+    }
+
     @Override
     public LimitEntity getLimit(String id) {
         String getLimitQuery = GET_LIMIT_BASE + "WHERE " + "limit_record.pk_limit_record=? "

cuebot/src/main/java/com/imageworks/spcue/service/AdminManager.java

@@ -15,6 +15,9 @@
 
 package com.imageworks.spcue.service;
 
+import java.util.Collection;
+import java.util.List;
+
 import com.imageworks.spcue.AllocationEntity;
 import com.imageworks.spcue.AllocationInterface;
 import com.imageworks.spcue.DepartmentInterface;
@@ -116,6 +119,14 @@ SubscriptionInterface createSubscription(ShowInterface show, AllocationInterface
 
     LimitInterface findLimit(String name);
 
+    /**
+     * Returns the subset of the given limit names that have not been created yet.
+     *
+     * @param names
+     * @return names that do not exist
+     */
+    List<String> findMissingLimitNames(Collection<String> names);
+
     LimitInterface getLimit(String id);
 
     void setLimitName(LimitInterface limit, String name);

cuebot/src/main/java/com/imageworks/spcue/service/AdminManagerService.java

@@ -15,6 +15,9 @@
 
 package com.imageworks.spcue.service;
 
+import java.util.Collection;
+import java.util.List;
+
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.LogManager;
 import org.springframework.transaction.annotation.Propagation;
@@ -260,6 +263,11 @@ public LimitInterface findLimit(String name) {
         return limitDao.findLimit(name);
     }
 
+    @Override
+    public List<String> findMissingLimitNames(Collection<String> names) {
+        return limitDao.findMissingLimitNames(names);
+    }
+
     @Override
     public LimitInterface getLimit(String id) {
         return limitDao.getLimit(id);

cuebot/src/main/java/com/imageworks/spcue/service/JobLauncher.java

@@ -17,7 +17,12 @@
 
 import java.io.File;
 import java.util.HashSet;
+import java.util.LinkedHashMap;
+import java.util.LinkedHashSet;
+import java.util.List;
+import java.util.Map;
 import java.util.Set;
+import java.util.stream.Collectors;
 
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.LogManager;
@@ -29,6 +34,7 @@
 import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
 
 import com.imageworks.spcue.BuildableJob;
+import com.imageworks.spcue.BuildableLayer;
 import com.imageworks.spcue.EntityCreationError;
 import com.imageworks.spcue.JobDetail;
 import com.imageworks.spcue.LocalHostAssignment;
@@ -163,6 +169,51 @@ public void verifyJobSpec(JobSpec spec) {
                     + " does not exist. Please contact "
                     + "administrator of your OpenCue deployment to have this show " + "created.");
         }
+
+        verifyLimits(spec);
+    }
+
+    /**
+     * Fails the launch if the spec references limits that have not been created yet, naming every
+     * missing limit and the layers that use it.
+     */
+    private void verifyLimits(JobSpec spec) {
+
+        Map<String, Set<String>> layersByLimit = new LinkedHashMap<String, Set<String>>();
+
+        for (BuildableJob job : spec.getJobs()) {
+            collectLimits(job, layersByLimit);
+            if (job.getPostJob() != null) {
+                collectLimits(job.getPostJob(), layersByLimit);
+            }
+        }
+
+        if (layersByLimit.isEmpty()) {
+            return;
+        }
+
+        List<String> missingLimits = adminManager.findMissingLimitNames(layersByLimit.keySet());
+        if (missingLimits.isEmpty()) {
+            return;
+        }
+
+        String detail = missingLimits.stream()
+                .map(limitName -> limitName + " (used by "
+                        + String.join(", ", layersByLimit.get(limitName)) + ")")
+                .collect(Collectors.joining("; "));
+
+        throw new EntityCreationError("The following limits do not exist: " + detail
+                + ". Please contact administrator of your OpenCue deployment to have these limits"
+                + " created, or remove them from the job spec.");
+    }
+
+    private void collectLimits(BuildableJob job, Map<String, Set<String>> layersByLimit) {
+        for (BuildableLayer buildableLayer : job.getBuildableLayers()) {
+            for (String limitName : buildableLayer.layerDetail.limits) {
+                layersByLimit.computeIfAbsent(limitName, name -> new LinkedHashSet<String>())
+                        .add(job.detail.name + "/" + buildableLayer.layerDetail.name);
+            }
+        }
     }
 
     public void queueAndLaunch(final JobSpec spec) {

cuebot/src/main/java/com/imageworks/spcue/service/JobManagerService.java

@@ -200,6 +200,24 @@ public void launchJobSpec(JobSpec spec) {
         }
     }
 
+    /**
+     * Links a layer to the limits it references, failing the launch if any of them do not exist.
+     */
+    private void addLayerLimits(LayerDetail layer) {
+        if (layer.limits.isEmpty()) {
+            return;
+        }
+
+        List<String> missingLimits = limitDao.findMissingLimitNames(layer.limits);
+        if (!missingLimits.isEmpty()) {
+            throw new JobLaunchException("layer " + layer.name
+                    + " references limits that do not exist: " + String.join(", ", missingLimits)
+                    + ". These limits have to be created before the job can be launched.");
+        }
+
+        layer.limits.forEach(ln -> addLayerLimit(layer, limitDao.findLimit(ln).getLimitId()));
+    }
+
     @Transactional(propagation = Propagation.REQUIRED)
     public JobDetail createJob(BuildableJob buildableJob) {
 
@@ -277,8 +295,7 @@ public JobDetail createJob(BuildableJob buildableJob) {
                 logger.info("creating layer " + layer.name + " range: " + layer.range);
                 layerDao.insertLayerDetail(layer);
                 layerDao.insertLayerEnvironment(layer, buildableLayer.env);
-                layer.limits.stream()
-                        .forEach(ln -> addLayerLimit(layer, limitDao.findLimit(ln).getLimitId()));
+                addLayerLimits(layer);
                 layer.outputs.stream().forEach(ln -> registerLayerOutput(layer, ln));
                 frameDao.insertFrames(layer, frames);
             }

cuebot/src/test/java/com/imageworks/spcue/test/dao/postgres/LimitDaoTests.java

@@ -15,6 +15,9 @@
 
 package com.imageworks.spcue.test.dao.postgres;
 
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
 import javax.annotation.Resource;
 
 import org.junit.Rule;
@@ -89,6 +92,22 @@ public void testFindLimit() {
         assertEquals(limit.maxValue, LIMIT_MAX_VALUE);
     }
 
+    @Test
+    @Transactional
+    @Rollback(true)
+    public void testFindMissingLimitNames() {
+        limitDao.createLimit(LIMIT_NAME, LIMIT_MAX_VALUE);
+
+        assertEquals(Collections.emptyList(),
+                limitDao.findMissingLimitNames(Collections.emptyList()));
+        assertEquals(Collections.emptyList(),
+                limitDao.findMissingLimitNames(Arrays.asList(LIMIT_NAME)));
+
+        List<String> missing =
+                limitDao.findMissingLimitNames(Arrays.asList("nope", LIMIT_NAME, "nada", "nope"));
+        assertEquals(Arrays.asList("nope", "nada"), missing);
+    }
+
     @Test
     @Transactional
     @Rollback(true)

cuebot/src/test/java/com/imageworks/spcue/test/service/JobLauncherLimitTests.java

@@ -0,0 +1,137 @@
+/*
+ * 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.service;
+
+import java.io.File;
+import java.util.List;
+import javax.annotation.Resource;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.springframework.test.annotation.Rollback;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.AbstractTransactionalJUnit4SpringContextTests;
+import org.springframework.test.context.support.AnnotationConfigContextLoader;
+import org.springframework.transaction.annotation.Transactional;
+
+import com.imageworks.spcue.EntityCreationError;
+import com.imageworks.spcue.JobDetail;
+import com.imageworks.spcue.JobLaunchException;
+import com.imageworks.spcue.LayerInterface;
+import com.imageworks.spcue.config.TestAppConfig;
+import com.imageworks.spcue.dao.LayerDao;
+import com.imageworks.spcue.dao.LimitDao;
+import com.imageworks.spcue.service.JobLauncher;
+import com.imageworks.spcue.service.JobManager;
+import com.imageworks.spcue.service.JobSpec;
+
+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)
+public class JobLauncherLimitTests extends AbstractTransactionalJUnit4SpringContextTests {
+
+    @Resource
+    JobLauncher jobLauncher;
+
+    @Resource
+    JobManager jobManager;
+
+    @Resource
+    LayerDao layerDao;
+
+    @Resource
+    LimitDao limitDao;
+
+    private static final String JOB_NAME = "pipe-dev.cue-testuser_shell_v1";
+
+    private static final File SPEC = new File("src/test/resources/conf/jobspec/jobspec_limit.xml");
+
+    @Before
+    public void setTestMode() {
+        jobLauncher.testMode = true;
+    }
+
+    @Test
+    @Transactional
+    @Rollback(true)
+    public void testLaunchRejectsUndefinedLimits() {
+        try {
+            jobLauncher.launch(SPEC);
+            fail("expected the launch to be rejected");
+        } catch (EntityCreationError e) {
+            // Both undefined limits are reported, along with the layers using them.
+            assertTrue(e.getMessage(),
+                    e.getMessage().contains("util (used by " + JOB_NAME + "/pass_1_preprocess)"));
+            assertTrue(e.getMessage(),
+                    e.getMessage().contains("arnold (used by " + JOB_NAME + "/pass_1)"));
+        }
+
+        assertFalse("no job should have been created", jobManager.isJobPending(JOB_NAME));
+    }
+
+    @Test
+    @Transactional
+    @Rollback(true)
+    public void testLaunchReportsOnlyUndefinedLimits() {
+        limitDao.createLimit("util", 15);
+
+        try {
+            jobLauncher.launch(SPEC);
+            fail("expected the launch to be rejected");
+        } catch (EntityCreationError e) {
+            assertTrue(e.getMessage(), e.getMessage().contains("arnold"));
+            assertFalse(e.getMessage(), e.getMessage().contains("util (used by"));
+        }
+    }
+
+    @Test
+    @Transactional
+    @Rollback(true)
+    public void testLaunchSucceedsWhenLimitsExist() {
+        limitDao.createLimit("util", 15);
+        limitDao.createLimit("arnold", 20);
+
+        jobLauncher.launch(SPEC);
+
+        JobDetail job = jobManager.findJobDetail(JOB_NAME);
+        LayerInterface layer = layerDao.findLayer(job, "pass_1");
+        List<String> limits = layerDao.getLimitNames(layer);
+        assertEquals(1, limits.size());
+        assertEquals("arnold", limits.get(0));
+    }
+
+    /**
+     * Job creation has to fail closed on its own, since it is also reachable without going through
+     * the launcher's verification step.
+     */
+    @Test
+    @Transactional
+    @Rollback(true)
+    public void testCreateJobRejectsUndefinedLimits() {
+        JobSpec spec = jobLauncher.parse(SPEC);
+
+        try {
+            jobManager.launchJobSpec(spec);
+            fail("expected job creation to be rejected");
+        } catch (JobLaunchException e) {
+            assertTrue(e.getMessage(),
+                    e.getMessage().contains("references limits that do not exist: util"));
+        }
+    }
+}