IngestJobPipeline.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.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashSet;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Level;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Stream;
import javax.annotation.concurrent.GuardedBy;
import javax.swing.JOptionPane;
import org.netbeans.api.progress.ProgressHandle;
import org.openide.util.Cancellable;
import org.openide.util.NbBundle;
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.ingest.IngestTasksScheduler.IngestJobTasksSnapshot;
import org.sleuthkit.datamodel.AbstractFile;
import org.sleuthkit.datamodel.Content;
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.DataArtifact;
import org.sleuthkit.datamodel.DataSource;

final class IngestJobPipeline {

    private static final String AUTOPSY_MODULE_PREFIX = "org.sleuthkit.autopsy";

    private static final Logger logger = Logger.getLogger(IngestJobPipeline.class.getName());

    /*
     * A regular expression for identifying the proxy classes Jython generates
     * for ingest module factories written using Python. For example:
     * org.python.proxies.GPX_Parser_Module$GPXParserFileIngestModuleFactory$14
     */
    private static final Pattern JYTHON_MODULE_REGEX = Pattern.compile("org\\.python\\.proxies\\.(.+?)\\$(.+?)(\\$[0-9]*)?$");

    /*
     * These fields define an ingest pipeline: the parent ingest job, a pipeline
     * ID, the user's ingest job settings, and the data source to be analyzed.
     * Optionally, there is a set of files to be analyzed instead of analyzing
     * all of the files in the data source.
     *
     * The pipeline ID is used to associate the pipeline with the ingest tasks
     * that the ingest task scheduler creates for the ingest job. The ingest job
     * ID cannot be used for this purpose because the parent ingest job may have
     * more than one data source and each data source gets its own pipeline.
     */
    private final IngestJob parentJob;
    private static final AtomicLong nextPipelineId = new AtomicLong(0L);
    private final long pipelineId;
    private final IngestJobSettings settings;
    private DataSource dataSource;
    private final List<AbstractFile> files;

    /*
     * An ingest pipeline runs its ingest modules in stages.
     */
    private static enum Stages {
        /*
         * The pipeline is instantiating ingest modules and loading them into
         * its child ingest module pipelines.
         */
        INITIALIZATION,
        /*
         * This stage is unique to a streaming mode ingest job. The pipeline is
         * running file ingest modules on files streamed to it via
         * addStreamedFiles(). If configured to have data artifact ingest
         * modules, the pipeline is also running them on data artifacts
         * generated by the analysis of the streamed files. This stage ends when
         * the data source is streamed to the pipeline via
         * addStreamedDataSource().
         */
        FIRST_STAGE_STREAMING,
        /*
         * The pipeline is running the following three types of ingest modules:
         * higher priority data source level ingest modules, file ingest
         * modules, and data artifact ingest modules.
         */
        FIRST_STAGE,
        SECOND_STAGE,
        FINALIZATION
    };
    private volatile Stages stage = IngestJobPipeline.Stages.INITIALIZATION;

    /*
     * The stage field is volatile to allow it to be read by multiple threads.
     * This lock is used not to guard the stage field, but to make stage
     * transitions atomic.
     */
    private final Object stageTransitionLock = new Object();

    /*
     * An ingest pipeline has separate data source level ingest module pipelines
     * for the first and second stages. Longer running, lower priority modules
     * belong in the second stage pipeline.
     */
    private DataSourceIngestPipeline firstStageDataSourceIngestPipeline;
    private DataSourceIngestPipeline secondStageDataSourceIngestPipeline;
    private volatile DataSourceIngestPipeline currentDataSourceIngestPipeline;

    /*
     * An ingest pipeline has a collection of identical file ingest module
     * pipelines, one for each file ingest thread in the ingest manager. The
     * file ingest threads take and return file ingest pipeline copies from a
     * blocking queue as they work through the file ingest tasks for the ingest
     * job. Additionally, a fixed list of all of the file ingest module
     * pipelines is used to bypass the blocking queue when cycling through the
     * pipelines to make ingest progress snapshots.
     */
    private final LinkedBlockingQueue<FileIngestPipeline> fileIngestPipelinesQueue = new LinkedBlockingQueue<>();
    private final List<FileIngestPipeline> fileIngestPipelines = new ArrayList<>();

    /*
     * An ingest pipeline has a single data artifact ingest module pipeline.
     */
    private DataArtifactIngestPipeline artifactIngestPipeline;

    /*
     * An ingest pipeline supports cancellation of analysis by individual data
     * source level ingest modules or cancellation of all remaining analysis by
     * all of its ingest modules. Cancellation works by setting flags that are
     * checked by the ingest module pipelines every time they transition from
     * one module to another. Ingest modules are also expected to check these
     * flags (via the ingest job context) and stop processing if they are set.
     * This approach to cancellation means that there can be a variable length
     * delay between a cancellation request and its fulfillment. Analysis
     * already completed at the time that cancellation occurs is not discarded.
     */
    private volatile boolean currentDataSourceIngestModuleCancelled;
    private final List<String> cancelledDataSourceIngestModules = new CopyOnWriteArrayList<>();
    private volatile boolean cancelled;
    private volatile IngestJob.CancellationReason cancellationReason = IngestJob.CancellationReason.NOT_CANCELLED;

    /*
     * An ingest pipeline interacts with the ingest task scheduler to create
     * ingest tasks for analyzing the data source, files and data artifacts that
     * are the subject of the ingest job. The scheduler queues the tasks for the
     * ingest manager's ingest threads. The ingest tasks are the units of work
     * for the ingest pipeline's child ingest module pipelines.
     */
    private static final IngestTasksScheduler taskScheduler = IngestTasksScheduler.getInstance();

    /*
     * If running with a GUI, an ingest pipeline reports analysis progress and
     * allows a user to cancel all or part of the analysis using progress bars
     * in the lower right hand corner of the main application window.
     */
    private final boolean doUI;
    private final Object dataSourceIngestProgressLock = new Object();
    private ProgressHandle dataSourceIngestProgressBar;
    private final Object fileIngestProgressLock = new Object();
    private final List<String> filesInProgress = new ArrayList<>();
    private long estimatedFilesToProcess;
    private long processedFiles;
    private ProgressHandle fileIngestProgressBar;
    private final Object artifactIngestProgressLock = new Object();
    private ProgressHandle artifactIngestProgressBar;

    /*
     * Ingest job details are tracked using this object and are recorded in the
     * case database when the pipeline starts up and shuts down.
     */
    private volatile IngestJobInfo ingestJobInfo;

    private final long createTime;

    /*
     * An ingest pipeline allows ingest module pipelines to register and
     * unregister the ingest thread they are running in when a scheduled ingest
     * pause occurs and the threads are made to sleep. This allows interruption
     * of these threads if the ingest job is canceled.
     */
    private final Object threadRegistrationLock = new Object();
    @GuardedBy("threadRegistrationLock")
    private final Set<Thread> pausedIngestThreads = new HashSet<>();

    IngestJobPipeline(IngestJob parentJob, Content dataSource, IngestJobSettings settings) throws InterruptedException {
        this(parentJob, dataSource, Collections.emptyList(), settings);
    }

    IngestJobPipeline(IngestJob parentJob, Content dataSource, List<AbstractFile> files, IngestJobSettings settings) throws InterruptedException {
        if (!(dataSource instanceof DataSource)) {
            throw new IllegalArgumentException("Passed dataSource that does not implement the DataSource interface"); //NON-NLS
        }
        this.parentJob = parentJob;
        pipelineId = IngestJobPipeline.nextPipelineId.getAndIncrement();
        this.dataSource = (DataSource) dataSource;
        this.files = new ArrayList<>();
        this.files.addAll(files);
        this.settings = settings;
        doUI = RuntimeProperties.runningWithGUI();
        createTime = new Date().getTime();
        stage = Stages.INITIALIZATION;
        createIngestModulePipelines();
    }

    private static void addToIngestPipelineTemplate(final List<IngestModuleTemplate> sortedModules, final Map<String, IngestModuleTemplate> javaModules, final Map<String, IngestModuleTemplate> jythonModules) {
        final List<IngestModuleTemplate> autopsyModules = new ArrayList<>();
        final List<IngestModuleTemplate> thirdPartyModules = new ArrayList<>();
        Stream.concat(javaModules.entrySet().stream(), jythonModules.entrySet().stream()).forEach((templateEntry) -> {
            if (templateEntry.getKey().startsWith(AUTOPSY_MODULE_PREFIX)) {
                autopsyModules.add(templateEntry.getValue());
            } else {
                thirdPartyModules.add(templateEntry.getValue());
            }
        });
        sortedModules.addAll(autopsyModules);
        sortedModules.addAll(thirdPartyModules);
    }

    private static String getModuleNameFromJythonClassName(String className) {
        Matcher m = JYTHON_MODULE_REGEX.matcher(className);
        if (m.find()) {
            return String.format("%s.%s", m.group(1), m.group(2)); //NON-NLS
        } else {
            return null;
        }
    }

    private static void addModuleTemplateToSortingMap(Map<String, IngestModuleTemplate> mapping, Map<String, IngestModuleTemplate> jythonMapping, IngestModuleTemplate template) {
        String className = template.getModuleFactory().getClass().getCanonicalName();
        String jythonName = getModuleNameFromJythonClassName(className);
        if (jythonName != null) {
            jythonMapping.put(jythonName, template);
        } else {
            mapping.put(className, template);
        }
    }

    private void createIngestModulePipelines() throws InterruptedException {
        /*
         * Get the enabled ingest module templates from the ingest job settings.
         */
        List<IngestModuleTemplate> enabledTemplates = settings.getEnabledIngestModuleTemplates();

        Map<String, IngestModuleTemplate> javaDataSourceModuleTemplates = new LinkedHashMap<>();
        Map<String, IngestModuleTemplate> jythonDataSourceModuleTemplates = new LinkedHashMap<>();
        Map<String, IngestModuleTemplate> javaFileModuleTemplates = new LinkedHashMap<>();
        Map<String, IngestModuleTemplate> jythonFileModuleTemplates = new LinkedHashMap<>();
        Map<String, IngestModuleTemplate> javaArtifactModuleTemplates = new LinkedHashMap<>();
        Map<String, IngestModuleTemplate> jythonArtifactModuleTemplates = new LinkedHashMap<>();
        for (IngestModuleTemplate template : enabledTemplates) {
            if (template.isDataSourceIngestModuleTemplate()) {
                addModuleTemplateToSortingMap(javaDataSourceModuleTemplates, jythonDataSourceModuleTemplates, template);
            }
            if (template.isFileIngestModuleTemplate()) {
                addModuleTemplateToSortingMap(javaFileModuleTemplates, jythonFileModuleTemplates, template);
            }
//            if (template.isDataArtifactIngestModuleTemplate()) {
//                addModuleTemplateToSortingMap(javaArtifactModuleTemplates, jythonArtifactModuleTemplates, template);
//            }
        }

        IngestPipelinesConfiguration pipelineConfig = IngestPipelinesConfiguration.getInstance();
        List<IngestModuleTemplate> firstStageDataSourcePipelineTemplate = createIngestPipelineTemplate(javaDataSourceModuleTemplates, jythonDataSourceModuleTemplates, pipelineConfig.getStageOneDataSourceIngestPipelineConfig());
        List<IngestModuleTemplate> secondStageDataSourcePipelineTemplate = createIngestPipelineTemplate(javaDataSourceModuleTemplates, jythonDataSourceModuleTemplates, pipelineConfig.getStageTwoDataSourceIngestPipelineConfig());
        List<IngestModuleTemplate> filePipelineTemplate = createIngestPipelineTemplate(javaFileModuleTemplates, jythonFileModuleTemplates, pipelineConfig.getFileIngestPipelineConfig());
        List<IngestModuleTemplate> artifactPipelineTemplate = new ArrayList<>();

        addToIngestPipelineTemplate(firstStageDataSourcePipelineTemplate, javaDataSourceModuleTemplates, jythonDataSourceModuleTemplates);
        addToIngestPipelineTemplate(filePipelineTemplate, javaFileModuleTemplates, jythonFileModuleTemplates);
        addToIngestPipelineTemplate(artifactPipelineTemplate, javaArtifactModuleTemplates, jythonArtifactModuleTemplates);

        firstStageDataSourceIngestPipeline = new DataSourceIngestPipeline(this, firstStageDataSourcePipelineTemplate);
        secondStageDataSourceIngestPipeline = new DataSourceIngestPipeline(this, secondStageDataSourcePipelineTemplate);
        int numberOfFileIngestThreads = IngestManager.getInstance().getNumberOfFileIngestThreads();
        for (int i = 0; i < numberOfFileIngestThreads; ++i) {
            FileIngestPipeline pipeline = new FileIngestPipeline(this, filePipelineTemplate);
            fileIngestPipelinesQueue.put(pipeline);
            fileIngestPipelines.add(pipeline);
        }
        artifactIngestPipeline = new DataArtifactIngestPipeline(this, artifactPipelineTemplate);
    }

    private static List<IngestModuleTemplate> createIngestPipelineTemplate(Map<String, IngestModuleTemplate> javaIngestModuleTemplates, Map<String, IngestModuleTemplate> jythonIngestModuleTemplates, List<String> pipelineConfig) {
        List<IngestModuleTemplate> pipelineTemplate = new ArrayList<>();
        for (String moduleClassName : pipelineConfig) {
            if (javaIngestModuleTemplates.containsKey(moduleClassName)) {
                pipelineTemplate.add(javaIngestModuleTemplates.remove(moduleClassName));
            } else if (jythonIngestModuleTemplates.containsKey(moduleClassName)) {
                pipelineTemplate.add(jythonIngestModuleTemplates.remove(moduleClassName));
            }
        }
        return pipelineTemplate;
    }

    long getId() {
        return pipelineId;
    }

    String getExecutionContext() {
        return settings.getExecutionContext();
    }

    DataSource getDataSource() {
        return dataSource;
    }

    boolean shouldProcessUnallocatedSpace() {
        return settings.getProcessUnallocatedSpace();
    }

    FilesSet getFileIngestFilter() {
        return settings.getFileFilter();
    }

    boolean hasIngestModules() {
        return hasFileIngestModules()
                || hasFirstStageDataSourceIngestModules()
                || hasSecondStageDataSourceIngestModules()
                || hasDataArtifactIngestModules();
    }

    boolean hasDataSourceIngestModules() {
        if (stage == Stages.SECOND_STAGE) {
            return hasSecondStageDataSourceIngestModules();
        } else {
            return hasFirstStageDataSourceIngestModules();
        }
    }

    private boolean hasFirstStageDataSourceIngestModules() {
        return (firstStageDataSourceIngestPipeline.isEmpty() == false);
    }

    private boolean hasSecondStageDataSourceIngestModules() {
        return (secondStageDataSourceIngestPipeline.isEmpty() == false);
    }

    boolean hasFileIngestModules() {
        if (!fileIngestPipelines.isEmpty()) {
            return !fileIngestPipelines.get(0).isEmpty();
        }
        return false;
    }

    boolean hasDataArtifactIngestModules() {
        return (artifactIngestPipeline.isEmpty() == false);
    }

    List<IngestModuleError> startUp() {
        List<IngestModuleError> errors = startUpIngestModulePipelines();
        if (errors.isEmpty()) {
            recordIngestJobStartUpInfo();
            if (hasFirstStageDataSourceIngestModules() || hasFileIngestModules() || hasDataArtifactIngestModules()) {
                if (parentJob.getIngestMode() == IngestJob.Mode.STREAMING) {
                    startFirstStageInStreamingMode();
                } else {
                    startFirstStageInBatchMode();
                }
            } else if (hasSecondStageDataSourceIngestModules()) {
                startSecondStage();
            }
        }
        return errors;
    }

    void recordIngestJobStartUpInfo() {
        try {
            SleuthkitCase caseDb = Case.getCurrentCase().getSleuthkitCase();
            List<IngestModuleInfo> ingestModuleInfoList = new ArrayList<>();
            for (IngestModuleTemplate module : settings.getEnabledIngestModuleTemplates()) {
                IngestModuleType moduleType = getIngestModuleTemplateType(module);
                IngestModuleInfo moduleInfo = caseDb.addIngestModule(module.getModuleName(), FactoryClassNameNormalizer.normalize(module.getModuleFactory().getClass().getCanonicalName()), moduleType, module.getModuleFactory().getModuleVersionNumber());
                ingestModuleInfoList.add(moduleInfo);
            }
            ingestJobInfo = caseDb.addIngestJob(dataSource, 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 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 List<IngestModuleError> startUpIngestModulePipelines() {
        List<IngestModuleError> errors = new ArrayList<>();
        errors.addAll(startUpIngestModulePipeline(firstStageDataSourceIngestPipeline));
        errors.addAll(startUpIngestModulePipeline(secondStageDataSourceIngestPipeline));
        for (FileIngestPipeline pipeline : fileIngestPipelines) {
            List<IngestModuleError> filePipelineErrors = startUpIngestModulePipeline(pipeline);
            if (!filePipelineErrors.isEmpty()) {
                /*
                 * If one file pipeline copy can't start up, assume that none of
                 * them will be able to start up for the same reason.
                 */
                errors.addAll(filePipelineErrors);
                break;
            }
        }
        errors.addAll(startUpIngestModulePipeline(artifactIngestPipeline));
        return errors;
    }

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

    private void startFirstStageInBatchMode() {
        synchronized (stageTransitionLock) {
            logInfoMessage("Starting first stage analysis in batch mode"); //NON-NLS        
            stage = Stages.FIRST_STAGE;

            /*
             * Do a count of the files the data source processor has added to
             * the case database. This estimate will be used for ingest progress
             * snapshots and for the file ingest progress bar if running with a
             * GUI.
             */
            if (hasFileIngestModules()) {
                long filesToProcess;
                if (files.isEmpty()) {
                    filesToProcess = dataSource.accept(new GetFilesCountVisitor());
                } else {
                    filesToProcess = files.size();
                }
                synchronized (fileIngestProgressLock) {
                    estimatedFilesToProcess = filesToProcess;
                }
            }

            /*
             * If running with a GUI, start ingest progress bars in the lower
             * right hand corner of the main application window.
             */
            if (doUI) {
                if (hasFileIngestModules()) {
                    startFileIngestProgressBar();
                }
                if (hasFirstStageDataSourceIngestModules()) {
                    startDataSourceIngestProgressBar();
                }
                if (hasDataArtifactIngestModules()) {
                    startArtifactIngestProgressBar();
                }
            }

            /*
             * Make the first stage data source level ingest pipeline the
             * current data source level pipeline.
             */
            currentDataSourceIngestPipeline = firstStageDataSourceIngestPipeline;

            /*
             * Schedule the first stage ingest tasks and then immediately check
             * for stage completion. This is necessary because it is possible
             * that zero tasks will actually make it to task execution due to
             * the file filter or other ingest job settings. In that case, there
             * will never be a stage completion check in an ingest thread
             * executing an ingest task, so such a job would run forever without
             * a check here.
             */
            if (!files.isEmpty() && hasFileIngestModules()) {
                taskScheduler.scheduleFileIngestTasks(this, files);
            } else if (hasFirstStageDataSourceIngestModules() || hasFileIngestModules() || hasDataArtifactIngestModules()) {
                taskScheduler.scheduleIngestTasks(this);
            }
            checkForStageCompleted();
        }
    }

    private void startFirstStageInStreamingMode() {
        synchronized (stageTransitionLock) {
            logInfoMessage("Starting first stage analysis in streaming mode"); //NON-NLS
            stage = Stages.FIRST_STAGE_STREAMING;

            if (doUI) {
                /*
                 * If running with a GUI, start ingest progress bars in the
                 * lower right hand corner of the main application window.
                 */
                if (hasFileIngestModules()) {
                    /*
                     * Note that because estimated files remaining to process
                     * still has its initial value of zero, the progress bar
                     * will start in the "indeterminate" state. An estimate of
                     * the files to process can be computed in
                     */
                    startFileIngestProgressBar();
                }
                if (hasDataArtifactIngestModules()) {
                    startArtifactIngestProgressBar();
                }
            }

            if (hasDataArtifactIngestModules()) {
                /*
                 * Schedule artifact ingest tasks for any artifacts currently in
                 * the case database. This needs to be done before any files or
                 * the data source are streamed in to avoid analyzing data
                 * artifacts added to the case database by the data source level
                 * or file level ingest tasks.
                 */
                taskScheduler.scheduleDataArtifactIngestTasks(this);
            }
        }
    }

    void addStreamedDataSource() {
        synchronized (stageTransitionLock) {
            logInfoMessage("Starting full first stage analysis in streaming mode"); //NON-NLS
            stage = IngestJobPipeline.Stages.FIRST_STAGE;
            currentDataSourceIngestPipeline = firstStageDataSourceIngestPipeline;

            if (hasFileIngestModules()) {
                /*
                 * Do a count of the files the data source processor has added
                 * to the case database. This estimate will be used for ingest
                 * progress snapshots and for the file ingest progress bar if
                 * running with a GUI.
                 */
                long filesToProcess = dataSource.accept(new GetFilesCountVisitor());
                synchronized (fileIngestProgressLock) {
                    estimatedFilesToProcess = filesToProcess;
                    if (doUI && fileIngestProgressBar != null) {
                        fileIngestProgressBar.switchToDeterminate((int) estimatedFilesToProcess);
                    }
                }
            }

            if (doUI) {
                if (hasFirstStageDataSourceIngestModules()) {
                    startDataSourceIngestProgressBar();
                }
            }

            currentDataSourceIngestPipeline = firstStageDataSourceIngestPipeline;
            if (hasFirstStageDataSourceIngestModules()) {
                IngestJobPipeline.taskScheduler.scheduleDataSourceIngestTask(this);
            } 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
                 * executed, there will never be a stage completion check in an
                 * ingest thread executing an ingest task, so such a job would
                 * run forever without a check here.
                 */
                checkForStageCompleted();
            }
        }
    }

    private void startSecondStage() {
        synchronized (stageTransitionLock) {
            if (hasSecondStageDataSourceIngestModules()) {
                logInfoMessage(String.format("Starting second stage ingest task pipelines for %s (objID=%d, jobID=%d)", dataSource.getName(), dataSource.getId(), parentJob.getId())); //NON-NLS
                stage = IngestJobPipeline.Stages.SECOND_STAGE;

                if (doUI) {
                    startDataSourceIngestProgressBar();
                }

                currentDataSourceIngestPipeline = secondStageDataSourceIngestPipeline;
                taskScheduler.scheduleDataSourceIngestTask(this);
            }
        }
    }

    private void startArtifactIngestProgressBar() {
        if (doUI) {
            synchronized (artifactIngestProgressLock) {
                String displayName = NbBundle.getMessage(this.getClass(), "IngestJob.progress.dataArtifactIngest.displayName", this.dataSource.getName());
                artifactIngestProgressBar = ProgressHandle.createHandle(displayName, new Cancellable() {
                    @Override
                    public boolean cancel() {
                        IngestJobPipeline.this.cancel(IngestJob.CancellationReason.USER_CANCELLED);
                        return true;
                    }
                });
                artifactIngestProgressBar.start();
                artifactIngestProgressBar.switchToIndeterminate();
            }
        }
    }

    private void startDataSourceIngestProgressBar() {
        if (this.doUI) {
            synchronized (this.dataSourceIngestProgressLock) {
                String displayName = NbBundle.getMessage(this.getClass(),
                        "IngestJob.progress.dataSourceIngest.initialDisplayName",
                        this.dataSource.getName());
                this.dataSourceIngestProgressBar = ProgressHandle.createHandle(displayName, new Cancellable() {
                    @Override
                    public boolean cancel() {
                        // If this method is called, the user has already pressed 
                        // the cancel button on the progress bar and the OK button
                        // of a cancelation confirmation dialog supplied by 
                        // NetBeans. What remains to be done is to 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(IngestJobPipeline.this.getClass(), "IngestJob.cancellationDialog.title");
                        JOptionPane.showConfirmDialog(WindowManager.getDefault().getMainWindow(), panel, dialogTitle, JOptionPane.OK_OPTION, JOptionPane.PLAIN_MESSAGE);
                        if (panel.cancelAllDataSourceIngestModules()) {
                            IngestJobPipeline.this.cancel(IngestJob.CancellationReason.USER_CANCELLED);
                        } else {
                            IngestJobPipeline.this.cancelCurrentDataSourceIngestModule();
                        }
                        return true;
                    }
                });
                this.dataSourceIngestProgressBar.start();
                this.dataSourceIngestProgressBar.switchToIndeterminate();
            }
        }
    }

    private void startFileIngestProgressBar() {
        if (this.doUI) {
            synchronized (this.fileIngestProgressLock) {
                String displayName = NbBundle.getMessage(this.getClass(),
                        "IngestJob.progress.fileIngest.displayName",
                        this.dataSource.getName());
                this.fileIngestProgressBar = ProgressHandle.createHandle(displayName, new Cancellable() {
                    @Override
                    public boolean cancel() {
                        // If this method is called, the user has already pressed 
                        // the cancel button on the progress bar and the OK button
                        // of a cancelation confirmation dialog supplied by 
                        // NetBeans. 
                        IngestJobPipeline.this.cancel(IngestJob.CancellationReason.USER_CANCELLED);
                        return true;
                    }
                });
                this.fileIngestProgressBar.start();
                this.fileIngestProgressBar.switchToDeterminate((int) this.estimatedFilesToProcess);
            }
        }
    }

    private void checkForStageCompleted() {
        synchronized (stageTransitionLock) {
            if (stage == Stages.FIRST_STAGE_STREAMING) {
                return;
            }
            if (taskScheduler.currentTasksAreCompleted(this)) {
                switch (stage) {
                    case FIRST_STAGE:
                        finishFirstStage();
                        break;
                    case SECOND_STAGE:
                        shutDown();
                        break;
                }
            }
        }
    }

    private void finishFirstStage() {
        synchronized (stageTransitionLock) {
            logInfoMessage("Finished first stage analysis"); //NON-NLS        

            shutDownIngestModulePipeline(currentDataSourceIngestPipeline);
            while (!fileIngestPipelinesQueue.isEmpty()) {
                FileIngestPipeline pipeline = fileIngestPipelinesQueue.poll();
                shutDownIngestModulePipeline(pipeline);
            }

            if (doUI) {
                synchronized (dataSourceIngestProgressLock) {
                    if (dataSourceIngestProgressBar != null) {
                        dataSourceIngestProgressBar.finish();
                        dataSourceIngestProgressBar = null;
                    }
                }

                synchronized (fileIngestProgressLock) {
                    if (fileIngestProgressBar != null) {
                        fileIngestProgressBar.finish();
                        fileIngestProgressBar = null;
                    }
                }
            }

            if (!cancelled && hasSecondStageDataSourceIngestModules()) {
                startSecondStage();
            } else {
                shutDown();
            }
        }
    }

    private void shutDown() {
        synchronized (stageTransitionLock) {
            logInfoMessage("Finished all tasks"); //NON-NLS        
            stage = IngestJobPipeline.Stages.FINALIZATION;

            shutDownIngestModulePipeline(currentDataSourceIngestPipeline);
            shutDownIngestModulePipeline(artifactIngestPipeline);

            if (doUI) {
                synchronized (dataSourceIngestProgressLock) {
                    if (dataSourceIngestProgressBar != null) {
                        dataSourceIngestProgressBar.finish();
                        dataSourceIngestProgressBar = null;
                    }
                }

                synchronized (fileIngestProgressLock) {
                    if (fileIngestProgressBar != null) {
                        fileIngestProgressBar.finish();
                        fileIngestProgressBar = null;
                    }
                }

                synchronized (artifactIngestProgressLock) {
                    if (artifactIngestProgressBar != null) {
                        artifactIngestProgressBar.finish();
                        artifactIngestProgressBar = null;
                    }
                }
            }

            if (ingestJobInfo != null) {
                if (cancelled) {
                    try {
                        ingestJobInfo.setIngestJobStatus(IngestJobStatusType.CANCELLED);
                    } catch (TskCoreException ex) {
                        logErrorMessage(Level.WARNING, "Failed to update ingest job status in case database", ex);
                    }
                } else {
                    try {
                        ingestJobInfo.setIngestJobStatus(IngestJobStatusType.COMPLETED);
                    } catch (TskCoreException ex) {
                        logErrorMessage(Level.WARNING, "Failed to update ingest job status in case database", ex);
                    }
                }
                try {
                    ingestJobInfo.setEndDateTime(new Date());
                } catch (TskCoreException ex) {
                    logErrorMessage(Level.WARNING, "Failed to set job end date in case database", ex);
                }
            }
        }

        parentJob.notifyIngestPipelineShutDown(this);
    }

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

    void execute(DataSourceIngestTask task) {
        try {
            if (!isCancelled()) {
                List<IngestModuleError> errors = new ArrayList<>();
                errors.addAll(currentDataSourceIngestPipeline.executeTask(task));
                if (!errors.isEmpty()) {
                    logIngestModuleErrors(errors);
                }
            }
        } finally {
            taskScheduler.notifyTaskCompleted(task);
            checkForStageCompleted();
        }
    }

    void execute(FileIngestTask task) {
        try {
            if (!isCancelled()) {
                FileIngestPipeline pipeline = fileIngestPipelinesQueue.take();
                if (!pipeline.isEmpty()) {
                    /*
                     * Get the file from the task. If the file was "streamed,"
                     * 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);
                        fileIngestPipelinesQueue.put(pipeline);
                        return;
                    }

                    synchronized (fileIngestProgressLock) {
                        ++processedFiles;
                        if (doUI) {
                            if (processedFiles <= estimatedFilesToProcess) {
                                fileIngestProgressBar.progress(file.getName(), (int) processedFiles);
                            } else {
                                fileIngestProgressBar.progress(file.getName(), (int) estimatedFilesToProcess);
                            }
                            filesInProgress.add(file.getName());
                        }
                    }

                    List<IngestModuleError> errors = new ArrayList<>();
                    errors.addAll(pipeline.executeTask(task));
                    if (!errors.isEmpty()) {
                        logIngestModuleErrors(errors, file);
                    }

                    if (doUI && !cancelled) {
                        synchronized (fileIngestProgressLock) {
                            filesInProgress.remove(file.getName());
                            if (filesInProgress.size() > 0) {
                                fileIngestProgressBar.progress(filesInProgress.get(0));
                            } else {
                                fileIngestProgressBar.progress("");
                            }
                        }
                    }
                }
                fileIngestPipelinesQueue.put(pipeline);
            }
        } catch (InterruptedException ex) {
            logger.log(Level.SEVERE, String.format("Unexpected interrupt of file ingest thread during execution of file ingest job (file obj ID = %d)", task.getFileId()), ex);
            Thread.currentThread().interrupt(); // Reset thread interrupted flag
        } finally {
            taskScheduler.notifyTaskCompleted(task);
            checkForStageCompleted();
        }
    }

    void execute(DataArtifactIngestTask task) {
        try {
            if (!isCancelled() && !artifactIngestPipeline.isEmpty()) {
                List<IngestModuleError> errors = new ArrayList<>();
                errors.addAll(artifactIngestPipeline.executeTask(task));
                if (!errors.isEmpty()) {
                    logIngestModuleErrors(errors);
                }
            }
        } finally {
            taskScheduler.notifyTaskCompleted(task);
            checkForStageCompleted();
        }
    }

    void addStreamedFiles(List<Long> fileObjIds) {
        if (hasFileIngestModules()) {
            if (stage.equals(Stages.FIRST_STAGE_STREAMING)) {
                IngestJobPipeline.taskScheduler.scheduleStreamedFileIngestTasks(this, fileObjIds);
            } else {
                logErrorMessage(Level.SEVERE, "Adding streaming files to job during stage " + stage.toString() + " not supported");
            }
        }
    }

    void addFiles(List<AbstractFile> files) {
        if (stage.equals(Stages.FIRST_STAGE_STREAMING)
                || stage.equals(Stages.FIRST_STAGE)) {
            taskScheduler.fastTrackFileIngestTasks(this, files);
        } else {
            logErrorMessage(Level.SEVERE, "Adding streaming files to job during stage " + stage.toString() + " not supported");
        }

        checkForStageCompleted();
    }

    void addDataArtifacts(List<DataArtifact> artifacts) {
        List<DataArtifact> artifactsToAnalyze = new ArrayList<>(artifacts);
        if (stage.equals(Stages.FIRST_STAGE_STREAMING)
                || stage.equals(Stages.FIRST_STAGE)
                || stage.equals(Stages.SECOND_STAGE)) {
            taskScheduler.scheduleDataArtifactIngestTasks(this, artifactsToAnalyze);
        } else {
            logErrorMessage(Level.SEVERE, "Adding streaming files to job during stage " + stage.toString() + " not supported");
        }

        checkForStageCompleted();
    }

    void updateDataSourceIngestProgressBarDisplayName(String displayName) {
        if (this.doUI && !this.cancelled) {
            synchronized (this.dataSourceIngestProgressLock) {
                this.dataSourceIngestProgressBar.setDisplayName(displayName);
            }
        }
    }

    void switchDataSourceIngestProgressBarToDeterminate(int workUnits) {
        if (this.doUI && !this.cancelled) {
            synchronized (this.dataSourceIngestProgressLock) {
                if (null != this.dataSourceIngestProgressBar) {
                    this.dataSourceIngestProgressBar.switchToDeterminate(workUnits);
                }
            }
        }
    }

    void switchDataSourceIngestProgressBarToIndeterminate() {
        if (this.doUI && !this.cancelled) {
            synchronized (this.dataSourceIngestProgressLock) {
                if (null != this.dataSourceIngestProgressBar) {
                    this.dataSourceIngestProgressBar.switchToIndeterminate();
                }
            }
        }
    }

    void advanceDataSourceIngestProgressBar(int workUnits) {
        if (doUI && !cancelled) {
            synchronized (dataSourceIngestProgressLock) {
                if (null != dataSourceIngestProgressBar) {
                    dataSourceIngestProgressBar.progress("", workUnits);
                }
            }
        }
    }

    void advanceDataSourceIngestProgressBar(String currentTask) {
        if (doUI && !cancelled) {
            synchronized (dataSourceIngestProgressLock) {
                if (null != dataSourceIngestProgressBar) {
                    dataSourceIngestProgressBar.progress(currentTask);
                }
            }
        }
    }

    void advanceDataSourceIngestProgressBar(String currentTask, int workUnits) {
        if (this.doUI && !this.cancelled) {
            synchronized (this.fileIngestProgressLock) {
                this.dataSourceIngestProgressBar.progress(currentTask, workUnits);
            }
        }
    }

    boolean currentDataSourceIngestModuleIsCancelled() {
        return this.currentDataSourceIngestModuleCancelled;
    }

    void currentDataSourceIngestModuleCancellationCompleted(String moduleDisplayName) {
        this.currentDataSourceIngestModuleCancelled = false;
        this.cancelledDataSourceIngestModules.add(moduleDisplayName);

        if (this.doUI) {
            synchronized (this.dataSourceIngestProgressLock) {
                this.dataSourceIngestProgressBar.finish();
                this.dataSourceIngestProgressBar = null;
                this.startDataSourceIngestProgressBar();
            }
        }
    }

    DataSourceIngestPipeline.DataSourcePipelineModule getCurrentDataSourceIngestModule() {
        if (currentDataSourceIngestPipeline != null) {
            return (DataSourceIngestPipeline.DataSourcePipelineModule) currentDataSourceIngestPipeline.getCurrentlyRunningModule();
        } else {
            return null;
        }
    }

    void cancelCurrentDataSourceIngestModule() {
        this.currentDataSourceIngestModuleCancelled = true;
    }

    void cancel(IngestJob.CancellationReason reason) {
        this.cancelled = true;
        this.cancellationReason = reason;
        IngestJobPipeline.taskScheduler.cancelPendingFileTasksForIngestJob(this);

        if (this.doUI) {
            synchronized (this.dataSourceIngestProgressLock) {
                if (null != dataSourceIngestProgressBar) {
                    dataSourceIngestProgressBar.setDisplayName(NbBundle.getMessage(this.getClass(), "IngestJob.progress.dataSourceIngest.initialDisplayName", this.dataSource.getName()));
                    dataSourceIngestProgressBar.progress(NbBundle.getMessage(this.getClass(), "IngestJob.progress.cancelling"));
                }
            }

            synchronized (this.fileIngestProgressLock) {
                if (null != this.fileIngestProgressBar) {
                    this.fileIngestProgressBar.setDisplayName(NbBundle.getMessage(this.getClass(), "IngestJob.progress.fileIngest.displayName", this.dataSource.getName()));
                    this.fileIngestProgressBar.progress(NbBundle.getMessage(this.getClass(), "IngestJob.progress.cancelling"));
                }
            }
        }

        synchronized (threadRegistrationLock) {
            for (Thread thread : pausedIngestThreads) {
                thread.interrupt();
            }
            pausedIngestThreads.clear();
        }

        // If a data source had no tasks in progress it may now be complete.
        checkForStageCompleted();
    }

    boolean isCancelled() {
        return this.cancelled;
    }

    IngestJob.CancellationReason getCancellationReason() {
        return this.cancellationReason;
    }

    private void logInfoMessage(String message) {
        logger.log(Level.INFO, String.format("%s (data source = %s, objId = %d, pipeline id = %d, ingest job id = %d)", message, this.dataSource.getName(), this.dataSource.getId(), pipelineId, ingestJobInfo.getIngestJobId())); //NON-NLS        
    }

    private void logErrorMessage(Level level, String message, Throwable throwable) {
        logger.log(level, String.format("%s (data source = %s, objId = %d, pipeline id = %d, ingest job id = %d)", message, this.dataSource.getName(), this.dataSource.getId(), pipelineId, ingestJobInfo.getIngestJobId()), throwable); //NON-NLS
    }

    private void logErrorMessage(Level level, String message) {
        logger.log(level, String.format("%s (data source = %s, objId = %d, pipeline id = %d, ingest job id %d)", message, this.dataSource.getName(), this.dataSource.getId(), pipelineId, ingestJobInfo.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
        }
    }

    Snapshot getSnapshot(boolean getIngestTasksSnapshot) {
        boolean fileIngestRunning = false;
        Date fileIngestStartTime = null;
        for (FileIngestPipeline pipeline : this.fileIngestPipelines) {
            if (pipeline.isRunning()) {
                fileIngestRunning = true;
            }
            Date pipelineStartTime = pipeline.getStartTime();
            if (null != pipelineStartTime && (null == fileIngestStartTime || pipelineStartTime.before(fileIngestStartTime))) {
                fileIngestStartTime = pipelineStartTime;
            }
        }

        long processedFilesCount = 0;
        long estimatedFilesToProcessCount = 0;
        long snapShotTime = new Date().getTime();
        IngestJobTasksSnapshot tasksSnapshot = null;
        if (getIngestTasksSnapshot) {
            synchronized (fileIngestProgressLock) {
                processedFilesCount = this.processedFiles;
                estimatedFilesToProcessCount = this.estimatedFilesToProcess;
                snapShotTime = new Date().getTime();
            }
            tasksSnapshot = taskScheduler.getTasksSnapshotForJob(pipelineId);
        }

        return new Snapshot(dataSource.getName(),
                pipelineId, createTime,
                getCurrentDataSourceIngestModule(),
                fileIngestRunning, fileIngestStartTime,
                cancelled, cancellationReason, cancelledDataSourceIngestModules,
                processedFilesCount, estimatedFilesToProcessCount, snapShotTime, tasksSnapshot);
    }

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

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

}