JobManagerService.java

/*
 * 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.service;

import java.util.List;

import com.google.common.collect.Sets;
import org.apache.logging.log4j.Logger;
import org.apache.logging.log4j.LogManager;
import org.springframework.dao.DataAccessException;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;

import com.imageworks.spcue.BuildableDependency;
import com.imageworks.spcue.BuildableJob;
import com.imageworks.spcue.BuildableLayer;
import com.imageworks.spcue.DispatchFrame;
import com.imageworks.spcue.DispatchJob;
import com.imageworks.spcue.EntityRetrievalException;
import com.imageworks.spcue.ExecutionSummary;
import com.imageworks.spcue.FrameDetail;
import com.imageworks.spcue.FrameInterface;
import com.imageworks.spcue.FrameStateTotals;
import com.imageworks.spcue.GroupDetail;
import com.imageworks.spcue.JobDetail;
import com.imageworks.spcue.JobInterface;
import com.imageworks.spcue.JobLaunchException;
import com.imageworks.spcue.LayerDetail;
import com.imageworks.spcue.LayerInterface;
import com.imageworks.spcue.LimitEntity;
import com.imageworks.spcue.LimitInterface;
import com.imageworks.spcue.ShowEntity;
import com.imageworks.spcue.ThreadStats;
import com.imageworks.spcue.dao.FacilityDao;
import com.imageworks.spcue.dao.FrameDao;
import com.imageworks.spcue.dao.GroupDao;
import com.imageworks.spcue.dao.HostDao;
import com.imageworks.spcue.dao.JobDao;
import com.imageworks.spcue.dao.LayerDao;
import com.imageworks.spcue.dao.LimitDao;
import com.imageworks.spcue.dao.ShowDao;
import com.imageworks.spcue.dao.criteria.FrameSearchInterface;
import com.imageworks.spcue.dispatcher.Dispatcher;
import com.imageworks.spcue.grpc.job.CheckpointState;
import com.imageworks.spcue.grpc.job.FrameState;
import com.imageworks.spcue.grpc.job.JobState;
import com.imageworks.spcue.grpc.job.Order;
import com.imageworks.spcue.grpc.limit.Limit;
import com.imageworks.spcue.util.CueUtil;
import com.imageworks.spcue.util.FrameSet;
import com.imageworks.spcue.util.JobLogUtil;
import com.imageworks.spcue.util.Convert;

@Transactional
public class JobManagerService implements JobManager {

    private static final Logger logger = LogManager.getLogger(JobManagerService.class);

    private JobDao jobDao;
    private ShowDao showDao;
    private FrameDao frameDao;
    private LayerDao layerDao;
    private LimitDao limitDao;
    private HostDao hostDao;
    private DependManager dependManager;
    private FilterManager filterManager;
    private GroupDao groupDao;
    private FacilityDao facilityDao;
    private JobLogUtil jobLogUtil;

    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public boolean isJobComplete(JobInterface job) {
        return jobDao.isJobComplete(job);
    }

    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public boolean isLayerComplete(LayerInterface layer) {
        return layerDao.isLayerComplete(layer);
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public boolean isLayerThreadable(LayerInterface layer) {
        return layerDao.isThreadable(layer);
    }

    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public boolean isJobPending(String name) {
        return jobDao.exists(name);
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public void removeJob(JobInterface job) {
        jobDao.deleteJob(job);
    }

    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public JobDetail getJobDetail(String id) {
        return jobDao.getJobDetail(id);
    }

    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public JobInterface getJob(String id) {
        return jobDao.getJob(id);
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public JobDetail findJobDetail(String name) {
        return jobDao.findJobDetail(name);
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public JobInterface findJob(String name) {
        return jobDao.findJob(name);
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public boolean isOverMinCores(JobInterface job) {
        return jobDao.isOverMinCores(job);
    }

    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public DispatchJob getDispatchJob(String id) {
        return jobDao.getDispatchJob(id);
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public FrameInterface getFrame(String id) {
        return frameDao.getFrame(id);
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public FrameInterface findFrame(LayerInterface layer, int number) {
        return frameDao.findFrame(layer, number);
    }

    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public DispatchFrame getDispatchFrame(String id) {
        return frameDao.getDispatchFrame(id);
    }

    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public JobDetail findLastJob(String name) {
        return jobDao.findLastJob(name);
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public void setJobPaused(JobInterface job, boolean paused) {
        jobDao.updatePaused(job, paused);
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public void launchJobSpec(JobSpec spec) {

        for (BuildableJob job : spec.getJobs()) {

            JobDetail d = createJob(job);
            if (job.maxCoresOverride != null) {
                jobDao.updateMaxCores(d,
                        Convert.coresToWholeCoreUnits(job.maxCoresOverride.intValue()));
            }
            if (job.maxGpusOverride != null) {
                jobDao.updateMaxGpus(d, job.maxGpusOverride.intValue());
            }
            if (job.getPostJob() != null) {
                BuildableJob postJob = job.getPostJob();
                postJob.env.put("CUE_PARENT_JOB_ID", d.id);
                postJob.env.put("CUE_PARENT_JOB", d.name);
                createJob(postJob);
                jobDao.mapPostJob(job);
            }
        }

        for (BuildableDependency dep : spec.getDepends()) {
            dep.setLaunchDepend(true);
            dependManager.createDepend(dep);
        }

        for (BuildableJob job : spec.getJobs()) {
            jobDao.activateJob(job.detail, JobState.PENDING);
            if (job.getPostJob() != null) {
                jobDao.activateJob(job.getPostJob().detail, JobState.POSTED);
            }
        }
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public JobDetail createJob(BuildableJob buildableJob) {

        logger.info("creating new job: " + buildableJob.detail.name);
        long startTime = System.currentTimeMillis();

        if (jobDao.exists(buildableJob.detail.name)) {
            throw new JobLaunchException(
                    "error launching job, active job already exists: " + buildableJob.detail.name);
        }

        if (buildableJob.getBuildableLayers().size() < 1) {
            throw new JobLaunchException("error launching job, there were no layers defined!");
        }

        JobDetail job = buildableJob.detail;

        try {
            /*
             * Get the last job with the same name and try to use the memory settings for that job.
             * Do this before inserting the new job we'll find this job as the last job.
             */
            JobDetail lastJob = null;
            try {
                lastJob = findLastJob(job.name);
                logger.info("Last job " + job.name + " was found as " + lastJob.name);
            } catch (Exception e) {
                logger.info("Last job " + job.name + " was NOT found");
                // don't have another version of the job in the DB.
            }

            ShowEntity show = showDao.findShowDetail(job.showName);
            if (!job.isPaused) {
                job.isPaused = show.paused;
            }

            job.showId = show.id;
            job.logDir = job.name;

            /*
             * The job gets inserted into the root group and unknown department.
             */
            GroupDetail rootGroup = groupDao.getRootGroupDetail(job);
            job.groupId = rootGroup.id;
            job.deptId = rootGroup.deptId;

            resolveFacility(job);

            jobDao.insertJob(job, jobLogUtil);
            jobDao.insertEnvironment(job, buildableJob.env);

            for (BuildableLayer buildableLayer : buildableJob.getBuildableLayers()) {

                LayerDetail layer = buildableLayer.layerDetail;
                layer.jobId = job.id;
                layer.showId = show.id;

                /** Not accurate anymore */
                List<Integer> frames = CueUtil.normalizeFrameRange(layer.range, layer.chunkSize);
                layer.totalFrameCount = frames.size();

                if (lastJob != null && !buildableLayer.isMemoryOverride) {
                    long pastMaxRSS = layerDao.findPastMaxRSS(lastJob, layer.name);
                    if (pastMaxRSS > 0) {
                        logger.info("found new maxRSS for layer: " + layer.name + " " + pastMaxRSS);
                        layer.minimumMemory = pastMaxRSS;
                    }
                }

                if (layer.minimumCores > 0
                        && layer.minimumCores < Dispatcher.CORE_POINTS_RESERVED_MIN) {
                    layer.minimumCores = Dispatcher.CORE_POINTS_RESERVED_MIN;
                }

                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()));
                layer.outputs.stream().forEach(ln -> registerLayerOutput(layer, ln));
                frameDao.insertFrames(layer, frames);
            }

            // The priority of a job is set on it's resource entry.
            // To update it we set the priority after it's been inserted.
            jobDao.updatePriority(job, job.priority);

            /*
             * Finally, run any filters on the job which may set the job's priority.
             */
            filterManager.runFiltersOnJob(job);

            CueUtil.logDuration(startTime, "created job " + job.getName() + " " + job.getId());
            return job;

        } catch (Exception e) {
            logger.info("error launching job: " + job.name + "," + e);
            throw new JobLaunchException("error launching job: " + job.name + "," + e, e);
        }
    }

    private void resolveFacility(JobDetail job) {
        try {
            if (job.facilityName == null) {
                job.facilityId = facilityDao.getDefaultFacility().getId();
            } else {
                job.facilityId = facilityDao.getFacility(job.facilityName).getId();
            }
        } catch (Exception e) {
            throw new EntityRetrievalException("failed to find facility: " + job.facilityName, e);
        }
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public boolean shutdownJob(JobInterface job) {
        // See JobManagerSupport
        if (jobDao.updateJobFinished(job)) {
            logger.info("shutting down job: " + job.getName());
            jobDao.activatePostJob(job);
            logger.info("activating post jobs");
            return true;
        }
        return false;
    }

    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public List<FrameInterface> findFrames(FrameSearchInterface r) {
        return frameDao.findFrames(r);
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public void updateFrameState(FrameInterface frame, FrameState state) {
        frameDao.updateFrameState(frame, state);
    }

    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public LayerDetail getLayerDetail(String id) {
        return layerDao.getLayerDetail(id);
    }

    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public LayerInterface getLayer(String id) {
        return layerDao.getLayer(id);
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public void updateJobState(JobInterface job, JobState state) {
        jobDao.updateState(job, state);
    }

    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public FrameDetail getFrameDetail(String id) {
        return frameDao.getFrameDetail(id);
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public void markFrameAsWaiting(FrameInterface frame) {
        frameDao.markFrameAsWaiting(frame);
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public void markFrameAsDepend(FrameInterface frame) {
        frameDao.markFrameAsDepend(frame);
    }

    /**
     * Creates a new job log directory. This is only called when launching a job.
     *
     * @param job
     */
    @Transactional(propagation = Propagation.NEVER)
    public void createJobLogDirectory(JobDetail job) {
        if (!jobLogUtil.createJobLogDirectory(job.logDir)) {
            throw new JobLaunchException("error launching job, unable to create log directory");
        }
    }

    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public List<LayerInterface> getLayers(JobInterface job) {
        return layerDao.getLayers(job);
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public void increaseLayerMemoryRequirement(LayerInterface layer, long memKb) {
        layerDao.increaseLayerMinMemory(layer, memKb);
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public void reorderLayer(LayerInterface layer, FrameSet frameSet, Order order) {
        switch (order) {
            case FIRST:
                frameDao.reorderFramesFirst(layer, frameSet);
                break;
            case LAST:
                frameDao.reorderFramesLast(layer, frameSet);
                break;
            case REVERSE:
                frameDao.reorderLayerReverse(layer, frameSet);
                break;
        }
    }

    @Transactional(propagation = Propagation.NOT_SUPPORTED)
    public void staggerLayer(LayerInterface layer, String range, int stagger) {
        frameDao.staggerLayer(layer, range, stagger);
    }

    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public List<LayerDetail> getLayerDetails(JobInterface job) {
        return layerDao.getLayerDetails(job);
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public List<ThreadStats> getThreadStats(LayerInterface layer) {
        return layerDao.getThreadStats(layer);
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public void optimizeLayer(LayerInterface layer, int cores, long maxRss, int runTime) {
        layerDao.balanceLayerMinMemory(layer, maxRss);
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED)
    public void enableMemoryOptimizer(LayerInterface layer, boolean state) {
        layerDao.enableMemoryOptimizer(layer, state);
    }

    @Override
    public void appendLayerTag(LayerInterface layer, String tag) {
        layerDao.appendLayerTags(layer, tag);
    }

    @Override
    public void setLayerTag(LayerInterface layer, String tag) {
        layerDao.updateLayerTags(layer, Sets.newHashSet(tag));
    }

    @Override
    public void setLayerDispatchOrder(LayerInterface layer, int order) {
        layerDao.updateLayerDispatchOrder(layer, order);
    }

    @Override
    public void setLayerMinCores(LayerInterface layer, int coreUnits) {
        layerDao.updateLayerMinCores(layer, coreUnits);
    }

    @Override
    public void setLayerMaxCores(LayerInterface layer, int coreUnits) {
        layerDao.updateLayerMaxCores(layer, coreUnits);
    }

    @Override
    public void setLayerMinGpus(LayerInterface layer, int gpu) {
        layerDao.updateLayerMinGpus(layer, gpu);
    }

    @Override
    public void setLayerMaxGpus(LayerInterface layer, int gpu) {
        layerDao.updateLayerMaxGpus(layer, gpu);
    }

    @Override
    public void addLayerLimit(LayerInterface layer, String limitId) {
        layerDao.addLimit(layer, limitId);
    }

    @Override
    public void dropLayerLimit(LayerInterface layer, String limitId) {
        layerDao.dropLimit(layer, limitId);
    }

    @Override
    public List<LimitEntity> getLayerLimits(LayerInterface layer) {
        return layerDao.getLimits(layer);
    }

    @Override
    public void registerLayerOutput(LayerInterface layer, String filespec) {
        try {
            layerDao.insertLayerOutput(layer, filespec);
        } catch (DataAccessException e) {
            // Fail quietly but log it.
            logger.warn("Failed to add layer output: " + filespec + "," + e);
        }
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public List<String> getLayerOutputs(LayerInterface layer) {
        return layerDao.getLayerOutputs(layer);
    }

    @Override
    @Transactional(propagation = Propagation.SUPPORTS)
    public void updateCheckpointState(FrameInterface frame, CheckpointState state) {

        if (frameDao.updateFrameCheckpointState(frame, state)) {
            logger.info(
                    "Checkpoint state of frame " + frame.getId() + " set to " + state.toString());
        } else {
            logger.warn("Failed to set checkpoint state of " + frame.getId() + " to "
                    + state.toString());
        }
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public FrameDetail findHighestMemoryFrame(JobInterface job) {
        return frameDao.findHighestMemoryFrame(job);
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public FrameDetail findLongestFrame(JobInterface job) {
        return frameDao.findLongestFrame(job);
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public FrameDetail findLowestMemoryFrame(JobInterface job) {
        return frameDao.findLowestMemoryFrame(job);
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public FrameDetail findShortestFrame(JobInterface job) {
        return frameDao.findShortestFrame(job);
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public ExecutionSummary getExecutionSummary(JobInterface job) {
        return jobDao.getExecutionSummary(job);
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public FrameStateTotals getFrameStateTotals(JobInterface job) {
        return jobDao.getFrameStateTotals(job);
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public ExecutionSummary getExecutionSummary(LayerInterface layer) {
        return layerDao.getExecutionSummary(layer);
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public FrameStateTotals getFrameStateTotals(LayerInterface layer) {
        return layerDao.getFrameStateTotals(layer);
    }

    @Override
    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public List<FrameInterface> getStaleCheckpoints(int cutoffTimeSec) {
        return frameDao.getStaleCheckpoints(cutoffTimeSec);
    }

    @Transactional(propagation = Propagation.REQUIRED)
    public void updateEmail(JobInterface job, String email) {
        jobDao.updateEmail(job, email);
    }

    @Transactional(propagation = Propagation.REQUIRED, readOnly = true)
    public String getEmail(JobInterface job) {
        return jobDao.getEmail(job);
    }

    public DependManager getDependManager() {
        return dependManager;
    }

    public void setDependManager(DependManager dependManager) {
        this.dependManager = dependManager;
    }

    public FrameDao getFrameDao() {
        return frameDao;
    }

    public void setFrameDao(FrameDao frameDao) {
        this.frameDao = frameDao;
    }

    public LayerDao getLayerDao() {
        return layerDao;
    }

    public void setLayerDao(LayerDao layerDao) {
        this.layerDao = layerDao;
    }

    public LimitDao getLimitDao() {
        return limitDao;
    }

    public void setLimitDao(LimitDao limitDao) {
        this.limitDao = limitDao;
    }

    public ShowDao getShowDao() {
        return showDao;
    }

    public void setShowDao(ShowDao showDao) {
        this.showDao = showDao;
    }

    public JobDao getJobDao() {
        return jobDao;
    }

    public void setJobDao(JobDao workDao) {
        this.jobDao = workDao;
    }

    public FilterManager getFilterManager() {
        return filterManager;
    }

    public void setFilterManager(FilterManager filterManager) {
        this.filterManager = filterManager;
    }

    public GroupDao getGroupDao() {
        return groupDao;
    }

    public void setGroupDao(GroupDao groupDao) {
        this.groupDao = groupDao;
    }

    public FacilityDao getFacilityDao() {
        return facilityDao;
    }

    public void setFacilityDao(FacilityDao facilityDao) {
        this.facilityDao = facilityDao;
    }

    public HostDao getHostDao() {
        return hostDao;
    }

    public void setHostDao(HostDao hostDao) {
        this.hostDao = hostDao;
    }

    public JobLogUtil getJobLogUtil() {
        return jobLogUtil;
    }

    public void setJobLogUtil(JobLogUtil jobLogUtil) {
        this.jobLogUtil = jobLogUtil;
    }
}