def _MergeTaskStorage(self, storage_writer):
    """Merges a task storage with the session storage.

    This function checks all task stores that are ready to merge and updates
    the scheduled tasks. Note that to prevent this function holding up
    the task scheduling loop only the first available task storage is merged.

    Args:
      storage_writer (StorageWriter): storage writer for a session storage used
          to merge task storage.
    """
    if self._processing_profiler:
      self._processing_profiler.StartTiming('merge_check')

    for task_identifier in storage_writer.GetProcessedTaskIdentifiers():
      try:
        task = self._task_manager.GetProcessedTaskByIdentifier(task_identifier)

        self._task_manager.SampleTaskStatus(task, 'processed')

        to_merge = self._task_manager.CheckTaskToMerge(task)
        if not to_merge:
          storage_writer.RemoveProcessedTaskStorage(task)

          self._task_manager.RemoveTask(task)
          self._task_manager.SampleTaskStatus(task, 'removed_processed')

        else:
          storage_writer.PrepareMergeTaskStorage(task)
          self._task_manager.UpdateTaskAsPendingMerge(task)

      except KeyError:
        logger.error(
            'Unable to retrieve task: {0:s} to prepare it to be merged.'.format(
                task_identifier))
        continue

    if self._processing_profiler:
      self._processing_profiler.StopTiming('merge_check')

    task = None
    if not self._storage_merge_reader_on_hold:
      task = self._task_manager.GetTaskPendingMerge(self._merge_task)

    # Limit the number of attribute containers from a single task-based
    # storage file that are merged per loop to keep tasks flowing.
    if task or self._storage_merge_reader:
      self._status = definitions.STATUS_INDICATOR_MERGING

      if self._processing_profiler:
        self._processing_profiler.StartTiming('merge')

      if task:
        if self._storage_merge_reader:
          self._merge_task_on_hold = self._merge_task
          self._storage_merge_reader_on_hold = self._storage_merge_reader

          self._task_manager.SampleTaskStatus(
              self._merge_task_on_hold, 'merge_on_hold')

        self._merge_task = task
        try:
          self._storage_merge_reader = storage_writer.StartMergeTaskStorage(
              task)

          self._task_manager.SampleTaskStatus(task, 'merge_started')

        except IOError as exception:
          logger.error((
              'Unable to merge results of task: {0:s} '
              'with error: {1!s}').format(task.identifier, exception))
          self._storage_merge_reader = None

      if self._storage_merge_reader:
        fully_merged = self._storage_merge_reader.MergeAttributeContainers(
            maximum_number_of_containers=self._MAXIMUM_NUMBER_OF_CONTAINERS)
      else:
        # TODO: Do something more sensible when this happens, perhaps
        # retrying the task once that is implemented. For now, we mark the task
        # as fully merged because we can't continue with it.
        fully_merged = True

      if self._processing_profiler:
        self._processing_profiler.StopTiming('merge')

      if fully_merged:
        try:
          self._task_manager.CompleteTask(self._merge_task)

        except KeyError as exception:
          logger.error(
              'Unable to complete task: {0:s} with error: {1!s}'.format(
                  self._merge_task.identifier, exception))

        if not self._storage_merge_reader_on_hold:
          self._merge_task = None
          self._storage_merge_reader = None
        else:
          self._merge_task = self._merge_task_on_hold
          self._storage_merge_reader = self._storage_merge_reader_on_hold

          self._merge_task_on_hold = None
          self._storage_merge_reader_on_hold = None

          self._task_manager.SampleTaskStatus(
              self._merge_task, 'merge_resumed')

      self._status = definitions.STATUS_INDICATOR_RUNNING
      self._number_of_produced_events = storage_writer.number_of_events
      self._number_of_produced_sources = storage_writer.number_of_event_sources
      self._number_of_produced_warnings = storage_writer.number_of_warnings
def _MergeTaskStorage ( self , storage_writer ) : if self . _processing_profiler : self . _processing_profiler . StartTiming ( 'merge_check' ) for task_identifier in storage_writer . GetProcessedTaskIdentifiers ( ) : try : task = self . _task_manager . GetProcessedTaskByIdentifier ( task_identifier ) self . _task_manager . SampleTaskStatus ( task , 'processed' ) to_merge = self . _task_manager . CheckTaskToMerge ( task ) if not to_merge : storage_writer . RemoveProcessedTaskStorage ( task ) self . _task_manager . RemoveTask ( task ) self . _task_manager . SampleTaskStatus ( task , 'removed_processed' ) else : storage_writer . PrepareMergeTaskStorage ( task ) self . _task_manager . UpdateTaskAsPendingMerge ( task ) except KeyError : logger . error ( 'Unable to retrieve task: {0:s} to prepare it to be merged.' . format ( task_identifier ) ) continue if self . _processing_profiler : self . _processing_profiler . StopTiming ( 'merge_check' ) task = None if not self . _storage_merge_reader_on_hold : task = self . _task_manager . GetTaskPendingMerge ( self . _merge_task ) if task or self . _storage_merge_reader : self . _status = definitions . STATUS_INDICATOR_MERGING if self . _processing_profiler : self . _processing_profiler . StartTiming ( 'merge' ) if task : if self . _storage_merge_reader : self . _merge_task_on_hold = self . _merge_task self . _storage_merge_reader_on_hold = self . _storage_merge_reader self . _task_manager . SampleTaskStatus ( self . _merge_task_on_hold , 'merge_on_hold' ) self . _merge_task = task try : self . _storage_merge_reader = storage_writer . StartMergeTaskStorage ( task ) self . _task_manager . SampleTaskStatus ( task , 'merge_started' ) except IOError as exception : logger . error ( ( 'Unable to merge results of task: {0:s} ' 'with error: {1!s}' ) . format ( task . identifier , exception ) ) self . _storage_merge_reader = None if self . _storage_merge_reader : fully_merged = self . _storage_merge_reader . MergeAttributeContainers ( maximum_number_of_containers = self . _MAXIMUM_NUMBER_OF_CONTAINERS ) else : fully_merged = True if self . _processing_profiler : self . _processing_profiler . StopTiming ( 'merge' ) if fully_merged : try : self . _task_manager . CompleteTask ( self . _merge_task ) except KeyError as exception : logger . error ( 'Unable to complete task: {0:s} with error: {1!s}' . format ( self . _merge_task . identifier , exception ) ) if not self . _storage_merge_reader_on_hold : self . _merge_task = None self . _storage_merge_reader = None else : self . _merge_task = self . _merge_task_on_hold self . _storage_merge_reader = self . _storage_merge_reader_on_hold self . _merge_task_on_hold = None self . _storage_merge_reader_on_hold = None self . _task_manager . SampleTaskStatus ( self . _merge_task , 'merge_resumed' ) self . _status = definitions . STATUS_INDICATOR_RUNNING self . _number_of_produced_events = storage_writer . number_of_events self . _number_of_produced_sources = storage_writer . number_of_event_sources self . _number_of_produced_warnings = storage_writer . number_of_warnings
Merges a task storage with the session storage.

    This function checks all task stores that are ready to merge and updates
    the scheduled tasks. Note that to prevent this function holding up
    the task scheduling loop only the first available task storage is merged.

    Args:
      storage_writer (StorageWriter): storage writer for a session storage used
          to merge task storage.