@flow.StateHandler()
  def Start(self):
    # If max_running_time is not specified, set it to 60% of this job's
    # lifetime.
    if not self.state.args.max_running_time:
      self.state.args.max_running_time = rdfvalue.Duration(
          "%ds" % int(ProcessHuntResultsCronFlow.lifetime.seconds * 0.6))

    self.start_time = rdfvalue.RDFDatetime().Now()

    exceptions_by_hunt = {}
    freeze_timestamp = rdfvalue.RDFDatetime().Now()
    for results_urn in aff4.ResultsOutputCollection.QueryNotifications(
        timestamp=freeze_timestamp, token=self.token):

      aff4.ResultsOutputCollection.DeleteNotifications(
          [results_urn], end=results_urn.age, token=self.token)

      # Feed the results to output plugins
      try:
        results = aff4.FACTORY.Open(
            results_urn, aff4_type="ResultsOutputCollection", token=self.token)
      except aff4.InstantiationError:  # Collection does not exist.
        continue

      exceptions_by_plugin = self.ProcessHuntResults(results, freeze_timestamp)
      if exceptions_by_plugin:
        hunt_urn = results.Get(results.Schema.RESULTS_SOURCE)
        exceptions_by_hunt[hunt_urn] = exceptions_by_plugin

      lease_time = config_lib.CONFIG["Worker.compaction_lease_time"]
      try:
        with aff4.FACTORY.OpenWithLock(results_urn, blocking=False,
                                       aff4_type="ResultsOutputCollection",
                                       lease_time=lease_time,
                                       token=self.token) as results:
          num_compacted = results.Compact(callback=self.HeartBeat,
                                          timestamp=freeze_timestamp)
          stats.STATS.IncrementCounter("hunt_results_compacted",
                                       delta=num_compacted)
          logging.debug("Compacted %d results in %s.", num_compacted,
                        results_urn)
      except aff4.LockError:
        logging.error("Trying to compact a collection that's already "
                      "locked: %s", results_urn)
        stats.STATS.IncrementCounter("hunt_results_compaction_locking_errors")

      if self.CheckIfRunningTooLong():
        self.Log("Running for too long, skipping rest of hunts.")
        break

    if exceptions_by_hunt:
      e = process_results.ResultsProcessingError()
      for hunt_urn, exceptions_by_plugin in exceptions_by_hunt.items():
        for plugin_name, exceptions in exceptions_by_plugin.items():
          for exception in exceptions:
            e.RegisterSubException(hunt_urn, plugin_name, exception)
      raise e
Start state of the flow.
dataset/ETHPy150Open google/grr/grr/lib/hunts/standard.py/ProcessHuntResultsCronFlow.Start