IngestJobExecutor.java

Go to the documentation of this file.

/*
 * Autopsy Forensic Browser
 *
 * Copyright 2014-2021 Basis Technology Corp.
 * Contact: carrier <at> sleuthkit <dot> org
 *
 * 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 org.sleuthkit.autopsy.ingest;
 
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.logging.Level;
import javax.annotation.concurrent.GuardedBy;
import javax.swing.JOptionPane;
import javax.swing.SwingUtilities;
import org.netbeans.api.progress.ProgressHandle;
import org.openide.util.Cancellable;
import org.openide.util.NbBundle;
import org.openide.util.NbBundle.Messages;
import org.openide.windows.WindowManager;
import org.sleuthkit.autopsy.casemodule.Case;
import org.sleuthkit.autopsy.core.RuntimeProperties;
import org.sleuthkit.autopsy.coreutils.Logger;
import org.sleuthkit.autopsy.coreutils.NetworkUtils;
import org.sleuthkit.autopsy.coreutils.ThreadConfined;
import org.sleuthkit.datamodel.AbstractFile;
import org.sleuthkit.datamodel.IngestJobInfo;
import org.sleuthkit.datamodel.IngestJobInfo.IngestJobStatusType;
import org.sleuthkit.datamodel.IngestModuleInfo;
import org.sleuthkit.datamodel.IngestModuleInfo.IngestModuleType;
import org.sleuthkit.datamodel.SleuthkitCase;
import org.sleuthkit.datamodel.TskCoreException;
import org.sleuthkit.autopsy.modules.interestingitems.FilesSet;
import org.sleuthkit.autopsy.python.FactoryClassNameNormalizer;
import org.sleuthkit.datamodel.AnalysisResult;
import org.sleuthkit.datamodel.DataArtifact;
import org.sleuthkit.datamodel.DataSource;

final class IngestJobExecutor {
 
    private static enum IngestJobState {
        PIPELINES_STARTING_UP,
        ACCEPTING_STREAMED_CONTENT_AND_ANALYZING,
        ANALYZING,
        PIPELINES_SHUTTING_DOWN
    };
    private static final Logger logger = Logger.getLogger(IngestJobExecutor.class.getName());
    private final IngestJob ingestJob;
    private final long createTime;
    private final boolean usingNetBeansGUI;
    private final IngestTasksScheduler taskScheduler = IngestTasksScheduler.getInstance();
    private final Object threadRegistrationLock = new Object();
    @GuardedBy("threadRegistrationLock")
    private final Set<Thread> pausedIngestThreads = new HashSet<>();
    private final List<String> cancelledDataSourceIngestModules = new CopyOnWriteArrayList<>();
    private final Object tierTransitionLock = new Object();
    private final List<IngestModuleTier> ingestModuleTiers = new ArrayList<>();
    private volatile int moduleTierIndex = 0;
    private volatile IngestJobState jobState = IngestJobExecutor.IngestJobState.PIPELINES_STARTING_UP;
    private volatile long estimatedFilesToProcess = 0;
    private volatile long processedFiles = 0;
    private volatile boolean currentDataSourceIngestModuleCancelled = false;
    private volatile boolean jobCancelled = false;
    private volatile IngestJob.CancellationReason cancellationReason = IngestJob.CancellationReason.NOT_CANCELLED;
    private volatile IngestJobInfo casDbingestJobInfo;
    @ThreadConfined(type = ThreadConfined.ThreadType.AWT)
    private ProgressHandle dataSourceIngestProgressBar;
    @ThreadConfined(type = ThreadConfined.ThreadType.AWT)
    private final List<String> filesInProgress = new ArrayList<>();
    @ThreadConfined(type = ThreadConfined.ThreadType.AWT)
    private ProgressHandle fileIngestProgressBar;
    @ThreadConfined(type = ThreadConfined.ThreadType.AWT)
    private ProgressHandle artifactIngestProgressBar;
    @ThreadConfined(type = ThreadConfined.ThreadType.AWT)
    private ProgressHandle resultIngestProgressBar;

    IngestJobExecutor(IngestJob ingestJob) throws InterruptedException {
        this.ingestJob = ingestJob;
        createTime = new Date().getTime();
        /*
         * If running in the NetBeans thick client application version of
         * Autopsy, NetBeans progress handles (i.e., progress bars) are used to
         * display ingest job progress in the lower right hand corner of the
         * main application window. A layer of abstraction to allow alternate
         * representations of progress could be used here, as it is in some
         * other places in the application (see implementations and usage of the
         * org.sleuthkit.autopsy.progress.ProgressIndicator interface).
         */
        usingNetBeansGUI = RuntimeProperties.runningWithGUI();
    }

    long getIngestJobId() {
        return ingestJob.getId();
    }

    String getExecutionContext() {
        return ingestJob.getSettings().getExecutionContext();
    }

    DataSource getDataSource() {
        return ingestJob.getDataSource();
    }

    boolean shouldProcessUnallocatedSpace() {
        return ingestJob.getSettings().getProcessUnallocatedSpace();
    }

    FilesSet getFileIngestFilter() {
        return ingestJob.getSettings().getFileFilter();
    }

    List<IngestModuleError> startUp() throws InterruptedException {
        jobState = IngestJobState.PIPELINES_STARTING_UP;
        ingestModuleTiers.addAll(IngestModuleTierBuilder.buildIngestModuleTiers(ingestJob.getSettings(), this));
        List<IngestModuleError> errors = startUpIngestModulePipelines();
        if (errors.isEmpty()) {
            recordIngestJobStartUpInfo();
            /*
             * Start up and execution of the first ingest module tier requires
             * some special treatment due to the differences between streaming
             * and batch mode ingest jobs. Subsequent tiers can be handled
             * generically.
             */
            if (ingestJob.getIngestMode() == IngestJob.Mode.STREAMING) {
                startStreamingModeAnalysis();
            } else {
                startBatchModeAnalysis();
            }
        }
        return errors;
    }

    private List<IngestModuleError> startUpIngestModulePipelines() {
        List<IngestModuleError> errors = new ArrayList<>();
        for (IngestModuleTier moduleTier : ingestModuleTiers) {
            Optional<DataSourceIngestPipeline> dataSourcePipeline = moduleTier.getDataSourceIngestPipeline();
            if (dataSourcePipeline.isPresent()) {
                errors.addAll(startUpIngestModulePipeline(dataSourcePipeline.get()));
            }
 
            for (FileIngestPipeline pipeline : moduleTier.getFileIngestPipelines()) {
                List<IngestModuleError> filePipelineErrors = startUpIngestModulePipeline(pipeline);
                if (!filePipelineErrors.isEmpty()) {
                    /*
                     * If one file pipeline copy can't start up, assume that
                     * none of the other copies will be able to start up, for
                     * the same reason.
                     */
                    errors.addAll(filePipelineErrors);
                    break;
                }
            }
 
            Optional<DataArtifactIngestPipeline> dataArtifactPipeline = moduleTier.getDataArtifactIngestPipeline();
            if (dataArtifactPipeline.isPresent()) {
                errors.addAll(startUpIngestModulePipeline(dataArtifactPipeline.get()));
            }
 
            Optional<AnalysisResultIngestPipeline> analysisResultPipeline = moduleTier.getAnalysisResultIngestPipeline();
            if (analysisResultPipeline.isPresent()) {
                errors.addAll(startUpIngestModulePipeline(analysisResultPipeline.get()));
            }
        }
        return errors;
    }

    private List<IngestModuleError> startUpIngestModulePipeline(IngestPipeline<?> pipeline) {
        List<IngestModuleError> startUpErrors = pipeline.startUp();
        if (!startUpErrors.isEmpty()) {
            List<IngestModuleError> shutDownErrors = pipeline.shutDown();
            if (!shutDownErrors.isEmpty()) {
                logIngestModuleErrors(shutDownErrors);
            }
        }
        return startUpErrors;
    }

    private void recordIngestJobStartUpInfo() {
        try {
            SleuthkitCase caseDb = Case.getCurrentCase().getSleuthkitCase();
            List<IngestModuleInfo> ingestModuleInfoList = new ArrayList<>();
            for (IngestModuleTemplate module : ingestJob.getSettings().getEnabledIngestModuleTemplates()) {
                IngestModuleType moduleType = getIngestModuleTemplateType(module);
                IngestModuleInfo moduleInfo = caseDb.addIngestModule(module.getModuleName(), FactoryClassNameNormalizer.normalize(module.getModuleFactory().getClass().getCanonicalName()), moduleType, module.getModuleFactory().getModuleVersionNumber());
                ingestModuleInfoList.add(moduleInfo);
            }
            casDbingestJobInfo = caseDb.addIngestJob(ingestJob.getDataSource(), NetworkUtils.getLocalHostName(), ingestModuleInfoList, new Date(this.createTime), new Date(0), IngestJobStatusType.STARTED, "");
        } catch (TskCoreException ex) {
            logErrorMessage(Level.SEVERE, "Failed to add ingest job info to case database", ex); //NON-NLS
        }
    }

    private static IngestModuleType getIngestModuleTemplateType(IngestModuleTemplate moduleTemplate) {
        IngestModuleType type = null;
        if (moduleTemplate.isDataSourceIngestModuleTemplate()) {
            type = IngestModuleType.DATA_SOURCE_LEVEL;
        }
        if (moduleTemplate.isFileIngestModuleTemplate()) {
            if (type == null) {
                type = IngestModuleType.FILE_LEVEL;
            } else {
                type = IngestModuleType.MULTIPLE;
            }
        }
        if (moduleTemplate.isDataArtifactIngestModuleTemplate()) {
            if (type == null) {
                type = IngestModuleType.DATA_ARTIFACT;
            } else {
                type = IngestModuleType.MULTIPLE;
            }
        }
        return type;
    }

    private void startBatchModeAnalysis() {
        synchronized (tierTransitionLock) {
            logInfoMessage("Starting ingest job in file batch mode"); //NON-NLS            
            jobState = IngestJobState.ANALYZING;
            IngestModuleTier currentTier = ingestModuleTiers.get(moduleTierIndex);
 
            if (currentTier.hasDataSourceIngestModules()) {
                startDataSourceIngestProgressBar();
                taskScheduler.scheduleDataSourceIngestTask(this);
            }
 
            if (currentTier.hasFileIngestModules()) {
                estimateFilesToProcess();
                startFileIngestProgressBar(true);
                taskScheduler.scheduleFileIngestTasks(this, ingestJob.getFiles());
            }
 
            if (currentTier.hasDataArtifactIngestModules()) {
                /*
                 * Analysis of any data artifacts already in the case database
                 * (possibly added by the DSP) will be performed.
                 */
                startDataArtifactIngestProgressBar();
                taskScheduler.scheduleDataArtifactIngestTasks(this);
            }
 
            if (currentTier.hasAnalysisResultIngestModules()) {
                /*
                 * Analysis of any analysis results already in the case database
                 * (possibly added by the DSP) will be performed.
                 */
                startAnalysisResultIngestProgressBar();
                taskScheduler.scheduleAnalysisResultIngestTasks(this);
            }
 
            /*
             * Check for analysis completion. This is necessary because it is
             * possible that none of the tasks that were just scheduled will
             * actually make it to task execution, due to the file filter or
             * other ingest job settings. If that happens, there will never be
             * another analysis completion check for this job in an ingest
             * thread executing an ingest task, so such a job would run forever,
             * doing nothing, without a check here.
             */
            checkForTierCompleted(moduleTierIndex);
        }
    }

    private void estimateFilesToProcess() {
        estimatedFilesToProcess = 0;
        processedFiles = 0;
        if (ingestModuleTiers.get(moduleTierIndex).hasFileIngestModules()) {
            /*
             * Do an estimate of the total number of files to be analyzed. This
             * will be used to estimate of how many files remain to be analyzed
             * as each file ingest task is completed. The numbers are estimates
             * because analysis can add carved and/or derived files to the job.
             */
            List<AbstractFile> files = ingestJob.getFiles();
            if (files.isEmpty()) {
                /*
                 * Do a count of the files from the data source that the data
                 * source processor (DSP) has added to the case database.
                 */
                estimatedFilesToProcess = ingestJob.getDataSource().accept(new GetFilesCountVisitor());
            } else {
                /*
                 * Otherwise, this job is analyzing a user-specified subset of
                 * the files in the data source.
                 */
                estimatedFilesToProcess = files.size();
            }
        }
    }

    private void startStreamingModeAnalysis() {
        synchronized (tierTransitionLock) {
            logInfoMessage("Starting ingest job in file streaming mode"); //NON-NLS
            jobState = IngestJobState.ACCEPTING_STREAMED_CONTENT_AND_ANALYZING;
            IngestModuleTier currentTier = ingestModuleTiers.get(moduleTierIndex);
 
            if (currentTier.hasFileIngestModules()) {
                /*
                 * Start the file ingest progress bar, but do not schedule any
                 * file or data source ingest tasks. File ingest tasks will
                 * instead be scheduled as files are streamed in via
                 * addStreamedFiles(), and a data source ingest task will be
                 * scheduled later, via addStreamedDataSource().
                 */
                startFileIngestProgressBar(false);
            }
 
            if (currentTier.hasDataArtifactIngestModules()) {
                /*
                 * Start the data artifact progress bar and schedule ingest
                 * tasks for any data artifacts currently in the case database.
                 * This needs to be done BEFORE any files or the data source are
                 * streamed in to ensure that any data artifacts added to the
                 * case database by the file and data source ingest tasks are
                 * not analyzed twice. This works here because the ingest
                 * manager has not yet returned the ingest stream object that is
                 * used to call addStreamedFiles() and addStreamedDataSource().
                 */
                startDataArtifactIngestProgressBar();
                taskScheduler.scheduleDataArtifactIngestTasks(this);
            }
 
            if (currentTier.hasAnalysisResultIngestModules()) {
                /*
                 * Start the analysis result progress bar and schedule ingest
                 * tasks for any analysis results currently in the case
                 * database. This needs to be done BEFORE any files or the data
                 * source are streamed in to ensure that any analysis results
                 * added to the case database by the file and data source ingest
                 * tasks are not analyzed twice. This works here because the
                 * ingest manager has not yet returned the ingest stream object
                 * that is used to call addStreamedFiles() and
                 * addStreamedDataSource().
                 */
                startAnalysisResultIngestProgressBar();
                taskScheduler.scheduleAnalysisResultIngestTasks(this);
            }
        }
    }

    void addStreamedDataSource() {
        synchronized (tierTransitionLock) {
            logInfoMessage("Data source received in streaming mode ingest job"); //NON-NLS
            jobState = IngestJobExecutor.IngestJobState.ANALYZING;
            IngestModuleTier currentTier = ingestModuleTiers.get(moduleTierIndex);
 
            if (currentTier.hasFileIngestModules()) {
                estimateFilesToProcess();
                switchFileIngestProgressBarToDeterminate();
                // We don't need to schedule file tasks here because they've already been
                // added as the data source was being processed
            }
 
            if (currentTier.hasDataSourceIngestModules()) {
                taskScheduler.scheduleDataSourceIngestTask(this);
                startDataSourceIngestProgressBar();
            } else {
                /*
                 * If no data source level ingest task is scheduled at this
                 * time, and all of the file level and artifact ingest tasks
                 * scheduled during the initial file streaming stage have
                 * already been executed, there will never be a stage completion
                 * check in an ingest thread executing an ingest task for this
                 * job, so such a job would run forever, doing nothing, without
                 * a check here.
                 */
                checkForTierCompleted(moduleTierIndex);
            }
        }
    }

    private void checkForTierCompleted(int currentTier) {
        synchronized (tierTransitionLock) {
            if (jobState.equals(IngestJobState.ACCEPTING_STREAMED_CONTENT_AND_ANALYZING)) {
                return;
            }
            if (currentTier < moduleTierIndex) {
                // We likely had a leftover task from the previous tier. Since we've already
                // advanced to the next tier, ignore it.
                return;
            }
            if (taskScheduler.currentTasksAreCompleted(getIngestJobId())) {
                do {
                    shutDownCurrentTier();
                    moduleTierIndex++;
                    if (moduleTierIndex < ingestModuleTiers.size()) {
                        startAnalysisForCurrentTier();
                    } else {
                        shutDown();
                        break;
                    }
                } while (taskScheduler.currentTasksAreCompleted(getIngestJobId())); // Loop again immediately in case the new tier is empty
            }
        }
    }

    private void startAnalysisForCurrentTier() {
        logInfoMessage(String.format("Scheduling ingest tasks for tier %s of ingest job", moduleTierIndex)); //NON-NLS        
        jobState = IngestJobExecutor.IngestJobState.ANALYZING;
        IngestModuleTier currentTier = ingestModuleTiers.get(moduleTierIndex);
 
        if (currentTier.hasDataSourceIngestModules()) {
            startDataSourceIngestProgressBar();
            taskScheduler.scheduleDataSourceIngestTask(this);
        }
 
        if (currentTier.hasFileIngestModules()) {
            estimateFilesToProcess();
            startFileIngestProgressBar(true);
            taskScheduler.scheduleFileIngestTasks(this, ingestJob.getFiles());
        }
 
        if (currentTier.hasDataArtifactIngestModules()) {
            startDataArtifactIngestProgressBar();
        }
 
        if (currentTier.hasAnalysisResultIngestModules()) {
            startDataArtifactIngestProgressBar();
        }
    }

    void execute(DataSourceIngestTask task) {
        try {
            if (!isCancelled()) {
                Optional<DataSourceIngestPipeline> pipeline = ingestModuleTiers.get(moduleTierIndex).getDataSourceIngestPipeline();
                if (pipeline.isPresent()) {
                    List<IngestModuleError> errors = new ArrayList<>();
                    errors.addAll(pipeline.get().performTask(task));
                    if (!errors.isEmpty()) {
                        logIngestModuleErrors(errors);
                    }
                }
            }
        } finally {
            // Save the module tier assocaited with this task since it could change after
            // notifyTaskComplete
            int currentTier = moduleTierIndex;
            taskScheduler.notifyTaskCompleted(task);
            checkForTierCompleted(currentTier);
        }
    }

    void execute(FileIngestTask task) {
        try {
            if (!isCancelled()) {
                FileIngestPipeline pipeline = ingestModuleTiers.get(moduleTierIndex).takeFileIngestPipeline();
                if (!pipeline.isEmpty()) {
                    /*
                     * Get the file from the task. If the file was streamed in,
                     * the task may only have the file object ID, and a trip to
                     * the case database will be required.
                     */
                    AbstractFile file;
                    try {
                        file = task.getFile();
                    } catch (TskCoreException ex) {
                        List<IngestModuleError> errors = new ArrayList<>();
                        errors.add(new IngestModuleError("Ingest Pipeline", ex));
                        logIngestModuleErrors(errors);
                        ingestModuleTiers.get(moduleTierIndex).returnFileIngestPipeleine(pipeline);
                        return;
                    }

                    final String fileName = file.getName();
                    processedFiles++;
                    updateFileProgressBarForFileTaskStarted(fileName);
                    List<IngestModuleError> errors = new ArrayList<>();
                    errors.addAll(pipeline.performTask(task));
                    if (!errors.isEmpty()) {
                        logIngestModuleErrors(errors, file);
                    }
                    updateFileProgressBarForFileTaskCompleted(fileName);
                }
                ingestModuleTiers.get(moduleTierIndex).returnFileIngestPipeleine(pipeline);
            }
        } catch (InterruptedException ex) {
            logger.log(Level.SEVERE, String.format("File ingest thread interrupted during execution of file ingest job (file object ID = %d, thread ID = %d)", task.getFileId(), task.getThreadId()), ex);
            Thread.currentThread().interrupt();
        } finally {
            // Save the module tier assocaited with this task since it could change after
            // notifyTaskComplete
            int currentTier = moduleTierIndex;
            taskScheduler.notifyTaskCompleted(task);
            checkForTierCompleted(currentTier);
        }
    }

    void execute(DataArtifactIngestTask task) {
        try {
            if (!isCancelled()) {
                Optional<DataArtifactIngestPipeline> pipeline = ingestModuleTiers.get(moduleTierIndex).getDataArtifactIngestPipeline();
                if (pipeline.isPresent()) {
                    List<IngestModuleError> errors = new ArrayList<>();
                    errors.addAll(pipeline.get().performTask(task));
                    if (!errors.isEmpty()) {
                        logIngestModuleErrors(errors);
                    }
                }
            }
        } finally {
            // Save the module tier assocaited with this task since it could change after
            // notifyTaskComplete
            int currentTier = moduleTierIndex;
            taskScheduler.notifyTaskCompleted(task);
            checkForTierCompleted(currentTier);
        }
    }

    void execute(AnalysisResultIngestTask task) {
        try {
            if (!isCancelled()) {
                Optional<AnalysisResultIngestPipeline> pipeline = ingestModuleTiers.get(moduleTierIndex).getAnalysisResultIngestPipeline();
                if (pipeline.isPresent()) {
                    List<IngestModuleError> errors = new ArrayList<>();
                    errors.addAll(pipeline.get().performTask(task));
                    if (!errors.isEmpty()) {
                        logIngestModuleErrors(errors);
                    }
                }
            }
        } finally {
            // Save the module tier assocaited with this task since it could change after
            // notifyTaskComplete
            int currentTier = moduleTierIndex;
            taskScheduler.notifyTaskCompleted(task);
            checkForTierCompleted(currentTier);
        }
    }

    void addStreamedFiles(List<Long> fileObjIds) {
        if (!isCancelled() && ingestModuleTiers.get(moduleTierIndex).hasFileIngestModules()) {
            if (jobState.equals(IngestJobState.ACCEPTING_STREAMED_CONTENT_AND_ANALYZING)) {
                taskScheduler.scheduleStreamedFileIngestTasks(this, fileObjIds);
            } else {
                logErrorMessage(Level.SEVERE, "Adding streaming files to job during stage " + jobState.toString() + " not supported");
            }
        }
    }

    void addFiles(List<AbstractFile> files) {
        if (!isCancelled() && ingestModuleTiers.get(moduleTierIndex).hasFileIngestModules()) {
            if (jobState.equals(IngestJobState.ACCEPTING_STREAMED_CONTENT_AND_ANALYZING) || jobState.equals(IngestJobState.ANALYZING)) {
                taskScheduler.scheduleHighPriorityFileIngestTasks(this, files);
            } else {
                logErrorMessage(Level.SEVERE, "Adding files to job during stage " + jobState.toString() + " not supported");
            }
        }
    }

    void addDataArtifacts(List<DataArtifact> artifacts) {
        if (!isCancelled() && ingestModuleTiers.get(moduleTierIndex).hasDataArtifactIngestModules()) {
            switch (jobState) {
                case ACCEPTING_STREAMED_CONTENT_AND_ANALYZING:
                case ANALYZING:
                    taskScheduler.scheduleDataArtifactIngestTasks(this, artifacts);
                    break;
                case PIPELINES_SHUTTING_DOWN:
                    /*
                    * Don't log an error if there is an attempt to add an
                    * data artifact ingest task in a pipeline shut down
                    * state. This is a work around for dealing with data
                    * artifacts generated by a final keyword search carried out
                    * during ingest module shut down by simply ignoring them.
                    * (Currently these are credit card accounts generated by 
                    * keyword search). Other ideas were to add 
                    * a startShutDown() phase to the ingest module 
                    * life cycle (complicated), or to add a flag
                    * to keyword hit processing to suppress posting the keyword
                    * hit analysis results / data artifacts to the blackboard during a final
                    * search (API changes required to allow firing of the event
                    * to make any GUI refresh).
                    */
                    break;
                default:
                    logErrorMessage(Level.SEVERE, "Attempt to add data artifacts to job during stage " + jobState.toString() + " not supported");
                    break;
            }
        }
    }

    void addAnalysisResults(List<AnalysisResult> results) {
        if (!isCancelled() && ingestModuleTiers.get(moduleTierIndex).hasAnalysisResultIngestModules()) {
            switch (jobState) {
                case ACCEPTING_STREAMED_CONTENT_AND_ANALYZING:
                case ANALYZING:
                    taskScheduler.scheduleAnalysisResultIngestTasks(this, results);
                    break;
                case PIPELINES_SHUTTING_DOWN:
                    /*
                     * Don't log an error if there is an attempt to add an
                     * analysis result ingest task in a pipeline shut down
                     * state. This is a work around for dealing with analysis
                     * results generated by a final keyword search carried out
                     * during ingest module shut down by simply ignoring them.
                     * Other ideas were to add a startShutDown() phase to the
                     * ingest module life cycle (complicated), or to add a flag
                     * to keyword hit processing to suppress posting the keyword
                     * hit analysis results to the blackboard during a final
                     * search (API changes required to allow firing of the event
                     * to make any GUI refresh).
                     */
                    break;
                default:
                    logErrorMessage(Level.SEVERE, "Attempt to add analysis results to job during stage " + jobState.toString() + " not supported");
            }
        }
    }

    private void shutDownCurrentTier() {
        // Note that this method is only called while holding the tierTransitionLock, so moduleTierIndex can not change
        // during execution.
        if (moduleTierIndex >= ingestModuleTiers.size()) {
            logErrorMessage(Level.SEVERE, "shutDownCurrentTier called with out-of-bounds moduleTierIndex (" + moduleTierIndex + ")");
            return;
        }
        logInfoMessage(String.format("Finished all ingest tasks for tier %s of ingest job", moduleTierIndex)); //NON-NLS        
        jobState = IngestJobExecutor.IngestJobState.PIPELINES_SHUTTING_DOWN;
        IngestModuleTier moduleTier = ingestModuleTiers.get(moduleTierIndex);
 
        Optional<DataSourceIngestPipeline> dataSourcePipeline = moduleTier.getDataSourceIngestPipeline();
        if (dataSourcePipeline.isPresent()) {
            shutDownIngestModulePipeline(dataSourcePipeline.get());
        }
 
        for (FileIngestPipeline pipeline : moduleTier.getFileIngestPipelines()) {
            shutDownIngestModulePipeline(pipeline);
        }
 
        Optional<DataArtifactIngestPipeline> dataArtifactPipeline = moduleTier.getDataArtifactIngestPipeline();
        if (dataArtifactPipeline.isPresent()) {
            shutDownIngestModulePipeline(dataArtifactPipeline.get());
        }
 
        Optional<AnalysisResultIngestPipeline> analysisResultPipeline = moduleTier.getAnalysisResultIngestPipeline();
        if (analysisResultPipeline.isPresent()) {
            shutDownIngestModulePipeline(analysisResultPipeline.get());
        }
 
        finishAllProgressBars();
    }

    private <T extends IngestTask> void shutDownIngestModulePipeline(IngestPipeline<T> pipeline) {
        if (pipeline.isRunning()) {
            List<IngestModuleError> errors = new ArrayList<>();
            errors.addAll(pipeline.shutDown());
            if (!errors.isEmpty()) {
                logIngestModuleErrors(errors);
            }
        }
    }

    private void shutDown() {
        logInfoMessage("Finished all ingest tasks for ingest job"); //NON-NLS        
        try {
            if (casDbingestJobInfo != null) {
                if (jobCancelled) {
                    casDbingestJobInfo.setIngestJobStatus(IngestJobStatusType.CANCELLED);
                } else {
                    casDbingestJobInfo.setIngestJobStatus(IngestJobStatusType.COMPLETED);
                }
                casDbingestJobInfo.setEndDateTime(new Date());
            }
        } catch (TskCoreException ex) {
            logErrorMessage(Level.WARNING, "Failed to set job end date in case database", ex);
        }
 
        ingestJob.notifyIngestPipelinesShutDown();
    }

    DataSourceIngestPipeline.DataSourcePipelineModule getCurrentDataSourceIngestModule() {
        Optional<DataSourceIngestPipeline> pipeline = getCurrentDataSourceIngestPipelines();
        if (pipeline.isPresent()) {
            return (DataSourceIngestPipeline.DataSourcePipelineModule) pipeline.get().getCurrentlyRunningModule();
        } else {
            return null;
        }
    }

    void cancelCurrentDataSourceIngestModule() {
        currentDataSourceIngestModuleCancelled = true;
    }

    boolean currentDataSourceIngestModuleIsCancelled() {
        return currentDataSourceIngestModuleCancelled;
    }

    void currentDataSourceIngestModuleCancellationCompleted(String moduleDisplayName) {
        currentDataSourceIngestModuleCancelled = false;
        cancelledDataSourceIngestModules.add(moduleDisplayName);
        if (usingNetBeansGUI && !jobCancelled) {
            try {
                // use invokeAndWait to ensure synchronous behavior.  
                // See JIRA-8298 for more information.
                SwingUtilities.invokeAndWait(() -> {
                    dataSourceIngestProgressBar.finish();
                    dataSourceIngestProgressBar = null;
                    startDataSourceIngestProgressBar();
                });
            } catch (InvocationTargetException | InterruptedException ex) {
                logger.log(Level.WARNING, "Cancellation worker cancelled.", ex);
            }
        }
    }

    void cancel(IngestJob.CancellationReason reason) {
        jobCancelled = true;
        cancellationReason = reason;
        displayCancellingProgressMessages();
        taskScheduler.cancelPendingFileTasksForIngestJob(getIngestJobId());
        synchronized (threadRegistrationLock) {
            for (Thread thread : pausedIngestThreads) {
                thread.interrupt();
            }
            pausedIngestThreads.clear();
        }
        checkForTierCompleted(moduleTierIndex);
    }

    boolean isCancelled() {
        return jobCancelled;
    }

    IngestJob.CancellationReason getCancellationReason() {
        return cancellationReason;
    }

    private void startDataSourceIngestProgressBar() {
        if (usingNetBeansGUI) {
            SwingUtilities.invokeLater(() -> {
                dataSourceIngestProgressBar = ProgressHandle.createHandle(NbBundle.getMessage(this.getClass(), "IngestJob.progress.dataSourceIngest.initialDisplayName", ingestJob.getDataSource().getName()), new Cancellable() {
                    @Override
                    public boolean cancel() {
                        /*
                         * The user has already pressed the cancel button on
                         * this progress bar, and the OK button of a cancelation
                         * confirmation dialog supplied by NetBeans. Find out
                         * whether the user wants to cancel only the currently
                         * executing data source ingest module or the entire
                         * ingest job.
                         */
                        DataSourceIngestCancellationPanel panel = new DataSourceIngestCancellationPanel();
                        String dialogTitle = NbBundle.getMessage(IngestJobExecutor.this.getClass(), "IngestJob.cancellationDialog.title");
                        JOptionPane.showConfirmDialog(WindowManager.getDefault().getMainWindow(), panel, dialogTitle, JOptionPane.OK_OPTION, JOptionPane.PLAIN_MESSAGE);
                        if (panel.cancelAllDataSourceIngestModules()) {
                            new Thread(() -> {
                                IngestJobExecutor.this.cancel(IngestJob.CancellationReason.USER_CANCELLED);
                            }).start();
                        } else {
                            new Thread(() -> {
                                IngestJobExecutor.this.cancelCurrentDataSourceIngestModule();
                            }).start();
                        }
                        return true;
                    }
                });
                dataSourceIngestProgressBar.start();
                dataSourceIngestProgressBar.switchToIndeterminate();
            });
        }
    }

    void changeDataSourceIngestProgressBarTitle(String title) {
        if (usingNetBeansGUI && !jobCancelled) {
            SwingUtilities.invokeLater(() -> {
                if (dataSourceIngestProgressBar != null) {
                    dataSourceIngestProgressBar.setDisplayName(title);
                }
            });
        }
    }

    void switchDataSourceIngestProgressBarToIndeterminate() {
        if (usingNetBeansGUI && !jobCancelled) {
            SwingUtilities.invokeLater(() -> {
                if (dataSourceIngestProgressBar != null) {
                    dataSourceIngestProgressBar.switchToIndeterminate();
                    dataSourceIngestProgressBar.progress(""); //NON-NLS
                }
            });
        }
    }

    void switchDataSourceIngestProgressBarToDeterminate(int workUnitsToDo) {
        if (usingNetBeansGUI && !jobCancelled) {
            SwingUtilities.invokeLater(() -> {
                if (dataSourceIngestProgressBar != null) {
                    dataSourceIngestProgressBar.switchToDeterminate(workUnitsToDo);
                }
            });
        }
    }

    void updateDataSourceIngestProgressBar(String newText, int workUnitsDone) {
        if (usingNetBeansGUI && !jobCancelled) {
            SwingUtilities.invokeLater(() -> {
                if (dataSourceIngestProgressBar != null) {
                    dataSourceIngestProgressBar.progress(newText, workUnitsDone);
                }
            });
        }
    }

    void updateDataSourceIngestProgressBarText(String newText) {
        if (usingNetBeansGUI && !jobCancelled) {
            SwingUtilities.invokeLater(() -> {
                if (dataSourceIngestProgressBar != null) {
                    dataSourceIngestProgressBar.progress(newText);
                }
            });
        }
    }

    void updateDataSourceIngestProgressBar(int workUnitsDone) {
        if (usingNetBeansGUI && !jobCancelled) {
            SwingUtilities.invokeLater(() -> {
                if (dataSourceIngestProgressBar != null) {
                    dataSourceIngestProgressBar.progress("", workUnitsDone);
                }
            });
        }
    }

    private void startFileIngestProgressBar(boolean useDeterminateMode) {
        if (usingNetBeansGUI) {
            SwingUtilities.invokeLater(() -> {
                fileIngestProgressBar = ProgressHandle.createHandle(NbBundle.getMessage(getClass(), "IngestJob.progress.fileIngest.displayName", ingestJob.getDataSource().getName()), new Cancellable() {
                    @Override
                    public boolean cancel() {
                        new Thread(() -> {
                            IngestJobExecutor.this.cancel(IngestJob.CancellationReason.USER_CANCELLED);
                        }).start();
                        return true;
                    }
                });
                if (useDeterminateMode) {
                    fileIngestProgressBar.start((int) estimatedFilesToProcess);
                } else {
                    fileIngestProgressBar.start();
                }
            });
        }
    }

    private void switchFileIngestProgressBarToDeterminate() {
        if (usingNetBeansGUI) {
            SwingUtilities.invokeLater(() -> {
                if (fileIngestProgressBar != null) {
                    fileIngestProgressBar.switchToDeterminate((int) estimatedFilesToProcess);
                }
            });
        }
    }

    private void updateFileProgressBarForFileTaskStarted(String fileName) {
        if (usingNetBeansGUI && !jobCancelled) {
            SwingUtilities.invokeLater(() -> {
                /*
                 * If processedFiles exceeds estimatedFilesToProcess, i.e., the
                 * max work units set for the progress bar, the progress bar
                 * will go into an infinite loop throwing
                 * IllegalArgumentExceptions in the EDT (NetBeans bug). Also, a
                 * check-then-act race condition needs to be avoided here. This
                 * can be done without guarding processedFiles and
                 * estimatedFilesToProcess with the same lock because
                 * estimatedFilesToProcess does not change after it is used to
                 * switch the progress bar to determinate mode.
                 */
                long processedFilesCapture = processedFiles;
                if (processedFilesCapture <= estimatedFilesToProcess) {
                    fileIngestProgressBar.progress(fileName, (int) processedFilesCapture);
                } else {
                    fileIngestProgressBar.progress(fileName, (int) estimatedFilesToProcess);
                }
                filesInProgress.add(fileName);
            });
        }
    }

    private void updateFileProgressBarForFileTaskCompleted(String completedFileName) {
        if (usingNetBeansGUI && !jobCancelled) {
            SwingUtilities.invokeLater(() -> {
                filesInProgress.remove(completedFileName);
                /*
                 * Display the name of another file in progress, or the empty
                 * string if there are none.
                 */
                if (filesInProgress.size() > 0) {
                    fileIngestProgressBar.progress(filesInProgress.get(0));
                } else {
                    fileIngestProgressBar.progress(""); // NON-NLS
                }
            });
        }
    }

    private void startDataArtifactIngestProgressBar() {
        if (usingNetBeansGUI) {
            SwingUtilities.invokeLater(() -> {
                artifactIngestProgressBar = ProgressHandle.createHandle(NbBundle.getMessage(this.getClass(), "IngestJob.progress.dataArtifactIngest.displayName", ingestJob.getDataSource().getName()), new Cancellable() {
                    @Override
                    public boolean cancel() {
                        new Thread(() -> {
                            IngestJobExecutor.this.cancel(IngestJob.CancellationReason.USER_CANCELLED);
                        }).start();
                        return true;
                    }
                });
                artifactIngestProgressBar.start();
                artifactIngestProgressBar.switchToIndeterminate();
            });
        }
    }

    @NbBundle.Messages({
        "# {0} - data source name",
        "IngestJob_progress_analysisResultIngest_displayName=Analyzing analysis results from {0}"
    })
    private void startAnalysisResultIngestProgressBar() {
        if (usingNetBeansGUI) {
            SwingUtilities.invokeLater(() -> {
                resultIngestProgressBar = ProgressHandle.createHandle(Bundle.IngestJob_progress_analysisResultIngest_displayName(ingestJob.getDataSource().getName()), new Cancellable() {
                    @Override
                    public boolean cancel() {
                        new Thread(() -> {
                            IngestJobExecutor.this.cancel(IngestJob.CancellationReason.USER_CANCELLED);
                        }).start();
                        return true;
                    }
                });
                resultIngestProgressBar.start();
                resultIngestProgressBar.switchToIndeterminate();
            });
        }
    }

    private void displayCancellingProgressMessages() {
        if (usingNetBeansGUI) {
            SwingUtilities.invokeLater(() -> {
                if (dataSourceIngestProgressBar != null) {
                    dataSourceIngestProgressBar.setDisplayName(NbBundle.getMessage(getClass(), "IngestJob.progress.dataSourceIngest.initialDisplayName", ingestJob.getDataSource().getName()));
                    dataSourceIngestProgressBar.progress(NbBundle.getMessage(getClass(), "IngestJob.progress.cancelling"));
                }
                if (fileIngestProgressBar != null) {
                    fileIngestProgressBar.setDisplayName(NbBundle.getMessage(getClass(), "IngestJob.progress.fileIngest.displayName", ingestJob.getDataSource().getName()));
                    fileIngestProgressBar.progress(NbBundle.getMessage(getClass(), "IngestJob.progress.cancelling"));
                }
                if (artifactIngestProgressBar != null) {
                    artifactIngestProgressBar.setDisplayName(NbBundle.getMessage(getClass(), "IngestJob.progress.dataArtifactIngest.displayName", ingestJob.getDataSource().getName()));
                    artifactIngestProgressBar.progress(NbBundle.getMessage(getClass(), "IngestJob.progress.cancelling"));
                }
                if (resultIngestProgressBar != null) {
                    resultIngestProgressBar.setDisplayName(Bundle.IngestJob_progress_analysisResultIngest_displayName(ingestJob.getDataSource().getName()));
                    resultIngestProgressBar.progress(NbBundle.getMessage(getClass(), "IngestJob.progress.cancelling"));
                }
            });
        }
    }

    private void finishAllProgressBars() {
        if (usingNetBeansGUI) {
            SwingUtilities.invokeLater(() -> {
                if (dataSourceIngestProgressBar != null) {
                    dataSourceIngestProgressBar.finish();
                    dataSourceIngestProgressBar = null;
                }
 
                if (fileIngestProgressBar != null) {
                    fileIngestProgressBar.finish();
                    fileIngestProgressBar = null;
                }
 
                if (artifactIngestProgressBar != null) {
                    artifactIngestProgressBar.finish();
                    artifactIngestProgressBar = null;
                }
 
                if (resultIngestProgressBar != null) {
                    resultIngestProgressBar.finish();
                    resultIngestProgressBar = null;
                }
            });
        }
    }

    private void logInfoMessage(String message) {
        logger.log(Level.INFO, String.format("%s (data source = %s, data source object ID = %d, job ID = %d)", message, ingestJob.getDataSource().getName(), ingestJob.getDataSource().getId(), getIngestJobId())); //NON-NLS        
    }

    private void logErrorMessage(Level level, String message, Throwable throwable) {
        logger.log(level, String.format("%s (data source = %s, data source object ID = %d, ingest job ID = %d)", message, ingestJob.getDataSource().getName(), ingestJob.getDataSource().getId(), getIngestJobId()), throwable); //NON-NLS
    }

    private void logErrorMessage(Level level, String message) {
        logger.log(level, String.format("%s (data source = %s, data source object ID= %d, ingest job ID %d)", message, ingestJob.getDataSource().getName(), ingestJob.getDataSource().getId(), getIngestJobId())); //NON-NLS
    }

    private void logIngestModuleErrors(List<IngestModuleError> errors) {
        for (IngestModuleError error : errors) {
            logErrorMessage(Level.SEVERE, String.format("%s experienced an error during analysis", error.getModuleDisplayName()), error.getThrowable()); //NON-NLS
        }
    }

    private void logIngestModuleErrors(List<IngestModuleError> errors, AbstractFile file) {
        for (IngestModuleError error : errors) {
            logErrorMessage(Level.SEVERE, String.format("%s experienced an error during analysis while processing file %s (object ID = %d)", error.getModuleDisplayName(), file.getName(), file.getId()), error.getThrowable()); //NON-NLS
        }
    }
    
    Optional<List<FileIngestPipeline>> getCurrentFileIngestPipelines() {
        // Make a local copy in case the tier increments
        int currentModuleTierIndex = moduleTierIndex;
        if (currentModuleTierIndex < ingestModuleTiers.size()) {
            return Optional.of(ingestModuleTiers.get(currentModuleTierIndex).getFileIngestPipelines());
        }
        return Optional.empty();
    }
    
    Optional<DataSourceIngestPipeline> getCurrentDataSourceIngestPipelines() {
        // Make a local copy in case the tier increments
        int currentModuleTierIndex = moduleTierIndex;
        if (currentModuleTierIndex < ingestModuleTiers.size()) {
            return ingestModuleTiers.get(currentModuleTierIndex).getDataSourceIngestPipeline();
        }
        return Optional.empty();
    }
    

    @Messages({
        "IngestJobExecutor_progress_snapshot_currentTier_shutDown_modifier=shut down",
        "# {0} - tier number",
        "# {1} - job state modifer",
        "IngestJobExecutor_progress_snapshot_currentTier=Tier {0} {1}"
    })
    IngestJobProgressSnapshot getIngestJobProgressSnapshot(boolean includeIngestTasksSnapshot) {
        /*
         * Determine whether file ingest is running at the time of this snapshot
         * and determine the earliest file ingest module pipeline start time, if
         * file ingest was started at all.
         */
        boolean fileIngestRunning = false;
        Date fileIngestStartTime = null;
        Optional<List<FileIngestPipeline>> fileIngestPipelines = getCurrentFileIngestPipelines();
        if (!fileIngestPipelines.isPresent()) {
            // If there are no currently running pipelines, use the original set.
            fileIngestPipelines = Optional.of(ingestModuleTiers.get(0).getFileIngestPipelines());
        }
        for (FileIngestPipeline pipeline : fileIngestPipelines.get()) {
            if (pipeline.isRunning()) {
                fileIngestRunning = true;
            }
            Date pipelineStartTime = pipeline.getStartTime();
            if (pipelineStartTime != null && (fileIngestStartTime == null || pipelineStartTime.before(fileIngestStartTime))) {
                fileIngestStartTime = pipelineStartTime;
            }
        }
 
        long processedFilesCount = 0;
        long estimatedFilesToProcessCount = 0;
        long snapShotTime = new Date().getTime();
        IngestTasksScheduler.IngestTasksSnapshot tasksSnapshot = null;
        if (includeIngestTasksSnapshot) {
            processedFilesCount = processedFiles;
            estimatedFilesToProcessCount = estimatedFilesToProcess;
            snapShotTime = new Date().getTime();
            tasksSnapshot = taskScheduler.getTasksSnapshotForJob(getIngestJobId());
        }
        return new IngestJobProgressSnapshot(
                ingestJob.getDataSource().getName(),
                getIngestJobId(),
                createTime,
                Bundle.IngestJobExecutor_progress_snapshot_currentTier(moduleTierIndex, jobState.equals(IngestJobState.PIPELINES_SHUTTING_DOWN) ? Bundle.IngestJobExecutor_progress_snapshot_currentTier_shutDown_modifier() : ""),
                getCurrentDataSourceIngestModule(),
                fileIngestRunning,
                fileIngestStartTime,
                jobCancelled,
                cancellationReason,
                cancelledDataSourceIngestModules,
                processedFilesCount,
                estimatedFilesToProcessCount,
                snapShotTime,
                tasksSnapshot);
    }

    void registerPausedIngestThread(Thread thread) {
        synchronized (threadRegistrationLock) {
            pausedIngestThreads.add(thread);
        }
    }

    void unregisterPausedIngestThread(Thread thread) {
        synchronized (threadRegistrationLock) {
            pausedIngestThreads.remove(thread);
        }
    }
 
}