Autopsy  4.21.0
Graphical digital forensics platform for The Sleuth Kit and other tools.
IngestManager.java
Go to the documentation of this file.
1 /*
2  * Autopsy Forensic Browser
3  *
4  * Copyright 2012-2021 Basis Technology Corp.
5  * Contact: carrier <at> sleuthkit <dot> org
6  *
7  * Licensed under the Apache License, Version 2.0 (the "License");
8  * you may not use this file except in compliance with the License.
9  * You may obtain a copy of the License at
10  *
11  * http://www.apache.org/licenses/LICENSE-2.0
12  *
13  * Unless required by applicable law or agreed to in writing, software
14  * distributed under the License is distributed on an "AS IS" BASIS,
15  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16  * See the License for the specific language governing permissions and
17  * limitations under the License.
18  */
19 package org.sleuthkit.autopsy.ingest;
20 
21 import com.google.common.eventbus.Subscribe;
22 import com.google.common.util.concurrent.ThreadFactoryBuilder;
23 import java.awt.EventQueue;
24 import java.beans.PropertyChangeEvent;
25 import java.beans.PropertyChangeListener;
26 import java.io.Serializable;
27 import java.lang.reflect.InvocationTargetException;
28 import java.util.ArrayList;
29 import java.util.Collection;
30 import java.util.Collections;
31 import java.util.Date;
32 import java.util.EnumSet;
33 import java.util.HashMap;
34 import java.util.HashSet;
35 import java.util.List;
36 import java.util.Map;
37 import java.util.Optional;
38 import java.util.Set;
39 import java.util.concurrent.Callable;
40 import java.util.concurrent.ConcurrentHashMap;
41 import java.util.concurrent.ExecutorService;
42 import java.util.concurrent.Executors;
43 import java.util.concurrent.Future;
44 import java.util.concurrent.atomic.AtomicLong;
45 import java.util.logging.Level;
46 import java.util.stream.Collectors;
47 import java.util.stream.Stream;
48 import javax.annotation.concurrent.GuardedBy;
49 import javax.annotation.concurrent.Immutable;
50 import javax.annotation.concurrent.ThreadSafe;
51 import javax.swing.JOptionPane;
52 import javax.swing.SwingUtilities;
53 import org.netbeans.api.progress.ProgressHandle;
54 import org.openide.util.Cancellable;
55 import org.openide.util.NbBundle;
56 import org.openide.windows.WindowManager;
72 import org.sleuthkit.datamodel.AbstractFile;
73 import org.sleuthkit.datamodel.AnalysisResult;
74 import org.sleuthkit.datamodel.Blackboard;
75 import org.sleuthkit.datamodel.BlackboardArtifact;
76 import org.sleuthkit.datamodel.Content;
77 import org.sleuthkit.datamodel.DataArtifact;
78 import org.sleuthkit.datamodel.DataSource;
79 import org.sleuthkit.datamodel.TskCoreException;
80 
120 @ThreadSafe
122 
123  private final static Logger logger = Logger.getLogger(IngestManager.class.getName());
124  private final static String INGEST_JOB_EVENT_CHANNEL_NAME = "%s-Ingest-Job-Events"; //NON-NLS
125  private final static Set<String> INGEST_JOB_EVENT_NAMES = Stream.of(IngestJobEvent.values()).map(IngestJobEvent::toString).collect(Collectors.toSet());
126  private final static String INGEST_MODULE_EVENT_CHANNEL_NAME = "%s-Ingest-Module-Events"; //NON-NLS
127  private final static Set<String> INGEST_MODULE_EVENT_NAMES = Stream.of(IngestModuleEvent.values()).map(IngestModuleEvent::toString).collect(Collectors.toSet());
128  private final static int MAX_ERROR_MESSAGE_POSTS = 200;
129  @GuardedBy("IngestManager.class")
130  private static IngestManager instance;
131  private final int numberOfFileIngestThreads;
132  private final AtomicLong nextIngestManagerTaskId = new AtomicLong(0L);
133  private final ExecutorService startIngestJobsExecutor = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().setNameFormat("IM-start-ingest-jobs-%d").build()); //NON-NLS;
134  @GuardedBy("startIngestJobFutures")
135  private final Map<Long, Future<Void>> startIngestJobFutures = new ConcurrentHashMap<>();
136  @GuardedBy("ingestJobsById")
137  private final Map<Long, IngestJob> ingestJobsById = new HashMap<>();
138  private final ExecutorService dataSourceLevelIngestJobTasksExecutor;
139  private final ExecutorService fileLevelIngestJobTasksExecutor;
140  private final ExecutorService dataArtifactIngestTasksExecutor;
141  private final ExecutorService analysisResultIngestTasksExecutor;
142  private final ExecutorService eventPublishingExecutor = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().setNameFormat("IM-ingest-events-%d").build()); //NON-NLS;
146  private final AutopsyEventPublisher moduleEventPublisher = new AutopsyEventPublisher();
147  private final Object ingestMessageBoxLock = new Object();
148  private final AtomicLong ingestErrorMessagePosts = new AtomicLong(0L);
149  private final ConcurrentHashMap<Long, IngestThreadActivitySnapshot> ingestThreadActivitySnapshots = new ConcurrentHashMap<>();
150  private final ConcurrentHashMap<String, Long> ingestModuleRunTimes = new ConcurrentHashMap<>();
151  private volatile IngestMessageTopComponent ingestMessageBox;
152  private volatile boolean caseIsOpen;
153 
160  public synchronized static IngestManager getInstance() {
161  if (null == instance) {
162  instance = new IngestManager();
163  instance.subscribeToServiceMonitorEvents();
164  instance.subscribeToCaseEvents();
165  }
166  return instance;
167  }
168 
173  private IngestManager() {
174  dataSourceLevelIngestJobTasksExecutor = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().setNameFormat("IM-data-source-ingest-%d").build()); //NON-NLS;
175  long threadId = nextIngestManagerTaskId.incrementAndGet();
176  dataSourceLevelIngestJobTasksExecutor.submit(new ExecuteIngestJobTasksTask(threadId, IngestTasksScheduler.getInstance().getDataSourceIngestTaskQueue()));
177  ingestThreadActivitySnapshots.put(threadId, new IngestThreadActivitySnapshot(threadId));
178 
180  fileLevelIngestJobTasksExecutor = Executors.newFixedThreadPool(numberOfFileIngestThreads, new ThreadFactoryBuilder().setNameFormat("IM-file-ingest-%d").build()); //NON-NLS
181  for (int i = 0; i < numberOfFileIngestThreads; ++i) {
182  threadId = nextIngestManagerTaskId.incrementAndGet();
183  fileLevelIngestJobTasksExecutor.submit(new ExecuteIngestJobTasksTask(threadId, IngestTasksScheduler.getInstance().getFileIngestTaskQueue()));
184  ingestThreadActivitySnapshots.put(threadId, new IngestThreadActivitySnapshot(threadId));
185  }
186 
187  dataArtifactIngestTasksExecutor = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().setNameFormat("IM-data-artifact-ingest-%d").build()); //NON-NLS;
188  threadId = nextIngestManagerTaskId.incrementAndGet();
189  dataArtifactIngestTasksExecutor.submit(new ExecuteIngestJobTasksTask(threadId, IngestTasksScheduler.getInstance().getDataArtifactIngestTaskQueue()));
190  ingestThreadActivitySnapshots.put(threadId, new IngestThreadActivitySnapshot(threadId));
191 
192  analysisResultIngestTasksExecutor = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder().setNameFormat("IM-analysis-result-ingest-%d").build()); //NON-NLS;
193  threadId = nextIngestManagerTaskId.incrementAndGet();
194  analysisResultIngestTasksExecutor.submit(new ExecuteIngestJobTasksTask(threadId, IngestTasksScheduler.getInstance().getAnalysisResultIngestTaskQueue()));
195  ingestThreadActivitySnapshots.put(threadId, new IngestThreadActivitySnapshot(threadId));
196  }
197 
203  PropertyChangeListener propChangeListener = (PropertyChangeEvent evt) -> {
204  if (evt.getNewValue().equals(ServicesMonitor.ServiceStatus.DOWN.toString())) {
205  /*
206  * The application services considered to be key services are
207  * only necessary for multi-user cases.
208  */
209  try {
211  return;
212  }
213  } catch (NoCurrentCaseException noCaseOpenException) {
214  return;
215  }
216 
217  String serviceDisplayName = ServicesMonitor.Service.valueOf(evt.getPropertyName()).getDisplayName();
218  logger.log(Level.SEVERE, "Service {0} is down, cancelling all running ingest jobs", serviceDisplayName); //NON-NLS
220  EventQueue.invokeLater(new Runnable() {
221  @Override
222  public void run() {
223  JOptionPane.showMessageDialog(WindowManager.getDefault().getMainWindow(),
224  NbBundle.getMessage(this.getClass(), "IngestManager.cancellingIngest.msgDlg.text"),
225  NbBundle.getMessage(this.getClass(), "IngestManager.serviceIsDown.msgDlg.text", serviceDisplayName),
226  JOptionPane.ERROR_MESSAGE);
227  }
228  });
229  }
231  }
232  };
233 
234  /*
235  * The key services for multi-user cases are currently the case database
236  * server and the Solr server. The Solr server is a key service not
237  * because search is essential, but because the coordination service
238  * (ZooKeeper) is running embedded within the Solr server.
239  */
240  Set<String> servicesList = new HashSet<>();
241  servicesList.add(ServicesMonitor.Service.REMOTE_CASE_DATABASE.toString());
242  servicesList.add(ServicesMonitor.Service.REMOTE_KEYWORD_SEARCH.toString());
243  this.servicesMonitor.addSubscriber(servicesList, propChangeListener);
244  }
245 
250  private void subscribeToCaseEvents() {
251  Case.addEventTypeSubscriber(EnumSet.of(Case.Events.CURRENT_CASE), (PropertyChangeEvent event) -> {
252  if (event.getNewValue() != null) {
253  handleCaseOpened();
254  } else {
255  handleCaseClosed();
256  }
257  });
258  }
259 
268  void handleCaseOpened() {
269  caseIsOpen = true;
271  try {
272  Case openedCase = Case.getCurrentCaseThrows();
273  String channelPrefix = openedCase.getName();
274  if (Case.CaseType.MULTI_USER_CASE == openedCase.getCaseType()) {
275  jobEventPublisher.openRemoteEventChannel(String.format(INGEST_JOB_EVENT_CHANNEL_NAME, channelPrefix));
276  moduleEventPublisher.openRemoteEventChannel(String.format(INGEST_MODULE_EVENT_CHANNEL_NAME, channelPrefix));
277  }
278  openedCase.getSleuthkitCase().registerForEvents(this);
279  } catch (NoCurrentCaseException | AutopsyEventException ex) {
280  logger.log(Level.SEVERE, "Failed to open remote events channel", ex); //NON-NLS
281  MessageNotifyUtil.Notify.error(NbBundle.getMessage(IngestManager.class, "IngestManager.OpenEventChannel.Fail.Title"),
282  NbBundle.getMessage(IngestManager.class, "IngestManager.OpenEventChannel.Fail.ErrMsg"));
283  }
284  }
285 
292  @Subscribe
293  void handleArtifactsPosted(Blackboard.ArtifactsPostedEvent tskEvent) {
294  /*
295  * Add any new data artifacts included in the event to the source ingest
296  * job for possible analysis.
297  */
298  List<DataArtifact> newDataArtifacts = new ArrayList<>();
299  List<AnalysisResult> newAnalysisResults = new ArrayList<>();
300  Collection<BlackboardArtifact> newArtifacts = tskEvent.getArtifacts();
301  for (BlackboardArtifact artifact : newArtifacts) {
302  if (artifact instanceof DataArtifact) {
303  newDataArtifacts.add((DataArtifact) artifact);
304  } else {
305  newAnalysisResults.add((AnalysisResult) artifact);
306  }
307  }
308  if (!newDataArtifacts.isEmpty() || !newAnalysisResults.isEmpty()) {
309  IngestJob ingestJob = null;
310  Optional<Long> ingestJobId = tskEvent.getIngestJobId();
311  if (ingestJobId.isPresent()) {
312  synchronized (ingestJobsById) {
313  ingestJob = ingestJobsById.get(ingestJobId.get());
314  }
315  } else {
316  /*
317  * There are four use cases where the ingest job ID returned by
318  * the event is expected be null:
319  *
320  * 1. The artifacts are being posted by a data source proccessor
321  * (DSP) module that runs before the ingest job is created,
322  * i.e., a DSP that does not support streaming ingest and has no
323  * noton of an ingest job ID. In this use case, the event is
324  * handled synchronously. The DSP calls
325  * Blackboard.postArtifacts(), which puts the event on the event
326  * bus to which this method subscribes, so the event will be
327  * handled here before the DSP completes and calls
328  * DataSourceProcessorCallback.done(). This means the code below
329  * will execute before the ingest job is created, so it will not
330  * find an ingest job to which to add the artifacts. However,
331  * the artifacts WILL be analyzed after the ingest job is
332  * started, when the ingest job executor, working in batch mode,
333  * schedules ingest tasks for all of the data artifacts in the
334  * case database. There is a slight risk that the wrong ingest
335  * job will be selected if multiple ingests of the same data
336  * source are in progress.
337  *
338  * 2. The artifacts were posted by an ingest module that either
339  * has not been updated to use the current
340  * Blackboard.postArtifacts() API, or is using it incorrectly.
341  * In this use case, the code below should be able to find the
342  * ingest job to which to add the artifacts via their data
343  * source. There is a slight risk that the wrong ingest job will
344  * be selected if multiple ingests of the same data source are
345  * in progress.
346  *
347  * 3. The portable case generator uses a
348  * CommunicationArtifactsHelper constructed with a null ingest
349  * job ID, and the CommunicatonsArtifactHelper posts artifacts.
350  * Ingest of that data source might be running, in which case
351  * the data artifact will be analyzed. It also might be analyzed
352  * by a subsequent ingest job for the data source. This is an
353  * acceptable edge case.
354  *
355  * 4. The user can manually create timeline events with the
356  * timeline tool, which posts the TSK_TL_EVENT data artifacts.
357  * The user selects the data source for these artifacts. Ingest
358  * of that data source might be running, in which case the data
359  * artifact will be analyzed. It also might be analyzed by a
360  * subsequent ingest job for the data source. This is an
361  * acceptable edge case.
362  *
363  * 5. The user can manually run ad hoc keyword searches,
364  * which post TSK_KEYWORD_HIT analysis results. Ingest
365  * of that data source might be running, in which case the analysis
366  * results will be analyzed. They also might be analyzed by a
367  * subsequent ingest job for the data source. This is an
368  * acceptable edge case.
369  */
370  BlackboardArtifact artifact = newArtifacts.iterator().next();
371  if (artifact != null) {
372  try {
373  Content artifactDataSource = artifact.getDataSource();
374  synchronized (ingestJobsById) {
375  for (IngestJob job : ingestJobsById.values()) {
376  Content dataSource = job.getDataSource();
377  if (artifactDataSource.getId() == dataSource.getId()) {
378  ingestJob = job;
379  break;
380  }
381  }
382  }
383  } catch (TskCoreException ex) {
384  logger.log(Level.SEVERE, String.format("Failed to get data source for blackboard artifact (object ID = %d)", artifact.getId()), ex); //NON-NLS
385  }
386  }
387  }
388  if (ingestJob != null) {
389  if (!newDataArtifacts.isEmpty()) {
390  ingestJob.addDataArtifacts(newDataArtifacts);
391  }
392  if (!newAnalysisResults.isEmpty()) {
393  ingestJob.addAnalysisResults(newAnalysisResults);
394  }
395  }
396  }
397 
398  /*
399  * Publish Autopsy events for the new artifacts, one event per artifact
400  * type.
401  */
402  for (BlackboardArtifact.Type artifactType : tskEvent.getArtifactTypes()) {
403  ModuleDataEvent legacyEvent = new ModuleDataEvent(tskEvent.getModuleName(), artifactType, tskEvent.getArtifacts(artifactType));
404  AutopsyEvent autopsyEvent = new BlackboardPostEvent(legacyEvent);
405  eventPublishingExecutor.submit(new PublishEventTask(autopsyEvent, moduleEventPublisher));
406  }
407  }
408 
418  void handleCaseClosed() {
419  /*
420  * TODO (JIRA-2227): IngestManager should wait for cancelled ingest jobs
421  * to complete when a case is closed.
422  */
423  cancelAllIngestJobs(IngestJob.CancellationReason.CASE_CLOSED);
424  Case.getCurrentCase().getSleuthkitCase().unregisterForEvents(this);
427  caseIsOpen = false;
429  }
430 
442  public IngestStream openIngestStream(DataSource dataSource, IngestJobSettings settings) throws TskCoreException {
443  if (!(dataSource instanceof DataSource)) {
444  throw new IllegalArgumentException("dataSource argument does not implement the DataSource interface"); //NON-NLS
445  }
446  IngestJob job = new IngestJob(dataSource, IngestJob.Mode.STREAMING, settings);
447  IngestJobInputStream stream = new IngestJobInputStream(job);
448  if (stream.getIngestJobStartResult().getJob() != null) {
449  return stream;
450  } else if (stream.getIngestJobStartResult().getModuleErrors().isEmpty()) {
451  for (IngestModuleError error : stream.getIngestJobStartResult().getModuleErrors()) {
452  logger.log(Level.SEVERE, String.format("%s ingest module startup error for %s", error.getModuleDisplayName(), dataSource.getName()), error.getThrowable());
453  }
454  throw new TskCoreException("Error starting ingest modules");
455  } else {
456  throw new TskCoreException("Error starting ingest modules", stream.getIngestJobStartResult().getStartupException());
457  }
458  }
459 
468  }
469 
476  public void queueIngestJob(Collection<Content> dataSources, IngestJobSettings settings) {
477  if (caseIsOpen) {
478  List<AbstractFile> emptyFilesSubset = new ArrayList<>();
479  for (Content dataSource : dataSources) {
480  queueIngestJob(dataSource, emptyFilesSubset, settings);
481  }
482  }
483  }
484 
494  public void queueIngestJob(Content dataSource, List<AbstractFile> files, IngestJobSettings settings) {
495  if (!(dataSource instanceof DataSource)) {
496  throw new IllegalArgumentException("dataSource argument does not implement the DataSource interface"); //NON-NLS
497  }
498  if (caseIsOpen) {
499  IngestJob job = new IngestJob((DataSource) dataSource, files, settings);
500  if (job.hasIngestPipeline()) {
501  long taskId = nextIngestManagerTaskId.incrementAndGet();
502  Future<Void> task = startIngestJobsExecutor.submit(new StartIngestJobTask(taskId, job));
503  synchronized (startIngestJobFutures) {
504  startIngestJobFutures.put(taskId, task);
505  }
506  }
507  }
508  }
509 
524  public IngestJobStartResult beginIngestJob(Collection<Content> dataSources, IngestJobSettings settings) {
525  List<DataSource> verifiedDataSources = new ArrayList<>();
526  for (Content content : dataSources) {
527  if (!(content instanceof DataSource)) {
528  throw new IllegalArgumentException("Content object in dataSources argument does not implement the DataSource interface"); //NON-NLS
529  }
530  DataSource verifiedDataSource = (DataSource) content;
531  verifiedDataSources.add(verifiedDataSource);
532  }
533  IngestJobStartResult startResult = null;
534  if (caseIsOpen) {
535  for (DataSource dataSource : verifiedDataSources) {
536  List<IngestJob> startedJobs = new ArrayList<>();
537  IngestJob job = new IngestJob(dataSource, IngestJob.Mode.BATCH, settings);
538  if (job.hasIngestPipeline()) {
539  startResult = startIngestJob(job);
540  if (startResult.getModuleErrors().isEmpty() && startResult.getStartupException() == null) {
541  startedJobs.add(job);
542  } else {
543  for (IngestJob jobToCancel : startedJobs) {
545  }
546  break;
547  }
548  } else {
549  startResult = new IngestJobStartResult(null, new IngestManagerException("No ingest pipeline created, likely due to no ingest modules being enabled"), null); //NON-NLS
550  break;
551  }
552  }
553  } else {
554  startResult = new IngestJobStartResult(null, new IngestManagerException("No case open"), null); //NON-NLS
555  }
556  return startResult;
557  }
558 
567  @NbBundle.Messages({
568  "IngestManager.startupErr.dlgTitle=Ingest Module Startup Failure",
569  "IngestManager.startupErr.dlgMsg=Unable to start up one or more ingest modules, ingest cancelled.",
570  "IngestManager.startupErr.dlgSolution=Please disable the failed modules or fix the errors before restarting ingest.",
571  "IngestManager.startupErr.dlgErrorList=Errors:"
572  })
573  IngestJobStartResult startIngestJob(IngestJob job) {
574 
575  // initialize IngestMessageInbox, if it hasn't been initialized yet. This can't be done in
576  // the constructor because that ends up freezing the UI on startup (JIRA-7345).
577  if (SwingUtilities.isEventDispatchThread()) {
578  initIngestMessageInbox();
579  } else {
580  try {
581  SwingUtilities.invokeAndWait(() -> initIngestMessageInbox());
582  } catch (InterruptedException ex) {
583  // ignore interruptions
584  } catch (InvocationTargetException ex) {
585  logger.log(Level.WARNING, "There was an error starting ingest message inbox", ex);
586  }
587  }
588 
589  List<IngestModuleError> errors = null;
590  Case openCase;
591  try {
592  openCase = Case.getCurrentCaseThrows();
593  } catch (NoCurrentCaseException ex) {
594  return new IngestJobStartResult(null, new IngestManagerException("Exception while getting open case.", ex), Collections.<IngestModuleError>emptyList()); //NON-NLS
595  }
596  if (openCase.getCaseType() == Case.CaseType.MULTI_USER_CASE) {
597  try {
598  if (!servicesMonitor.getServiceStatus(ServicesMonitor.Service.REMOTE_CASE_DATABASE.toString()).equals(ServicesMonitor.ServiceStatus.UP.toString())) {
599  if (RuntimeProperties.runningWithGUI()) {
600  EventQueue.invokeLater(new Runnable() {
601  @Override
602  public void run() {
603  String serviceDisplayName = ServicesMonitor.Service.REMOTE_CASE_DATABASE.getDisplayName();
604  JOptionPane.showMessageDialog(WindowManager.getDefault().getMainWindow(),
605  NbBundle.getMessage(this.getClass(), "IngestManager.cancellingIngest.msgDlg.text"),
606  NbBundle.getMessage(this.getClass(), "IngestManager.serviceIsDown.msgDlg.text", serviceDisplayName),
607  JOptionPane.ERROR_MESSAGE);
608  }
609  });
610  }
611  return new IngestJobStartResult(null, new IngestManagerException("Ingest aborted. Remote database is down"), Collections.<IngestModuleError>emptyList()); //NON-NLS
612  }
613  } catch (ServicesMonitor.ServicesMonitorException ex) {
614  return new IngestJobStartResult(null, new IngestManagerException("Database server is down", ex), Collections.<IngestModuleError>emptyList()); //NON-NLS
615  }
616  }
617 
618  if (!ingestMonitor.isRunning()) {
619  ingestMonitor.start();
620  }
621 
622  synchronized (ingestJobsById) {
623  ingestJobsById.put(job.getId(), job);
624  }
625  IngestManager.logger.log(Level.INFO, String.format("Starting ingest job %d at %s", job.getId(), new Date().getTime())); //NON-NLS
626  try {
627  errors = job.start();
628  } catch (InterruptedException ex) {
629  return new IngestJobStartResult(null, new IngestManagerException("Interrupted while starting ingest", ex), errors); //NON-NLS
630  }
631  if (errors.isEmpty()) {
632  this.fireIngestJobStarted(job.getId());
633  } else {
634  synchronized (ingestJobsById) {
635  this.ingestJobsById.remove(job.getId());
636  }
637  for (IngestModuleError error : errors) {
638  logger.log(Level.SEVERE, String.format("Error starting %s ingest module for job %d", error.getModuleDisplayName(), job.getId()), error.getThrowable()); //NON-NLS
639  }
640  IngestManager.logger.log(Level.SEVERE, "Ingest job {0} could not be started", job.getId()); //NON-NLS
641  if (RuntimeProperties.runningWithGUI()) {
642  final StringBuilder message = new StringBuilder(1024);
643  message.append(Bundle.IngestManager_startupErr_dlgMsg()).append("\n"); //NON-NLS
644  message.append(Bundle.IngestManager_startupErr_dlgSolution()).append("\n\n"); //NON-NLS
645  message.append(Bundle.IngestManager_startupErr_dlgErrorList()).append("\n"); //NON-NLS
646  for (IngestModuleError error : errors) {
647  String moduleName = error.getModuleDisplayName();
648  String errorMessage = error.getThrowable().getLocalizedMessage();
649  message.append(moduleName).append(": ").append(errorMessage).append("\n"); //NON-NLS
650  }
651  message.append("\n\n");
652  EventQueue.invokeLater(() -> {
653  JOptionPane.showMessageDialog(WindowManager.getDefault().getMainWindow(), message, Bundle.IngestManager_startupErr_dlgTitle(), JOptionPane.ERROR_MESSAGE);
654  });
655  }
656  return new IngestJobStartResult(null, new IngestManagerException("Errors occurred while starting ingest"), errors); //NON-NLS
657  }
658 
659  return new IngestJobStartResult(job, null, errors);
660  }
661 
667  void finishIngestJob(IngestJob job) {
668  long jobId = job.getId();
669  synchronized (ingestJobsById) {
670  ingestJobsById.remove(jobId);
671  }
672  if (!job.isCancelled()) {
673  IngestManager.logger.log(Level.INFO, String.format("Ingest job %d completed at %s", job.getId(), new Date().getTime())); //NON-NLS
674  fireIngestJobCompleted(jobId);
675  } else {
676  IngestManager.logger.log(Level.INFO, String.format("Ingest job %d cancelled at %s", job.getId(), new Date().getTime())); //NON-NLS
677  fireIngestJobCancelled(jobId);
678  }
679  }
680 
687  public boolean isIngestRunning() {
688  synchronized (ingestJobsById) {
689  return !ingestJobsById.isEmpty();
690  }
691  }
692 
699  synchronized (startIngestJobFutures) {
700  startIngestJobFutures.values().forEach((handle) -> {
701  handle.cancel(true);
702  });
703  }
704  synchronized (ingestJobsById) {
705  this.ingestJobsById.values().forEach((job) -> {
706  job.cancel(reason);
707  });
708  }
709  }
710 
716  public void addIngestJobEventListener(final PropertyChangeListener listener) {
717  jobEventPublisher.addSubscriber(INGEST_JOB_EVENT_NAMES, listener);
718  }
719 
727  public void addIngestJobEventListener(Set<IngestJobEvent> eventTypes, final PropertyChangeListener listener) {
728  eventTypes.forEach((IngestJobEvent event) -> {
729  jobEventPublisher.addSubscriber(event.toString(), listener);
730  });
731  }
732 
738  public void removeIngestJobEventListener(final PropertyChangeListener listener) {
739  jobEventPublisher.removeSubscriber(INGEST_JOB_EVENT_NAMES, listener);
740  }
741 
748  public void removeIngestJobEventListener(Set<IngestJobEvent> eventTypes, final PropertyChangeListener listener) {
749  eventTypes.forEach((IngestJobEvent event) -> {
750  jobEventPublisher.removeSubscriber(event.toString(), listener);
751  });
752  }
753 
759  public void addIngestModuleEventListener(final PropertyChangeListener listener) {
760  moduleEventPublisher.addSubscriber(INGEST_MODULE_EVENT_NAMES, listener);
761  }
762 
770  public void addIngestModuleEventListener(Set<IngestModuleEvent> eventTypes, final PropertyChangeListener listener) {
771  eventTypes.forEach((IngestModuleEvent event) -> {
772  moduleEventPublisher.addSubscriber(event.toString(), listener);
773  });
774  }
775 
781  public void removeIngestModuleEventListener(final PropertyChangeListener listener) {
782  moduleEventPublisher.removeSubscriber(INGEST_MODULE_EVENT_NAMES, listener);
783  }
784 
791  public void removeIngestModuleEventListener(Set<IngestModuleEvent> eventTypes, final PropertyChangeListener listener) {
792  moduleEventPublisher.removeSubscriber(INGEST_MODULE_EVENT_NAMES, listener);
793  }
794 
800  void fireIngestJobStarted(long ingestJobId) {
801  AutopsyEvent event = new AutopsyEvent(IngestJobEvent.STARTED.toString(), ingestJobId, null);
802  eventPublishingExecutor.submit(new PublishEventTask(event, jobEventPublisher));
803  }
804 
810  void fireIngestJobCompleted(long ingestJobId) {
811  AutopsyEvent event = new AutopsyEvent(IngestJobEvent.COMPLETED.toString(), ingestJobId, null);
812  eventPublishingExecutor.submit(new PublishEventTask(event, jobEventPublisher));
813  }
814 
820  void fireIngestJobCancelled(long ingestJobId) {
821  AutopsyEvent event = new AutopsyEvent(IngestJobEvent.CANCELLED.toString(), ingestJobId, null);
822  eventPublishingExecutor.submit(new PublishEventTask(event, jobEventPublisher));
823  }
824 
832  void fireDataSourceAnalysisStarted(long ingestJobId, Content dataSource) {
833  AutopsyEvent event = new DataSourceAnalysisStartedEvent(ingestJobId, dataSource);
834  eventPublishingExecutor.submit(new PublishEventTask(event, jobEventPublisher));
835  }
836 
844  void fireDataSourceAnalysisCompleted(long ingestJobId, Content dataSource) {
845  AutopsyEvent event = new DataSourceAnalysisCompletedEvent(ingestJobId, dataSource, DataSourceAnalysisCompletedEvent.Reason.ANALYSIS_COMPLETED);
846  eventPublishingExecutor.submit(new PublishEventTask(event, jobEventPublisher));
847  }
848 
856  void fireDataSourceAnalysisCancelled(long ingestJobId, Content dataSource) {
857  AutopsyEvent event = new DataSourceAnalysisCompletedEvent(ingestJobId, dataSource, DataSourceAnalysisCompletedEvent.Reason.ANALYSIS_CANCELLED);
858  eventPublishingExecutor.submit(new PublishEventTask(event, jobEventPublisher));
859  }
860 
867  void fireFileIngestDone(AbstractFile file) {
868  AutopsyEvent event = new FileAnalyzedEvent(file);
869  eventPublishingExecutor.submit(new PublishEventTask(event, moduleEventPublisher));
870  }
871 
879  void fireIngestModuleContentEvent(ModuleContentEvent moduleContentEvent) {
880  AutopsyEvent event = new ContentChangedEvent(moduleContentEvent);
881  eventPublishingExecutor.submit(new PublishEventTask(event, moduleEventPublisher));
882  }
883 
892  void initIngestMessageInbox() {
893  synchronized (this.ingestMessageBoxLock) {
894  ingestMessageBox = IngestMessageTopComponent.findInstance();
895  }
896  }
897 
903  void postIngestMessage(IngestMessage message) {
904  synchronized (this.ingestMessageBoxLock) {
905  if (ingestMessageBox != null && RuntimeProperties.runningWithGUI()) {
906  if (message.getMessageType() != IngestMessage.MessageType.ERROR && message.getMessageType() != IngestMessage.MessageType.WARNING) {
907  ingestMessageBox.displayMessage(message);
908  } else {
909  long errorPosts = ingestErrorMessagePosts.incrementAndGet();
910  if (errorPosts <= MAX_ERROR_MESSAGE_POSTS) {
911  ingestMessageBox.displayMessage(message);
912  } else if (errorPosts == MAX_ERROR_MESSAGE_POSTS + 1) {
913  IngestMessage errorMessageLimitReachedMessage = IngestMessage.createErrorMessage(
914  NbBundle.getMessage(this.getClass(), "IngestManager.IngestMessage.ErrorMessageLimitReached.title"),
915  NbBundle.getMessage(this.getClass(), "IngestManager.IngestMessage.ErrorMessageLimitReached.subject"),
916  NbBundle.getMessage(this.getClass(), "IngestManager.IngestMessage.ErrorMessageLimitReached.msg", MAX_ERROR_MESSAGE_POSTS));
917  ingestMessageBox.displayMessage(errorMessageLimitReachedMessage);
918  }
919  }
920  }
921  }
922  }
923 
924  /*
925  * Clears the ingest messages inbox.
926  */
927  private void clearIngestMessageBox() {
928  synchronized (this.ingestMessageBoxLock) {
929  if (null != ingestMessageBox) {
930  ingestMessageBox.clearMessages();
931  }
933  }
934  }
935 
946  void setIngestTaskProgress(IngestTask task, String currentModuleName) {
947  IngestThreadActivitySnapshot prevSnap = ingestThreadActivitySnapshots.get(task.getThreadId());
948  IngestThreadActivitySnapshot newSnap = new IngestThreadActivitySnapshot(task.getThreadId(), task.getIngestJobExecutor().getIngestJobId(), currentModuleName, task.getDataSource(), task.getContentName());
949  ingestThreadActivitySnapshots.put(task.getThreadId(), newSnap);
950  incrementModuleRunTime(prevSnap.getModuleDisplayName(), newSnap.getStartTime().getTime() - prevSnap.getStartTime().getTime());
951  }
952 
960  void setIngestTaskProgressCompleted(IngestTask task) {
961  IngestThreadActivitySnapshot prevSnap = ingestThreadActivitySnapshots.get(task.getThreadId());
962  IngestThreadActivitySnapshot newSnap = new IngestThreadActivitySnapshot(task.getThreadId());
963  ingestThreadActivitySnapshots.put(task.getThreadId(), newSnap);
964  incrementModuleRunTime(prevSnap.getModuleDisplayName(), newSnap.getStartTime().getTime() - prevSnap.getStartTime().getTime());
965  }
966 
973  void incrementModuleRunTime(String moduleDisplayName, Long duration) {
974  if (moduleDisplayName.equals("IDLE")) { //NON-NLS
975  return;
976  }
977 
978  synchronized (ingestModuleRunTimes) {
979  Long prevTimeL = ingestModuleRunTimes.get(moduleDisplayName);
980  long prevTime = 0;
981  if (prevTimeL != null) {
982  prevTime = prevTimeL;
983  }
984  prevTime += duration;
985  ingestModuleRunTimes.put(moduleDisplayName, prevTime);
986  }
987  }
988 
994  @Override
995  public Map<String, Long> getModuleRunTimes() {
996  synchronized (ingestModuleRunTimes) {
997  Map<String, Long> times = new HashMap<>(ingestModuleRunTimes);
998  return times;
999  }
1000  }
1001 
1008  @Override
1009  public List<IngestThreadActivitySnapshot> getIngestThreadActivitySnapshots() {
1010  return new ArrayList<>(ingestThreadActivitySnapshots.values());
1011  }
1012 
1018  @Override
1019  public List<IngestJobProgressSnapshot> getIngestJobSnapshots() {
1020  List<IngestJobProgressSnapshot> snapShots = new ArrayList<>();
1021  synchronized (ingestJobsById) {
1022  ingestJobsById.values().forEach((job) -> {
1023  IngestJobProgressSnapshot snapshot = job.getDiagnosticStatsSnapshot();
1024  if (snapshot != null) {
1025  snapShots.add(snapshot);
1026  }
1027  });
1028  }
1029  return snapShots;
1030  }
1031 
1038  long getFreeDiskSpace() {
1039  if (ingestMonitor != null) {
1040  return ingestMonitor.getFreeSpace();
1041  } else {
1042  return -1;
1043  }
1044  }
1045 
1049  private final class StartIngestJobTask implements Callable<Void> {
1050 
1051  private final long threadId;
1052  private final IngestJob job;
1053  private ProgressHandle progress;
1054 
1055  StartIngestJobTask(long threadId, IngestJob job) {
1056  this.threadId = threadId;
1057  this.job = job;
1058  }
1059 
1060  @Override
1061  public Void call() {
1062  try {
1063  if (Thread.currentThread().isInterrupted()) {
1064  synchronized (ingestJobsById) {
1065  ingestJobsById.remove(job.getId());
1066  }
1067  return null;
1068  }
1069 
1071  final String displayName = NbBundle.getMessage(this.getClass(), "IngestManager.StartIngestJobsTask.run.displayName");
1072  this.progress = ProgressHandle.createHandle(displayName, new Cancellable() {
1073  @Override
1074  public boolean cancel() {
1075  if (progress != null) {
1076  progress.setDisplayName(NbBundle.getMessage(this.getClass(), "IngestManager.StartIngestJobsTask.run.cancelling", displayName));
1077  }
1078  synchronized (startIngestJobFutures) {
1079  Future<?> handle = startIngestJobFutures.remove(threadId);
1080  handle.cancel(true);
1081  }
1082  return true;
1083  }
1084  });
1085  progress.start();
1086  }
1087 
1088  startIngestJob(job);
1089  return null;
1090 
1091  } finally {
1092  if (null != progress) {
1093  progress.finish();
1094  }
1095  synchronized (startIngestJobFutures) {
1096  startIngestJobFutures.remove(threadId);
1097  }
1098  }
1099  }
1100 
1101  }
1102 
1106  private final class ExecuteIngestJobTasksTask implements Runnable {
1107 
1108  private final long threadId;
1109  private final BlockingIngestTaskQueue tasks;
1110 
1111  ExecuteIngestJobTasksTask(long threadId, BlockingIngestTaskQueue tasks) {
1112  this.threadId = threadId;
1113  this.tasks = tasks;
1114  }
1115 
1116  @Override
1117  public void run() {
1118  while (true) {
1119  try {
1120  IngestTask task = tasks.getNextTask(); // Blocks.
1121  task.execute(threadId);
1122  } catch (InterruptedException ex) {
1123  break;
1124  }
1125  if (Thread.currentThread().isInterrupted()) {
1126  break;
1127  }
1128  }
1129  }
1130  }
1131 
1135  private static final class PublishEventTask implements Runnable {
1136 
1137  private final AutopsyEvent event;
1139 
1148  this.event = event;
1149  this.publisher = publisher;
1150  }
1151 
1152  @Override
1153  public void run() {
1154  publisher.publish(event);
1155  }
1156 
1157  }
1158 
1162  @Immutable
1163  public static final class IngestThreadActivitySnapshot implements Serializable {
1164 
1165  private static final long serialVersionUID = 1L;
1166 
1167  private final long threadId;
1168  private final Date startTime;
1169  private final String moduleDisplayName;
1170  private final String dataSourceName;
1171  private final String fileName;
1172  private final long jobId;
1173 
1179  IngestThreadActivitySnapshot(long threadId) {
1180  this.threadId = threadId;
1181  startTime = new Date();
1182  this.moduleDisplayName = NbBundle.getMessage(this.getClass(), "IngestManager.IngestThreadActivitySnapshot.idleThread");
1183  this.dataSourceName = "";
1184  this.fileName = "";
1185  this.jobId = 0;
1186  }
1187 
1201  IngestThreadActivitySnapshot(long threadId, long jobId, String moduleDisplayName, Content dataSource) {
1202  this.threadId = threadId;
1203  this.jobId = jobId;
1204  startTime = new Date();
1205  this.moduleDisplayName = moduleDisplayName;
1206  this.dataSourceName = dataSource.getName();
1207  this.fileName = "";
1208  }
1209 
1225  IngestThreadActivitySnapshot(long threadId, long jobId, String moduleDisplayName, Content dataSource, String fileName) {
1226  this.threadId = threadId;
1227  this.jobId = jobId;
1228  startTime = new Date();
1229  this.moduleDisplayName = moduleDisplayName;
1230  this.dataSourceName = dataSource.getName();
1231  this.fileName = fileName;
1232  }
1233 
1240  long getIngestJobId() {
1241  return jobId;
1242  }
1243 
1249  long getThreadId() {
1250  return threadId;
1251  }
1252 
1258  Date getStartTime() {
1259  return new Date(startTime.getTime());
1260  }
1261 
1267  String getModuleDisplayName() {
1268  return moduleDisplayName;
1269  }
1270 
1277  String getDataSourceName() {
1278  return dataSourceName;
1279  }
1280 
1286  String getFileName() {
1287  return fileName;
1288  }
1289 
1290  }
1291 
1295  public enum IngestJobEvent {
1296 
1333  }
1334 
1338  public enum IngestModuleEvent {
1339 
1361  }
1362 
1366  public final static class IngestManagerException extends Exception {
1367 
1368  private static final long serialVersionUID = 1L;
1369 
1375  private IngestManagerException(String message) {
1376  super(message);
1377  }
1378 
1385  private IngestManagerException(String message, Throwable cause) {
1386  super(message, cause);
1387  }
1388  }
1389 
1398  @Deprecated
1399  public static void addPropertyChangeListener(final PropertyChangeListener listener) {
1402  }
1403 
1412  @Deprecated
1413  public static void removePropertyChangeListener(final PropertyChangeListener listener) {
1416  }
1417 
1428  @Deprecated
1429  public synchronized IngestJob startIngestJob(Collection<Content> dataSources, IngestJobSettings settings) {
1430  return beginIngestJob(dataSources, settings).getJob();
1431  }
1432 
1439  @Deprecated
1440  public void cancelAllIngestJobs() {
1442  }
1443 
1444 }
final ConcurrentHashMap< String, Long > ingestModuleRunTimes
void addIngestModuleEventListener(Set< IngestModuleEvent > eventTypes, final PropertyChangeListener listener)
final Map< Long, Future< Void > > startIngestJobFutures
List< IngestJobProgressSnapshot > getIngestJobSnapshots()
void removeIngestModuleEventListener(final PropertyChangeListener listener)
IngestStream openIngestStream(DataSource dataSource, IngestJobSettings settings)
List< IngestThreadActivitySnapshot > getIngestThreadActivitySnapshots()
void queueIngestJob(Content dataSource, List< AbstractFile > files, IngestJobSettings settings)
static synchronized IngestManager getInstance()
void removeIngestModuleEventListener(Set< IngestModuleEvent > eventTypes, final PropertyChangeListener listener)
final ExecutorService dataSourceLevelIngestJobTasksExecutor
final ExecutorService dataArtifactIngestTasksExecutor
static void addPropertyChangeListener(final PropertyChangeListener listener)
IngestJobStartResult beginIngestJob(Collection< Content > dataSources, IngestJobSettings settings)
void addSubscriber(Set< String > eventNames, PropertyChangeListener subscriber)
static void removePropertyChangeListener(final PropertyChangeListener listener)
static final Set< String > INGEST_MODULE_EVENT_NAMES
volatile IngestMessageTopComponent ingestMessageBox
void addSubscriber(PropertyChangeListener subscriber)
void removeIngestJobEventListener(final PropertyChangeListener listener)
void addIngestJobEventListener(Set< IngestJobEvent > eventTypes, final PropertyChangeListener listener)
final ExecutorService analysisResultIngestTasksExecutor
static final Set< String > INGEST_JOB_EVENT_NAMES
final AutopsyEventPublisher moduleEventPublisher
synchronized void openRemoteEventChannel(String channelName)
void addIngestJobEventListener(final PropertyChangeListener listener)
void queueIngestJob(Collection< Content > dataSources, IngestJobSettings settings)
void removeSubscriber(Set< String > eventNames, PropertyChangeListener subscriber)
void addIngestModuleEventListener(final PropertyChangeListener listener)
synchronized static Logger getLogger(String name)
Definition: Logger.java:124
static void addEventTypeSubscriber(Set< Events > eventTypes, PropertyChangeListener subscriber)
Definition: Case.java:708
synchronized IngestJob startIngestJob(Collection< Content > dataSources, IngestJobSettings settings)
final ConcurrentHashMap< Long, IngestThreadActivitySnapshot > ingestThreadActivitySnapshots
void cancelAllIngestJobs(IngestJob.CancellationReason reason)
IngestManager.IngestManagerException getStartupException()
final ExecutorService fileLevelIngestJobTasksExecutor
final Map< Long, IngestJob > ingestJobsById
void removeIngestJobEventListener(Set< IngestJobEvent > eventTypes, final PropertyChangeListener listener)
final AutopsyEventPublisher jobEventPublisher

Copyright © 2012-2022 Basis Technology. Generated on: Tue Feb 6 2024
This work is licensed under a Creative Commons Attribution-Share Alike 3.0 United States License.