def METHOD_NAME():
    module = AnsibleModule(
        argument_spec=dict(
            command=dict(required=False, default=None, choices=[
                'ping', 'kv_test', 'join', 'plan', 'commit']),
            config_dir=dict(default='/etc/riak', type='path'),
            http_conn=dict(required=False, default='127.0.0.1:8098'),
            target_node=dict(default='riak@127.0.0.1', required=False),
            wait_for_handoffs=dict(default=0, type='int'),
            wait_for_ring=dict(default=0, type='int'),
            wait_for_service=dict(
                required=False, default=None, choices=['kv']),
            validate_certs=dict(default=True, type='bool'))
    )
    command = module.params.get('command')
    http_conn = module.params.get('http_conn')
    target_node = module.params.get('target_node')
    wait_for_handoffs = module.params.get('wait_for_handoffs')
    wait_for_ring = module.params.get('wait_for_ring')
    wait_for_service = module.params.get('wait_for_service')
    # make sure riak commands are on the path
    riak_bin = module.get_bin_path('riak')
    riak_admin_bin = module.get_bin_path('riak-admin')
    timeout = time.time() + 120
    while True:
        if time.time() > timeout:
            module.fail_json(msg='Timeout, could not fetch Riak stats.')
        (response, info) = fetch_url(module, 'http://%s/stats' % (http_conn), force=True, timeout=5)
        if info['status'] == 200:
            stats_raw = response.read()
            break
        time.sleep(5)
    # here we attempt to load those stats,
    try:
        stats = json.loads(stats_raw)
    except Exception:
        module.fail_json(msg='Could not parse Riak stats.')
    node_name = stats['nodename']
    nodes = stats['ring_members']
    ring_size = stats['ring_creation_size']
    rc, out, err = module.run_command([riak_bin, 'version'])
    version = out.strip()
    result = dict(node_name=node_name,
                  nodes=nodes,
                  ring_size=ring_size,
                  version=version)
    if command == 'ping':
        cmd = '%s ping %s' % (riak_bin, target_node)
        rc, out, err = module.run_command(cmd)
        if rc == 0:
            result['ping'] = out
        else:
            module.fail_json(msg=out)
    elif command == 'kv_test':
        cmd = '%s test' % riak_admin_bin
        rc, out, err = module.run_command(cmd)
        if rc == 0:
            result['kv_test'] = out
        else:
            module.fail_json(msg=out)
    elif command == 'join':
        if nodes.count(node_name) == 1 and len(nodes) > 1:
            result['join'] = 'Node is already in cluster or staged to be in cluster.'
        else:
            cmd = '%s cluster join %s' % (riak_admin_bin, target_node)
            rc, out, err = module.run_command(cmd)
            if rc == 0:
                result['join'] = out
                result['changed'] = True
            else:
                module.fail_json(msg=out)
    elif command == 'plan':
        cmd = '%s cluster plan' % riak_admin_bin
        rc, out, err = module.run_command(cmd)
        if rc == 0:
            result['plan'] = out
            if 'Staged Changes' in out:
                result['changed'] = True
        else:
            module.fail_json(msg=out)
    elif command == 'commit':
        cmd = '%s cluster commit' % riak_admin_bin
        rc, out, err = module.run_command(cmd)
        if rc == 0:
            result['commit'] = out
            result['changed'] = True
        else:
            module.fail_json(msg=out)