'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.'

@classmethod
def StartFlow(cls, args=None, runner_args=None, parent_flow=None, sync=True, token=None, **kwargs):

    if (runner_args is None):
        runner_args = rdf_flows.FlowRunnerArgs()
    cls.FilterArgsFromSemanticProtobuf(runner_args, kwargs)
    if runner_args.start_time:
        sync = False
    if (runner_args.flow_name not in GRRFlow.classes):
        stats.STATS.IncrementCounter('grr_flow_invalid_flow_count')
        raise RuntimeError(('Unable    to    locate    flow    %s' % runner_args.flow_name))
    if (not token):
        raise access_control.UnauthorizedAccess('A    token    must    be    specified.')
    token = token.SetUID()
    token.expiry = rdfvalue.RDFDatetime.FromHumanReadable('2997-01-01')
    flow_cls = aff4.AFF4Object.classes.get(runner_args.flow_name)
    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.')
    flow_obj = aff4.FACTORY.Create(None, flow_cls, token=token)
    if (args is None):
        args = flow_obj.args_type()
    cls.FilterArgsFromSemanticProtobuf(args, kwargs)
    args.Validate()
    flow_obj.args = args
    flow_obj.runner_args = runner_args
    if kwargs:
        raise type_info.UnknownArg(('Unknown    parameters    to    StartFlow:    %s' % kwargs))
    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:
        flow_obj.Start()
    else:
        runner.CallState(next_state='Start', start_time=runner_args.start_time)
    if (not runner.OutstandingRequests()):
        flow_obj.Terminate()
    flow_obj.Close()
    if (parent_flow is None):
        events.Events.PublishEvent('Audit', 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
