Repository: google/turbinia
Pull request #412 (2019-07-27 00:09:31)

Generate basic task execution time statistics
- Adds run_time attribute of Task to be stored in Datastore/Redis
- Also convert all date formatting to use same format
- Also deserializes date output from datastore/redis

Regular stats output looks like:
```
Execution time statistics for Turbinia:

All Tasks: Count: 13, Min: 0:00:00, Mean: 0:00:09, Max: 0:01:36
Successful Tasks: Count: 10, Min: 0:00:01, Mean: 0:00:10, Max: 0:01:36
Failed Tasks: Count: 3, Min: 0:00:00, Mean: 0:00:00, Max: 0:00:08
Total Request Time: Count: 1, Min: 0:02:54, Mean: 0:02:54, Max: 0:02:54
Task type FileArtifactExtractionTask: Count: 5, Min: 0:00:09, Mean: 0:00:09, Max: 0:00:10
Task type GrepTask: Count: 2, Min: 0:00:00, Mean: 0:00:00, Max: 0:00:00
Task type HadoopAnalysisTask: Count: 1, Min: 0:00:10, Mean: 0:00:10, Max: 0:00:10
Task type JenkinsAnalysisTask: Count: 1, Min: 0:00:08, Mean: 0:00:08, Max: 0:00:08
Task type PlasoTask: Count: 1, Min: 0:00:18, Mean: 0:00:18, Max: 0:00:18
Task type PsortTask: Count: 1, Min: 0:00:01, Mean: 0:00:01, Max: 0:00:01
Task type StringsAsciiTask: Count: 1, Min: 0:01:36, Mean: 0:01:36, Max: 0:01:36
Task type StringsUnicodeTask: Count: 1, Min: 0:01:33, Mean: 0:01:33, Max: 0:01:33
Worker turbinia-instance-group-4-2345: Count: 6, Min: 0:00:00, Mean: 0:00:10, Max: 0:01:36
Worker turbinia-instance-group-4-1234: Count: 7, Min: 0:00:00, Mean: 0:00:09, Max: 0:01:33
User aaronp: Count: 13, Min: 0:00:00, Mean: 0:00:09, Max: 0:01:36
```

And CSV stats output looks like
```
stat_type, count, min, mean, max
All Tasks, 13, 0:00:00, 0:00:09, 0:01:36
Successful Tasks, 10, 0:00:01, 0:00:10, 0:01:36
Failed Tasks, 3, 0:00:00, 0:00:00, 0:00:08
Total Request Time, 1, 0:02:54, 0:02:54, 0:02:54
Task type FileArtifactExtractionTask, 5, 0:00:09, 0:00:09, 0:00:10
Task type GrepTask, 2, 0:00:00, 0:00:00, 0:00:00
Task type HadoopAnalysisTask, 1, 0:00:10, 0:00:10, 0:00:10
Task type JenkinsAnalysisTask, 1, 0:00:08, 0:00:08, 0:00:08
Task type PlasoTask, 1, 0:00:18, 0:00:18, 0:00:18
Task type PsortTask, 1, 0:00:01, 0:00:01, 0:00:01
Task type StringsAsciiTask, 1, 0:01:36, 0:01:36, 0:01:36
Task type StringsUnicodeTask, 1, 0:01:33, 0:01:33, 0:01:33
Worker turbinia-instance-group-4-1234, 6, 0:00:00, 0:00:10, 0:01:36
Worker turbinia-instance-group-4-2345, 7, 0:00:00, 0:00:09, 0:01:33
User aaronp, 13, 0:00:00, 0:00:09, 0:01:36
```


Patch:
diff --git a/turbinia/client.py b/turbinia/client.py
index 9be1827..473bd18 100644
--- a/turbinia/client.py
+++ b/turbinia/client.py
@@ -22,12 +22,14 @@ from datetime import timedelta
 import json
 import logging
 from operator import itemgetter
+from operator import attrgetter
 import os
 import stat
 import time
 
 from turbinia import config
 from turbinia.config import logger
+from turbinia.config import DATETIME_FORMAT
 from turbinia import task_manager
 from turbinia import TurbiniaException
 from turbinia.lib import text_formatter as fmt
@@ -111,6 +113,78 @@ def check_directory(directory):
           'Can not add write permissions to {0:s}'.format(directory))
 
 
+class TurbiniaStats(object):
+  """Statistics for Turbinia task execution.
+
+  Attributes:
+    count(int): The number of tasks
+    min(datetime.timedelta): The minimum run time of all tasks
+    max(datetime.timedelta): The maximum run time of all tasks
+    mean(datetime.timedelta): The mean run time of all tasks
+    tasks(list): A list of tasks to calculate stats for
+  """
+
+  def __init__(self, description=None):
+    self.description = description
+    self.min = None
+    self.mean = None
+    self.max = None
+    self.tasks = []
+
+  def __str__(self):
+    return self.format_stats()
+
+  @property
+  def count(self):
+    """Gets a count of the tasks in this stats object.
+
+    Returns:
+      Int of task count.
+    """
+    return len(self.tasks)
+
+  def add_task(self, task):
+    """Add a task result dict.
+
+    Args:
+      task(dict): The task results we want to count stats for.
+    """
+    self.tasks.append(task)
+
+  def calculate_stats(self):
+    """Calculates statistics of the current tasks."""
+    if not self.tasks:
+      return
+
+    sorted_tasks = sorted(self.tasks, key=itemgetter('run_time'))
+    self.min = sorted_tasks[0]['run_time']
+    self.max = sorted_tasks[len(sorted_tasks) - 1]['run_time']
+    self.mean = sorted_tasks[len(sorted_tasks) // 2]['run_time']
+
+    # Remove the microseconds to keep things cleaner
+    self.min = self.min - timedelta(microseconds=self.min.microseconds)
+    self.max = self.max - timedelta(microseconds=self.max.microseconds)
+    self.mean = self.mean - timedelta(microseconds=self.mean.microseconds)
+
+  def format_stats(self):
+    """Formats statistics data.
+
+    Returns:
+      String of statistics data
+    """
+    return '{0:s}: Count: {1:d}, Min: {2!s}, Mean: {3!s}, Max: {4!s}'.format(
+        self.description, self.count, self.min, self.mean, self.max)
+
+  def format_stats_csv(self):
+    """Formats statistics data into CSV output.
+
+    Returns:
+      String of statistics data in CSV format
+    """
+    return '{0:s}, {1:d}, {2!s}, {3!s}, {4!s}'.format(
+        self.description, self.count, self.min, self.mean, self.max)
+
+
 class TurbiniaClient(object):
   """Client class for Turbinia.
 
@@ -229,7 +303,7 @@ class TurbiniaClient(object):
       start_time = datetime.now() - timedelta(days=days)
       # Format this like '1990-01-01T00:00:00z' so we can cast it directly to a
       # javascript Date() object in the cloud function.
-      start_string = start_time.strftime('%Y-%m-%dT%H:%M:%S')
+      start_string = start_time.strftime(DATETIME_FORMAT)
       func_args.update({'start_time': start_string})
     elif task_id:
       func_args.update({'task_id': task_id})
@@ -254,9 +328,19 @@ class TurbiniaClient(object):
       results = json.loads(response['result'])
     except (TypeError, ValueError) as e:
       raise TurbiniaException(
-          'Could not deserialize result from GCF: [{0!s}]'.format(e))
+          'Could not deserialize result [{0!s}] from GCF: [{1!s}]'.format(
+              response.get('result'), e))
+
+    # Convert run_time/last_update back into datetime objects
+    task_data = results[0]
+    for task in task_data:
+      if task.get('run_time'):
+        task['run_time'] = timedelta(seconds=task['run_time'])
+      if task.get('last_update'):
+        task['last_update'] = datetime.strptime(
+            task['last_update'], DATETIME_FORMAT)
 
-    return results[0]
+    return task_data
 
   def format_task_detail(self, task, show_files=False):
     """Formats a single task in detail.
@@ -310,6 +394,174 @@ class TurbiniaClient(object):
       report.append('')
     return report
 
+  def get_task_statistics(
+      self, instance, project, region, days=0, task_id=None, request_id=None,
+      user=None):
+    """Gathers statistics for Turbinia execution data.
+
+    Args:
+      instance (string): The Turbinia instance name (by default the same as the
+          INSTANCE_ID in the config).
+      project (string): The name of the project.
+      region (string): The name of the zone to execute in.
+      days (int): The number of days we want history for.
+      task_id (string): The Id of the task.
+      request_id (string): The Id of the request we want tasks for.
+      user (string): The user of the request we want tasks for.
+
+    Returns:
+      task_stats(dict): Mapping of statistic names to values
+    """
+    task_results = self.get_task_data(
+        instance, project, region, days, task_id, request_id, user)
+    if not task_results:
+      return {}
+
+    task_stats = {
+        'all_tasks': TurbiniaStats('All Tasks'),
+        'successful_tasks': TurbiniaStats('Successful Tasks'),
+        'failed_tasks': TurbiniaStats('Failed Tasks'),
+        'requests': TurbiniaStats('Total Request Time'),
+        # The following are dicts mapping the user/worker/type names to their
+        # respective TurbiniaStats() objects.
+        # Total wall-time for all tasks of a given type
+        'tasks_per_type': {},
+        # Total wall-time for all tasks per Worker
+        'tasks_per_worker': {},
+        # Total wall-time for all tasks per User
+        'tasks_per_user': {},
+    }
+
+    # map of request ids to [min time, max time]
+    requests = {}
+
+    for task in task_results:
+      request_id = task.get('request_id')
+      task_type = task.get('name')
+      worker = task.get('worker_name')
+      user = task.get('requester')
+      if not task.get('run_time'):
+        log.debug(
+            'Ignoring task {0:s} in statistics because the run_time is not '
+            'set, and it is required to calculate stats'.format(
+                task.get('name')))
+        continue
+
+      # Stats for all/successful/failed tasks
+      task_stats['all_tasks'].add_task(task)
+      if task.get('successful') is True:
+        task_stats['successful_tasks'].add_task(task)
+      elif task.get('successful') is False:
+        task_stats['failed_tasks'].add_task(task)
+
+      # Stats for Tasks per Task type.
+      if task_type in task_stats['tasks_per_type']:
+        task_type_stats = task_stats['tasks_per_type'].get(task_type)
+      else:
+        task_type_stats = TurbiniaStats('Task type {0:s}'.format(task_type))
+        task_stats['tasks_per_type'][task_type] = task_type_stats
+      task_type_stats.add_task(task)
+
+      # Stats per worker.
+      if worker in task_stats['tasks_per_worker']:
+        worker_stats = task_stats['tasks_per_worker'].get(worker)
+      else:
+        worker_stats = TurbiniaStats('Worker {0:s}'.format(worker))
+        task_stats['tasks_per_worker'][worker] = worker_stats
+      worker_stats.add_task(task)
+
+      # Stats per submitting User.
+      if user in task_stats['tasks_per_user']:
+        user_stats = task_stats['tasks_per_user'].get(user)
+      else:
+        user_stats = TurbiniaStats('User {0:s}'.format(user))
+        task_stats['tasks_per_user'][user] = user_stats
+      user_stats.add_task(task)
+
+      # Stats for the total request.  This will, for each request, calculate the
+      # start time of the earliest task and the stop time of the latest task.
+      # This will give the overall run time covering all tasks in the request.
+      task_start_time = task['last_update'] - task['run_time']
+      task_stop_time = task['last_update']
+      if request_id in requests:
+        start_time, stop_time = requests[request_id]
+        if task_start_time < start_time:
+          requests[request_id][0] = task_start_time
+        if task_stop_time > stop_time:
+          requests[request_id][1] = task_stop_time
+      else:
+        requests[request_id] = [task_start_time, task_stop_time]
+
+    # Add a fake task result for each request with our calculated times to the
+    # stats module
+    for min_time, max_time in requests.values():
+      task = {}
+      task['run_time'] = max_time - min_time
+      task_stats['requests'].add_task(task)
+
+    # Go over all stat objects and calculate them
+    for stat_obj in task_stats.values():
+      if isinstance(stat_obj, dict):
+        for inner_stat_obj in stat_obj.values():
+          inner_stat_obj.calculate_stats()
+      else:
+        stat_obj.calculate_stats()
+
+    return task_stats
+
+  def format_task_statistics(
+      self, instance, project, region, days=0, task_id=None, request_id=None,
+      user=None, csv=False):
+    """Formats statistics for Turbinia execution data.
+
+    Args:
+      instance (string): The Turbinia instance name (by default the same as the
+          INSTANCE_ID in the config).
+      project (string): The name of the project.
+      region (string): The name of the zone to execute in.
+      days (int): The number of days we want history for.
+      task_id (string): The Id of the task.
+      request_id (string): The Id of the request we want tasks for.
+      user (string): The user of the request we want tasks for.
+      csv (bool): Whether we want the output in CSV format.
+
+    Returns:
+      String of task statistics report
+    """
+    task_stats = self.get_task_statistics(
+        instance, project, region, days, task_id, request_id, user)
+    if not task_stats:
+      return 'No tasks found'
+
+    stats_order = [
+        'all_tasks', 'successful_tasks', 'failed_tasks', 'requests',
+        'tasks_per_type', 'tasks_per_worker', 'tasks_per_user'
+    ]
+
+    if csv:
+      report = ['stat_type, count, min, mean, max']
+    else:
+      report = ['Execution time statistics for Turbinia:', '']
+    for stat_name in stats_order:
+      stat_obj = task_stats[stat_name]
+      if isinstance(stat_obj, dict):
+        # Sort by description so that we get consistent report output
+        inner_stat_objs = sorted(
+            stat_obj.values(), key=attrgetter('description'))
+        for inner_stat_obj in inner_stat_objs:
+          if csv:
+            report.append(inner_stat_obj.format_stats_csv())
+          else:
+            report.append(inner_stat_obj.format_stats())
+      else:
+        if csv:
+          report.append(stat_obj.format_stats_csv())
+        else:
+          report.append(stat_obj.format_stats())
+
+    report.append('')
+    return '\n'.join(report)
+
   def format_task_status(
       self, instance, project, region, days=0, task_id=None, request_id=None,
       user=None, all_fields=False, full_report=False,
diff --git a/turbinia/config/__init__.py b/turbinia/config/__init__.py
index d9f8817..b3462d3 100644
--- a/turbinia/config/__init__.py
+++ b/turbinia/config/__init__.py
@@ -22,7 +22,7 @@ import logging
 import os
 import sys
 
-log = logging.getLogger('turbinia')
+DATETIME_FORMAT = '%Y-%m-%dT%H:%M:%S.%fZ'
 
 # Look for config files with these names
 CONFIGFILES = ['.turbiniarc', 'turbinia.conf', 'turbinia_config.py']
@@ -80,6 +80,8 @@ ENVCONFIGVAR = 'TURBINIA_CONFIG_PATH'
 
 CONFIG = None
 
+log = logging.getLogger('turbinia')
+
 
 class TurbiniaConfigException(Exception):
   """Exception for Turbinia configuration."""
diff --git a/turbinia/evidence.py b/turbinia/evidence.py
index 278036f..61a5bfc 100644
--- a/turbinia/evidence.py
+++ b/turbinia/evidence.py
@@ -100,6 +100,9 @@ class Evidence(object):
         relevant to us.
   """
 
+  # The list of attributes a given piece of Evidence requires to be set
+  REQUIRED_ATTRIBUTES = []
+
   def __init__(
       self, name=None, description=None, source=None, local_path=None,
       tags=None, request_id=None):
@@ -193,6 +196,26 @@ class Evidence(object):
     if self.parent_evidence:
       self.parent_evidence.postprocess()
 
+  def validate(self):
+    """Runs validation to verify evidence meets minimum requirements.
+
+    This default implementation will just check that the attributes listed in
+    REQUIRED_ATTRIBUTES are set, but other evidence types can override this
+    method to implement their own more stringent checks as needed.  This is
+    called by the worker, prior to the pre/post-processors running.
+
+    Raises:
+      TurbiniaException: If validation fails
+    """
+    for attribute in self.REQUIRED_ATTRIBUTES:
+      attribute_value = getattr(self, attribute, None)
+      if not attribute_value:
+        message = (
+            'Evidence validation failed: Required attribute {0:s} for class '
+            '{1:s} is not set. Please check original request.'.format(
+                attribute, self.name))
+        raise TurbiniaException(message)
+
 
 class Directory(Evidence):
   """Filesystem directory evidence."""
@@ -208,6 +231,8 @@ class ChromiumProfile(Evidence):
     format: Output format (default is sqlite, other options are xlsx and jsonl)
   """
 
+  REQUIRED_ATTRIBUTES = ['browser_type', 'output_format']
+
   def __init__(self, browser_type=None, output_format=None, *args, **kwargs):
     """Initialization for chromium profile evidence object."""
     super(ChromiumProfile, self).__init__(*args, **kwargs)
@@ -274,6 +299,8 @@ class BitlockerDisk(EncryptedDisk):
     unencrypted_path: A string to the unencrypted local path
   """
 
+  REQUIRED_ATTRIBUTES = ['recovery_key', 'password']
+
   def __init__(self, recovery_key=None, password=None, *args, **kwargs):
     """Initialization for Bitlocker disk evidence object"""
     self.recovery_key = recovery_key
@@ -292,6 +319,8 @@ class APFSEncryptedDisk(EncryptedDisk):
     unencrypted_path: A string to the unencrypted local path
   """
 
+  REQUIRED_ATTRIBUTES = ['recovery_key', 'password']
+
   def __init__(self, recovery_key=None, password=None, *args, **kwargs):
     """Initialization for Bitlocker disk evidence object"""
     self.recovery_key = recovery_key
@@ -309,6 +338,8 @@ class GoogleCloudDisk(RawDisk):
     disk_name: The cloud disk name.
   """
 
+  REQUIRED_ATTRIBUTES = ['disk_name', 'project', 'zone']
+
   def __init__(self, project=None, zone=None, disk_name=None, *args, **kwargs):
     """Initialization for Google Cloud Disk."""
     self.project = project
@@ -337,6 +368,8 @@ class GoogleCloudDiskRawEmbedded(GoogleCloudDisk):
     embedded_path: The path of the raw disk image inside the Persistent Disk
   """
 
+  REQUIRED_ATTRIBUTES = ['disk_name', 'project', 'zone', 'embedded_path']
+
   def __init__(self, embedded_path=None, *args, **kwargs):
     """Initialization for Google Cloud Disk."""
     self.embedded_path = embedded_path
@@ -404,6 +437,8 @@ class FilteredTextFile(TextFile):
 class ExportedFileArtifact(Evidence):
   """Exported file artifact."""
 
+  REQUIRED_ATTRIBUTES = ['artifact_name']
+
   def __init__(self, artifact_name=None):
     """Initializes an exported file artifact."""
     super(ExportedFileArtifact, self).__init__()
@@ -424,6 +459,8 @@ class RawMemory(Evidence):
     module_list (list): Module used for the analysis
     """
 
+  REQUIRED_ATTRIBUTES = ['module_list', 'profile']
+
   def __init__(self, module_list=None, profile=None, *args, **kwargs):
     """Initialization for raw memory evidence object."""
     super(RawMemory, self).__init__(*args, **kwargs)
diff --git a/turbinia/lib/libcloudforensics.py b/turbinia/lib/libcloudforensics.py
index 532ae1e..7ac071d 100644
--- a/turbinia/lib/libcloudforensics.py
+++ b/turbinia/lib/libcloudforensics.py
@@ -32,6 +32,7 @@ import time
 from apiclient.discovery import build
 from googleapiclient.errors import HttpError
 from oauth2client.client import GoogleCredentials
+from oauth2client.client import ApplicationDefaultCredentialsError
 
 log = logging.getLogger('turbinia')
 
@@ -45,8 +46,16 @@ def create_service(service_name, api_version):
 
   Returns:
     API service resource (apiclient.discovery.Resource)
+
+  Raises:
+    RuntimeError: If Application Default Credentials could not be obtained.
   """
-  credentials = GoogleCredentials.get_application_default()
+  try:
+    credentials = GoogleCredentials.get_application_default()
+  except ApplicationDefaultCredentialsError as error:
+    raise RuntimeError(
+        'Could not get application default credentials: {0!s}\n'
+        'Have you run $ gcloud auth application-default login?'.format(error))
   return build(
       service_name, api_version, credentials=credentials, cache_discovery=False)
 
@@ -286,6 +295,9 @@ class GoogleCloudProject(object):
     except HttpError as exception:
       if exception.resp.status == 409:
         raise RuntimeError('Disk {0:s} already exists'.format(disk_name))
+      raise RuntimeError(
+          'Unknown error (status: {0:d}) occurred when creating '
+          'disk from snapshot:\n{1!s}'.format(exception.resp.status, exception))
     self.gce_operation(operation, zone=self.default_zone, block=True)
     return GoogleComputeDisk(
         project=self, zone=self.default_zone, name=disk_name)
diff --git a/turbinia/state_manager.py b/turbinia/state_manager.py
index c4d4889..d21801a 100644
--- a/turbinia/state_manager.py
+++ b/turbinia/state_manager.py
@@ -29,6 +29,7 @@ from datetime import timedelta
 import six
 
 from turbinia import config
+from turbinia.config import DATETIME_FORMAT
 from turbinia import TurbiniaException
 from turbinia.workers import TurbiniaTask
 from turbinia.workers import TurbiniaTaskResult
@@ -44,7 +45,6 @@ else:
       config.STATE_MANAGER)
   raise TurbiniaException(msg)
 
-DATETIME_FORMAT = '%Y-%m-%dT%H:%M:%S'
 MAX_DATASTORE_STRLEN = 1500
 log = logging.getLogger('turbinia')
 
@@ -101,6 +101,10 @@ class BaseStateManager(object):
         if isinstance(task_dict[attr], six.binary_type):
           task_dict[attr] = six.u(task_dict[attr])
 
+    # We'll store the run_time as seconds instead of a timedelta()
+    if task_dict.get('run_time'):
+      task_dict['run_time'] = task_dict['run_time'].total_seconds()
+
     # Set all non-existent keys to None
     all_attrs = set(
         TurbiniaTask.STORED_ATTRIBUTES + TurbiniaTaskResult.STORED_ATTRIBUTES)
@@ -154,7 +158,13 @@ class DatastoreStateManager(BaseStateManager):
 
   def __init__(self):
     config.LoadConfig()
-    self.client = datastore.Client(project=config.TURBINIA_PROJECT)
+    try:
+      self.client = datastore.Client(project=config.TURBINIA_PROJECT)
+    except EnvironmentError as e:
+      message = (
+          'Could not create Datastore client: {0!s}\n'
+          'Have you run $ gcloud auth application-default login?'.format(e))
+      raise TurbiniaException(message)
 
   def _validate_data(self, data):
     for key, value in iter(data.items()):
@@ -233,14 +243,20 @@ class RedisStateManager(BaseStateManager):
         if json.loads(self.client.get(task)).get('instance') == instance or
         not instance
     ]
+
+    # Convert relevant date attributes back into dates/timedeltas
+    for task in tasks:
+      if task.get('last_update'):
+        task['last_update'] = datetime.strptime(
+            task.get('last_update'), DATETIME_FORMAT)
+      if task.get('run_time'):
+        task['run_time'] = datetime.timedelta(seconds=task['run_time'])
+
     # pylint: disable=no-else-return
     if days:
       start_time = datetime.now() - timedelta(days=days)
       # Redis only supports strings; we convert to/from datetime here and below
-      return [
-          task for task in tasks if datetime.strptime(
-              task.get('last_update'), DATETIME_FORMAT) > start_time
-      ]
+      return [task for task in tasks if task.get('last_update') > start_time]
     elif task_id:
       return [task for task in tasks if task.get('task_id') == task_id]
     elif request_id:
@@ -256,6 +272,8 @@ class RedisStateManager(BaseStateManager):
     task_data = self.get_task_dict(task)
     task_data['last_update'] = task_data['last_update'].strftime(
         DATETIME_FORMAT)
+    if task_data['run_time']:
+      task_data['run_time'] = task_data['run_time'].total_seconds()
     # Need to use json.dumps, else redis returns single quoted string which
     # is invalid json
     if not self.client.set(key, json.dumps(task_data)):
@@ -268,6 +286,8 @@ class RedisStateManager(BaseStateManager):
     task_data = self.get_task_dict(task)
     task_data['last_update'] = task_data['last_update'].strftime(
         DATETIME_FORMAT)
+    if task_data['run_time']:
+      task_data['run_time'] = task_data['run_time'].total_seconds()
     # nx=True prevents overwriting (i.e. no unintentional task clobbering)
     if not self.client.set(key, json.dumps(task_data), nx=True):
       log.error(
diff --git a/turbinia/turbiniactl.py b/turbinia/turbiniactl.py
index ec04a42..e0be322 100644
--- a/turbinia/turbiniactl.py
+++ b/turbinia/turbiniactl.py
@@ -288,6 +288,10 @@ def main():
   parser_status.add_argument(
       '-c', '--close_tasks', action='store_true',
       help='Close tasks based on Request ID or Task ID', required=False)
+  parser_status.add_argument(
+      '-C', '--csv', action='store_true',
+      help='When used with --statistics, the output will be in CSV format',
+      required=False)
   parser_status.add_argument(
       '-d', '--days_history', default=0, type=int,
       help='Number of days of history to show', required=False)
@@ -309,6 +313,9 @@ def main():
       '-R', '--full_report',
       help='Generate full markdown report instead of just a summary',
       action='store_true', required=False)
+  parser_status.add_argument(
+      '-s', '--statistics', help='Generate statistics only',
+      action='store_true', required=False)
   parser_status.add_argument(
       '-t', '--task_id', help='Show task for given Task ID', required=False)
   parser_status.add_argument(
@@ -465,6 +472,14 @@ def main():
         )
         sys.exit(1)
 
+    if args.statistics:
+      print(
+          client.format_task_statistics(
+              instance=config.INSTANCE_ID, project=config.TURBINIA_PROJECT,
+              region=region, days=args.days_history, task_id=args.task_id,
+              request_id=args.request_id, user=args.user, csv=args.csv))
+      sys.exit(0)
+
     if args.wait and args.request_id:
       client.wait_for_request(
           instance=config.INSTANCE_ID, project=config.TURBINIA_PROJECT,
diff --git a/turbinia/workers/__init__.py b/turbinia/workers/__init__.py
index ce65023..c8ef47e 100644
--- a/turbinia/workers/__init__.py
+++ b/turbinia/workers/__init__.py
@@ -34,6 +34,7 @@ import turbinia
 import filelock
 
 from turbinia import config
+from turbinia.config import DATETIME_FORMAT
 from turbinia.evidence import evidence_decode
 from turbinia import output_manager
 from turbinia import TurbiniaException
@@ -82,8 +83,8 @@ class TurbiniaTaskResult(object):
 
   # The list of attributes that we will persist into storage
   STORED_ATTRIBUTES = [
-      'worker_name', 'report_data', 'report_priority', 'status', 'saved_paths',
-      'successful'
+      'worker_name', 'report_data', 'report_priority', 'run_time', 'status',
+      'saved_paths', 'successful'
   ]
 
   def __init__(
@@ -266,7 +267,7 @@ class TurbiniaTaskResult(object):
       dict: Object dictionary that is JSON serializable.
     """
     self.run_time = self.run_time.total_seconds() if self.run_time else None
-    self.start_time = str(self.start_time)
+    self.start_time = self.start_time.strftime(DATETIME_FORMAT)
     if self.input_evidence:
       self.input_evidence = self.input_evidence.serialize()
     self.evidence = [x.serialize() for x in self.evidence]
@@ -286,8 +287,7 @@ class TurbiniaTaskResult(object):
     result.__dict__.update(input_dict)
     if result.run_time:
       result.run_time = timedelta(seconds=result.run_time)
-    result.start_time = datetime.strptime(
-        result.start_time, '%Y-%m-%d %H:%M:%S.%f')
+    result.start_time = datetime.strptime(result.start_time, DATETIME_FORMAT)
     if result.input_evidence:
       result.input_evidence = evidence_decode(result.input_evidence)
     result.evidence = [evidence_decode(x) for x in result.evidence]
@@ -330,6 +330,7 @@ class TurbiniaTask(object):
       self.base_output_dir = base_output_dir
     else:
       self.base_output_dir = config.OUTPUT_DIR
+
     self.id = uuid.uuid4().hex
     self.last_update = datetime.now()
     self.name = name if name else self.__class__.__name__
@@ -353,7 +354,7 @@ class TurbiniaTask(object):
     """
     task_copy = deepcopy(self.__dict__)
     task_copy['output_manager'] = self.output_manager.__dict__
-    task_copy['last_update'] = str(self.last_update)
+    task_copy['last_update'] = self.last_update.strftime(DATETIME_FORMAT)
     return task_copy
 
   @classmethod
@@ -381,7 +382,7 @@ class TurbiniaTask(object):
     task.output_manager = output_manager.OutputManager()
     task.output_manager.__dict__.update(input_dict['output_manager'])
     task.last_update = datetime.strptime(
-        input_dict['last_update'], '%Y-%m-%d %H:%M:%S.%f')
+        input_dict['last_update'], DATETIME_FORMAT)
     return task
 
   def execute(
@@ -592,6 +593,7 @@ class TurbiniaTask(object):
       try:
         self.result = self.setup(evidence)
         original_result_id = self.result.id
+        evidence.validate()
 
         if self.turbinia_version != turbinia.__version__:
           message = (


Test patch:
diff --git a/turbinia/client_test.py b/turbinia/client_test.py
index e2bd8d4..24c9a64 100644
--- a/turbinia/client_test.py
+++ b/turbinia/client_test.py
@@ -17,6 +17,8 @@
 from __future__ import unicode_literals
 
 from datetime import datetime
+from datetime import timedelta
+import json
 import unittest
 import os
 import shutil
@@ -28,6 +30,7 @@ import mock
 from turbinia import config
 from turbinia.client import TurbiniaClient
 from turbinia.client import TurbiniaServer
+from turbinia.client import TurbiniaStats
 from turbinia.client import TurbiniaPsqWorker
 from turbinia import TurbiniaException
 
@@ -58,7 +61,7 @@ LONG_REPORT = textwrap.dedent(
     ## TaskName2
     * **Status:** This second fake task executed
     * Task Id: 0xfakeTaskId2
-    * Executed on worker fake_worker
+    * Executed on worker fake_worker2
 
     ### Task Reported Data
     #### Fake High priority Report
@@ -83,7 +86,7 @@ LONG_REPORT_FILES = textwrap.dedent(
     ## TaskName2
     * **Status:** This second fake task executed
     * Task Id: 0xfakeTaskId2
-    * Executed on worker fake_worker
+    * Executed on worker fake_worker2
 
     ### Task Reported Data
     #### Fake High priority Report
@@ -110,20 +113,55 @@ LONG_REPORT_FILES = textwrap.dedent(
     * None
 """)
 
+STATISTICS_REPORT = textwrap.dedent(
+    """\
+    Execution time statistics for Turbinia:
+
+    All Tasks: Count: 3, Min: 0:01:00, Mean: 0:03:00, Max: 0:05:00
+    Successful Tasks: Count: 2, Min: 0:01:00, Mean: 0:05:00, Max: 0:05:00
+    Failed Tasks: Count: 1, Min: 0:03:00, Mean: 0:03:00, Max: 0:03:00
+    Total Request Time: Count: 2, Min: 0:03:00, Mean: 0:21:00, Max: 0:21:00
+    Task type TaskName: Count: 1, Min: 0:01:00, Mean: 0:01:00, Max: 0:01:00
+    Task type TaskName2: Count: 1, Min: 0:05:00, Mean: 0:05:00, Max: 0:05:00
+    Task type TaskName3: Count: 1, Min: 0:03:00, Mean: 0:03:00, Max: 0:03:00
+    Worker fake_worker: Count: 2, Min: 0:01:00, Mean: 0:03:00, Max: 0:03:00
+    Worker fake_worker2: Count: 1, Min: 0:05:00, Mean: 0:05:00, Max: 0:05:00
+    User myuser: Count: 2, Min: 0:01:00, Mean: 0:05:00, Max: 0:05:00
+    User myuser2: Count: 1, Min: 0:03:00, Mean: 0:03:00, Max: 0:03:00
+""")
+
+STATISTICS_REPORT_CSV = textwrap.dedent(
+    """\
+    stat_type, count, min, mean, max
+    All Tasks, 3, 0:01:00, 0:03:00, 0:05:00
+    Successful Tasks, 2, 0:01:00, 0:05:00, 0:05:00
+    Failed Tasks, 1, 0:03:00, 0:03:00, 0:03:00
+    Total Request Time, 2, 0:03:00, 0:21:00, 0:21:00
+    Task type TaskName, 1, 0:01:00, 0:01:00, 0:01:00
+    Task type TaskName2, 1, 0:05:00, 0:05:00, 0:05:00
+    Task type TaskName3, 1, 0:03:00, 0:03:00, 0:03:00
+    Worker fake_worker, 2, 0:01:00, 0:03:00, 0:03:00
+    Worker fake_worker2, 1, 0:05:00, 0:05:00, 0:05:00
+    User myuser, 2, 0:01:00, 0:05:00, 0:05:00
+    User myuser2, 1, 0:03:00, 0:03:00, 0:03:00
+""")
+
 
 class TestTurbiniaClient(unittest.TestCase):
   """Test Turbinia client class."""
 
   def setUp(self):
+    last_update = datetime.now()
     self.task_data = [
         {
             'id': '0xfakeTaskId',
             'instance': 'MyTurbiniaInstance',
-            'last_update': datetime.now(),
+            'last_update': last_update,
             'name': 'TaskName',
             'report_data': '#### Fake Low priority Report\n* Fake Bullet',
             'report_priority': 80,
             'request_id': '0xFakeRequestId',
+            'run_time': timedelta(minutes=1),
             'saved_paths': ['/no/path/', '/fake/path'],
             'status': 'This fake task executed',
             'successful': True,
@@ -132,28 +170,30 @@ class TestTurbiniaClient(unittest.TestCase):
         }, {
             'id': '0xfakeTaskId2',
             'instance': 'MyTurbiniaInstance',
-            'last_update': datetime.now(),
+            'last_update': last_update + timedelta(minutes=20),
             'name': 'TaskName2',
             'report_data': '#### Fake High priority Report\n* Fake Bullet',
             'report_priority': 10,
             'request_id': '0xFakeRequestId',
+            'run_time': timedelta(minutes=5),
             'saved_paths': ['/no/path/2', '/fake/path/2'],
             'status': 'This second fake task executed',
             'successful': True,
             'requester': 'myuser',
-            'worker_name': 'fake_worker'
+            'worker_name': 'fake_worker2'
         }, {
             'id': '0xfakeTaskId3',
             'instance': 'MyTurbiniaInstance',
-            'last_update': datetime.now(),
+            'last_update': last_update,
             'name': 'TaskName3',
             'report_data': '',
             'report_priority': 80,
-            'request_id': '0xFakeRequestId',
+            'request_id': '0xFakeRequestId2',
+            'run_time': timedelta(minutes=3),
             'saved_paths': ['/no/path/3', '/fake/path/3'],
             'status': 'Third Task Failed...',
             'successful': False,
-            'requester': 'myuser',
+            'requester': 'myuser2',
             'worker_name': 'fake_worker'
         }
     ] # yapf: disable
@@ -173,11 +213,20 @@ class TestTurbiniaClient(unittest.TestCase):
   def testTurbiniaClientGetTaskData(self, _, __, mock_cloud_function):
     """Basic test for client.get_task_data"""
     # ExecuteFunction returns a dict with a 'result' key that has a json-encoded
-    # list.
-    function_return = {'result': '["bar", "baz"]'}
+    # list.  This contains our task data, which is a list of dicts.
+    run_time = timedelta(seconds=3)
+    test_task_data = [{'bar': 'bar2', 'run_time': run_time.total_seconds()}]
+    gcf_result = [test_task_data, 'Unused GCF data']
+    gcf_result = json.dumps(gcf_result)
+    function_return = {'result': gcf_result}
     mock_cloud_function.return_value = function_return
     client = TurbiniaClient()
-    self.assertEqual(client.get_task_data("inst", "proj", "reg"), "bar")
+    task_data = client.get_task_data('inst', 'proj', 'reg')
+    # get_task_data() converts this back into a timedelta(). We returned it
+    # seconds from the GCF function call because that is what it is stored in
+    # Datastore as.
+    test_task_data[0]['run_time'] = run_time
+    self.assertEqual(task_data, test_task_data)
 
   @mock.patch('turbinia.client.GoogleCloudFunction.ExecuteFunction')
   @mock.patch('turbinia.client.task_manager.PSQTaskManager._backend_setup')
@@ -200,6 +249,64 @@ class TestTurbiniaClient(unittest.TestCase):
     self.assertRaises(
         TurbiniaException, client.get_task_data, "inst", "proj", "reg")
 
+  @mock.patch('turbinia.client.GoogleCloudFunction.ExecuteFunction')
+  @mock.patch('turbinia.client.task_manager.PSQTaskManager._backend_setup')
+  @mock.patch('turbinia.state_manager.get_state_manager')
+  def testClientFormatTaskStatistics(self, _, __, ___):
+    """Tests format_task_statistics() report output."""
+    client = TurbiniaClient()
+    client.get_task_data = mock.MagicMock()
+    client.get_task_data.return_value = self.task_data
+    stats_report = client.format_task_statistics('inst', 'proj', 'reg')
+    self.maxDiff = None
+    self.assertEqual(stats_report, STATISTICS_REPORT)
+
+  @mock.patch('turbinia.client.GoogleCloudFunction.ExecuteFunction')
+  @mock.patch('turbinia.client.task_manager.PSQTaskManager._backend_setup')
+  @mock.patch('turbinia.state_manager.get_state_manager')
+  def testClientFormatTaskStatisticsCsv(self, _, __, ___):
+    """Tests format_task_statistics() CSV report output."""
+    client = TurbiniaClient()
+    client.get_task_data = mock.MagicMock()
+    client.get_task_data.return_value = self.task_data
+    stats_report = client.format_task_statistics(
+        'inst', 'proj', 'reg', csv=True)
+    self.maxDiff = None
+    self.assertEqual(stats_report, STATISTICS_REPORT_CSV)
+
+  @mock.patch('turbinia.client.GoogleCloudFunction.ExecuteFunction')
+  @mock.patch('turbinia.client.task_manager.PSQTaskManager._backend_setup')
+  @mock.patch('turbinia.state_manager.get_state_manager')
+  def testClientGetTaskStatistics(self, _, __, ___):
+    """Tests get_task_statistics() basic functionality."""
+    client = TurbiniaClient()
+    client.get_task_data = mock.MagicMock()
+    client.get_task_data.return_value = self.task_data
+    task_stats = client.get_task_statistics('inst', 'proj', 'reg')
+
+    # Make sure we have the right number of tasks for all sections
+    self.assertEqual(task_stats['all_tasks'].count, 3)
+    self.assertEqual(task_stats['successful_tasks'].count, 2)
+    self.assertEqual(task_stats['failed_tasks'].count, 1)
+    self.assertEqual(task_stats['requests'].count, 2)
+    self.assertEqual(len(task_stats['tasks_per_user']), 2)
+    self.assertEqual(len(task_stats['tasks_per_worker']), 2)
+    self.assertEqual(len(task_stats['tasks_per_type']), 3)
+
+    # Checking min/mean/max
+    self.assertEqual(task_stats['all_tasks'].min, timedelta(minutes=1))
+    self.assertEqual(task_stats['all_tasks'].mean, timedelta(minutes=3))
+    self.assertEqual(task_stats['all_tasks'].max, timedelta(minutes=5))
+    # Delta for this is 21 minutes because the last_update for 0xfakeTaskId2 is
+    # 20 minutes later than the first task, and the first task ran for 1 minute.
+    self.assertEqual(task_stats['requests'].max, timedelta(minutes=21))
+    self.assertEqual(
+        task_stats['tasks_per_user']['myuser'].max, timedelta(minutes=5))
+    self.assertEqual(
+        task_stats['tasks_per_worker']['fake_worker'].max, timedelta(minutes=3))
+    self.assertEqual(
+        task_stats['tasks_per_type']['TaskName2'].mean, timedelta(minutes=5))
+
   @mock.patch('turbinia.client.GoogleCloudFunction.ExecuteFunction')
   @mock.patch('turbinia.client.task_manager.PSQTaskManager._backend_setup')
   @mock.patch('turbinia.state_manager.get_state_manager')
@@ -217,8 +324,8 @@ class TestTurbiniaClient(unittest.TestCase):
   @mock.patch('turbinia.client.GoogleCloudFunction.ExecuteFunction')
   @mock.patch('turbinia.client.task_manager.PSQTaskManager._backend_setup')
   @mock.patch('turbinia.state_manager.get_state_manager')
-  def testClientFormatTaskStatus(self, _, __, ___):
-    """Tests format_task_status() has valid output."""
+  def testClientFormatTaskStatusShortReport(self, _, __, ___):
+    """Tests format_task_status() has valid output with short report."""
     client = TurbiniaClient()
     client.get_task_data = mock.MagicMock()
     client.get_task_data.return_value = self.task_data
@@ -249,6 +356,71 @@ class TestTurbiniaClient(unittest.TestCase):
     self.assertEqual(result.strip(), LONG_REPORT_FILES.strip())
 
 
+class TestTurbiniaStats(unittest.TestCase):
+  """Test TurbiniaStats class."""
+
+  def testTurbiniaStatsAddTask(self):
+    """Tests TurbiniaStats.add_task() method."""
+    test_task = {'run_time': None, 'last_update': None}
+    stats = TurbiniaStats()
+    stats.add_task(test_task)
+    self.assertIn(test_task, stats.tasks)
+    self.assertEqual(stats.count, 1)
+
+  def testTurbiniaStatsCalculateStats(self):
+    """Tests TurbiniaStats.calculateStats() method."""
+    last_update = datetime.now()
+    test_task1 = {'run_time': timedelta(minutes=3), 'last_update': last_update}
+    test_task2 = {'run_time': timedelta(minutes=5), 'last_update': last_update}
+    test_task3 = {'run_time': timedelta(minutes=1), 'last_update': last_update}
+
+    stats = TurbiniaStats()
+    stats.add_task(test_task1)
+    stats.add_task(test_task2)
+    stats.add_task(test_task3)
+    stats.calculate_stats()
+
+    self.assertEqual(stats.min, timedelta(minutes=1))
+    self.assertEqual(stats.mean, timedelta(minutes=3))
+    self.assertEqual(stats.max, timedelta(minutes=5))
+    self.assertEqual(stats.count, 3)
+
+  def testTurbiniaStatsCalculateStatsEmpty(self):
+    """Tests that calculate_stats() works when no tasks are added."""
+    stats = TurbiniaStats()
+    stats.calculate_stats()
+    self.assertEqual(stats.count, 0)
+    self.assertEqual(stats.min, None)
+
+  def testTurbiniaStatsFormatStats(self):
+    """Tests TurbiniaStats.format_stats() returns valid output."""
+    test_output = (
+        'Test Task Results: Count: 1, Min: 0:03:00, Mean: 0:03:00, '
+        'Max: 0:03:00')
+    test_task1 = {
+        'run_time': timedelta(minutes=3),
+        'last_update': datetime.now()
+    }
+    stats = TurbiniaStats('Test Task Results')
+    stats.add_task(test_task1)
+    stats.calculate_stats()
+    report = stats.format_stats()
+    self.assertEqual(report, test_output)
+
+  def testTurbiniaStatsFormatStatsCsv(self):
+    """Tests TurbiniaStats.format_stats() returns valid CSV output."""
+    test_output = ('Test Task Results, 1, 0:03:00, 0:03:00, 0:03:00')
+    test_task1 = {
+        'run_time': timedelta(minutes=3),
+        'last_update': datetime.now()
+    }
+    stats = TurbiniaStats('Test Task Results')
+    stats.add_task(test_task1)
+    stats.calculate_stats()
+    report = stats.format_stats_csv()
+    self.assertEqual(report, test_output)
+
+
 class TestTurbiniaServer(unittest.TestCase):
   """Test Turbinia Server class."""
 
diff --git a/turbinia/evidence_test.py b/turbinia/evidence_test.py
index 15a9c35..d1c02f4 100644
--- a/turbinia/evidence_test.py
+++ b/turbinia/evidence_test.py
@@ -46,3 +46,24 @@ class TestTurbiniaEvidence(unittest.TestCase):
     """Test that evidence_decode throws error on dict with no type attribute."""
     test = {1: 2, 3: 4}
     self.assertRaises(TurbiniaException, evidence.evidence_decode, test)
+
+  def testEvidenceValidation(self):
+    """Test successful evidence validation."""
+    rawdisk = evidence.RawDisk(
+        name='My Evidence', local_path='/tmp/foo', mount_path='/mnt/foo')
+    rawdisk.REQUIRED_ATTRIBUTES = ['name', 'local_path', 'mount_path']
+    rawdisk.validate()
+
+  def testEvidenceValidationEmptyAttribute(self):
+    """Test failed evidence validation with an empty attribute."""
+    rawdisk = evidence.RawDisk(
+        name='My Evidence', local_path=None, mount_path='/mnt/foo')
+    rawdisk.REQUIRED_ATTRIBUTES = ['name', 'local_path', 'mount_path']
+    self.assertRaises(TurbiniaException, rawdisk.validate)
+
+  def testEvidenceValidationNoAttribute(self):
+    """Test failed evidence validation with no attribute."""
+    rawdisk = evidence.RawDisk(
+        name='My Evidence', local_path='/tmp/foo', mount_path='/mnt/foo')
+    rawdisk.REQUIRED_ATTRIBUTES = ['doesnotexist']
+    self.assertRaises(TurbiniaException, rawdisk.validate)
diff --git a/turbinia/workers/workers_test.py b/turbinia/workers/workers_test.py
index 65a1c8a..1b45874 100644
--- a/turbinia/workers/workers_test.py
+++ b/turbinia/workers/workers_test.py
@@ -191,6 +191,18 @@ class TestTurbiniaTask(TestTurbiniaTaskBase):
     self.assertEqual(type(new_result), TurbiniaTaskResult)
     self.assertNotEqual(new_result.error, {})
 
+  @mock.patch('turbinia.workers.evidence_decode')
+  def testTurbiniaTaskEvidenceValidationFailure(self, evidence_decode_mock):
+    """Tests Task fails when evidence validation fails."""
+    self.setResults()
+    test_evidence = evidence.RawDisk()
+    test_evidence.REQUIRED_ATTRIBUTES = ['doesnotexist']
+    evidence_decode_mock.return_value = test_evidence
+    test_result = self.task.run_wrapper(test_evidence.__dict__)
+    test_result = TurbiniaTaskResult.deserialize(test_result)
+    self.assertFalse(test_result.successful)
+    self.assertIn('validation failed', test_result.status)
+
   @mock.patch('turbinia.workers.subprocess.Popen')
   def testTurbiniaTaskExecute(self, popen_mock):
     """Test execution with success case."""