[cuebot] Licensing limits

VERSION.in

@@ -1 +1 @@
-1.28
+1.29

cuebot/src/main/java/com/imageworks/common/spring/remoting/GrpcServer.java

@@ -28,6 +28,7 @@
 import com.imageworks.spcue.servant.ManageHost;
 import com.imageworks.spcue.servant.ManageJob;
 import com.imageworks.spcue.servant.ManageLayer;
+import com.imageworks.spcue.servant.ManageLicense;
 import com.imageworks.spcue.servant.ManageLimit;
 import com.imageworks.spcue.servant.ManageMatcher;
 import com.imageworks.spcue.servant.ManageOwner;
@@ -110,6 +111,7 @@ public void start() throws IOException {
                 .addService(applicationContext.getBean("manageHost", ManageHost.class))
                 .addService(applicationContext.getBean("manageJob", ManageJob.class))
                 .addService(applicationContext.getBean("manageLayer", ManageLayer.class))
+                .addService(applicationContext.getBean("manageLicense", ManageLicense.class))
                 .addService(applicationContext.getBean("manageLimit", ManageLimit.class))
                 .addService(applicationContext.getBean("manageMatcher", ManageMatcher.class))
                 .addService(applicationContext.getBean("manageOwner", ManageOwner.class))

cuebot/src/main/java/com/imageworks/spcue/config/AppConfig.java

@@ -28,6 +28,7 @@
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Conditional;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
 import org.springframework.context.annotation.ImportResource;
 import org.springframework.context.annotation.Primary;
 import org.springframework.context.annotation.PropertySource;
@@ -42,6 +43,7 @@
         "classpath:conf/spring/applicationContext-monitoring.xml",
         "classpath:conf/spring/applicationContext-accounting.xml"})
 @EnableConfigurationProperties
+@Import(LicenseConfig.class)
 @PropertySource({"classpath:opencue.properties"})
 public class AppConfig {
 

cuebot/src/main/java/com/imageworks/spcue/config/LicenseConfig.java

@@ -0,0 +1,52 @@
+
+/*
+ * 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.config;
+
+import javax.sql.DataSource;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.core.env.Environment;
+import org.springframework.jdbc.core.JdbcTemplate;
+
+import com.imageworks.spcue.dispatcher.LicenseBookingGate;
+import com.imageworks.spcue.dispatcher.LicenseSource;
+
+/**
+ * Live application licensing (CUE_LICENSES) beans, shared between the main and test contexts so the
+ * XML-defined dispatcher beans can reference them in both. Everything here is inert unless
+ * scheduler.license.provider is configured AND a layer declares licenses in its environment.
+ */
+@Configuration
+public class LicenseConfig {
+
+    /**
+     * Live view of floating application licenses. The poller is a no-op unless
+     * scheduler.license.provider is configured; a layer declaring licenses in its environment is
+     * what turns gating on.
+     */
+    @Bean(initMethod = "start", destroyMethod = "stop")
+    public LicenseSource licenseSource(Environment env, DataSource cueDataSource) {
+        return new LicenseSource(env, new JdbcTemplate(cueDataSource));
+    }
+
+    /** Applies the license budgets to the legacy booking path. */
+    @Bean
+    public LicenseBookingGate licenseBookingGate(LicenseSource licenseSource,
+            DataSource cueDataSource) {
+        return new LicenseBookingGate(new JdbcTemplate(cueDataSource), licenseSource);
+    }
+}

cuebot/src/main/java/com/imageworks/spcue/dispatcher/CoreUnitDispatcher.java

@@ -85,6 +85,12 @@ public class CoreUnitDispatcher implements Dispatcher {
 
     private HostManager hostManager;
 
+    /**
+     * Gates dispatch on live application-license budgets (CUE_LICENSES). Optional: when unset,
+     * dispatch behaves exactly as before.
+     */
+    private LicenseBookingGate licenseBookingGate;
+
     public boolean testMode = false;
 
     private final long MEM_RESERVED_MIN;
@@ -135,6 +141,10 @@ private Cache<String, String> getOrCreateJobLock() {
     private List<VirtualProc> dispatchJobs(DispatchHost host, Set<String> jobs) {
         List<VirtualProc> procs = new ArrayList<VirtualProc>();
 
+        // One license session for the whole booking pass: its budget snapshot is
+        // read at most once and its pass-local accounting spans every job below.
+        LicenseBookingGate.Session licenseSession =
+                licenseBookingGate == null ? null : licenseBookingGate.newSession(host.getName());
         try {
             for (String jobid : jobs) {
 
@@ -157,7 +167,7 @@ private List<VirtualProc> dispatchJobs(DispatchHost host, Set<String> jobs) {
 
                 DispatchJob job = jobManager.getDispatchJob(jobid);
                 try {
-                    procs.addAll(dispatchHost(host, job));
+                    procs.addAll(dispatchHost(host, job, licenseSession));
                 } catch (JobDispatchException e) {
                     logger.info("job dispatch exception," + e);
                 }
@@ -241,6 +251,12 @@ public List<VirtualProc> dispatchHost(DispatchHost host, GroupInterface group) {
 
     @Override
     public List<VirtualProc> dispatchHost(DispatchHost host, JobInterface job) {
+        return dispatchHost(host, job,
+                licenseBookingGate == null ? null : licenseBookingGate.newSession(host.getName()));
+    }
+
+    private List<VirtualProc> dispatchHost(DispatchHost host, JobInterface job,
+            LicenseBookingGate.Session licenseSession) {
 
         List<VirtualProc> procs = new ArrayList<VirtualProc>();
 
@@ -258,6 +274,14 @@ public List<VirtualProc> dispatchHost(DispatchHost host, JobInterface job) {
                 env.getProperty("dispatcher.frame.selfish.services", "").split(",");
         for (DispatchFrame frame : frames) {
 
+            // Hold frames whose layer needs an application license with no free
+            // seat; other layers of the job may still book, so skip, not break.
+            if (licenseSession != null && !licenseSession.canBook(frame.getLayerId())) {
+                logger.debug("Cannot dispatch frame " + frame.getName()
+                        + ", no application license seat free.");
+                continue;
+            }
+
             VirtualProc proc = VirtualProc.build(host, frame, selfishServices);
 
             if (frame.minCores <= 0 && !proc.canHandleNegativeCoresRequest) {
@@ -291,6 +315,9 @@ public void wrapDispatchFrame() {
             }.execute();
 
             if (success) {
+                if (licenseSession != null) {
+                    licenseSession.booked(frame.getLayerId());
+                }
                 procs.add(proc);
 
                 DispatchSupport.bookedProcs.getAndIncrement();
@@ -322,9 +349,14 @@ public void wrapDispatchFrame() {
 
     public void dispatchProcToJob(VirtualProc proc, JobInterface job) {
 
+        LicenseBookingGate.Session licenseSession =
+                licenseBookingGate == null ? null : licenseBookingGate.newSession(proc.hostName);
         // Do not throttle this method
         for (DispatchFrame frame : dispatchSupport.findNextDispatchFrames(job, proc,
                 getIntProperty("dispatcher.frame_query_max"))) {
+            if (licenseSession != null && !licenseSession.canBook(frame.getLayerId())) {
+                continue;
+            }
             try {
                 boolean success = new DispatchFrameTemplate(proc, job, frame, true) {
                     public void wrapDispatchFrame() {
@@ -422,6 +454,14 @@ public void setRqdClient(RqdClient rqdClient) {
         this.rqdClient = rqdClient;
     }
 
+    public LicenseBookingGate getLicenseBookingGate() {
+        return licenseBookingGate;
+    }
+
+    public void setLicenseBookingGate(LicenseBookingGate licenseBookingGate) {
+        this.licenseBookingGate = licenseBookingGate;
+    }
+
     private abstract class DispatchFrameTemplate {
         protected VirtualProc proc;
         protected JobInterface job;

cuebot/src/main/java/com/imageworks/spcue/dispatcher/FrameCompleteHandler.java

@@ -16,10 +16,16 @@
 package com.imageworks.spcue.dispatcher;
 
 import java.sql.Timestamp;
+import java.util.Collections;
 import java.util.EnumSet;
+import java.util.HashSet;
 import java.util.Random;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 
+import com.google.common.cache.Cache;
+import com.google.common.cache.CacheBuilder;
 import org.apache.logging.log4j.Logger;
 import org.apache.logging.log4j.LogManager;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -129,11 +135,95 @@ public void setSatisfyDependOnlyOnFrameSuccess(boolean satisfyDependOnlyOnFrameS
         this.satisfyDependOnlyOnFrameSuccess = satisfyDependOnlyOnFrameSuccess;
     }
 
+    /**
+     * Exit statuses that mean "the application could not get a license", from
+     * {@code scheduler.license.denied_exit_statuses}. Vendor specific, so it is a site setting;
+     * EMPTY by default, which leaves frame-completion behaviour exactly as it was.
+     *
+     * Static because {@link #determineFrameState} is static and is the natural place for the
+     * decision. Written once when this bean is constructed, long before any report can arrive, and
+     * only read afterwards.
+     */
+    private static volatile Set<Integer> licenseDeniedStatuses = Collections.emptySet();
+
+    /**
+     * How many times one frame may be requeued for a license-denied exit without spending a retry,
+     * from {@code scheduler.license.denied_requeue_limit}. The booking gate holds licensed layers
+     * while their pool is full, so genuine denials are rare races; a frame denied over and over is
+     * misconfigured (wrong feature name, bad local license setup) and without a bound it would
+     * requeue forever, never marching to DEAD. Past the limit the vendor's own exit status is
+     * persisted again, so ordinary retry accounting takes over. Zero or negative means unbounded
+     * (the pre-limit behaviour).
+     */
+    private static volatile int licenseDeniedRequeueLimit = 10;
+
+    /**
+     * License-denied requeues per frame id, backing the limit above. Static for the same reason as
+     * {@link #licenseDeniedStatuses}. Entries expire so a frame retried much later starts fresh,
+     * and the size bound keeps a misbehaving farm from growing it without limit.
+     */
+    private static final Cache<String, Long> licenseDeniedRequeues = CacheBuilder.newBuilder()
+            .maximumSize(50_000).expireAfterWrite(12, TimeUnit.HOURS).build();
+
     @Autowired
     public FrameCompleteHandler(Environment env) {
         this.env = env;
         satisfyDependOnlyOnFrameSuccess =
                 env.getProperty("depend.satisfy_only_on_frame_success", Boolean.class, true);
+        licenseDeniedStatuses =
+                parseStatuses(env.getProperty("scheduler.license.denied_exit_statuses", ""));
+        licenseDeniedRequeueLimit =
+                env.getProperty("scheduler.license.denied_requeue_limit", Integer.class, 10);
+        if (!licenseDeniedStatuses.isEmpty()) {
+            logger.info("license-denied exit statuses (requeued without spending a retry, up to "
+                    + licenseDeniedRequeueLimit + " times per frame): " + licenseDeniedStatuses);
+        }
+    }
+
+    /** Parse a comma separated list of exit statuses, ignoring blanks and junk. */
+    private static Set<Integer> parseStatuses(String csv) {
+        if (csv == null || csv.trim().isEmpty())
+            return Collections.emptySet();
+        Set<Integer> out = new HashSet<>();
+        for (String part : csv.split(",")) {
+            String s = part.trim();
+            if (s.isEmpty())
+                continue;
+            try {
+                out.add(Integer.valueOf(s));
+            } catch (NumberFormatException e) {
+                logger.warn("ignoring non-numeric scheduler.license.denied_exit_statuses entry '"
+                        + s + "'");
+            }
+        }
+        return out;
+    }
+
+    /**
+     * Did this frame exit because no application license was free? Always false unless the site
+     * configured the statuses, so this cannot change behaviour on its own.
+     */
+    private static boolean isLicenseDenied(int exitStatus) {
+        return licenseDeniedStatuses.contains(exitStatus);
+    }
+
+    /**
+     * Does this frame still have license-denied requeue budget left? Read by
+     * {@link #determineFrameState} and by the exit-status persistence, both against the same count;
+     * {@link #countLicenseDeniedRequeue} moves the count afterwards, once per report.
+     */
+    static boolean underLicenseDeniedLimit(String frameId) {
+        if (licenseDeniedRequeueLimit <= 0) {
+            return true;
+        }
+        Long count = licenseDeniedRequeues.getIfPresent(frameId);
+        return count == null || count < licenseDeniedRequeueLimit;
+    }
+
+    /** Record one license-denied requeue for this frame. */
+    static void countLicenseDeniedRequeue(String frameId) {
+        Long count = licenseDeniedRequeues.getIfPresent(frameId);
+        licenseDeniedRequeues.put(frameId, count == null ? 1L : count + 1);
     }
 
     /**
@@ -170,7 +260,23 @@ public void handleFrameCompleteReport(final FrameCompleteReport report) {
             // not able to override what has been set by the previous logic.
             int exitStatus = report.getExitStatus();
             if (frameDetail.exitStatus == Dispatcher.EXIT_STATUS_MEMORY_FAILURE) {
+                // The OOM pre-mark wins over everything, including a license-denied
+                // exit code: the frame was killed for memory, and storing
+                // MEMORY_FAILURE is what drives the retry-with-more-memory logic.
                 exitStatus = frameDetail.exitStatus;
+            } else if (isLicenseDenied(exitStatus)
+                    && underLicenseDeniedLimit(report.getFrame().getFrameId())) {
+                // A frame that died because no application license was free is
+                // requeued by determineFrameState above. Persist it as SKIP_RETRY so
+                // the retry counter is not incremented when it runs again (the
+                // increment reads the frame's STORED exit status, so recording the
+                // vendor's own code here would spend a retry on a queue wait).
+                // Bounded per frame: past the limit the vendor status is stored and
+                // ordinary retry accounting resumes (see licenseDeniedRequeueLimit).
+                logger.info("frame " + frame.getName() + " could not get a license (exit "
+                        + exitStatus + "); requeueing without spending a retry");
+                countLicenseDeniedRequeue(report.getFrame().getFrameId());
+                exitStatus = FrameExitStatus.SKIP_RETRY_VALUE;
             }
 
             if (dispatchSupport.stopFrame(frame, newFrameState, exitStatus,
@@ -623,7 +729,20 @@ else if (frame.state.equals(FrameState.DEAD)) {
             long lastUpdate = (r - report.getFrame().getLluTime()) / 60;
 
             FrameState newState = FrameState.WAITING;
-            if (report.getExitStatus() == FrameExitStatus.SKIP_RETRY_VALUE
+            if (isLicenseDenied(report.getExitStatus())
+                    && underLicenseDeniedLimit(frame.getFrameId())) {
+                // The application could not get a license. That is a resource
+                // being contended, not a broken frame: requeue it and do NOT
+                // burn a retry (the caller persists SKIP_RETRY for that), or a
+                // busy license pool would march every frame to DEAD. Headroom
+                // and the dispatcher's live gate are what avoid this; this
+                // catches the race they cannot -- an artist taking the last seat
+                // between the license sample and the actual checkout. Bounded
+                // per frame (underLicenseDeniedLimit): a frame denied over and
+                // over is misconfigured, not queue-unlucky, and falls back to
+                // ordinary retry accounting below.
+                newState = FrameState.WAITING;
+            } else if (report.getExitStatus() == FrameExitStatus.SKIP_RETRY_VALUE
                     || (job.maxRetries != 0 && report.getExitSignal() == 119)) {
                 report = FrameCompleteReport.newBuilder(report)
                         .setExitStatus(FrameExitStatus.SKIP_RETRY_VALUE).build();