IngestManager.java

Go to the documentation of this file.

/*
 * Autopsy Forensic Browser
 *
 * Copyright 2012-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 com.google.common.eventbus.Subscribe;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.awt.EventQueue;
import java.awt.GraphicsEnvironment;
import java.beans.PropertyChangeEvent;
import java.beans.PropertyChangeListener;
import java.io.Serializable;
import java.lang.reflect.InvocationTargetException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicLong;
import java.util.logging.Level;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import javax.annotation.concurrent.GuardedBy;
import javax.annotation.concurrent.Immutable;
import javax.annotation.concurrent.ThreadSafe;
import javax.swing.JOptionPane;
import javax.swing.SwingUtilities;
import org.netbeans.api.progress.ProgressHandle;
import org.openide.util.Cancellable;
import org.openide.util.NbBundle;
import org.openide.windows.WindowManager;
import org.sleuthkit.autopsy.casemodule.Case;
import org.sleuthkit.autopsy.casemodule.NoCurrentCaseException;
import org.sleuthkit.autopsy.core.RuntimeProperties;
import org.sleuthkit.autopsy.core.ServicesMonitor;
import org.sleuthkit.autopsy.core.UserPreferences;
import org.sleuthkit.autopsy.coreutils.Logger;
import org.sleuthkit.autopsy.coreutils.MessageNotifyUtil;
import org.sleuthkit.autopsy.events.AutopsyEvent;
import org.sleuthkit.autopsy.events.AutopsyEventException;
import org.sleuthkit.autopsy.events.AutopsyEventPublisher;
import org.sleuthkit.autopsy.ingest.events.BlackboardPostEvent;
import org.sleuthkit.autopsy.ingest.events.ContentChangedEvent;
import org.sleuthkit.autopsy.ingest.events.DataSourceAnalysisCompletedEvent;
import org.sleuthkit.autopsy.ingest.events.DataSourceAnalysisStartedEvent;
import org.sleuthkit.autopsy.ingest.events.FileAnalyzedEvent;
import org.sleuthkit.datamodel.AbstractFile;
import org.sleuthkit.datamodel.AnalysisResult;
import org.sleuthkit.datamodel.Blackboard;
import org.sleuthkit.datamodel.BlackboardArtifact;
import org.sleuthkit.datamodel.Content;
import org.sleuthkit.datamodel.DataArtifact;
import org.sleuthkit.datamodel.DataSource;
import org.sleuthkit.datamodel.TskCoreException;

@ThreadSafe
public class IngestManager implements IngestProgressSnapshotProvider {
 
    private final static Logger logger = Logger.getLogger(IngestManager.class.getName());
    private final static String INGEST_JOB_EVENT_CHANNEL_NAME = "%s-Ingest-Job-Events"; //NON-NLS
    private final static Set<String> INGEST_JOB_EVENT_NAMES = Stream.of(IngestJobEvent.values()).map(IngestJobEvent::toString).collect(Collectors.toSet());
    private final static String INGEST_MODULE_EVENT_CHANNEL_NAME = "%s-Ingest-Module-Events"; //NON-NLS
    private final static Set<String> INGEST_MODULE_EVENT_NAMES = Stream.of(IngestModuleEvent.values()).map(IngestModuleEvent::toString).collect(Collectors.toSet());
    private final static int MAX_ERROR_MESSAGE_POSTS = 200;
    @GuardedBy("IngestManager.class")
    private static IngestManager instance;
    private final int numberOfFileIngestThreads;
    private final AtomicLong nextIngestManagerTaskId = new AtomicLong(0L);
    private final ExecutorService startIngestJobsExecutor = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().setNameFormat("IM-start-ingest-jobs-%d").build()); //NON-NLS;
    @GuardedBy("startIngestJobFutures")
    private final Map<Long, Future<Void>> startIngestJobFutures = new ConcurrentHashMap<>();
    @GuardedBy("ingestJobsById")
    private final Map<Long, IngestJob> ingestJobsById = new HashMap<>();
    private final ExecutorService dataSourceLevelIngestJobTasksExecutor;
    private final ExecutorService fileLevelIngestJobTasksExecutor;
    private final ExecutorService dataArtifactIngestTasksExecutor;
    private final ExecutorService analysisResultIngestTasksExecutor;
    private final ExecutorService eventPublishingExecutor = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().setNameFormat("IM-ingest-events-%d").build()); //NON-NLS;
    private final IngestMonitor ingestMonitor = new IngestMonitor();
    private final ServicesMonitor servicesMonitor = ServicesMonitor.getInstance();
    private final AutopsyEventPublisher jobEventPublisher = new AutopsyEventPublisher();
    private final AutopsyEventPublisher moduleEventPublisher = new AutopsyEventPublisher();
    private final Object ingestMessageBoxLock = new Object();
    private final AtomicLong ingestErrorMessagePosts = new AtomicLong(0L);
    private final ConcurrentHashMap<Long, IngestThreadActivitySnapshot> ingestThreadActivitySnapshots = new ConcurrentHashMap<>();
    private final ConcurrentHashMap<String, Long> ingestModuleRunTimes = new ConcurrentHashMap<>();
    private volatile IngestMessageTopComponent ingestMessageBox;
    private volatile boolean caseIsOpen;

    public synchronized static IngestManager getInstance() {
        if (null == instance) {
            instance = new IngestManager();
            instance.subscribeToServiceMonitorEvents();
            instance.subscribeToCaseEvents();
        }
        return instance;
    }

    private IngestManager() {
        dataSourceLevelIngestJobTasksExecutor = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().setNameFormat("IM-data-source-ingest-%d").build()); //NON-NLS;        
        long threadId = nextIngestManagerTaskId.incrementAndGet();
        dataSourceLevelIngestJobTasksExecutor.submit(new ExecuteIngestJobTasksTask(threadId, IngestTasksScheduler.getInstance().getDataSourceIngestTaskQueue()));
        ingestThreadActivitySnapshots.put(threadId, new IngestThreadActivitySnapshot(threadId));
 
        numberOfFileIngestThreads = UserPreferences.numberOfFileIngestThreads();
        fileLevelIngestJobTasksExecutor = Executors.newFixedThreadPool(numberOfFileIngestThreads, new ThreadFactoryBuilder().setNameFormat("IM-file-ingest-%d").build()); //NON-NLS
        for (int i = 0; i < numberOfFileIngestThreads; ++i) {
            threadId = nextIngestManagerTaskId.incrementAndGet();
            fileLevelIngestJobTasksExecutor.submit(new ExecuteIngestJobTasksTask(threadId, IngestTasksScheduler.getInstance().getFileIngestTaskQueue()));
            ingestThreadActivitySnapshots.put(threadId, new IngestThreadActivitySnapshot(threadId));
        }
 
        dataArtifactIngestTasksExecutor = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().setNameFormat("IM-data-artifact-ingest-%d").build()); //NON-NLS;        
        threadId = nextIngestManagerTaskId.incrementAndGet();
        dataArtifactIngestTasksExecutor.submit(new ExecuteIngestJobTasksTask(threadId, IngestTasksScheduler.getInstance().getDataArtifactIngestTaskQueue()));
        ingestThreadActivitySnapshots.put(threadId, new IngestThreadActivitySnapshot(threadId));
 
        analysisResultIngestTasksExecutor = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().setNameFormat("IM-analysis-result-ingest-%d").build()); //NON-NLS;        
        threadId = nextIngestManagerTaskId.incrementAndGet();
        analysisResultIngestTasksExecutor.submit(new ExecuteIngestJobTasksTask(threadId, IngestTasksScheduler.getInstance().getAnalysisResultIngestTaskQueue()));
        ingestThreadActivitySnapshots.put(threadId, new IngestThreadActivitySnapshot(threadId));
    }

    private void subscribeToServiceMonitorEvents() {
        PropertyChangeListener propChangeListener = (PropertyChangeEvent evt) -> {
            if (evt.getNewValue().equals(ServicesMonitor.ServiceStatus.DOWN.toString())) {
                /*
                 * The application services considered to be key services are
                 * only necessary for multi-user cases.
                 */
                try {
                    if (Case.getCurrentCaseThrows().getCaseType() != Case.CaseType.MULTI_USER_CASE) {
                        return;
                    }
                } catch (NoCurrentCaseException noCaseOpenException) {
                    return;
                }
 
                String serviceDisplayName = ServicesMonitor.Service.valueOf(evt.getPropertyName()).getDisplayName();
                logger.log(Level.SEVERE, "Service {0} is down, cancelling all running ingest jobs", serviceDisplayName); //NON-NLS
                if (isIngestRunning() && RuntimeProperties.runningWithGUI()) {
                    EventQueue.invokeLater(new Runnable() {
                        @Override
                        public void run() {
                            JOptionPane.showMessageDialog(WindowManager.getDefault().getMainWindow(),
                                    NbBundle.getMessage(this.getClass(), "IngestManager.cancellingIngest.msgDlg.text"),
                                    NbBundle.getMessage(this.getClass(), "IngestManager.serviceIsDown.msgDlg.text", serviceDisplayName),
                                    JOptionPane.ERROR_MESSAGE);
                        }
                    });
                }
                cancelAllIngestJobs(IngestJob.CancellationReason.SERVICES_DOWN);
            }
        };
 
        /*
         * The key services for multi-user cases are currently the case database
         * server and the Solr server. The Solr server is a key service not
         * because search is essential, but because the coordination service
         * (ZooKeeper) is running embedded within the Solr server.
         */
        Set<String> servicesList = new HashSet<>();
        servicesList.add(ServicesMonitor.Service.REMOTE_CASE_DATABASE.toString());
        servicesList.add(ServicesMonitor.Service.REMOTE_KEYWORD_SEARCH.toString());
        this.servicesMonitor.addSubscriber(servicesList, propChangeListener);
    }

    private void subscribeToCaseEvents() {
        Case.addEventTypeSubscriber(EnumSet.of(Case.Events.CURRENT_CASE), (PropertyChangeEvent event) -> {
            if (event.getNewValue() != null) {
                handleCaseOpened();
            } else {
                handleCaseClosed();
            }
        });
    }

    void handleCaseOpened() {
        caseIsOpen = true;
        clearIngestMessageBox();
        try {
            Case openedCase = Case.getCurrentCaseThrows();
            String channelPrefix = openedCase.getName();
            if (Case.CaseType.MULTI_USER_CASE == openedCase.getCaseType()) {
                jobEventPublisher.openRemoteEventChannel(String.format(INGEST_JOB_EVENT_CHANNEL_NAME, channelPrefix));
                moduleEventPublisher.openRemoteEventChannel(String.format(INGEST_MODULE_EVENT_CHANNEL_NAME, channelPrefix));
            }
            openedCase.getSleuthkitCase().registerForEvents(this);
        } catch (NoCurrentCaseException | AutopsyEventException ex) {
            logger.log(Level.SEVERE, "Failed to open remote events channel", ex); //NON-NLS
            MessageNotifyUtil.Notify.error(NbBundle.getMessage(IngestManager.class, "IngestManager.OpenEventChannel.Fail.Title"),
                    NbBundle.getMessage(IngestManager.class, "IngestManager.OpenEventChannel.Fail.ErrMsg"));
        }
    }

    @Subscribe
    void handleArtifactsPosted(Blackboard.ArtifactsPostedEvent tskEvent) {
        /*
         * Add any new data artifacts included in the event to the source ingest
         * job for possible analysis.
         */
        List<DataArtifact> newDataArtifacts = new ArrayList<>();
        List<AnalysisResult> newAnalysisResults = new ArrayList<>();
        Collection<BlackboardArtifact> newArtifacts = tskEvent.getArtifacts();
        for (BlackboardArtifact artifact : newArtifacts) {
            if (artifact instanceof DataArtifact) {
                newDataArtifacts.add((DataArtifact) artifact);
            } else {
                newAnalysisResults.add((AnalysisResult) artifact);
            }
        }
        if (!newDataArtifacts.isEmpty() || !newAnalysisResults.isEmpty()) {
            IngestJob ingestJob = null;
            Optional<Long> ingestJobId = tskEvent.getIngestJobId();
            if (ingestJobId.isPresent()) {
                synchronized (ingestJobsById) {
                    ingestJob = ingestJobsById.get(ingestJobId.get());
                }
            } else {
                /*
                 * There are four use cases where the ingest job ID returned by
                 * the event is expected be null:
                 *
                 * 1. The artifacts are being posted by a data source proccessor
                 * (DSP) module that runs before the ingest job is created,
                 * i.e., a DSP that does not support streaming ingest and has no
                 * noton of an ingest job ID. In this use case, the event is
                 * handled synchronously. The DSP calls
                 * Blackboard.postArtifacts(), which puts the event on the event
                 * bus to which this method subscribes, so the event will be
                 * handled here before the DSP completes and calls
                 * DataSourceProcessorCallback.done(). This means the code below
                 * will execute before the ingest job is created, so it will not
                 * find an ingest job to which to add the artifacts. However,
                 * the artifacts WILL be analyzed after the ingest job is
                 * started, when the ingest job executor, working in batch mode,
                 * schedules ingest tasks for all of the data artifacts in the
                 * case database. There is a slight risk that the wrong ingest
                 * job will be selected if multiple ingests of the same data
                 * source are in progress.
                 *
                 * 2. The artifacts were posted by an ingest module that either
                 * has not been updated to use the current
                 * Blackboard.postArtifacts() API, or is using it incorrectly.
                 * In this use case, the code below should be able to find the
                 * ingest job to which to add the artifacts via their data
                 * source. There is a slight risk that the wrong ingest job will
                 * be selected if multiple ingests of the same data source are
                 * in progress.
                 *
                 * 3. The portable case generator uses a
                 * CommunicationArtifactsHelper constructed with a null ingest
                 * job ID, and the CommunicatonsArtifactHelper posts artifacts.
                 * Ingest of that data source might be running, in which case
                 * the data artifact will be analyzed. It also might be analyzed
                 * by a subsequent ingest job for the data source. This is an
                 * acceptable edge case.
                 *
                 * 4. The user can manually create timeline events with the
                 * timeline tool, which posts the TSK_TL_EVENT data artifacts.
                 * The user selects the data source for these artifacts. Ingest
                 * of that data source might be running, in which case the data
                 * artifact will be analyzed. It also might be analyzed by a
                 * subsequent ingest job for the data source. This is an
                 * acceptable edge case.
                 * 
                 * 5. The user can manually run ad hoc keyword searches,
                 * which post TSK_KEYWORD_HIT analysis results. Ingest
                 * of that data source might be running, in which case the analysis
                 * results will be analyzed. They also might be analyzed by a
                 * subsequent ingest job for the data source. This is an
                 * acceptable edge case.
                 */
                BlackboardArtifact artifact = newArtifacts.iterator().next();
                if (artifact != null) {
                    try {
                        Content artifactDataSource = artifact.getDataSource();
                        synchronized (ingestJobsById) {
                            for (IngestJob job : ingestJobsById.values()) {
                                Content dataSource = job.getDataSource();
                                if (artifactDataSource.getId() == dataSource.getId()) {
                                    ingestJob = job;
                                    break;
                                }
                            }
                        }
                    } catch (TskCoreException ex) {
                        logger.log(Level.SEVERE, String.format("Failed to get data source for blackboard artifact (object ID = %d)", artifact.getId()), ex); //NON-NLS
                    }
                }
            }
            if (ingestJob != null) {
                if (!newDataArtifacts.isEmpty()) {
                    ingestJob.addDataArtifacts(newDataArtifacts);
                }
                if (!newAnalysisResults.isEmpty()) {
                    ingestJob.addAnalysisResults(newAnalysisResults);
                }
            }
        }
 
        /*
         * Publish Autopsy events for the new artifacts, one event per artifact
         * type.
         */
        for (BlackboardArtifact.Type artifactType : tskEvent.getArtifactTypes()) {
            ModuleDataEvent legacyEvent = new ModuleDataEvent(tskEvent.getModuleName(), artifactType, tskEvent.getArtifacts(artifactType));
            AutopsyEvent autopsyEvent = new BlackboardPostEvent(legacyEvent);
            eventPublishingExecutor.submit(new PublishEventTask(autopsyEvent, moduleEventPublisher));
        }
    }

    void handleCaseClosed() {
        /*
         * TODO (JIRA-2227): IngestManager should wait for cancelled ingest jobs
         * to complete when a case is closed.
         */
        cancelAllIngestJobs(IngestJob.CancellationReason.CASE_CLOSED);
        Case.getCurrentCase().getSleuthkitCase().unregisterForEvents(this);
        jobEventPublisher.closeRemoteEventChannel();
        moduleEventPublisher.closeRemoteEventChannel();
        caseIsOpen = false;
        clearIngestMessageBox();
    }

    public IngestStream openIngestStream(DataSource dataSource, IngestJobSettings settings) throws TskCoreException {
        if (!(dataSource instanceof DataSource)) {
            throw new IllegalArgumentException("dataSource argument does not implement the DataSource interface"); //NON-NLS
        }
        IngestJob job = new IngestJob(dataSource, IngestJob.Mode.STREAMING, settings);
        IngestJobInputStream stream = new IngestJobInputStream(job);
        if (stream.getIngestJobStartResult().getJob() != null) {
            return stream;
        } else if (stream.getIngestJobStartResult().getModuleErrors().isEmpty()) {
            for (IngestModuleError error : stream.getIngestJobStartResult().getModuleErrors()) {
                logger.log(Level.SEVERE, String.format("%s ingest module startup error for %s", error.getModuleDisplayName(), dataSource.getName()), error.getThrowable());
            }
            throw new TskCoreException("Error starting ingest modules");
        } else {
            throw new TskCoreException("Error starting ingest modules", stream.getIngestJobStartResult().getStartupException());
        }
    }

    public int getNumberOfFileIngestThreads() {
        return numberOfFileIngestThreads;
    }

    public void queueIngestJob(Collection<Content> dataSources, IngestJobSettings settings) {
        if (caseIsOpen) {
            List<AbstractFile> emptyFilesSubset = new ArrayList<>();
            for (Content dataSource : dataSources) {
                queueIngestJob(dataSource, emptyFilesSubset, settings);
            }
        }
    }

    public void queueIngestJob(Content dataSource, List<AbstractFile> files, IngestJobSettings settings) {
        if (!(dataSource instanceof DataSource)) {
            throw new IllegalArgumentException("dataSource argument does not implement the DataSource interface"); //NON-NLS
        }
        if (caseIsOpen) {
            IngestJob job = new IngestJob((DataSource) dataSource, files, settings);
            if (job.hasIngestPipeline()) {
                long taskId = nextIngestManagerTaskId.incrementAndGet();
                Future<Void> task = startIngestJobsExecutor.submit(new StartIngestJobTask(taskId, job));
                synchronized (startIngestJobFutures) {
                    startIngestJobFutures.put(taskId, task);
                }
            }
        }
    }

    public IngestJobStartResult beginIngestJob(Collection<Content> dataSources, IngestJobSettings settings) {
        List<DataSource> verifiedDataSources = new ArrayList<>();
        for (Content content : dataSources) {
            if (!(content instanceof DataSource)) {
                throw new IllegalArgumentException("Content object in dataSources argument does not implement the DataSource interface"); //NON-NLS
            }
            DataSource verifiedDataSource = (DataSource) content;
            verifiedDataSources.add(verifiedDataSource);
        }
        IngestJobStartResult startResult = null;
        if (caseIsOpen) {
            for (DataSource dataSource : verifiedDataSources) {
                List<IngestJob> startedJobs = new ArrayList<>();
                IngestJob job = new IngestJob(dataSource, IngestJob.Mode.BATCH, settings);
                if (job.hasIngestPipeline()) {
                    startResult = startIngestJob(job);
                    if (startResult.getModuleErrors().isEmpty() && startResult.getStartupException() == null) {
                        startedJobs.add(job);
                    } else {
                        for (IngestJob jobToCancel : startedJobs) {
                            jobToCancel.cancel(IngestJob.CancellationReason.INGEST_MODULES_STARTUP_FAILED);
                        }
                        break;
                    }
                } else {
                    startResult = new IngestJobStartResult(null, new IngestManagerException("No ingest pipeline created, likely due to no ingest modules being enabled"), null); //NON-NLS
                    break;
                }
            }
        } else {
            startResult = new IngestJobStartResult(null, new IngestManagerException("No case open"), null); //NON-NLS
        }
        return startResult;
    }

    @NbBundle.Messages({
        "IngestManager.startupErr.dlgTitle=Ingest Module Startup Failure",
        "IngestManager.startupErr.dlgMsg=Unable to start up one or more ingest modules, ingest cancelled.",
        "IngestManager.startupErr.dlgSolution=Please disable the failed modules or fix the errors before restarting ingest.",
        "IngestManager.startupErr.dlgErrorList=Errors:"
    })
    IngestJobStartResult startIngestJob(IngestJob job) {
 
        // initialize IngestMessageInbox, if it hasn't been initialized yet. This can't be done in
        // the constructor because that ends up freezing the UI on startup (JIRA-7345).
        if (!GraphicsEnvironment.isHeadless()) {
            if (SwingUtilities.isEventDispatchThread()) {
                initIngestMessageInbox();
            } else {
                try {
                    SwingUtilities.invokeAndWait(() -> initIngestMessageInbox());
                } catch (InterruptedException ex) {
                    // ignore interruptions
                } catch (InvocationTargetException ex) {
                    logger.log(Level.WARNING, "There was an error starting ingest message inbox", ex);
                }
            }
        }
 
        List<IngestModuleError> errors = null;
        Case openCase;
        try {
            openCase = Case.getCurrentCaseThrows();
        } catch (NoCurrentCaseException ex) {
            return new IngestJobStartResult(null, new IngestManagerException("Exception while getting open case.", ex), Collections.<IngestModuleError>emptyList()); //NON-NLS
        }
        if (openCase.getCaseType() == Case.CaseType.MULTI_USER_CASE) {
            try {
                if (!servicesMonitor.getServiceStatus(ServicesMonitor.Service.REMOTE_CASE_DATABASE.toString()).equals(ServicesMonitor.ServiceStatus.UP.toString())) {
                    if (RuntimeProperties.runningWithGUI()) {
                        EventQueue.invokeLater(new Runnable() {
                            @Override
                            public void run() {
                                String serviceDisplayName = ServicesMonitor.Service.REMOTE_CASE_DATABASE.getDisplayName();
                                JOptionPane.showMessageDialog(WindowManager.getDefault().getMainWindow(),
                                        NbBundle.getMessage(this.getClass(), "IngestManager.cancellingIngest.msgDlg.text"),
                                        NbBundle.getMessage(this.getClass(), "IngestManager.serviceIsDown.msgDlg.text", serviceDisplayName),
                                        JOptionPane.ERROR_MESSAGE);
                            }
                        });
                    }
                    return new IngestJobStartResult(null, new IngestManagerException("Ingest aborted. Remote database is down"), Collections.<IngestModuleError>emptyList()); //NON-NLS
                }
            } catch (ServicesMonitor.ServicesMonitorException ex) {
                return new IngestJobStartResult(null, new IngestManagerException("Database server is down", ex), Collections.<IngestModuleError>emptyList()); //NON-NLS
            }
        }
 
        if (!ingestMonitor.isRunning()) {
            ingestMonitor.start();
        }
 
        synchronized (ingestJobsById) {
            ingestJobsById.put(job.getId(), job);
        }
        IngestManager.logger.log(Level.INFO, String.format("Starting ingest job %d at %s", job.getId(), new Date().getTime())); //NON-NLS
        try {
            errors = job.start();
        } catch (InterruptedException ex) {
            return new IngestJobStartResult(null, new IngestManagerException("Interrupted while starting ingest", ex), errors); //NON-NLS
        }
        if (errors.isEmpty()) {
            this.fireIngestJobStarted(job.getId());
        } else {
            synchronized (ingestJobsById) {
                this.ingestJobsById.remove(job.getId());
            }
            for (IngestModuleError error : errors) {
                logger.log(Level.SEVERE, String.format("Error starting %s ingest module for job %d", error.getModuleDisplayName(), job.getId()), error.getThrowable()); //NON-NLS
            }
            IngestManager.logger.log(Level.SEVERE, "Ingest job {0} could not be started", job.getId()); //NON-NLS
            if (RuntimeProperties.runningWithGUI()) {
                final StringBuilder message = new StringBuilder(1024);
                message.append(Bundle.IngestManager_startupErr_dlgMsg()).append("\n"); //NON-NLS
                message.append(Bundle.IngestManager_startupErr_dlgSolution()).append("\n\n"); //NON-NLS
                message.append(Bundle.IngestManager_startupErr_dlgErrorList()).append("\n"); //NON-NLS
                for (IngestModuleError error : errors) {
                    String moduleName = error.getModuleDisplayName();
                    String errorMessage = error.getThrowable().getLocalizedMessage();
                    message.append(moduleName).append(": ").append(errorMessage).append("\n"); //NON-NLS
                }
                message.append("\n\n");
                EventQueue.invokeLater(() -> {
                    JOptionPane.showMessageDialog(WindowManager.getDefault().getMainWindow(), message, Bundle.IngestManager_startupErr_dlgTitle(), JOptionPane.ERROR_MESSAGE);
                });
            }
            return new IngestJobStartResult(null, new IngestManagerException("Errors occurred while starting ingest"), errors); //NON-NLS
        }
 
        return new IngestJobStartResult(job, null, errors);
    }

    void finishIngestJob(IngestJob job) {
        long jobId = job.getId();
        synchronized (ingestJobsById) {
            ingestJobsById.remove(jobId);
        }
        if (!job.isCancelled()) {
            IngestManager.logger.log(Level.INFO, String.format("Ingest job %d completed at %s", job.getId(), new Date().getTime())); //NON-NLS            
            fireIngestJobCompleted(jobId);
        } else {
            IngestManager.logger.log(Level.INFO, String.format("Ingest job %d cancelled at %s", job.getId(), new Date().getTime())); //NON-NLS
            fireIngestJobCancelled(jobId);
        }
    }

    public boolean isIngestRunning() {
        synchronized (ingestJobsById) {
            return !ingestJobsById.isEmpty();
        }
    }

    public void cancelAllIngestJobs(IngestJob.CancellationReason reason) {
        synchronized (startIngestJobFutures) {
            startIngestJobFutures.values().forEach((handle) -> {
                handle.cancel(true);
            });
        }
        synchronized (ingestJobsById) {
            this.ingestJobsById.values().forEach((job) -> {
                job.cancel(reason);
            });
        }
    }

    public void addIngestJobEventListener(final PropertyChangeListener listener) {
        jobEventPublisher.addSubscriber(INGEST_JOB_EVENT_NAMES, listener);
    }

    public void addIngestJobEventListener(Set<IngestJobEvent> eventTypes, final PropertyChangeListener listener) {
        eventTypes.forEach((IngestJobEvent event) -> {
            jobEventPublisher.addSubscriber(event.toString(), listener);
        });
    }

    public void removeIngestJobEventListener(final PropertyChangeListener listener) {
        jobEventPublisher.removeSubscriber(INGEST_JOB_EVENT_NAMES, listener);
    }

    public void removeIngestJobEventListener(Set<IngestJobEvent> eventTypes, final PropertyChangeListener listener) {
        eventTypes.forEach((IngestJobEvent event) -> {
            jobEventPublisher.removeSubscriber(event.toString(), listener);
        });
    }

    public void addIngestModuleEventListener(final PropertyChangeListener listener) {
        moduleEventPublisher.addSubscriber(INGEST_MODULE_EVENT_NAMES, listener);
    }

    public void addIngestModuleEventListener(Set<IngestModuleEvent> eventTypes, final PropertyChangeListener listener) {
        eventTypes.forEach((IngestModuleEvent event) -> {
            moduleEventPublisher.addSubscriber(event.toString(), listener);
        });
    }

    public void removeIngestModuleEventListener(final PropertyChangeListener listener) {
        moduleEventPublisher.removeSubscriber(INGEST_MODULE_EVENT_NAMES, listener);
    }

    public void removeIngestModuleEventListener(Set<IngestModuleEvent> eventTypes, final PropertyChangeListener listener) {
        moduleEventPublisher.removeSubscriber(INGEST_MODULE_EVENT_NAMES, listener);
    }

    void fireIngestJobStarted(long ingestJobId) {
        AutopsyEvent event = new AutopsyEvent(IngestJobEvent.STARTED.toString(), ingestJobId, null);
        eventPublishingExecutor.submit(new PublishEventTask(event, jobEventPublisher));
    }

    void fireIngestJobCompleted(long ingestJobId) {
        AutopsyEvent event = new AutopsyEvent(IngestJobEvent.COMPLETED.toString(), ingestJobId, null);
        eventPublishingExecutor.submit(new PublishEventTask(event, jobEventPublisher));
    }

    void fireIngestJobCancelled(long ingestJobId) {
        AutopsyEvent event = new AutopsyEvent(IngestJobEvent.CANCELLED.toString(), ingestJobId, null);
        eventPublishingExecutor.submit(new PublishEventTask(event, jobEventPublisher));
    }

    void fireDataSourceAnalysisStarted(long ingestJobId, Content dataSource) {
        AutopsyEvent event = new DataSourceAnalysisStartedEvent(ingestJobId, dataSource);
        eventPublishingExecutor.submit(new PublishEventTask(event, jobEventPublisher));
    }

    void fireDataSourceAnalysisCompleted(long ingestJobId, Content dataSource) {
        AutopsyEvent event = new DataSourceAnalysisCompletedEvent(ingestJobId, dataSource, DataSourceAnalysisCompletedEvent.Reason.ANALYSIS_COMPLETED);
        eventPublishingExecutor.submit(new PublishEventTask(event, jobEventPublisher));
    }

    void fireDataSourceAnalysisCancelled(long ingestJobId, Content dataSource) {
        AutopsyEvent event = new DataSourceAnalysisCompletedEvent(ingestJobId, dataSource, DataSourceAnalysisCompletedEvent.Reason.ANALYSIS_CANCELLED);
        eventPublishingExecutor.submit(new PublishEventTask(event, jobEventPublisher));
    }

    void fireFileIngestDone(AbstractFile file) {
        AutopsyEvent event = new FileAnalyzedEvent(file);
        eventPublishingExecutor.submit(new PublishEventTask(event, moduleEventPublisher));
    }

    void fireIngestModuleContentEvent(ModuleContentEvent moduleContentEvent) {
        AutopsyEvent event = new ContentChangedEvent(moduleContentEvent);
        eventPublishingExecutor.submit(new PublishEventTask(event, moduleEventPublisher));
    }

    void initIngestMessageInbox() {
        synchronized (this.ingestMessageBoxLock) {
            ingestMessageBox = IngestMessageTopComponent.findInstance();
        }
    }

    void postIngestMessage(IngestMessage message) {
        synchronized (this.ingestMessageBoxLock) {
            if (ingestMessageBox != null && RuntimeProperties.runningWithGUI()) {
                if (message.getMessageType() != IngestMessage.MessageType.ERROR && message.getMessageType() != IngestMessage.MessageType.WARNING) {
                    ingestMessageBox.displayMessage(message);
                } else {
                    long errorPosts = ingestErrorMessagePosts.incrementAndGet();
                    if (errorPosts <= MAX_ERROR_MESSAGE_POSTS) {
                        ingestMessageBox.displayMessage(message);
                    } else if (errorPosts == MAX_ERROR_MESSAGE_POSTS + 1) {
                        IngestMessage errorMessageLimitReachedMessage = IngestMessage.createErrorMessage(
                                NbBundle.getMessage(this.getClass(), "IngestManager.IngestMessage.ErrorMessageLimitReached.title"),
                                NbBundle.getMessage(this.getClass(), "IngestManager.IngestMessage.ErrorMessageLimitReached.subject"),
                                NbBundle.getMessage(this.getClass(), "IngestManager.IngestMessage.ErrorMessageLimitReached.msg", MAX_ERROR_MESSAGE_POSTS));
                        ingestMessageBox.displayMessage(errorMessageLimitReachedMessage);
                    }
                }
            }
        }
    }
 
    /*
     * Clears the ingest messages inbox.
     */
    private void clearIngestMessageBox() {
        synchronized (this.ingestMessageBoxLock) {
            if (null != ingestMessageBox) {
                ingestMessageBox.clearMessages();
            }
            ingestErrorMessagePosts.set(0);
        }
    }

    void setIngestTaskProgress(IngestTask task, String currentModuleName) {
        IngestThreadActivitySnapshot prevSnap = ingestThreadActivitySnapshots.get(task.getThreadId());
        IngestThreadActivitySnapshot newSnap = new IngestThreadActivitySnapshot(task.getThreadId(), task.getIngestJobExecutor().getIngestJobId(), currentModuleName, task.getDataSource(), task.getContentName());
        ingestThreadActivitySnapshots.put(task.getThreadId(), newSnap);
        incrementModuleRunTime(prevSnap.getModuleDisplayName(), newSnap.getStartTime().getTime() - prevSnap.getStartTime().getTime());
    }

    void setIngestTaskProgressCompleted(IngestTask task) {
        IngestThreadActivitySnapshot prevSnap = ingestThreadActivitySnapshots.get(task.getThreadId());
        IngestThreadActivitySnapshot newSnap = new IngestThreadActivitySnapshot(task.getThreadId());
        ingestThreadActivitySnapshots.put(task.getThreadId(), newSnap);
        incrementModuleRunTime(prevSnap.getModuleDisplayName(), newSnap.getStartTime().getTime() - prevSnap.getStartTime().getTime());
    }

    void incrementModuleRunTime(String moduleDisplayName, Long duration) {
        if (moduleDisplayName.equals("IDLE")) { //NON-NLS
            return;
        }
 
        synchronized (ingestModuleRunTimes) {
            Long prevTimeL = ingestModuleRunTimes.get(moduleDisplayName);
            long prevTime = 0;
            if (prevTimeL != null) {
                prevTime = prevTimeL;
            }
            prevTime += duration;
            ingestModuleRunTimes.put(moduleDisplayName, prevTime);
        }
    }

    @Override
    public Map<String, Long> getModuleRunTimes() {
        synchronized (ingestModuleRunTimes) {
            Map<String, Long> times = new HashMap<>(ingestModuleRunTimes);
            return times;
        }
    }

    @Override
    public List<IngestThreadActivitySnapshot> getIngestThreadActivitySnapshots() {
        return new ArrayList<>(ingestThreadActivitySnapshots.values());
    }

    @Override
    public List<IngestJobProgressSnapshot> getIngestJobSnapshots() {
        List<IngestJobProgressSnapshot> snapShots = new ArrayList<>();
        synchronized (ingestJobsById) {
            ingestJobsById.values().forEach((job) -> {
                IngestJobProgressSnapshot snapshot = job.getDiagnosticStatsSnapshot();
                if (snapshot != null) {
                    snapShots.add(snapshot);
                }
            });
        }
        return snapShots;
    }

    long getFreeDiskSpace() {
        if (ingestMonitor != null) {
            return ingestMonitor.getFreeSpace();
        } else {
            return -1;
        }
    }

    private final class StartIngestJobTask implements Callable<Void> {
 
        private final long threadId;
        private final IngestJob job;
        private ProgressHandle progress;
 
        StartIngestJobTask(long threadId, IngestJob job) {
            this.threadId = threadId;
            this.job = job;
        }
 
        @Override
        public Void call() {
            try {
                if (Thread.currentThread().isInterrupted()) {
                    synchronized (ingestJobsById) {
                        ingestJobsById.remove(job.getId());
                    }
                    return null;
                }
 
                if (RuntimeProperties.runningWithGUI()) {
                    final String displayName = NbBundle.getMessage(this.getClass(), "IngestManager.StartIngestJobsTask.run.displayName");
                    this.progress = ProgressHandle.createHandle(displayName, new Cancellable() {
                        @Override
                        public boolean cancel() {
                            if (progress != null) {
                                progress.setDisplayName(NbBundle.getMessage(this.getClass(), "IngestManager.StartIngestJobsTask.run.cancelling", displayName));
                            }
                            synchronized (startIngestJobFutures) {
                                Future<?> handle = startIngestJobFutures.remove(threadId);
                                handle.cancel(true);
                            }
                            return true;
                        }
                    });
                    progress.start();
                }
 
                startIngestJob(job);
                return null;
 
            } finally {
                if (null != progress) {
                    progress.finish();
                }
                synchronized (startIngestJobFutures) {
                    startIngestJobFutures.remove(threadId);
                }
            }
        }
 
    }

    private final class ExecuteIngestJobTasksTask implements Runnable {
 
        private final long threadId;
        private final BlockingIngestTaskQueue tasks;
 
        ExecuteIngestJobTasksTask(long threadId, BlockingIngestTaskQueue tasks) {
            this.threadId = threadId;
            this.tasks = tasks;
        }
 
        @Override
        public void run() {
            while (true) {
                try {
                    IngestTask task = tasks.getNextTask(); // Blocks.
                    task.execute(threadId);
                } catch (InterruptedException ex) {
                    break;
                }
                if (Thread.currentThread().isInterrupted()) {
                    break;
                }
            }
        }
    }

    private static final class PublishEventTask implements Runnable {
 
        private final AutopsyEvent event;
        private final AutopsyEventPublisher publisher;

        PublishEventTask(AutopsyEvent event, AutopsyEventPublisher publisher) {
            this.event = event;
            this.publisher = publisher;
        }
 
        @Override
        public void run() {
            publisher.publish(event);
        }
 
    }

    @Immutable
    public static final class IngestThreadActivitySnapshot implements Serializable {
 
        private static final long serialVersionUID = 1L;
 
        private final long threadId;
        private final Date startTime;
        private final String moduleDisplayName;
        private final String dataSourceName;
        private final String fileName;
        private final long jobId;

        IngestThreadActivitySnapshot(long threadId) {
            this.threadId = threadId;
            startTime = new Date();
            this.moduleDisplayName = NbBundle.getMessage(this.getClass(), "IngestManager.IngestThreadActivitySnapshot.idleThread");
            this.dataSourceName = "";
            this.fileName = "";
            this.jobId = 0;
        }

        IngestThreadActivitySnapshot(long threadId, long jobId, String moduleDisplayName, Content dataSource) {
            this.threadId = threadId;
            this.jobId = jobId;
            startTime = new Date();
            this.moduleDisplayName = moduleDisplayName;
            this.dataSourceName = dataSource.getName();
            this.fileName = "";
        }

        IngestThreadActivitySnapshot(long threadId, long jobId, String moduleDisplayName, Content dataSource, String fileName) {
            this.threadId = threadId;
            this.jobId = jobId;
            startTime = new Date();
            this.moduleDisplayName = moduleDisplayName;
            this.dataSourceName = dataSource.getName();
            this.fileName = fileName;
        }

        long getIngestJobId() {
            return jobId;
        }

        long getThreadId() {
            return threadId;
        }

        Date getStartTime() {
            return new Date(startTime.getTime());
        }

        String getModuleDisplayName() {
            return moduleDisplayName;
        }

        String getDataSourceName() {
            return dataSourceName;
        }

        String getFileName() {
            return fileName;
        }
 
    }

    public enum IngestJobEvent {

        STARTED,
        COMPLETED,
        CANCELLED,
        DATA_SOURCE_ANALYSIS_STARTED,
        DATA_SOURCE_ANALYSIS_COMPLETED,
    }

    public enum IngestModuleEvent {

        DATA_ADDED,
        CONTENT_CHANGED,
        FILE_DONE,
    }

    public final static class IngestManagerException extends Exception {
 
        private static final long serialVersionUID = 1L;

        private IngestManagerException(String message) {
            super(message);
        }

        private IngestManagerException(String message, Throwable cause) {
            super(message, cause);
        }
    }

    @Deprecated
    public static void addPropertyChangeListener(final PropertyChangeListener listener) {
        instance.addIngestJobEventListener(listener);
        instance.addIngestModuleEventListener(listener);
    }

    @Deprecated
    public static void removePropertyChangeListener(final PropertyChangeListener listener) {
        instance.removeIngestJobEventListener(listener);
        instance.removeIngestModuleEventListener(listener);
    }

    @Deprecated
    public synchronized IngestJob startIngestJob(Collection<Content> dataSources, IngestJobSettings settings) {
        return beginIngestJob(dataSources, settings).getJob();
    }

    @Deprecated
    public void cancelAllIngestJobs() {
        cancelAllIngestJobs(IngestJob.CancellationReason.USER_CANCELLED);
    }
 
}