ovsdb-client: Make "monitor" command able to monitor all tables.

Users may find this feature useful, but the reason to add this feature is
to allow a test to use it in an upcoming commit.

Signed-off-by: Ben Pfaff <blp@nicira.com>

NEWS

@@ -51,6 +51,9 @@ Post-v2.0.0
      incorrectly that ovs-controller was a necessary or desirable part
      of an Open vSwitch deployment.
    - Added vlog option to export to a UDP syslog sink.
+   - ovsdb-client:
+     * The "monitor" command can now monitor all tables in a database,
+       instead of being limited to a single table.
 
 
 v2.0.0 - 15 Oct 2013

ovsdb/ovsdb-client.1.in

@@ -30,6 +30,8 @@ ovsdb\-client \- command-line interface to \fBovsdb-server\fR(1)
 \fBovsdb\-client \fR[\fIoptions\fR] \fBmonitor\fI \fR[\fIserver\fR] \fR[\fIdatabase\fR] \fItable\fR
 [\fIcolumn\fR[\fB,\fIcolumn\fR]...]...
 .br
+\fBovsdb\-client \fR[\fIoptions\fR] \fBmonitor\fI \fR[\fIserver\fR] \fR[\fIdatabase\fR] \fBALL\fR
+.br
 \fBovsdb\-client help\fR
 .IP "Output formatting options:"
 [\fB\-\-format=\fIformat\fR]
@@ -128,6 +130,14 @@ line.
 If \fB\-\-detach\fR is used with \fBmonitor\fR, then \fBovsdb\-client\fR
 detaches after it has successfully received and printed the initial
 contents of \fItable\fR.
+.
+.IP "\fBmonitor\fI \fR[\fIserver\fR] \fR[\fIdatabase\fR] \fBALL\fR"
+Connects to \fIserver\fR and monitors the contents of all tables in
+\fIdatabase\fR.  Prints initial values and all kinds of changes to all
+columns in the database.  The \fB\-\-detach\fR option causes
+\fBovsdb\-client\fR to detach after it successfully receives and
+prints the initial database contents.
+.
 .SH OPTIONS
 .SS "Output Formatting Options"
 Much of the output from \fBovsdb\-client\fR is in the form of tables.

ovsdb/ovsdb-client.c

@@ -37,12 +37,14 @@
 #include "ovsdb.h"
 #include "ovsdb-data.h"
 #include "ovsdb-error.h"
+#include "poll-loop.h"
 #include "sort.h"
 #include "svec.h"
 #include "stream.h"
 #include "stream-ssl.h"
 #include "table.h"
 #include "timeval.h"
+#include "unixctl.h"
 #include "util.h"
 #include "vlog.h"
 
@@ -253,6 +255,9 @@ usage(void)
            "    monitor contents of COLUMNs in TABLE in DATABASE on SERVER.\n"
            "    COLUMNs may include !initial, !insert, !delete, !modify\n"
            "    to avoid seeing the specified kinds of changes.\n"
+           "\n  monitor [SERVER] [DATABASE] ALL\n"
+           "    monitor all changes to all columns in all tables\n"
+           "    in DATBASE on SERVER.\n"
            "\n  dump [SERVER] [DATABASE]\n"
            "    dump contents of DATABASE on SERVER to stdout\n"
            "\nThe default SERVER is unix:%s/db.sock.\n"
@@ -503,6 +508,13 @@ do_transact(struct jsonrpc *rpc, const char *database OVS_UNUSED,
     putchar('\n');
     jsonrpc_msg_destroy(reply);
 }
+
+/* "monitor" command. */
+
+struct monitored_table {
+    struct ovsdb_table_schema *table;
+    struct ovsdb_column_set columns;
+};
 
 static void
 monitor_print_row(struct json *row, const char *type, const char *uuid,
@@ -533,30 +545,24 @@ monitor_print_row(struct json *row, const char *type, const char *uuid,
 }
 
 static void
-monitor_print(struct json *table_updates,
+monitor_print_table(struct json *table_update,
-              const struct ovsdb_table_schema *table,
+                    const struct monitored_table *mt, char *caption,
-              const struct ovsdb_column_set *columns, bool initial)
+                    bool initial)
 {
-    struct json *table_update;
+    const struct ovsdb_table_schema *table = mt->table;
+    const struct ovsdb_column_set *columns = &mt->columns;
     struct shash_node *node;
     struct table t;
     size_t i;
 
+    if (table_update->type != JSON_OBJECT) {
+        ovs_error(0, "<table-update> for table %s is not object", table->name);
+        return;
+    }
+
     table_init(&t);
     table_set_timestamp(&t, timestamp);
-
+    table_set_caption(&t, caption);
-    if (table_updates->type != JSON_OBJECT) {
-        ovs_error(0, "<table-updates> is not object");
-        return;
-    }
-    table_update = shash_find_data(json_object(table_updates), table->name);
-    if (!table_update) {
-        return;
-    }
-    if (table_update->type != JSON_OBJECT) {
-        ovs_error(0, "<table-update> is not object");
-        return;
-    }
 
     table_add_column(&t, "row");
     table_add_column(&t, "action");
@@ -588,6 +594,30 @@ monitor_print(struct json *table_updates,
     table_destroy(&t);
 }
 
+static void
+monitor_print(struct json *table_updates,
+              const struct monitored_table *mts, size_t n_mts,
+              bool initial)
+{
+    size_t i;
+
+    if (table_updates->type != JSON_OBJECT) {
+        ovs_error(0, "<table-updates> is not object");
+        return;
+    }
+
+    for (i = 0; i < n_mts; i++) {
+        const struct monitored_table *mt = &mts[i];
+        struct json *table_update = shash_find_data(json_object(table_updates),
+                                                    mt->table->name);
+        if (table_update) {
+            monitor_print_table(table_update, mt,
+                                n_mts > 1 ? xstrdup(mt->table->name) : NULL,
+                                initial);
+        }
+    }
+}
+
 static void
 add_column(const char *server, const struct ovsdb_column *column,
            struct ovsdb_column_set *columns, struct json *columns_json)
@@ -653,7 +683,7 @@ parse_monitor_columns(char *arg, const char *server, const char *database,
         }
         free(nodes);
 
-        add_column(server, ovsdb_table_schema_get_column(table,"_version"),
+        add_column(server, ovsdb_table_schema_get_column(table, "_version"),
                    columns, columns_json);
     }
 
@@ -670,24 +700,59 @@ parse_monitor_columns(char *arg, const char *server, const char *database,
 }
 
 static void
-do_monitor(struct jsonrpc *rpc, const char *database,
+ovsdb_client_exit(struct unixctl_conn *conn, int argc OVS_UNUSED,
-           int argc, char *argv[])
+                  const char *argv[] OVS_UNUSED, void *exiting_)
 {
-    const char *server = jsonrpc_get_name(rpc);
+    bool *exiting = exiting_;
-    const char *table_name = argv[0];
+    *exiting = true;
-    struct ovsdb_column_set columns = OVSDB_COLUMN_SET_INITIALIZER;
+    unixctl_command_reply(conn, NULL);
-    struct ovsdb_table_schema *table;
+}
-    struct ovsdb_schema *schema;
-    struct jsonrpc_msg *request;
-    struct json *monitor, *monitor_request_array,
-        *monitor_requests, *request_id;
 
-    schema = fetch_schema(rpc, database);
+static void
-    table = shash_find_data(&schema->tables, table_name);
+ovsdb_client_block(struct unixctl_conn *conn, int argc OVS_UNUSED,
-    if (!table) {
+                   const char *argv[] OVS_UNUSED, void *blocked_)
-        ovs_fatal(0, "%s: %s does not have a table named \"%s\"",
+{
-                  server, database, table_name);
+    bool *blocked = blocked_;
+
+    if (!*blocked) {
+        *blocked = true;
+        unixctl_command_reply(conn, NULL);
+    } else {
+        unixctl_command_reply(conn, "already blocking");
     }
+}
+
+static void
+ovsdb_client_unblock(struct unixctl_conn *conn, int argc OVS_UNUSED,
+                     const char *argv[] OVS_UNUSED, void *blocked_)
+{
+    bool *blocked = blocked_;
+
+    if (*blocked) {
+        *blocked = false;
+        unixctl_command_reply(conn, NULL);
+    } else {
+        unixctl_command_reply(conn, "already unblocked");
+    }
+}
+
+static void
+add_monitored_table(int argc, char *argv[],
+                    const char *server, const char *database,
+                    struct ovsdb_table_schema *table,
+                    struct json *monitor_requests,
+                    struct monitored_table **mts,
+                    size_t *n_mts, size_t *allocated_mts)
+{
+    struct json *monitor_request_array;
+    struct monitored_table *mt;
+
+    if (*n_mts >= *allocated_mts) {
+        *mts = x2nrealloc(*mts, allocated_mts, sizeof **mts);
+    }
+    mt = &(*mts)[(*n_mts)++];
+    mt->table = table;
+    ovsdb_column_set_init(&mt->columns);
 
     monitor_request_array = json_array_create_empty();
     if (argc > 1) {
@@ -697,58 +762,140 @@ do_monitor(struct jsonrpc *rpc, const char *database,
             json_array_add(
                 monitor_request_array,
                 parse_monitor_columns(argv[i], server, database, table,
-                                      &columns));
+                                      &mt->columns));
         }
     } else {
-        /* Allocate a writable empty string since parse_monitor_columns() is
+        /* Allocate a writable empty string since parse_monitor_columns()
-         * going to strtok() it and that's risky with literal "". */
+         * is going to strtok() it and that's risky with literal "". */
         char empty[] = "";
         json_array_add(
             monitor_request_array,
-            parse_monitor_columns(empty, server, database, table, &columns));
+            parse_monitor_columns(empty, server, database,
+                                  table, &mt->columns));
     }
 
+    json_object_put(monitor_requests, table->name, monitor_request_array);
+}
+
+static void
+do_monitor(struct jsonrpc *rpc, const char *database,
+           int argc, char *argv[])
+{
+    const char *server = jsonrpc_get_name(rpc);
+    const char *table_name = argv[0];
+    struct unixctl_server *unixctl;
+    struct ovsdb_schema *schema;
+    struct jsonrpc_msg *request;
+    struct json *monitor, *monitor_requests, *request_id;
+    bool exiting = false;
+    bool blocked = false;
+
+    struct monitored_table *mts;
+    size_t n_mts, allocated_mts;
+
+    daemon_save_fd(STDOUT_FILENO);
+    daemonize_start();
+    if (get_detach()) {
+        int error;
+
+        error = unixctl_server_create(NULL, &unixctl);
+        if (error) {
+            ovs_fatal(error, "failed to create unixctl server");
+        }
+
+        unixctl_command_register("exit", "", 0, 0,
+                                 ovsdb_client_exit, &exiting);
+        unixctl_command_register("ovsdb-client/block", "", 0, 0,
+                                 ovsdb_client_block, &blocked);
+        unixctl_command_register("ovsdb-client/unblock", "", 0, 0,
+                                 ovsdb_client_unblock, &blocked);
+    } else {
+        unixctl = NULL;
+    }
+
+    schema = fetch_schema(rpc, database);
+
     monitor_requests = json_object_create();
-    json_object_put(monitor_requests, table_name, monitor_request_array);
+
+    mts = NULL;
+    n_mts = allocated_mts = 0;
+    if (strcmp(table_name, "ALL")) {
+        struct ovsdb_table_schema *table;
+
+        table = shash_find_data(&schema->tables, table_name);
+        if (!table) {
+            ovs_fatal(0, "%s: %s does not have a table named \"%s\"",
+                      server, database, table_name);
+        }
+
+        add_monitored_table(argc, argv, server, database, table,
+                            monitor_requests, &mts, &n_mts, &allocated_mts);
+    } else {
+        size_t n = shash_count(&schema->tables);
+        const struct shash_node **nodes = shash_sort(&schema->tables);
+        size_t i;
+
+        for (i = 0; i < n; i++) {
+            struct ovsdb_table_schema *table = nodes[i]->data;
+
+            add_monitored_table(argc, argv, server, database, table,
+                                monitor_requests,
+                                &mts, &n_mts, &allocated_mts);
+        }
+        free(nodes);
+    }
 
     monitor = json_array_create_3(json_string_create(database),
                                   json_null_create(), monitor_requests);
     request = jsonrpc_create_request("monitor", monitor, NULL);
     request_id = json_clone(request->id);
     jsonrpc_send(rpc, request);
+
     for (;;) {
-        struct jsonrpc_msg *msg;
+        unixctl_server_run(unixctl);
-        int error;
+        while (!blocked) {
+            struct jsonrpc_msg *msg;
+            int error;
 
-        error = jsonrpc_recv_block(rpc, &msg);
+            error = jsonrpc_recv(rpc, &msg);
-        if (error) {
+            if (error == EAGAIN) {
-            ovsdb_schema_destroy(schema);
+                break;
-            ovs_fatal(error, "%s: receive failed", server);
+            } else if (error) {
-        }
+                ovs_fatal(error, "%s: receive failed", server);
-
-        if (msg->type == JSONRPC_REQUEST && !strcmp(msg->method, "echo")) {
-            jsonrpc_send(rpc, jsonrpc_create_reply(json_clone(msg->params),
-                                                   msg->id));
-        } else if (msg->type == JSONRPC_REPLY
-                   && json_equal(msg->id, request_id)) {
-            monitor_print(msg->result, table, &columns, true);
-            fflush(stdout);
-            if (get_detach()) {
-                daemon_save_fd(STDOUT_FILENO);
-                daemonize();
             }
-        } else if (msg->type == JSONRPC_NOTIFY
+
-                   && !strcmp(msg->method, "update")) {
+            if (msg->type == JSONRPC_REQUEST && !strcmp(msg->method, "echo")) {
-            struct json *params = msg->params;
+                jsonrpc_send(rpc, jsonrpc_create_reply(json_clone(msg->params),
-            if (params->type == JSON_ARRAY
+                                                       msg->id));
-                && params->u.array.n == 2
+            } else if (msg->type == JSONRPC_REPLY
-                && params->u.array.elems[0]->type == JSON_NULL) {
+                       && json_equal(msg->id, request_id)) {
-                monitor_print(params->u.array.elems[1],
+                monitor_print(msg->result, mts, n_mts, true);
-                              table, &columns, false);
                 fflush(stdout);
+                daemonize_complete();
+            } else if (msg->type == JSONRPC_NOTIFY
+                       && !strcmp(msg->method, "update")) {
+                struct json *params = msg->params;
+                if (params->type == JSON_ARRAY
+                    && params->u.array.n == 2
+                    && params->u.array.elems[0]->type == JSON_NULL) {
+                    monitor_print(params->u.array.elems[1], mts, n_mts, false);
+                    fflush(stdout);
+                }
             }
+            jsonrpc_msg_destroy(msg);
         }
-        jsonrpc_msg_destroy(msg);
+
+        if (exiting) {
+            break;
+        }
+
+        jsonrpc_run(rpc);
+        jsonrpc_wait(rpc);
+        if (!blocked) {
+            jsonrpc_recv_wait(rpc);
+        }
+        unixctl_server_wait(unixctl);
+        poll_block();
     }
 }