flow: Refactor ct_orig_tuple check in miniflow_extract().

[mirror_ovs.git] / ovn / utilities / ovn-sbctl.c

/*
 * Copyright (c) 2015, 2016, 2017 Nicira, Inc.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at:
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <config.h>

#include <ctype.h>
#include <errno.h>
#include <float.h>
#include <getopt.h>
#include <inttypes.h>
#include <signal.h>
#include <stdarg.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>

#include "colors.h"
#include "command-line.h"
#include "compiler.h"
#include "db-ctl-base.h"
#include "dirs.h"
#include "fatal-signal.h"
#include "openvswitch/dynamic-string.h"
#include "openvswitch/json.h"
#include "openvswitch/ofp-actions.h"
#include "openvswitch/ofp-print.h"
#include "openvswitch/shash.h"
#include "openvswitch/vconn.h"
#include "openvswitch/vlog.h"
#include "ovn/lib/ovn-sb-idl.h"
#include "ovn/lib/ovn-util.h"
#include "ovsdb-data.h"
#include "ovsdb-idl.h"
#include "poll-loop.h"
#include "process.h"
#include "sset.h"
#include "stream-ssl.h"
#include "stream.h"
#include "table.h"
#include "timeval.h"
#include "util.h"
#include "svec.h"

VLOG_DEFINE_THIS_MODULE(sbctl);

struct sbctl_context;

/* --db: The database server to contact. */
static const char *db;

/* --oneline: Write each command's output as a single line? */
static bool oneline;

/* --dry-run: Do not commit any changes. */
static bool dry_run;

/* --timeout: Time to wait for a connection to 'db'. */
static int timeout;

/* Format for table output. */
static struct table_style table_style = TABLE_STYLE_DEFAULT;

/* The IDL we're using and the current transaction, if any.
 * This is for use by sbctl_exit() only, to allow it to clean up.
 * Other code should use its context arguments. */
static struct ovsdb_idl *the_idl;
static struct ovsdb_idl_txn *the_idl_txn;
OVS_NO_RETURN static void sbctl_exit(int status);

static void sbctl_cmd_init(void);
OVS_NO_RETURN static void usage(void);
static void parse_options(int argc, char *argv[], struct shash *local_options);
static void run_prerequisites(struct ctl_command[], size_t n_commands,
                              struct ovsdb_idl *);
static bool do_sbctl(const char *args, struct ctl_command *, size_t n,
                     struct ovsdb_idl *);

int
main(int argc, char *argv[])
{
    struct ovsdb_idl *idl;
    struct ctl_command *commands;
    struct shash local_options;
    unsigned int seqno;
    size_t n_commands;
    char *args;

    set_program_name(argv[0]);
    fatal_ignore_sigpipe();
    vlog_set_levels(NULL, VLF_CONSOLE, VLL_WARN);
    vlog_set_levels_from_string_assert("reconnect:warn");

    sbctl_cmd_init();

    /* Log our arguments.  This is often valuable for debugging systems. */
    args = process_escape_args(argv);
    VLOG(ctl_might_write_to_db(argv) ? VLL_INFO : VLL_DBG, "Called as %s", args);

    /* Parse command line. */
    shash_init(&local_options);
    parse_options(argc, argv, &local_options);
    commands = ctl_parse_commands(argc - optind, argv + optind, &local_options,
                                  &n_commands);

    if (timeout) {
        time_alarm(timeout);
    }

    /* Initialize IDL. */
    idl = the_idl = ovsdb_idl_create(db, &sbrec_idl_class, false, false);
    run_prerequisites(commands, n_commands, idl);

    /* Execute the commands.
     *
     * 'seqno' is the database sequence number for which we last tried to
     * execute our transaction.  There's no point in trying to commit more than
     * once for any given sequence number, because if the transaction fails
     * it's because the database changed and we need to obtain an up-to-date
     * view of the database before we try the transaction again. */
    seqno = ovsdb_idl_get_seqno(idl);
    for (;;) {
        ovsdb_idl_run(idl);
        if (!ovsdb_idl_is_alive(idl)) {
            int retval = ovsdb_idl_get_last_error(idl);
            ctl_fatal("%s: database connection failed (%s)",
                        db, ovs_retval_to_string(retval));
        }

        if (seqno != ovsdb_idl_get_seqno(idl)) {
            seqno = ovsdb_idl_get_seqno(idl);
            if (do_sbctl(args, commands, n_commands, idl)) {
                free(args);
                exit(EXIT_SUCCESS);
            }
        }

        if (seqno == ovsdb_idl_get_seqno(idl)) {
            ovsdb_idl_wait(idl);
            poll_block();
        }
    }
}

static void
parse_options(int argc, char *argv[], struct shash *local_options)
{
    enum {
        OPT_DB = UCHAR_MAX + 1,
        OPT_ONELINE,
        OPT_NO_SYSLOG,
        OPT_DRY_RUN,
        OPT_LOCAL,
        OPT_COMMANDS,
        OPT_OPTIONS,
        VLOG_OPTION_ENUMS,
        TABLE_OPTION_ENUMS,
        SSL_OPTION_ENUMS,
    };
    static const struct option global_long_options[] = {
        {"db", required_argument, NULL, OPT_DB},
        {"no-syslog", no_argument, NULL, OPT_NO_SYSLOG},
        {"dry-run", no_argument, NULL, OPT_DRY_RUN},
        {"oneline", no_argument, NULL, OPT_ONELINE},
        {"timeout", required_argument, NULL, 't'},
        {"help", no_argument, NULL, 'h'},
        {"commands", no_argument, NULL, OPT_COMMANDS},
        {"options", no_argument, NULL, OPT_OPTIONS},
        {"version", no_argument, NULL, 'V'},
        VLOG_LONG_OPTIONS,
        STREAM_SSL_LONG_OPTIONS,
        TABLE_LONG_OPTIONS,
        {NULL, 0, NULL, 0},
    };
    const int n_global_long_options = ARRAY_SIZE(global_long_options) - 1;
    char *tmp, *short_options;

    struct option *options;
    size_t allocated_options;
    size_t n_options;
    size_t i;

    tmp = ovs_cmdl_long_options_to_short_options(global_long_options);
    short_options = xasprintf("+%s", tmp);
    free(tmp);

    /* We want to parse both global and command-specific options here, but
     * getopt_long() isn't too convenient for the job.  We copy our global
     * options into a dynamic array, then append all of the command-specific
     * options. */
    options = xmemdup(global_long_options, sizeof global_long_options);
    allocated_options = ARRAY_SIZE(global_long_options);
    n_options = n_global_long_options;
    ctl_add_cmd_options(&options, &n_options, &allocated_options, OPT_LOCAL);

    for (;;) {
        int idx;
        int c;

        c = getopt_long(argc, argv, short_options, options, &idx);
        if (c == -1) {
            break;
        }

        switch (c) {
        case OPT_DB:
            db = optarg;
            break;

        case OPT_ONELINE:
            oneline = true;
            break;

        case OPT_NO_SYSLOG:
            vlog_set_levels(&this_module, VLF_SYSLOG, VLL_WARN);
            break;

        case OPT_DRY_RUN:
            dry_run = true;
            break;

        case OPT_LOCAL:
            if (shash_find(local_options, options[idx].name)) {
                ctl_fatal("'%s' option specified multiple times",
                            options[idx].name);
            }
            shash_add_nocopy(local_options,
                             xasprintf("--%s", options[idx].name),
                             nullable_xstrdup(optarg));
            break;

        case 'h':
            usage();

        case OPT_COMMANDS:
            ctl_print_commands();

        case OPT_OPTIONS:
            ctl_print_options(global_long_options);

        case 'V':
            ovs_print_version(0, 0);
            printf("DB Schema %s\n", sbrec_get_db_version());
            exit(EXIT_SUCCESS);

        case 't':
            timeout = strtoul(optarg, NULL, 10);
            if (timeout < 0) {
                ctl_fatal("value %s on -t or --timeout is invalid", optarg);
            }
            break;

        VLOG_OPTION_HANDLERS
        TABLE_OPTION_HANDLERS(&table_style)
        STREAM_SSL_OPTION_HANDLERS

        case '?':
            exit(EXIT_FAILURE);

        default:
            abort();
        }
    }
    free(short_options);

    if (!db) {
        db = default_sb_db();
    }

    for (i = n_global_long_options; options[i].name; i++) {
        free(CONST_CAST(char *, options[i].name));
    }
    free(options);
}

static void
usage(void)
{
    printf("\
%s: OVN southbound DB management utility\n\
\n\
usage: %s [OPTIONS] COMMAND [ARG...]\n\
\n\
General commands:\n\
  show                        print overview of database contents\n\
\n\
Chassis commands:\n\
  chassis-add CHASSIS ENCAP-TYPE ENCAP-IP  create a new chassis named\n\
                                           CHASSIS with ENCAP-TYPE tunnels\n\
                                           and ENCAP-IP\n\
  chassis-del CHASSIS         delete CHASSIS and all of its encaps\n\
                              and gateway_ports\n\
\n\
Port binding commands:\n\
  lsp-bind PORT CHASSIS       bind logical port PORT to CHASSIS\n\
  lsp-unbind PORT             reset the port binding of logical port PORT\n\
\n\
Logical flow commands:\n\
  lflow-list [DATAPATH] [LFLOW...] List logical flows for DATAPATH\n\
  dump-flows [DATAPATH] [LFLOW...] Alias for lflow-list\n\
\n\
Connection commands:\n\
  get-connection             print the connections\n\
  del-connection             delete the connections\n\
  set-connection TARGET...   set the list of connections to TARGET...\n\
\n\
SSL commands:\n\
  get-ssl                     print the SSL configuration\n\
  del-ssl                     delete the SSL configuration\n\
  set-ssl PRIV-KEY CERT CA-CERT  set the SSL configuration\n\
\n\
%s\
\n\
Options:\n\
  --db=DATABASE               connect to DATABASE\n\
                              (default: %s)\n\
  -t, --timeout=SECS          wait at most SECS seconds\n\
  --dry-run                   do not commit changes to database\n\
  --oneline                   print exactly one line of output per command\n",
           program_name, program_name, ctl_get_db_cmd_usage(),
           default_sb_db());
    table_usage();
    vlog_usage();
    printf("\
  --no-syslog             equivalent to --verbose=sbctl:syslog:warn\n");
    printf("\n\
Other options:\n\
  -h, --help                  display this help message\n\
  -V, --version               display version information\n");
    stream_usage("database", true, true, false);
    exit(EXIT_SUCCESS);
}

\f
/* ovs-sbctl specific context.  Inherits the 'struct ctl_context' as base. */
struct sbctl_context {
    struct ctl_context base;

    /* A cache of the contents of the database.
     *
     * A command that needs to use any of this information must first call
     * sbctl_context_populate_cache().  A command that changes anything that
     * could invalidate the cache must either call
     * sbctl_context_invalidate_cache() or manually update the cache to
     * maintain its correctness. */
    bool cache_valid;
    /* Maps from chassis name to struct sbctl_chassis. */
    struct shash chassis;
    /* Maps from lport name to struct sbctl_port_binding. */
    struct shash port_bindings;
};

/* Casts 'base' into 'struct sbctl_context'. */
static struct sbctl_context *
sbctl_context_cast(struct ctl_context *base)
{
    return CONTAINER_OF(base, struct sbctl_context, base);
}

struct sbctl_chassis {
    const struct sbrec_chassis *ch_cfg;
};

struct sbctl_port_binding {
    const struct sbrec_port_binding *bd_cfg;
};

static void
sbctl_context_invalidate_cache(struct ctl_context *ctx)
{
    struct sbctl_context *sbctl_ctx = sbctl_context_cast(ctx);

    if (!sbctl_ctx->cache_valid) {
        return;
    }
    sbctl_ctx->cache_valid = false;
    shash_destroy_free_data(&sbctl_ctx->chassis);
    shash_destroy_free_data(&sbctl_ctx->port_bindings);
}

static void
sbctl_context_populate_cache(struct ctl_context *ctx)
{
    struct sbctl_context *sbctl_ctx = sbctl_context_cast(ctx);
    const struct sbrec_chassis *chassis_rec;
    const struct sbrec_port_binding *port_binding_rec;
    struct sset chassis, port_bindings;

    if (sbctl_ctx->cache_valid) {
        /* Cache is already populated. */
        return;
    }
    sbctl_ctx->cache_valid = true;
    shash_init(&sbctl_ctx->chassis);
    shash_init(&sbctl_ctx->port_bindings);
    sset_init(&chassis);
    SBREC_CHASSIS_FOR_EACH(chassis_rec, ctx->idl) {
        struct sbctl_chassis *ch;

        if (!sset_add(&chassis, chassis_rec->name)) {
            VLOG_WARN("database contains duplicate chassis name (%s)",
                      chassis_rec->name);
            continue;
        }

        ch = xmalloc(sizeof *ch);
        ch->ch_cfg = chassis_rec;
        shash_add(&sbctl_ctx->chassis, chassis_rec->name, ch);
    }
    sset_destroy(&chassis);

    sset_init(&port_bindings);
    SBREC_PORT_BINDING_FOR_EACH(port_binding_rec, ctx->idl) {
        struct sbctl_port_binding *bd;

        if (!sset_add(&port_bindings, port_binding_rec->logical_port)) {
            VLOG_WARN("database contains duplicate port binding for logical "
                      "port (%s)",
                      port_binding_rec->logical_port);
            continue;
        }

        bd = xmalloc(sizeof *bd);
        bd->bd_cfg = port_binding_rec;
        shash_add(&sbctl_ctx->port_bindings, port_binding_rec->logical_port,
                  bd);
    }
    sset_destroy(&port_bindings);
}

static void
check_conflicts(struct sbctl_context *sbctl_ctx, const char *name,
                char *msg)
{
    if (shash_find(&sbctl_ctx->chassis, name)) {
        ctl_fatal("%s because a chassis named %s already exists",
                    msg, name);
    }
    free(msg);
}

static struct sbctl_chassis *
find_chassis(struct sbctl_context *sbctl_ctx, const char *name,
             bool must_exist)
{
    struct sbctl_chassis *sbctl_ch;

    ovs_assert(sbctl_ctx->cache_valid);

    sbctl_ch = shash_find_data(&sbctl_ctx->chassis, name);
    if (must_exist && !sbctl_ch) {
        ctl_fatal("no chassis named %s", name);
    }

    return sbctl_ch;
}

static struct sbctl_port_binding *
find_port_binding(struct sbctl_context *sbctl_ctx, const char *name,
                  bool must_exist)
{
    struct sbctl_port_binding *bd;

    ovs_assert(sbctl_ctx->cache_valid);

    bd = shash_find_data(&sbctl_ctx->port_bindings, name);
    if (must_exist && !bd) {
        ctl_fatal("no port named %s", name);
    }

    return bd;
}

static void
pre_get_info(struct ctl_context *ctx)
{
    ovsdb_idl_add_column(ctx->idl, &sbrec_chassis_col_name);
    ovsdb_idl_add_column(ctx->idl, &sbrec_chassis_col_encaps);

    ovsdb_idl_add_column(ctx->idl, &sbrec_encap_col_type);
    ovsdb_idl_add_column(ctx->idl, &sbrec_encap_col_ip);

    ovsdb_idl_add_column(ctx->idl, &sbrec_port_binding_col_logical_port);
    ovsdb_idl_add_column(ctx->idl, &sbrec_port_binding_col_chassis);

    ovsdb_idl_add_column(ctx->idl, &sbrec_logical_flow_col_logical_datapath);
    ovsdb_idl_add_column(ctx->idl, &sbrec_logical_flow_col_pipeline);
    ovsdb_idl_add_column(ctx->idl, &sbrec_logical_flow_col_actions);
    ovsdb_idl_add_column(ctx->idl, &sbrec_logical_flow_col_priority);
    ovsdb_idl_add_column(ctx->idl, &sbrec_logical_flow_col_table_id);
    ovsdb_idl_add_column(ctx->idl, &sbrec_logical_flow_col_match);
    ovsdb_idl_add_column(ctx->idl, &sbrec_logical_flow_col_external_ids);

    ovsdb_idl_add_column(ctx->idl, &sbrec_datapath_binding_col_external_ids);
}

static struct cmd_show_table cmd_show_tables[] = {
    {&sbrec_table_chassis,
     &sbrec_chassis_col_name,
     {&sbrec_chassis_col_hostname,
      &sbrec_chassis_col_encaps,
      NULL},
     {&sbrec_table_port_binding,
      &sbrec_port_binding_col_logical_port,
      &sbrec_port_binding_col_chassis}},

    {&sbrec_table_encap,
     &sbrec_encap_col_type,
     {&sbrec_encap_col_ip,
      &sbrec_encap_col_options,
      NULL},
     {NULL, NULL, NULL}},

    {NULL, NULL, {NULL, NULL, NULL}, {NULL, NULL, NULL}},
};

static void
sbctl_init(struct ctl_context *ctx OVS_UNUSED)
{
}

static void
cmd_chassis_add(struct ctl_context *ctx)
{
    struct sbctl_context *sbctl_ctx = sbctl_context_cast(ctx);
    bool may_exist = shash_find(&ctx->options, "--may-exist") != NULL;
    const char *ch_name, *encap_types, *encap_ip;

    ch_name = ctx->argv[1];
    encap_types = ctx->argv[2];
    encap_ip = ctx->argv[3];

    sbctl_context_populate_cache(ctx);
    if (may_exist) {
        struct sbctl_chassis *sbctl_ch;

        sbctl_ch = find_chassis(sbctl_ctx, ch_name, false);
        if (sbctl_ch) {
            return;
        }
    }
    check_conflicts(sbctl_ctx, ch_name,
                    xasprintf("cannot create a chassis named %s", ch_name));

    struct sset encap_set;
    sset_from_delimited_string(&encap_set, encap_types, ",");

    size_t n_encaps = sset_count(&encap_set);
    struct sbrec_encap **encaps = xmalloc(n_encaps * sizeof *encaps);
    const struct smap options = SMAP_CONST1(&options, "csum", "true");
    const char *encap_type;
    int i = 0;
    SSET_FOR_EACH (encap_type, &encap_set){
        encaps[i] = sbrec_encap_insert(ctx->txn);

        sbrec_encap_set_type(encaps[i], encap_type);
        sbrec_encap_set_ip(encaps[i], encap_ip);
        sbrec_encap_set_options(encaps[i], &options);
        i++;
    }
    sset_destroy(&encap_set);

    struct sbrec_chassis *ch = sbrec_chassis_insert(ctx->txn);
    sbrec_chassis_set_name(ch, ch_name);
    sbrec_chassis_set_encaps(ch, encaps, n_encaps);
    free(encaps);

    sbctl_context_invalidate_cache(ctx);
}

static void
cmd_chassis_del(struct ctl_context *ctx)
{
    struct sbctl_context *sbctl_ctx = sbctl_context_cast(ctx);
    bool must_exist = !shash_find(&ctx->options, "--if-exists");
    struct sbctl_chassis *sbctl_ch;

    sbctl_context_populate_cache(ctx);
    sbctl_ch = find_chassis(sbctl_ctx, ctx->argv[1], must_exist);
    if (sbctl_ch) {
        if (sbctl_ch->ch_cfg) {
            size_t i;

            for (i = 0; i < sbctl_ch->ch_cfg->n_encaps; i++) {
                sbrec_encap_delete(sbctl_ch->ch_cfg->encaps[i]);
            }
            sbrec_chassis_delete(sbctl_ch->ch_cfg);
        }
        shash_find_and_delete(&sbctl_ctx->chassis, ctx->argv[1]);
        free(sbctl_ch);
    }
}

static void
cmd_lsp_bind(struct ctl_context *ctx)
{
    struct sbctl_context *sbctl_ctx = sbctl_context_cast(ctx);
    bool may_exist = shash_find(&ctx->options, "--may-exist") != NULL;
    struct sbctl_chassis *sbctl_ch;
    struct sbctl_port_binding *sbctl_bd;
    char *lport_name, *ch_name;

    /* port_binding must exist, chassis must exist! */
    lport_name = ctx->argv[1];
    ch_name = ctx->argv[2];

    sbctl_context_populate_cache(ctx);
    sbctl_bd = find_port_binding(sbctl_ctx, lport_name, true);
    sbctl_ch = find_chassis(sbctl_ctx, ch_name, true);

    if (sbctl_bd->bd_cfg->chassis) {
        if (may_exist && sbctl_bd->bd_cfg->chassis == sbctl_ch->ch_cfg) {
            return;
        } else {
            ctl_fatal("lport (%s) has already been binded to chassis (%s)",
                      lport_name, sbctl_bd->bd_cfg->chassis->name);
        }
    }
    sbrec_port_binding_set_chassis(sbctl_bd->bd_cfg, sbctl_ch->ch_cfg);
    sbctl_context_invalidate_cache(ctx);
}

static void
cmd_lsp_unbind(struct ctl_context *ctx)
{
    struct sbctl_context *sbctl_ctx = sbctl_context_cast(ctx);
    bool must_exist = !shash_find(&ctx->options, "--if-exists");
    struct sbctl_port_binding *sbctl_bd;
    char *lport_name;

    lport_name = ctx->argv[1];
    sbctl_context_populate_cache(ctx);
    sbctl_bd = find_port_binding(sbctl_ctx, lport_name, must_exist);
    if (sbctl_bd) {
        sbrec_port_binding_set_chassis(sbctl_bd->bd_cfg, NULL);
    }
}

enum {
    PL_INGRESS,
    PL_EGRESS,
};

/* Help ensure we catch any future pipeline values */
static int
pipeline_encode(const char *pl)
{
    if (!strcmp(pl, "ingress")) {
        return PL_INGRESS;
    } else if (!strcmp(pl, "egress")) {
        return PL_EGRESS;
    }

    OVS_NOT_REACHED();
}

static int
lflow_cmp(const void *lf1_, const void *lf2_)
{
    const struct sbrec_logical_flow *const *lf1p = lf1_;
    const struct sbrec_logical_flow *const *lf2p = lf2_;
    const struct sbrec_logical_flow *lf1 = *lf1p;
    const struct sbrec_logical_flow *lf2 = *lf2p;

    int pl1 = pipeline_encode(lf1->pipeline);
    int pl2 = pipeline_encode(lf2->pipeline);

#define CMP(expr) \
    do { \
        int res; \
        res = (expr); \
        if (res) { \
            return res; \
        } \
    } while (0)

    CMP(uuid_compare_3way(&lf1->logical_datapath->header_.uuid,
                          &lf2->logical_datapath->header_.uuid));
    CMP(pl1 - pl2);
    CMP(lf1->table_id > lf2->table_id ? 1 :
            (lf1->table_id < lf2->table_id ? -1 : 0));
    CMP(lf1->priority > lf2->priority ? -1 :
            (lf1->priority < lf2->priority ? 1 : 0));
    CMP(strcmp(lf1->match, lf2->match));

#undef CMP

    return 0;
}

static char *
parse_partial_uuid(char *s)
{
    /* Accept a full or partial UUID. */
    if (uuid_is_partial_string(s)) {
        return s;
    }

    /* Accept a full or partial UUID prefixed by 0x, since "ovs-ofctl
     * dump-flows" prints cookies prefixed by 0x. */
    if (s[0] == '0' && (s[1] == 'x' || s[1] == 'X')
        && uuid_is_partial_string(s + 2)) {
        return s + 2;
    }

    /* Not a (partial) UUID. */
    return NULL;
}

static const char *
strip_leading_zero(const char *s)
{
    return s + strspn(s, "0");
}

static bool
is_partial_uuid_match(const struct uuid *uuid, const char *match)
{
    char uuid_s[UUID_LEN + 1];
    snprintf(uuid_s, sizeof uuid_s, UUID_FMT, UUID_ARGS(uuid));

    /* We strip leading zeros because we want to accept cookie values derived
     * from UUIDs, and cookie values are printed without leading zeros because
     * they're just numbers. */
    const char *s1 = strip_leading_zero(uuid_s);
    const char *s2 = strip_leading_zero(match);

    return !strncmp(s1, s2, strlen(s2));
}

static char *
default_ovs(void)
{
    return xasprintf("unix:%s/br-int.mgmt", ovs_rundir());
}

static struct vconn *
sbctl_open_vconn(struct shash *options)
{
    struct shash_node *ovs = shash_find(options, "--ovs");
    if (!ovs) {
        return NULL;
    }

    char *remote = ovs->data ? xstrdup(ovs->data) : default_ovs();
    struct vconn *vconn;
    int retval = vconn_open_block(remote, 1 << OFP13_VERSION, 0, &vconn);
    if (retval) {
        VLOG_WARN("%s: connection failed (%s)", remote, ovs_strerror(retval));
    }
    free(remote);
    return vconn;
}

static void
sbctl_dump_openflow(struct vconn *vconn, const struct uuid *uuid, bool stats)
{
    struct ofputil_flow_stats_request fsr = {
        .cookie = htonll(uuid->parts[0]),
        .cookie_mask = OVS_BE64_MAX,
        .out_port = OFPP_ANY,
        .out_group = OFPG_ANY,
        .table_id = OFPTT_ALL,
    };

    struct ofputil_flow_stats *fses;
    size_t n_fses;
    int error = vconn_dump_flows(vconn, &fsr, OFPUTIL_P_OF13_OXM,
                                 &fses, &n_fses);
    if (error) {
        VLOG_WARN("%s: error obtaining flow stats (%s)",
                  vconn_get_name(vconn), ovs_strerror(error));
        return;
    }

    if (n_fses) {
        struct ds s = DS_EMPTY_INITIALIZER;
        for (size_t i = 0; i < n_fses; i++) {
            const struct ofputil_flow_stats *fs = &fses[i];

            ds_clear(&s);
            if (stats) {
                ofp_print_flow_stats(&s, fs, NULL);
            } else {
                ds_put_format(&s, " %stable=%s%"PRIu8" ",
                              colors.special, colors.end, fs->table_id);
                match_format(&fs->match, NULL, &s, OFP_DEFAULT_PRIORITY);
                if (ds_last(&s) != ' ') {
                    ds_put_char(&s, ' ');
                }

                ds_put_format(&s, "%sactions=%s", colors.actions, colors.end);
                ofpacts_format(fs->ofpacts, fs->ofpacts_len, NULL, &s);
            }
            printf("   %s\n", ds_cstr(&s));
        }
        ds_destroy(&s);
    }

    for (size_t i = 0; i < n_fses; i++) {
        free(CONST_CAST(struct ofpact *, fses[i].ofpacts));
    }
    free(fses);
}

static void
cmd_lflow_list(struct ctl_context *ctx)
{
    const struct sbrec_datapath_binding *datapath = NULL;
    if (ctx->argc > 1) {
        datapath = (const struct sbrec_datapath_binding *)
            ctl_get_row(ctx, &sbrec_table_datapath_binding,
                        ctx->argv[1], false);
        if (datapath) {
            ctx->argc--;
            ctx->argv++;
        }
    }

    for (size_t i = 1; i < ctx->argc; i++) {
        char *s = parse_partial_uuid(ctx->argv[i]);
        if (!s) {
            ctl_fatal("%s is not a UUID or the beginning of a UUID",
                      ctx->argv[i]);
        }
        ctx->argv[i] = s;
    }

    struct vconn *vconn = sbctl_open_vconn(&ctx->options);
    bool stats = shash_find(&ctx->options, "--stats") != NULL;

    const struct sbrec_logical_flow **lflows = NULL;
    size_t n_flows = 0;
    size_t n_capacity = 0;
    const struct sbrec_logical_flow *lflow;
    SBREC_LOGICAL_FLOW_FOR_EACH (lflow, ctx->idl) {
        if (datapath && lflow->logical_datapath != datapath) {
            continue;
        }

        if (n_flows == n_capacity) {
            lflows = x2nrealloc(lflows, &n_capacity, sizeof *lflows);
        }
        lflows[n_flows] = lflow;
        n_flows++;
    }
    qsort(lflows, n_flows, sizeof *lflows, lflow_cmp);

    bool print_uuid = shash_find(&ctx->options, "--uuid") != NULL;

    const struct sbrec_logical_flow *prev = NULL;
    for (size_t i = 0; i < n_flows; i++) {
        lflow = lflows[i];

        /* Figure out whether to print this particular flow.  By default, we
         * print all flows, but if any UUIDs were listed on the command line
         * then we only print the matching ones. */
        bool include;
        if (ctx->argc > 1) {
            include = false;
            for (size_t j = 1; j < ctx->argc; j++) {
                if (is_partial_uuid_match(&lflow->header_.uuid,
                                          ctx->argv[j])) {
                    include = true;
                    break;
                }
            }
        } else {
            include = true;
        }
        if (!include) {
            continue;
        }

        /* Print a header line for this datapath or pipeline, if we haven't
         * already done so. */
        if (!prev
            || prev->logical_datapath != lflow->logical_datapath
            || strcmp(prev->pipeline, lflow->pipeline)) {
            printf("Datapath:");

            const struct smap *ids = &lflow->logical_datapath->external_ids;
            const char *name = smap_get(ids, "name");
            const char *name2 = smap_get(ids, "name2");
            if (name && name2) {
                printf(" \"%s\" aka \"%s\"", name, name2);
            } else if (name || name2) {
                printf(" \"%s\"", name ? name : name2);
            }
            printf(" ("UUID_FMT")  Pipeline: %s\n",
                   UUID_ARGS(&lflow->logical_datapath->header_.uuid),
                   lflow->pipeline);
        }

        /* Print the flow. */
        printf("  ");
        if (print_uuid) {
            printf("uuid=0x%08"PRIx32", ", lflow->header_.uuid.parts[0]);
        }
        printf("table=%-2"PRId64"(%-19s), priority=%-5"PRId64
               ", match=(%s), action=(%s)\n",
               lflow->table_id,
               smap_get_def(&lflow->external_ids, "stage-name", ""),
               lflow->priority, lflow->match, lflow->actions);
        if (vconn) {
            sbctl_dump_openflow(vconn, &lflow->header_.uuid, stats);
        }
        prev = lflow;
    }

    vconn_close(vconn);
    free(lflows);
}

static void
verify_connections(struct ctl_context *ctx)
{
    const struct sbrec_sb_global *sb_global = sbrec_sb_global_first(ctx->idl);
    const struct sbrec_connection *conn;

    sbrec_sb_global_verify_connections(sb_global);

    SBREC_CONNECTION_FOR_EACH(conn, ctx->idl) {
        sbrec_connection_verify_target(conn);
    }
}

static void
pre_connection(struct ctl_context *ctx)
{
    ovsdb_idl_add_column(ctx->idl, &sbrec_sb_global_col_connections);
    ovsdb_idl_add_column(ctx->idl, &sbrec_connection_col_target);
    ovsdb_idl_add_column(ctx->idl, &sbrec_connection_col_read_only);
}

static void
cmd_get_connection(struct ctl_context *ctx)
{
    const struct sbrec_connection *conn;
    struct svec targets;
    size_t i;

    verify_connections(ctx);

    /* Print the targets in sorted order for reproducibility. */
    svec_init(&targets);

    SBREC_CONNECTION_FOR_EACH(conn, ctx->idl) {
        char *s;

        s = xasprintf("%s %s", conn->read_only ? "read-only" : "read-write",
                               conn->target);
        svec_add(&targets, s);
        free(s);
    }

    svec_sort_unique(&targets);
    for (i = 0; i < targets.n; i++) {
        ds_put_format(&ctx->output, "%s\n", targets.names[i]);
    }
    svec_destroy(&targets);
}

static void
delete_connections(struct ctl_context *ctx)
{
    const struct sbrec_sb_global *sb_global = sbrec_sb_global_first(ctx->idl);
    const struct sbrec_connection *conn, *next;

    /* Delete Manager rows pointed to by 'connection_options' column. */
    SBREC_CONNECTION_FOR_EACH_SAFE(conn, next, ctx->idl) {
        sbrec_connection_delete(conn);
    }

    /* Delete 'Manager' row refs in 'manager_options' column. */
    sbrec_sb_global_set_connections(sb_global, NULL, 0);
}

static void
cmd_del_connection(struct ctl_context *ctx)
{
    verify_connections(ctx);
    delete_connections(ctx);
}

static void
insert_connections(struct ctl_context *ctx, char *targets[], size_t n)
{
    const struct sbrec_sb_global *sb_global = sbrec_sb_global_first(ctx->idl);
    struct sbrec_connection **connections;
    size_t i, conns=0;
    bool read_only = false;

    /* Insert each connection in a new row in Connection table. */
    connections = xmalloc(n * sizeof *connections);
    for (i = 0; i < n; i++) {
        if (!strcmp(targets[i], "read-only")) {
            read_only = true;
            continue;
        } else if (!strcmp(targets[i], "read-write")) {
            read_only = false;
            continue;
        } else if (stream_verify_name(targets[i]) &&
                   pstream_verify_name(targets[i])) {
            VLOG_WARN("target type \"%s\" is possibly erroneous", targets[i]);
        }

        connections[conns] = sbrec_connection_insert(ctx->txn);
        sbrec_connection_set_target(connections[conns], targets[i]);
        sbrec_connection_set_read_only(connections[conns], read_only);
        conns++;
    }

    /* Store uuids of new connection rows in 'connection' column. */
    sbrec_sb_global_set_connections(sb_global, connections, conns);
    free(connections);
}

static void
cmd_set_connection(struct ctl_context *ctx)
{
    const size_t n = ctx->argc - 1;

    verify_connections(ctx);
    delete_connections(ctx);
    insert_connections(ctx, &ctx->argv[1], n);
}

static void
pre_cmd_get_ssl(struct ctl_context *ctx)
{
    ovsdb_idl_add_column(ctx->idl, &sbrec_sb_global_col_ssl);

    ovsdb_idl_add_column(ctx->idl, &sbrec_ssl_col_private_key);
    ovsdb_idl_add_column(ctx->idl, &sbrec_ssl_col_certificate);
    ovsdb_idl_add_column(ctx->idl, &sbrec_ssl_col_ca_cert);
    ovsdb_idl_add_column(ctx->idl, &sbrec_ssl_col_bootstrap_ca_cert);
}

static void
cmd_get_ssl(struct ctl_context *ctx)
{
    const struct sbrec_sb_global *sb_global = sbrec_sb_global_first(ctx->idl);
    const struct sbrec_ssl *ssl = sbrec_ssl_first(ctx->idl);

    sbrec_sb_global_verify_ssl(sb_global);
    if (ssl) {
        sbrec_ssl_verify_private_key(ssl);
        sbrec_ssl_verify_certificate(ssl);
        sbrec_ssl_verify_ca_cert(ssl);
        sbrec_ssl_verify_bootstrap_ca_cert(ssl);

        ds_put_format(&ctx->output, "Private key: %s\n", ssl->private_key);
        ds_put_format(&ctx->output, "Certificate: %s\n", ssl->certificate);
        ds_put_format(&ctx->output, "CA Certificate: %s\n", ssl->ca_cert);
        ds_put_format(&ctx->output, "Bootstrap: %s\n",
                ssl->bootstrap_ca_cert ? "true" : "false");
    }
}

static void
pre_cmd_del_ssl(struct ctl_context *ctx)
{
    ovsdb_idl_add_column(ctx->idl, &sbrec_sb_global_col_ssl);
}

static void
cmd_del_ssl(struct ctl_context *ctx)
{
    const struct sbrec_sb_global *sb_global = sbrec_sb_global_first(ctx->idl);
    const struct sbrec_ssl *ssl = sbrec_ssl_first(ctx->idl);

    if (ssl) {
        sbrec_sb_global_verify_ssl(sb_global);
        sbrec_ssl_delete(ssl);
        sbrec_sb_global_set_ssl(sb_global, NULL);
    }
}

static void
pre_cmd_set_ssl(struct ctl_context *ctx)
{
    ovsdb_idl_add_column(ctx->idl, &sbrec_sb_global_col_ssl);
}

static void
cmd_set_ssl(struct ctl_context *ctx)
{
    bool bootstrap = shash_find(&ctx->options, "--bootstrap");
    const struct sbrec_sb_global *sb_global = sbrec_sb_global_first(ctx->idl);
    const struct sbrec_ssl *ssl = sbrec_ssl_first(ctx->idl);

    sbrec_sb_global_verify_ssl(sb_global);
    if (ssl) {
        sbrec_ssl_delete(ssl);
    }
    ssl = sbrec_ssl_insert(ctx->txn);

    sbrec_ssl_set_private_key(ssl, ctx->argv[1]);
    sbrec_ssl_set_certificate(ssl, ctx->argv[2]);
    sbrec_ssl_set_ca_cert(ssl, ctx->argv[3]);

    sbrec_ssl_set_bootstrap_ca_cert(ssl, bootstrap);

    sbrec_sb_global_set_ssl(sb_global, ssl);
}

\f
static const struct ctl_table_class tables[SBREC_N_TABLES] = {
    [SBREC_TABLE_CHASSIS].row_ids[0] = {&sbrec_chassis_col_name, NULL, NULL},

    [SBREC_TABLE_DATAPATH_BINDING].row_ids
     = {{&sbrec_datapath_binding_col_external_ids, "name", NULL},
        {&sbrec_datapath_binding_col_external_ids, "name2", NULL},
        {&sbrec_datapath_binding_col_external_ids, "logical-switch", NULL},
        {&sbrec_datapath_binding_col_external_ids, "logical-router", NULL}},

    [SBREC_TABLE_PORT_BINDING].row_ids
     = {{&sbrec_port_binding_col_logical_port, NULL, NULL},
        {&sbrec_port_binding_col_external_ids, "name", NULL}},

    [SBREC_TABLE_MAC_BINDING].row_ids[0] =
    {&sbrec_mac_binding_col_logical_port, NULL, NULL},

    [SBREC_TABLE_ADDRESS_SET].row_ids[0]
    = {&sbrec_address_set_col_name, NULL, NULL},
};

\f
static void
sbctl_context_init_command(struct sbctl_context *sbctl_ctx,
                           struct ctl_command *command)
{
    ctl_context_init_command(&sbctl_ctx->base, command);
}

static void
sbctl_context_init(struct sbctl_context *sbctl_ctx,
                   struct ctl_command *command, struct ovsdb_idl *idl,
                   struct ovsdb_idl_txn *txn,
                   struct ovsdb_symbol_table *symtab)
{
    ctl_context_init(&sbctl_ctx->base, command, idl, txn, symtab,
                     sbctl_context_invalidate_cache);
    sbctl_ctx->cache_valid = false;
}

static void
sbctl_context_done_command(struct sbctl_context *sbctl_ctx,
                           struct ctl_command *command)
{
    ctl_context_done_command(&sbctl_ctx->base, command);
}

static void
sbctl_context_done(struct sbctl_context *sbctl_ctx,
                   struct ctl_command *command)
{
    ctl_context_done(&sbctl_ctx->base, command);
}

static void
run_prerequisites(struct ctl_command *commands, size_t n_commands,
                  struct ovsdb_idl *idl)
{
    ovsdb_idl_add_table(idl, &sbrec_table_sb_global);

    for (struct ctl_command *c = commands; c < &commands[n_commands]; c++) {
        if (c->syntax->prerequisites) {
            struct sbctl_context sbctl_ctx;

            ds_init(&c->output);
            c->table = NULL;

            sbctl_context_init(&sbctl_ctx, c, idl, NULL, NULL);
            (c->syntax->prerequisites)(&sbctl_ctx.base);
            sbctl_context_done(&sbctl_ctx, c);

            ovs_assert(!c->output.string);
            ovs_assert(!c->table);
        }
    }
}

static bool
do_sbctl(const char *args, struct ctl_command *commands, size_t n_commands,
         struct ovsdb_idl *idl)
{
    struct ovsdb_idl_txn *txn;
    enum ovsdb_idl_txn_status status;
    struct ovsdb_symbol_table *symtab;
    struct sbctl_context sbctl_ctx;
    struct ctl_command *c;
    struct shash_node *node;
    char *error = NULL;

    txn = the_idl_txn = ovsdb_idl_txn_create(idl);
    if (dry_run) {
        ovsdb_idl_txn_set_dry_run(txn);
    }

    ovsdb_idl_txn_add_comment(txn, "ovs-sbctl: %s", args);

    const struct sbrec_sb_global *sb = sbrec_sb_global_first(idl);
    if (!sb) {
        /* XXX add verification that table is empty */
        sb = sbrec_sb_global_insert(txn);
    }

    symtab = ovsdb_symbol_table_create();
    for (c = commands; c < &commands[n_commands]; c++) {
        ds_init(&c->output);
        c->table = NULL;
    }
    sbctl_context_init(&sbctl_ctx, NULL, idl, txn, symtab);
    for (c = commands; c < &commands[n_commands]; c++) {
        sbctl_context_init_command(&sbctl_ctx, c);
        if (c->syntax->run) {
            (c->syntax->run)(&sbctl_ctx.base);
        }
        sbctl_context_done_command(&sbctl_ctx, c);

        if (sbctl_ctx.base.try_again) {
            sbctl_context_done(&sbctl_ctx, NULL);
            goto try_again;
        }
    }
    sbctl_context_done(&sbctl_ctx, NULL);

    SHASH_FOR_EACH (node, &symtab->sh) {
        struct ovsdb_symbol *symbol = node->data;
        if (!symbol->created) {
            ctl_fatal("row id \"%s\" is referenced but never created (e.g. "
                      "with \"-- --id=%s create ...\")",
                      node->name, node->name);
        }
        if (!symbol->strong_ref) {
            if (!symbol->weak_ref) {
                VLOG_WARN("row id \"%s\" was created but no reference to it "
                          "was inserted, so it will not actually appear in "
                          "the database", node->name);
            } else {
                VLOG_WARN("row id \"%s\" was created but only a weak "
                          "reference to it was inserted, so it will not "
                          "actually appear in the database", node->name);
            }
        }
    }

    status = ovsdb_idl_txn_commit_block(txn);
    if (status == TXN_UNCHANGED || status == TXN_SUCCESS) {
        for (c = commands; c < &commands[n_commands]; c++) {
            if (c->syntax->postprocess) {
                sbctl_context_init(&sbctl_ctx, c, idl, txn, symtab);
                (c->syntax->postprocess)(&sbctl_ctx.base);
                sbctl_context_done(&sbctl_ctx, c);
            }
        }
    }
    error = xstrdup(ovsdb_idl_txn_get_error(txn));

    switch (status) {
    case TXN_UNCOMMITTED:
    case TXN_INCOMPLETE:
        OVS_NOT_REACHED();

    case TXN_ABORTED:
        /* Should not happen--we never call ovsdb_idl_txn_abort(). */
        ctl_fatal("transaction aborted");

    case TXN_UNCHANGED:
    case TXN_SUCCESS:
        break;

    case TXN_TRY_AGAIN:
        goto try_again;

    case TXN_ERROR:
        ctl_fatal("transaction error: %s", error);

    case TXN_NOT_LOCKED:
        /* Should not happen--we never call ovsdb_idl_set_lock(). */
        ctl_fatal("database not locked");

    default:
        OVS_NOT_REACHED();
    }
    free(error);

    ovsdb_symbol_table_destroy(symtab);

    for (c = commands; c < &commands[n_commands]; c++) {
        struct ds *ds = &c->output;

        if (c->table) {
            table_print(c->table, &table_style);
        } else if (oneline) {
            size_t j;

            ds_chomp(ds, '\n');
            for (j = 0; j < ds->length; j++) {
                int ch = ds->string[j];
                switch (ch) {
                case '\n':
                    fputs("\\n", stdout);
                    break;

                case '\\':
                    fputs("\\\\", stdout);
                    break;

                default:
                    putchar(ch);
                }
            }
            putchar('\n');
        } else {
            fputs(ds_cstr(ds), stdout);
        }
        ds_destroy(&c->output);
        table_destroy(c->table);
        free(c->table);

        shash_destroy_free_data(&c->options);
    }
    free(commands);
    ovsdb_idl_txn_destroy(txn);
    ovsdb_idl_destroy(idl);

    return true;

try_again:
    /* Our transaction needs to be rerun, or a prerequisite was not met.  Free
     * resources and return so that the caller can try again. */
    ovsdb_idl_txn_abort(txn);
    ovsdb_idl_txn_destroy(txn);
    the_idl_txn = NULL;

    ovsdb_symbol_table_destroy(symtab);
    for (c = commands; c < &commands[n_commands]; c++) {
        ds_destroy(&c->output);
        table_destroy(c->table);
        free(c->table);
    }
    free(error);
    return false;
}

/* Frees the current transaction and the underlying IDL and then calls
 * exit(status).
 *
 * Freeing the transaction and the IDL is not strictly necessary, but it makes
 * for a clean memory leak report from valgrind in the normal case.  That makes
 * it easier to notice real memory leaks. */
static void
sbctl_exit(int status)
{
    if (the_idl_txn) {
        ovsdb_idl_txn_abort(the_idl_txn);
        ovsdb_idl_txn_destroy(the_idl_txn);
    }
    ovsdb_idl_destroy(the_idl);
    exit(status);
}

static const struct ctl_command_syntax sbctl_commands[] = {
    { "init", 0, 0, "", NULL, sbctl_init, NULL, "", RW },

    /* Chassis commands. */
    {"chassis-add", 3, 3, "CHASSIS ENCAP-TYPE ENCAP-IP", pre_get_info,
     cmd_chassis_add, NULL, "--may-exist", RW},
    {"chassis-del", 1, 1, "CHASSIS", pre_get_info, cmd_chassis_del, NULL,
     "--if-exists", RW},

    /* Port binding commands. */
    {"lsp-bind", 2, 2, "PORT CHASSIS", pre_get_info, cmd_lsp_bind, NULL,
     "--may-exist", RW},
    {"lsp-unbind", 1, 1, "PORT", pre_get_info, cmd_lsp_unbind, NULL,
     "--if-exists", RW},

    /* Logical flow commands */
    {"lflow-list", 0, INT_MAX, "[DATAPATH] [LFLOW...]",
     pre_get_info, cmd_lflow_list, NULL,
     "--uuid,--ovs?,--stats", RO},
    {"dump-flows", 0, INT_MAX, "[DATAPATH] [LFLOW...]",
     pre_get_info, cmd_lflow_list, NULL,
     "--uuid,--ovs?,--stats", RO}, /* Friendly alias for lflow-list */

    /* Connection commands. */
    {"get-connection", 0, 0, "", pre_connection, cmd_get_connection, NULL, "", RO},
    {"del-connection", 0, 0, "", pre_connection, cmd_del_connection, NULL, "", RW},
    {"set-connection", 1, INT_MAX, "TARGET...", pre_connection, cmd_set_connection,
     NULL, "", RW},

    /* SSL commands. */
    {"get-ssl", 0, 0, "", pre_cmd_get_ssl, cmd_get_ssl, NULL, "", RO},
    {"del-ssl", 0, 0, "", pre_cmd_del_ssl, cmd_del_ssl, NULL, "", RW},
    {"set-ssl", 3, 3, "PRIVATE-KEY CERTIFICATE CA-CERT", pre_cmd_set_ssl,
     cmd_set_ssl, NULL, "--bootstrap", RW},

    {NULL, 0, 0, NULL, NULL, NULL, NULL, NULL, RO},
};

/* Registers sbctl and common db commands. */
static void
sbctl_cmd_init(void)
{
    ctl_init(sbrec_table_classes, tables, cmd_show_tables, sbctl_exit);
    ctl_register_commands(sbctl_commands);
}