IngestManager.java

Go to the documentation of this file.

/*
 * Autopsy Forensic Browser
 *
 * Copyright 2013-2014 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.util.concurrent.ThreadFactoryBuilder;
import java.awt.EventQueue;
import java.beans.PropertyChangeEvent;
import java.beans.PropertyChangeListener;
import java.beans.PropertyChangeSupport;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
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 javax.swing.JOptionPane;
import org.netbeans.api.progress.ProgressHandle;
import org.netbeans.api.progress.ProgressHandleFactory;
import org.openide.util.Cancellable;
import org.openide.util.NbBundle;
import org.sleuthkit.autopsy.casemodule.Case;
import org.sleuthkit.autopsy.core.UserPreferences;
import org.sleuthkit.autopsy.coreutils.Logger;
import org.sleuthkit.autopsy.coreutils.MessageNotifyUtil;
import org.sleuthkit.datamodel.AbstractFile;
import org.sleuthkit.datamodel.Content;

public class IngestManager {

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

    private final ConcurrentHashMap<Long, IngestJob> jobsById;

    private final AtomicLong nextThreadId;

    private final ConcurrentHashMap<Long, Future<Void>> ingestJobStarters;
    private final ExecutorService startIngestJobsThreadPool;

    private final ExecutorService dataSourceIngestThreadPool;
    private static final int MIN_NUMBER_OF_FILE_INGEST_THREADS = 1;
    private static final int MAX_NUMBER_OF_FILE_INGEST_THREADS = 16;
    private static final int DEFAULT_NUMBER_OF_FILE_INGEST_THREADS = 2;
    private int numberOfFileIngestThreads;
    private final ExecutorService fileIngestThreadPool;

    private final PropertyChangeSupport ingestJobEventPublisher;
    private final PropertyChangeSupport ingestModuleEventPublisher;
    private final ExecutorService fireIngestEventsThreadPool;

    private final IngestMonitor ingestMonitor;

    private static final int MAX_ERROR_MESSAGE_POSTS = 200;
    private volatile IngestMessageTopComponent ingestMessageBox;
    private final AtomicLong ingestErrorMessagePosts;

    private final ConcurrentHashMap<Long, IngestThreadActivitySnapshot> ingestThreadActivitySnapshots;
    private final ConcurrentHashMap<String, Long> ingestModuleRunTimes;

    private volatile boolean jobCreationIsEnabled;

    private volatile boolean runInteractively;

    public enum IngestJobEvent {

        STARTED,
        COMPLETED,
        CANCELLED,
    };

    public enum IngestModuleEvent {

        DATA_ADDED,
        CONTENT_CHANGED,
        FILE_DONE,
    };

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

    private IngestManager() {
        this.runInteractively = true;
        this.ingestModuleRunTimes = new ConcurrentHashMap<>();
        this.ingestThreadActivitySnapshots = new ConcurrentHashMap<>();
        this.ingestErrorMessagePosts = new AtomicLong(0L);
        this.ingestMonitor = new IngestMonitor();
        this.ingestModuleEventPublisher = new PropertyChangeSupport(IngestManager.class);
        this.fireIngestEventsThreadPool = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().setNameFormat("IM-ingest-events-%d").build()); //NON-NLS
        this.ingestJobEventPublisher = new PropertyChangeSupport(IngestManager.class);
        this.dataSourceIngestThreadPool = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().setNameFormat("IM-data-source-ingest-%d").build()); //NON-NLS
        this.startIngestJobsThreadPool = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().setNameFormat("IM-start-ingest-jobs-%d").build()); //NON-NLS
        this.nextThreadId = new AtomicLong(0L);
        this.jobsById = new ConcurrentHashMap<>();
        this.ingestJobStarters = new ConcurrentHashMap<>();

        this.startDataSourceIngestThread();

        numberOfFileIngestThreads = UserPreferences.numberOfFileIngestThreads();
        if ((numberOfFileIngestThreads < MIN_NUMBER_OF_FILE_INGEST_THREADS) || (numberOfFileIngestThreads > MAX_NUMBER_OF_FILE_INGEST_THREADS)) {
            numberOfFileIngestThreads = DEFAULT_NUMBER_OF_FILE_INGEST_THREADS;
            UserPreferences.setNumberOfFileIngestThreads(numberOfFileIngestThreads);
        }
        fileIngestThreadPool = Executors.newFixedThreadPool(numberOfFileIngestThreads, new ThreadFactoryBuilder().setNameFormat("IM-file-ingest-%d").build()); //NON-NLS
        for (int i = 0; i < numberOfFileIngestThreads; ++i) {
            startFileIngestThread();
        }
    }

    private void startDataSourceIngestThread() {
        long threadId = nextThreadId.incrementAndGet();
        dataSourceIngestThreadPool.submit(new IngestTaskExecuter(threadId, IngestTasksScheduler.getInstance().getDataSourceIngestTaskQueue()));
        ingestThreadActivitySnapshots.put(threadId, new IngestThreadActivitySnapshot(threadId));
    }

    private void startFileIngestThread() {
        long threadId = nextThreadId.incrementAndGet();
        fileIngestThreadPool.submit(new IngestTaskExecuter(threadId, IngestTasksScheduler.getInstance().getFileIngestTaskQueue()));
        ingestThreadActivitySnapshots.put(threadId, new IngestThreadActivitySnapshot(threadId));
    }

    private void subscribeToCaseEvents() {
        Case.addPropertyChangeListener(new PropertyChangeListener() {
            @Override
            public void propertyChange(PropertyChangeEvent event) {
                if (event.getPropertyName().equals(Case.Events.CURRENT_CASE.toString())) {
                    if (event.getNewValue() != null) {
                        handleCaseOpened();
                    } else {
                        handleCaseClosed();
                    }
                }
            }
        });
    }

    synchronized void handleCaseOpened() {
        this.jobCreationIsEnabled = true;
        clearIngestMessageBox();
    }

    synchronized void handleCaseClosed() {
        this.jobCreationIsEnabled = false;
        cancelAllIngestJobs();
        clearIngestMessageBox();
    }

    public synchronized void setRunInteractively(boolean runInteractively) {
        this.runInteractively = runInteractively;
    }

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

    synchronized void postIngestMessage(IngestMessage message) {
        if (ingestMessageBox != null && this.runInteractively) {
            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);
                }
            }
        }
    }

    private void clearIngestMessageBox() {
        if (ingestMessageBox != null) {
            ingestMessageBox.clearMessages();
        }
        ingestErrorMessagePosts.set(0);
    }

    public int getNumberOfFileIngestThreads() {
        return numberOfFileIngestThreads;
    }

    public synchronized void queueIngestJob(Collection<Content> dataSources, IngestJobSettings settings) {
        if (this.jobCreationIsEnabled) {
            IngestJob job = new IngestJob(dataSources, settings, this.runInteractively);
            if (job.hasIngestPipeline()) {
                long taskId = nextThreadId.incrementAndGet();
                Future<Void> task = startIngestJobsThreadPool.submit(new IngestJobStarter(taskId, job));
                ingestJobStarters.put(taskId, task);
            }
        }
    }

    public synchronized IngestJob startIngestJob(Collection<Content> dataSources, IngestJobSettings settings) {
        if (this.jobCreationIsEnabled) {
            IngestJob job = new IngestJob(dataSources, settings, this.runInteractively);
            if (job.hasIngestPipeline()) {
                if (this.startIngestJob(job)) {
                    return job;
                }
            }
        }
        return null;
    }

    private boolean startIngestJob(IngestJob job) {
        boolean success = false;
        if (this.jobCreationIsEnabled) {
            if (runInteractively && jobsById.size() == 1) {
                clearIngestMessageBox();
            }

            if (!ingestMonitor.isRunning()) {
                ingestMonitor.start();
            }

            this.jobsById.put(job.getId(), job);
            List<IngestModuleError> errors = job.start();
            if (errors.isEmpty()) {
                this.fireIngestJobStarted(job.getId());
                IngestManager.logger.log(Level.INFO, "Ingest job {0} started", job.getId()); //NON-NLS
                success = true;
            } else {
                this.jobsById.remove(job.getId());
                for (IngestModuleError error : errors) {
                    logger.log(Level.SEVERE, String.format("Error starting %s ingest module", error.getModuleDisplayName()), error.getModuleError()); //NON-NLS
                }
                IngestManager.logger.log(Level.INFO, "Ingest job {0} could not be started", job.getId()); //NON-NLS
                if (this.runInteractively) {
                    EventQueue.invokeLater(new Runnable() {

                        @Override
                        public void run() {
                            StringBuilder moduleStartUpErrors = new StringBuilder();
                            for (IngestModuleError error : errors) {
                                String moduleName = error.getModuleDisplayName();
                                moduleStartUpErrors.append(moduleName);
                                moduleStartUpErrors.append(": ");
                                moduleStartUpErrors.append(error.getModuleError().getLocalizedMessage());
                                moduleStartUpErrors.append("\n");
                            }
                            StringBuilder notifyMessage = new StringBuilder();
                            notifyMessage.append(NbBundle.getMessage(this.getClass(),
                                    "IngestManager.StartIngestJobsTask.run.startupErr.dlgMsg"));
                            notifyMessage.append("\n");
                            notifyMessage.append(NbBundle.getMessage(this.getClass(),
                                    "IngestManager.StartIngestJobsTask.run.startupErr.dlgSolution"));
                            notifyMessage.append("\n");
                            notifyMessage.append(NbBundle.getMessage(this.getClass(),
                                    "IngestManager.StartIngestJobsTask.run.startupErr.dlgErrorList",
                                    moduleStartUpErrors.toString()));
                            notifyMessage.append("\n\n");
                            JOptionPane.showMessageDialog(null, notifyMessage.toString(),
                                    NbBundle.getMessage(this.getClass(),
                                            "IngestManager.StartIngestJobsTask.run.startupErr.dlgTitle"), JOptionPane.ERROR_MESSAGE);
                        }
                    });
                }
            }
        }
        return success;
    }

    synchronized void finishIngestJob(IngestJob job) {
        long jobId = job.getId();
        this.jobsById.remove(jobId);
        if (!job.isCancelled()) {
            IngestManager.logger.log(Level.INFO, "Ingest job {0} completed", jobId); //NON-NLS
            this.fireIngestJobCompleted(jobId);
        } else {
            IngestManager.logger.log(Level.INFO, "Ingest job {0} cancelled", jobId); //NON-NLS
            this.fireIngestJobCancelled(jobId);
        }
    }

    public boolean isIngestRunning() {
        return !this.jobsById.isEmpty();
    }

    public synchronized void cancelAllIngestJobs() {
        // Stop creating new ingest jobs.
        for (Future<Void> handle : ingestJobStarters.values()) {
            handle.cancel(true);
        }

        // Cancel all the jobs already created. 
        for (IngestJob job : this.jobsById.values()) {
            job.cancel();
        }
    }

    public void addIngestJobEventListener(final PropertyChangeListener listener) {
        ingestJobEventPublisher.addPropertyChangeListener(listener);
    }

    public void removeIngestJobEventListener(final PropertyChangeListener listener) {
        ingestJobEventPublisher.removePropertyChangeListener(listener);
    }

    public void addIngestModuleEventListener(final PropertyChangeListener listener) {
        ingestModuleEventPublisher.addPropertyChangeListener(listener);
    }

    public void removeIngestModuleEventListener(final PropertyChangeListener listener) {
        ingestModuleEventPublisher.removePropertyChangeListener(listener);
    }

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

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

    void fireIngestJobStarted(long ingestJobId) {
        fireIngestEventsThreadPool.submit(new IngestEventPublisher(ingestJobEventPublisher, IngestJobEvent.STARTED, ingestJobId, null));
    }

    void fireIngestJobCompleted(long ingestJobId) {
        fireIngestEventsThreadPool.submit(new IngestEventPublisher(ingestJobEventPublisher, IngestJobEvent.COMPLETED, ingestJobId, null));
    }

    void fireIngestJobCancelled(long ingestJobId) {
        fireIngestEventsThreadPool.submit(new IngestEventPublisher(ingestJobEventPublisher, IngestJobEvent.CANCELLED, ingestJobId, null));
    }

    void fireFileIngestDone(AbstractFile file) {
        fireIngestEventsThreadPool.submit(new IngestEventPublisher(ingestModuleEventPublisher, IngestModuleEvent.FILE_DONE, file.getId(), file));
    }

    void fireIngestModuleDataEvent(ModuleDataEvent moduleDataEvent) {
        fireIngestEventsThreadPool.submit(new IngestEventPublisher(ingestModuleEventPublisher, IngestModuleEvent.DATA_ADDED, moduleDataEvent, null));
    }

    void fireIngestModuleContentEvent(ModuleContentEvent moduleContentEvent) {
        fireIngestEventsThreadPool.submit(new IngestEventPublisher(ingestModuleEventPublisher, IngestModuleEvent.CONTENT_CHANGED, moduleContentEvent, null));
    }

    void setIngestTaskProgress(DataSourceIngestTask task, String ingestModuleDisplayName) {
        ingestThreadActivitySnapshots.put(task.getThreadId(), new IngestThreadActivitySnapshot(task.getThreadId(), task.getIngestJob().getId(), ingestModuleDisplayName, task.getDataSource()));
    }

    void setIngestTaskProgress(FileIngestTask task, String ingestModuleDisplayName) {
        IngestThreadActivitySnapshot prevSnap = ingestThreadActivitySnapshots.get(task.getThreadId());
        IngestThreadActivitySnapshot newSnap = new IngestThreadActivitySnapshot(task.getThreadId(), task.getIngestJob().getId(), ingestModuleDisplayName, task.getDataSource(), task.getFile());
        ingestThreadActivitySnapshots.put(task.getThreadId(), newSnap);

        incrementModuleRunTime(prevSnap.getActivity(), newSnap.getStartTime().getTime() - prevSnap.getStartTime().getTime());
    }

    void setIngestTaskProgressCompleted(DataSourceIngestTask task) {
        ingestThreadActivitySnapshots.put(task.getThreadId(), new IngestThreadActivitySnapshot(task.getThreadId()));
    }

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

    private void incrementModuleRunTime(String moduleName, Long duration) {
        if (moduleName.equals("IDLE")) { //NON-NLS
            return;
        }

        synchronized (ingestModuleRunTimes) {
            Long prevTimeL = ingestModuleRunTimes.get(moduleName);
            long prevTime = 0;
            if (prevTimeL != null) {
                prevTime = prevTimeL;
            }
            prevTime += duration;
            ingestModuleRunTimes.put(moduleName, prevTime);
        }
    }

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

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

    List<DataSourceIngestJob.Snapshot> getIngestJobSnapshots() {
        List<DataSourceIngestJob.Snapshot> snapShots = new ArrayList<>();
        for (IngestJob job : this.jobsById.values()) {
            snapShots.addAll(job.getDataSourceIngestJobSnapshots());
        }
        return snapShots;
    }

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

    private final class IngestJobStarter implements Callable<Void> {

        private final long threadId;
        private final IngestJob job;
        private ProgressHandle progress;

        IngestJobStarter(long threadId, IngestJob job) {
            this.threadId = threadId;
            this.job = job;
        }

        @Override
        public Void call() {
            try {
                if (Thread.currentThread().isInterrupted()) {
                    jobsById.remove(job.getId());
                    return null;
                }

                if (runInteractively) {
                    final String displayName = NbBundle.getMessage(this.getClass(), "IngestManager.StartIngestJobsTask.run.displayName");
                    this.progress = ProgressHandleFactory.createHandle(displayName, new Cancellable() {
                        @Override
                        public boolean cancel() {
                            if (progress != null) {
                                progress.setDisplayName(NbBundle.getMessage(this.getClass(), "IngestManager.StartIngestJobsTask.run.cancelling", displayName));
                            }
                            Future<?> handle = ingestJobStarters.remove(threadId);
                            handle.cancel(true);
                            return true;
                        }
                    });
                    progress.start();
                }

                startIngestJob(job);
                return null;

            } finally {
                if (null != progress) {
                    progress.finish();
                }
                ingestJobStarters.remove(threadId);
            }
        }

    }

    private final class IngestTaskExecuter implements Runnable {

        private final long threadId;
        private final IngestTaskQueue tasks;

        IngestTaskExecuter(long threadId, IngestTaskQueue 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 IngestEventPublisher implements Runnable {

        private final PropertyChangeSupport publisher;
        private final IngestJobEvent jobEvent;
        private final IngestModuleEvent moduleEvent;
        private final Object oldValue;
        private final Object newValue;

        IngestEventPublisher(PropertyChangeSupport publisher, IngestJobEvent event, Object oldValue, Object newValue) {
            this.publisher = publisher;
            this.jobEvent = event;
            this.moduleEvent = null;
            this.oldValue = oldValue;
            this.newValue = newValue;
        }

        IngestEventPublisher(PropertyChangeSupport publisher, IngestModuleEvent event, Object oldValue, Object newValue) {
            this.publisher = publisher;
            this.jobEvent = null;
            this.moduleEvent = event;
            this.oldValue = oldValue;
            this.newValue = newValue;
        }

        @Override
        public void run() {
            try {
                publisher.firePropertyChange((jobEvent != null ? jobEvent.toString() : moduleEvent.toString()), oldValue, newValue);
            } catch (Exception e) {
                logger.log(Level.SEVERE, "Ingest manager listener threw exception", e); //NON-NLS
                MessageNotifyUtil.Notify.show(NbBundle.getMessage(IngestManager.class, "IngestManager.moduleErr"),
                        NbBundle.getMessage(IngestManager.class, "IngestManager.moduleErr.errListenToUpdates.msg"),
                        MessageNotifyUtil.MessageType.ERROR);
            }
        }
    }

    static final class IngestThreadActivitySnapshot {

        private final long threadId;
        private final Date startTime;
        private final String activity;
        private final String dataSourceName;
        private final String fileName;
        private final long jobId;

        // nothing is running on the thread
        IngestThreadActivitySnapshot(long threadId) {
            this.threadId = threadId;
            startTime = new Date();
            this.activity = NbBundle.getMessage(this.getClass(), "IngestManager.IngestThreadActivitySnapshot.idleThread");
            this.dataSourceName = "";
            this.fileName = "";
            this.jobId = 0;
        }

        // data souce thread
        IngestThreadActivitySnapshot(long threadId, long jobId, String activity, Content dataSource) {
            this.threadId = threadId;
            this.jobId = jobId;
            startTime = new Date();
            this.activity = activity;
            this.dataSourceName = dataSource.getName();
            this.fileName = "";
        }

        // file ingest thread
        IngestThreadActivitySnapshot(long threadId, long jobId, String activity, Content dataSource, AbstractFile file) {
            this.threadId = threadId;
            this.jobId = jobId;
            startTime = new Date();
            this.activity = activity;
            this.dataSourceName = dataSource.getName();
            this.fileName = file.getName();
        }

        long getJobId() {
            return jobId;
        }

        long getThreadId() {
            return threadId;
        }

        Date getStartTime() {
            return startTime;
        }

        String getActivity() {
            return activity;
        }

        String getDataSourceName() {
            return dataSourceName;
        }

        String getFileName() {
            return fileName;
        }

    }

}