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.casemodule.NoCurrentCaseException;
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;

final class IngestJobPipeline {

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

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

    // to match something like: "org.python.proxies.GPX_Parser_Module$GPXParserFileIngestModuleFactory$14"
    private static final Pattern JYTHON_REGEX = Pattern.compile("org\\.python\\.proxies\\.(.+?)\\$(.+?)(\\$[0-9]*)?$");

    private final IngestJob parentJob;
    private static final AtomicLong nextJobId = new AtomicLong(0L);
    private final long id;
    private final IngestJobSettings settings;
    private Content dataSource = null;
    private final IngestJob.Mode ingestMode;
    private final List<AbstractFile> files = new ArrayList<>();

    private static enum Stages {

        INITIALIZATION,
        FIRST_STAGE_FILES_ONLY,
        FIRST_STAGE_FILES_AND_DATASOURCE,
        SECOND_STAGE,
        FINALIZATION
    };
    private volatile Stages stage = IngestJobPipeline.Stages.INITIALIZATION;
    private final Object stageCompletionCheckLock = new Object();

    private final Object dataSourceIngestPipelineLock = new Object();
    private DataSourceIngestPipeline firstStageDataSourceIngestPipeline;
    private DataSourceIngestPipeline secondStageDataSourceIngestPipeline;
    private DataSourceIngestPipeline currentDataSourceIngestPipeline;

    private final LinkedBlockingQueue<FileIngestPipeline> fileIngestPipelinesQueue = new LinkedBlockingQueue<>();
    private final List<FileIngestPipeline> fileIngestPipelines = new ArrayList<>();

    private volatile boolean currentDataSourceIngestModuleCancelled;
    private final List<String> cancelledDataSourceIngestModules = new CopyOnWriteArrayList<>();
    private volatile boolean cancelled;
    private volatile IngestJob.CancellationReason cancellationReason = IngestJob.CancellationReason.NOT_CANCELLED;

    private static final IngestTasksScheduler taskScheduler = IngestTasksScheduler.getInstance();

    private final boolean doUI;

    private final Object dataSourceIngestProgressLock = new Object();
    private ProgressHandle dataSourceIngestProgress;

    private final Object fileIngestProgressLock = new Object();
    private final List<String> filesInProgress = new ArrayList<>();
    private long estimatedFilesToProcess;
    private long processedFiles;
    private ProgressHandle fileIngestProgress;
    private String currentFileIngestModule = "";
    private String currentFileIngestTask = "";
    private final List<IngestModuleInfo> ingestModules = new ArrayList<>();
    private volatile IngestJobInfo ingestJob;

    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) {
        this(parentJob, dataSource, Collections.emptyList(), settings);
    }

    IngestJobPipeline(IngestJob parentJob, Content dataSource, List<AbstractFile> files, IngestJobSettings settings) {
        this.parentJob = parentJob;
        this.id = IngestJobPipeline.nextJobId.getAndIncrement();
        this.dataSource = dataSource;
        this.files.addAll(files);
        this.ingestMode = parentJob.getIngestMode();
        this.settings = settings;
        this.doUI = RuntimeProperties.runningWithGUI();
        this.createTime = new Date().getTime();
        this.stage = Stages.INITIALIZATION;
        this.createIngestPipelines();
    }

    private static void addOrdered(final List<IngestModuleTemplate> dest,
            final Map<String, IngestModuleTemplate> src, final Map<String, IngestModuleTemplate> jythonSrc) {

        final List<IngestModuleTemplate> autopsyModules = new ArrayList<>();
        final List<IngestModuleTemplate> thirdPartyModules = new ArrayList<>();

        Stream.concat(src.entrySet().stream(), jythonSrc.entrySet().stream()).forEach((templateEntry) -> {
            if (templateEntry.getKey().startsWith(AUTOPSY_MODULE_PREFIX)) {
                autopsyModules.add(templateEntry.getValue());
            } else {
                thirdPartyModules.add(templateEntry.getValue());
            }
        });

        dest.addAll(autopsyModules);
        dest.addAll(thirdPartyModules);
    }

    private static String getJythonName(String canonicalName) {
        Matcher m = JYTHON_REGEX.matcher(canonicalName);
        if (m.find()) {
            return String.format("%s.%s", m.group(1), m.group(2));
        } else {
            return null;
        }
    }

    private static void addModule(Map<String, IngestModuleTemplate> mapping,
            Map<String, IngestModuleTemplate> jythonMapping, IngestModuleTemplate template) {

        String className = template.getModuleFactory().getClass().getCanonicalName();
        String jythonName = getJythonName(className);
        if (jythonName != null) {
            jythonMapping.put(jythonName, template);
        } else {
            mapping.put(className, template);
        }
    }

    private void createIngestPipelines() {
        List<IngestModuleTemplate> ingestModuleTemplates = this.settings.getEnabledIngestModuleTemplates();

        Map<String, IngestModuleTemplate> dataSourceModuleTemplates = new LinkedHashMap<>();
        Map<String, IngestModuleTemplate> fileModuleTemplates = new LinkedHashMap<>();

        // mappings for jython modules.  These mappings are only used to determine modules in the pipelineconfig.xml.
        Map<String, IngestModuleTemplate> jythonDataSourceModuleTemplates = new LinkedHashMap<>();
        Map<String, IngestModuleTemplate> jythonFileModuleTemplates = new LinkedHashMap<>();

        for (IngestModuleTemplate template : ingestModuleTemplates) {
            if (template.isDataSourceIngestModuleTemplate()) {
                addModule(dataSourceModuleTemplates, jythonDataSourceModuleTemplates, template);
            }
            if (template.isFileIngestModuleTemplate()) {
                addModule(fileModuleTemplates, jythonFileModuleTemplates, template);
            }
        }

        IngestPipelinesConfiguration pipelineConfigs = IngestPipelinesConfiguration.getInstance();
        List<IngestModuleTemplate> firstStageDataSourceModuleTemplates = IngestJobPipeline.getConfiguredIngestModuleTemplates(
                dataSourceModuleTemplates, jythonDataSourceModuleTemplates, pipelineConfigs.getStageOneDataSourceIngestPipelineConfig());

        List<IngestModuleTemplate> fileIngestModuleTemplates = IngestJobPipeline.getConfiguredIngestModuleTemplates(
                fileModuleTemplates, jythonFileModuleTemplates, pipelineConfigs.getFileIngestPipelineConfig());

        List<IngestModuleTemplate> secondStageDataSourceModuleTemplates = IngestJobPipeline.getConfiguredIngestModuleTemplates(
                dataSourceModuleTemplates, null, pipelineConfigs.getStageTwoDataSourceIngestPipelineConfig());

        addOrdered(firstStageDataSourceModuleTemplates, dataSourceModuleTemplates, jythonDataSourceModuleTemplates);
        addOrdered(fileIngestModuleTemplates, fileModuleTemplates, jythonFileModuleTemplates);

        this.firstStageDataSourceIngestPipeline = new DataSourceIngestPipeline(this, firstStageDataSourceModuleTemplates);
        this.secondStageDataSourceIngestPipeline = new DataSourceIngestPipeline(this, secondStageDataSourceModuleTemplates);

        try {
            int numberOfFileIngestThreads = IngestManager.getInstance().getNumberOfFileIngestThreads();
            for (int i = 0; i < numberOfFileIngestThreads; ++i) {
                FileIngestPipeline pipeline = new FileIngestPipeline(this, fileIngestModuleTemplates);
                this.fileIngestPipelinesQueue.put(pipeline);
                this.fileIngestPipelines.add(pipeline);
            }
        } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
        }
        try {
            SleuthkitCase skCase = Case.getCurrentCaseThrows().getSleuthkitCase();
            this.addIngestModules(firstStageDataSourceModuleTemplates, IngestModuleType.DATA_SOURCE_LEVEL, skCase);
            this.addIngestModules(fileIngestModuleTemplates, IngestModuleType.FILE_LEVEL, skCase);
            this.addIngestModules(secondStageDataSourceModuleTemplates, IngestModuleType.DATA_SOURCE_LEVEL, skCase);
        } catch (TskCoreException | NoCurrentCaseException ex) {
            logErrorMessage(Level.WARNING, "Failed to add ingest modules listing to case database", ex);
        }
    }

    private void addIngestModules(List<IngestModuleTemplate> templates, IngestModuleType type, SleuthkitCase skCase) throws TskCoreException {
        for (IngestModuleTemplate module : templates) {
            ingestModules.add(skCase.addIngestModule(module.getModuleName(), FactoryClassNameNormalizer.normalize(module.getModuleFactory().getClass().getCanonicalName()), type, module.getModuleFactory().getModuleVersionNumber()));
        }
    }

    private static List<IngestModuleTemplate> getConfiguredIngestModuleTemplates(
            Map<String, IngestModuleTemplate> ingestModuleTemplates, Map<String, IngestModuleTemplate> jythonIngestModuleTemplates, List<String> pipelineConfig) {
        List<IngestModuleTemplate> templates = new ArrayList<>();
        for (String moduleClassName : pipelineConfig) {
            if (ingestModuleTemplates != null && ingestModuleTemplates.containsKey(moduleClassName)) {
                templates.add(ingestModuleTemplates.remove(moduleClassName));
            } else if (jythonIngestModuleTemplates != null && jythonIngestModuleTemplates.containsKey(moduleClassName)) {
                templates.add(jythonIngestModuleTemplates.remove(moduleClassName));
            }
        }
        return templates;
    }

    long getId() {
        return this.id;
    }

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

    Content getDataSource() {
        return this.dataSource;
    }

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

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

    boolean hasIngestPipeline() {
        return this.hasFirstStageDataSourceIngestPipeline()
                || this.hasFileIngestPipeline()
                || this.hasSecondStageDataSourceIngestPipeline();
    }

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

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

    private boolean hasFileIngestPipeline() {
        if (!this.fileIngestPipelines.isEmpty()) {
            return !this.fileIngestPipelines.get(0).isEmpty();
        }
        return false;
    }

    List<IngestModuleError> start() {
        if (dataSource == null) {
            // TODO - Remove once data source is always present during initialization
            throw new IllegalStateException("Ingest started before setting data source");
        }
        List<IngestModuleError> errors = startUpIngestPipelines();
        if (errors.isEmpty()) {
            try {
                this.ingestJob = Case.getCurrentCaseThrows().getSleuthkitCase().addIngestJob(dataSource, NetworkUtils.getLocalHostName(), ingestModules, new Date(this.createTime), new Date(0), IngestJobStatusType.STARTED, "");
            } catch (TskCoreException | NoCurrentCaseException ex) {
                logErrorMessage(Level.WARNING, "Failed to add ingest job info to case database", ex); //NON-NLS
            }

            if (this.hasFirstStageDataSourceIngestPipeline() || this.hasFileIngestPipeline()) {
                if (ingestMode == IngestJob.Mode.BATCH) {
                    logInfoMessage("Starting first stage analysis"); //NON-NLS
                    this.startFirstStage();
                } else {
                    logInfoMessage("Preparing for first stage analysis"); //NON-NLS
                    this.startFileIngestStreaming();
                }
            } else if (this.hasSecondStageDataSourceIngestPipeline()) {
                logInfoMessage("Starting second stage analysis"); //NON-NLS
                this.startSecondStage();
            }
        }
        return errors;
    }

    private List<IngestModuleError> startUpIngestPipelines() {
        List<IngestModuleError> errors = new ArrayList<>();

        /*
         * Start the data-source-level ingest module pipelines.
         */
        errors.addAll(this.firstStageDataSourceIngestPipeline.startUp());
        errors.addAll(this.secondStageDataSourceIngestPipeline.startUp());

        /*
         * If the data-source-level ingest pipelines were successfully started,
         * start the file-level ingest pipelines (one per pipeline file ingest
         * thread).
         */
        if (errors.isEmpty()) {
            for (FileIngestPipeline pipeline : this.fileIngestPipelinesQueue) {
                errors.addAll(pipeline.startUp());
                if (!errors.isEmpty()) {
                    /*
                     * If there are start up errors, the ingest job will not
                     * proceed, so shut down any file ingest pipelines that did
                     * start up.
                     */
                    while (!this.fileIngestPipelinesQueue.isEmpty()) {
                        FileIngestPipeline startedPipeline = this.fileIngestPipelinesQueue.poll();
                        if (startedPipeline.isRunning()) {
                            List<IngestModuleError> shutDownErrors = startedPipeline.shutDown();
                            if (!shutDownErrors.isEmpty()) {
                                /*
                                 * The start up errors will ultimately be
                                 * reported to the user for possible remedy, but
                                 * the shut down errors are logged here.
                                 */
                                logIngestModuleErrors(shutDownErrors);
                            }
                        }
                    }
                    break;
                }
            }
        }            
        return errors;
    }

    private void startFirstStage() {
        this.stage = IngestJobPipeline.Stages.FIRST_STAGE_FILES_AND_DATASOURCE;

        if (this.hasFileIngestPipeline()) {
            synchronized (this.fileIngestProgressLock) {
                this.estimatedFilesToProcess = this.dataSource.accept(new GetFilesCountVisitor());
            }
        }

        if (this.doUI) {
            if (this.hasFirstStageDataSourceIngestPipeline()) {
                this.startDataSourceIngestProgressBar();
            }
            if (this.hasFileIngestPipeline()) {
                this.startFileIngestProgressBar();
            }
        }

        synchronized (this.dataSourceIngestPipelineLock) {
            this.currentDataSourceIngestPipeline = this.firstStageDataSourceIngestPipeline;
        }

        if (this.hasFirstStageDataSourceIngestPipeline() && this.hasFileIngestPipeline()) {
            logInfoMessage("Scheduling first stage data source and file level analysis tasks"); //NON-NLS
            IngestJobPipeline.taskScheduler.scheduleIngestTasks(this);
        } else if (this.hasFirstStageDataSourceIngestPipeline()) {
            logInfoMessage("Scheduling first stage data source level analysis tasks"); //NON-NLS
            IngestJobPipeline.taskScheduler.scheduleDataSourceIngestTask(this);
        } else {
            logInfoMessage("Scheduling file level analysis tasks, no first stage data source level analysis configured"); //NON-NLS
            IngestJobPipeline.taskScheduler.scheduleFileIngestTasks(this, this.files);

            this.checkForStageCompleted();
        }
    }

    private void startFileIngestStreaming() {
        synchronized (this.stageCompletionCheckLock) {
            this.stage = IngestJobPipeline.Stages.FIRST_STAGE_FILES_ONLY;
        }

        if (this.hasFileIngestPipeline()) {
            synchronized (this.fileIngestProgressLock) {
                this.estimatedFilesToProcess = 0; // Set to indeterminate until the data source is complete
            }
        }

        if (this.doUI) {
            if (this.hasFileIngestPipeline()) {
                this.startFileIngestProgressBar();
            }
        }

        logInfoMessage("Waiting for streaming files"); //NON-NLS
    }

    private void startDataSourceIngestStreaming() {

        // Now that the data source is complete, we can get the estimated number of
        // files and switch to a determinate progress bar.
        synchronized (fileIngestProgressLock) {
            if (null != this.fileIngestProgress) {
                estimatedFilesToProcess = dataSource.accept(new GetFilesCountVisitor());
                fileIngestProgress.switchToDeterminate((int) estimatedFilesToProcess);
            }
        }

        if (this.doUI) {
            if (this.hasFirstStageDataSourceIngestPipeline()) {
                this.startDataSourceIngestProgressBar();
            }
        }

        synchronized (this.dataSourceIngestPipelineLock) {
            this.currentDataSourceIngestPipeline = this.firstStageDataSourceIngestPipeline;
        }

        logInfoMessage("Scheduling first stage data source level analysis tasks"); //NON-NLS
        synchronized (this.stageCompletionCheckLock) {
            this.stage = IngestJobPipeline.Stages.FIRST_STAGE_FILES_AND_DATASOURCE;
            IngestJobPipeline.taskScheduler.scheduleDataSourceIngestTask(this);
        }
    }

    private void startSecondStage() {
        logInfoMessage("Starting second stage analysis"); //NON-NLS        
        this.stage = IngestJobPipeline.Stages.SECOND_STAGE;
        if (this.doUI) {
            this.startDataSourceIngestProgressBar();
        }
        synchronized (this.dataSourceIngestPipelineLock) {
            this.currentDataSourceIngestPipeline = this.secondStageDataSourceIngestPipeline;
        }
        logInfoMessage("Scheduling second stage data source level analysis tasks"); //NON-NLS        
        IngestJobPipeline.taskScheduler.scheduleDataSourceIngestTask(this);
    }

    private void startDataSourceIngestProgressBar() {
        if (this.doUI) {
            synchronized (this.dataSourceIngestProgressLock) {
                String displayName = NbBundle.getMessage(this.getClass(),
                        "IngestJob.progress.dataSourceIngest.initialDisplayName",
                        this.dataSource.getName());
                this.dataSourceIngestProgress = 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.dataSourceIngestProgress.start();
                this.dataSourceIngestProgress.switchToIndeterminate();
            }
        }
    }

    private void startFileIngestProgressBar() {
        if (this.doUI) {
            synchronized (this.fileIngestProgressLock) {
                String displayName = NbBundle.getMessage(this.getClass(),
                        "IngestJob.progress.fileIngest.displayName",
                        this.dataSource.getName());
                this.fileIngestProgress = 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.fileIngestProgress.start();
                this.fileIngestProgress.switchToDeterminate((int) this.estimatedFilesToProcess);
            }
        }
    }

    private void checkForStageCompleted() {
        if (ingestMode == IngestJob.Mode.BATCH) {
            checkForStageCompletedBatch();
        } else {
            checkForStageCompletedStreaming();
        }
    }

    private void checkForStageCompletedBatch() {
        synchronized (this.stageCompletionCheckLock) {
            if (IngestJobPipeline.taskScheduler.currentTasksAreCompleted(this)) {
                switch (this.stage) {
                    case FIRST_STAGE_FILES_AND_DATASOURCE:
                        this.finishFirstStage();
                        break;
                    case SECOND_STAGE:
                        this.finish();
                        break;
                }
            }
        }
    }

    private void checkForStageCompletedStreaming() {
        synchronized (this.stageCompletionCheckLock) {
            if (IngestJobPipeline.taskScheduler.currentTasksAreCompleted(this)) {
                switch (this.stage) {
                    case FIRST_STAGE_FILES_ONLY:
                        // Nothing to do here - need to wait for the data source
                        break;
                    case FIRST_STAGE_FILES_AND_DATASOURCE:
                        // Finish file and data source ingest, start second stage (if applicable)
                        this.finishFirstStage();
                        break;
                    case SECOND_STAGE:
                        this.finish();
                        break;
                }
            }
        }
    }

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

        // Shut down the file ingest pipelines. Note that no shut down is
        // required for the data source ingest pipeline because data source 
        // ingest modules do not have a shutdown() method.
        List<IngestModuleError> errors = new ArrayList<>();
        while (!this.fileIngestPipelinesQueue.isEmpty()) {
            FileIngestPipeline pipeline = fileIngestPipelinesQueue.poll();
            if (pipeline.isRunning()) {
                errors.addAll(pipeline.shutDown());
            }
        }
        if (!errors.isEmpty()) {
            logIngestModuleErrors(errors);
        }

        if (this.doUI) {
            // Finish the first stage data source ingest progress bar, if it hasn't 
            // already been finished.
            synchronized (this.dataSourceIngestProgressLock) {
                if (this.dataSourceIngestProgress != null) {
                    this.dataSourceIngestProgress.finish();
                    this.dataSourceIngestProgress = null;
                }
            }

            // Finish the file ingest progress bar, if it hasn't already 
            // been finished.
            synchronized (this.fileIngestProgressLock) {
                if (this.fileIngestProgress != null) {
                    this.fileIngestProgress.finish();
                    this.fileIngestProgress = null;
                }
            }
        }

        if (!this.cancelled && this.hasSecondStageDataSourceIngestPipeline()) {
            this.startSecondStage();
        } else {
            this.finish();
        }
    }

    private void finish() {
        logInfoMessage("Finished analysis"); //NON-NLS        
        this.stage = IngestJobPipeline.Stages.FINALIZATION;

        if (this.doUI) {
            // Finish the second stage data source ingest progress bar, if it hasn't 
            // already been finished.
            synchronized (this.dataSourceIngestProgressLock) {
                if (this.dataSourceIngestProgress != null) {
                    this.dataSourceIngestProgress.finish();
                    this.dataSourceIngestProgress = null;
                }
            }
        }
        if (ingestJob != null) {
            if (this.cancelled) {
                try {
                    ingestJob.setIngestJobStatus(IngestJobStatusType.CANCELLED);
                } catch (TskCoreException ex) {
                    logErrorMessage(Level.WARNING, "Failed to update ingest job status in case database", ex);
                }
            } else {
                try {
                    ingestJob.setIngestJobStatus(IngestJobStatusType.COMPLETED);
                } catch (TskCoreException ex) {
                    logErrorMessage(Level.WARNING, "Failed to update ingest job status in case database", ex);
                }
            }
            try {
                this.ingestJob.setEndDateTime(new Date());
            } catch (TskCoreException ex) {
                logErrorMessage(Level.WARNING, "Failed to set job end date in case database", ex);
            }
        }
        this.parentJob.ingestJobPipelineFinished(this);
    }

    void process(DataSourceIngestTask task) {
        try {
            synchronized (this.dataSourceIngestPipelineLock) {
                if (!this.isCancelled() && !this.currentDataSourceIngestPipeline.isEmpty()) {
                    List<IngestModuleError> errors = new ArrayList<>();
                    errors.addAll(this.currentDataSourceIngestPipeline.performTask(task));
                    if (!errors.isEmpty()) {
                        logIngestModuleErrors(errors);
                    }
                }
            }

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

        } finally {
            IngestJobPipeline.taskScheduler.notifyTaskCompleted(task);
            this.checkForStageCompleted();
        }
    }

    void process(FileIngestTask task) throws InterruptedException {
        try {
            if (!this.isCancelled()) {
                FileIngestPipeline pipeline = this.fileIngestPipelinesQueue.take();
                if (!pipeline.isEmpty()) {
                    AbstractFile file;
                    try {
                        file = task.getFile();
                    } catch (TskCoreException ex) {
                        // In practice, this task would never have been enqueued since the file
                        // lookup would have failed there.
                        List<IngestModuleError> errors = new ArrayList<>();
                        errors.add(new IngestModuleError("Ingest Job Pipeline", ex));
                        logIngestModuleErrors(errors);
                        this.fileIngestPipelinesQueue.put(pipeline);
                        return;
                    }

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

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

                    if (this.doUI && !this.cancelled) {
                        synchronized (this.fileIngestProgressLock) {
                            this.filesInProgress.remove(file.getName());
                            if (this.filesInProgress.size() > 0) {
                                this.fileIngestProgress.progress(this.filesInProgress.get(0));
                            } else {
                                this.fileIngestProgress.progress("");
                            }
                        }
                    }
                }
                this.fileIngestPipelinesQueue.put(pipeline);
            }
        } finally {
            IngestJobPipeline.taskScheduler.notifyTaskCompleted(task);
            this.checkForStageCompleted();
        }
    }

    void addStreamingIngestFiles(List<Long> fileObjIds) {

        // Return if there are no file ingest modules enabled.
        if (!hasFileIngestPipeline()) {
            return;
        }

        if (stage.equals(Stages.FIRST_STAGE_FILES_ONLY)) {
            IngestJobPipeline.taskScheduler.scheduleStreamedFileIngestTasks(this, fileObjIds);
        } else {
            logErrorMessage(Level.SEVERE, "Adding streaming files to job during stage " + stage.toString() + " not supported");
        }
    }

    void processStreamingIngestDataSource() {
        startDataSourceIngestStreaming();
        checkForStageCompleted();
    }

    void addFiles(List<AbstractFile> files) {
        if (stage.equals(Stages.FIRST_STAGE_FILES_ONLY)
                || stage.equals(Stages.FIRST_STAGE_FILES_AND_DATASOURCE)) {
            IngestJobPipeline.taskScheduler.fastTrackFileIngestTasks(this, files);
        } else {
            logErrorMessage(Level.SEVERE, "Adding files to job during second stage analysis not supported");
        }

        this.checkForStageCompleted();
    }

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

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

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

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

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

    void advanceDataSourceIngestProgressBar(String currentTask, int workUnits) {
        if (this.doUI && !this.cancelled) {
            synchronized (this.fileIngestProgressLock) {
                this.dataSourceIngestProgress.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.dataSourceIngestProgress.finish();
                this.dataSourceIngestProgress = null;
                this.startDataSourceIngestProgressBar();
            }
        }
    }

    DataSourceIngestPipeline.DataSourcePipelineModule getCurrentDataSourceIngestModule() {
        if (null != currentDataSourceIngestPipeline) {
            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.cancelPendingTasksForIngestJob(this);

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

            synchronized (this.fileIngestProgressLock) {
                if (null != this.fileIngestProgress) {
                    this.fileIngestProgress.setDisplayName(NbBundle.getMessage(this.getClass(), "IngestJob.progress.fileIngest.displayName", this.dataSource.getName()));
                    this.fileIngestProgress.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();
    }

    void setCurrentFileIngestModule(String moduleName, String taskName) {
        this.currentFileIngestModule = moduleName;
        this.currentFileIngestTask = taskName;
    }

    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(), id, ingestJob.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(), id, ingestJob.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(), id, ingestJob.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 = IngestJobPipeline.taskScheduler.getTasksSnapshotForJob(id);

        }

        return new Snapshot(this.dataSource.getName(), id, 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);
        }
    }

}