The main factory function for creating and executing a new flow.

  Args:
    args: An arg protocol buffer which is an instance of the required flow's
      args_type class attribute.
    runner_args: an instance of FlowRunnerArgs() protocol buffer which is used
      to initialize the runner for this flow.
    parent_flow: A parent flow or None if this is a top level flow.
    sync: If True, the Start method of this flow will be called inline.
      Otherwise we schedule the starting of this flow on another worker.
    token: Security credentials token identifying the user.
    **kwargs: If args or runner_args are not specified, we construct these
      protobufs from these keywords.

  Returns:
    the session id of the flow.

  Raises:
    RuntimeError: Unknown or invalid parameters were provided.

def StartAFF4Flow(args=None,
                  runner_args=None,
                  parent_flow=None,
                  sync=True,
                  token=None,
                  **kwargs):
  """The main factory function for creating and executing a new flow.

  Args:
    args: An arg protocol buffer which is an instance of the required flow's
      args_type class attribute.
    runner_args: an instance of FlowRunnerArgs() protocol buffer which is used
      to initialize the runner for this flow.
    parent_flow: A parent flow or None if this is a top level flow.
    sync: If True, the Start method of this flow will be called inline.
      Otherwise we schedule the starting of this flow on another worker.
    token: Security credentials token identifying the user.
    **kwargs: If args or runner_args are not specified, we construct these
      protobufs from these keywords.

  Returns:
    the session id of the flow.

  Raises:
    RuntimeError: Unknown or invalid parameters were provided.
  """
  # Build the runner args from the keywords.
  if runner_args is None:
    runner_args = rdf_flow_runner.FlowRunnerArgs()

  FilterArgsFromSemanticProtobuf(runner_args, kwargs)

  # Is the required flow a known flow?
  try:
    flow_cls = registry.AFF4FlowRegistry.FlowClassByName(runner_args.flow_name)
  except ValueError:
    stats_collector_instance.Get().IncrementCounter(
        "grr_flow_invalid_flow_count")
    raise RuntimeError("Unable to locate flow %s" % runner_args.flow_name)

  # If no token is specified, raise.
  if not token:
    raise access_control.UnauthorizedAccess("A token must be specified.")

  # For the flow itself we use a supervisor token.
  token = token.SetUID()

  # Extend the expiry time of this token indefinitely. Python on Windows only
  # supports dates up to the year 3000.
  token.expiry = rdfvalue.RDFDatetime.FromHumanReadable("2997-01-01")

  if flow_cls.category and not runner_args.client_id:
    raise RuntimeError("Flow with category (user-visible flow) has to be "
                       "started on a client, but runner_args.client_id "
                       "is missing.")

  # We create an anonymous AFF4 object first, The runner will then generate
  # the appropriate URN.
  flow_obj = aff4.FACTORY.Create(None, flow_cls, token=token)

  # Now parse the flow args into the new object from the keywords.
  if args is None:
    args = flow_obj.args_type()

  FilterArgsFromSemanticProtobuf(args, kwargs)

  # Check that the flow args are valid.
  args.Validate()

  # Store the flow args.
  flow_obj.args = args
  flow_obj.runner_args = runner_args

  # At this point we should exhaust all the keyword args. If any are left
  # over, we do not know what to do with them so raise.
  if kwargs:
    raise type_info.UnknownArg("Unknown parameters to StartAFF4Flow: %s" %
                               kwargs)

  # Create a flow runner to run this flow with.
  if parent_flow:
    parent_runner = parent_flow.runner
  else:
    parent_runner = None

  runner = flow_obj.CreateRunner(
      parent_runner=parent_runner, runner_args=runner_args)

  logging.info(u"Scheduling %s(%s) on %s", flow_obj.urn, runner_args.flow_name,
               runner_args.client_id)
  if sync:
    # Just run the first state inline. NOTE: Running synchronously means
    # that this runs on the thread that starts the flow. The advantage is
    # that that Start method can raise any errors immediately.
    flow_obj.Start()
  else:
    # Running Asynchronously: Schedule the start method on another worker.
    runner.CallState(next_state="Start")

  # The flow does not need to actually remain running.
  if not flow_obj.outstanding_requests:
    flow_obj.Terminate()

  flow_obj.Close()

  # Publish an audit event, only for top level flows.
  if parent_flow is None:
    events.Events.PublishEvent(
        "Audit",
        rdf_events.AuditEvent(
            user=token.username,
            action="RUN_FLOW",
            flow_name=runner_args.flow_name,
            urn=flow_obj.urn,
            client=runner_args.client_id),
        token=token)

  return flow_obj.urn