ovsdb-monitor: Support monitor_cond_since.

Support the new monitor method monitor_cond_since so that a client
can request monitoring start from a specific point instead of always
from beginning. This will reduce the cost at scenarios when server
is restarted/failed-over but client still has all existing data. In
these scenarios only new changes (and in most cases no change) needed
to be transfered to client. When ovsdb-server restarted, history
transactions are read from disk file; when ovsdb-server failed over,
history transactions exists already in the memory of the new server.

There are situations that the requested transaction may not be found.
For example, a transaction that is too old and has been discarded
from the maintained history list in memory, or the transactions on
disk has been compacted during ovsdb compaction. In those situations
the server fall backs to transfer all data start from begining.

For more details of the protocol change, see
Documentation/ref/ovsdb-server.7.rst.

This change includes both server side and ovsdb-client side changes
with the new protocol. IDLs using this capability will be added in
future patches.

Now the feature takes effect only for cluster mode of ovsdb-server,
because cluster mode is the only mode that supports unique transcation
uuid today. For other modes, the monitor_cond_since always fall back
to transfer all data with found = false. Support for those modes can
be added in the future.

Signed-off-by: Han Zhou <hzhou8@ebay.com>
Signed-off-by: Ben Pfaff <blp@ovn.org>

7 files changed, 315 insertions, 36 deletions

diff --git a/ovsdb/jsonrpc-server.c b/ovsdb/jsonrpc-server.c
index f9b7c27cc..4dda63a9d 100644
--- a/ovsdb/jsonrpc-server.c
+++ b/ovsdb/jsonrpc-server.c
@@ -986,13 +986,19 @@ ovsdb_jsonrpc_session_got_request(struct ovsdb_jsonrpc_session *s,
             ovsdb_jsonrpc_trigger_create(s, db, request);
         }
     } else if (!strcmp(request->method, "monitor") ||
-               (monitor_cond_enable__ && !strcmp(request->method,
-                                                 "monitor_cond"))) {
+               (monitor_cond_enable__ &&
+                (!strcmp(request->method, "monitor_cond") ||
+                 !strcmp(request->method, "monitor_cond_since")))) {
         struct ovsdb *db = ovsdb_jsonrpc_lookup_db(s, request, &reply);
         if (!reply) {
-            int l = strlen(request->method) - strlen("monitor");
-            enum ovsdb_monitor_version version = l ? OVSDB_MONITOR_V2
-                                                   : OVSDB_MONITOR_V1;
+            enum ovsdb_monitor_version version;
+            if (!strcmp(request->method, "monitor")) {
+                version = OVSDB_MONITOR_V1;
+            } else if (!strcmp(request->method, "monitor_cond")) {
+                version = OVSDB_MONITOR_V2;
+            } else {
+                version = OVSDB_MONITOR_V3;
+            }
             reply = ovsdb_jsonrpc_monitor_create(s, db, request->params,
                                                  version, request->id);
         }
@@ -1364,7 +1370,8 @@ ovsdb_jsonrpc_monitor_create(struct ovsdb_jsonrpc_session *s, struct ovsdb *db,
     struct shash_node *node;
     struct json *json;
 
-    if (json_array(params)->n != 3) {
+    if ((version == OVSDB_MONITOR_V2 && json_array(params)->n != 3) ||
+        (version == OVSDB_MONITOR_V3 && json_array(params)->n != 4)) {
         error = ovsdb_syntax_error(params, NULL, "invalid parameters");
         goto error;
     }
@@ -1385,7 +1392,7 @@ ovsdb_jsonrpc_monitor_create(struct ovsdb_jsonrpc_session *s, struct ovsdb *db,
     m->session = s;
     m->db = db;
     m->dbmon = ovsdb_monitor_create(db, m);
-    if (version == OVSDB_MONITOR_V2) {
+    if (version == OVSDB_MONITOR_V2 || version == OVSDB_MONITOR_V3) {
         m->condition = ovsdb_monitor_session_condition_create();
     }
     m->version = version;
@@ -1444,9 +1451,42 @@ ovsdb_jsonrpc_monitor_create(struct ovsdb_jsonrpc_session *s, struct ovsdb *db,
         ovsdb_monitor_condition_bind(m->dbmon, m->condition);
     }
 
-    ovsdb_monitor_get_initial(m->dbmon, &m->change_set);
-    json = ovsdb_jsonrpc_monitor_compose_update(m, true);
+    bool initial = false;
+    if (version == OVSDB_MONITOR_V3) {
+        struct json *last_id = params->array.elems[3];
+        if (last_id->type != JSON_STRING) {
+            error = ovsdb_syntax_error(last_id, NULL,
+                                       "last-txn-id must be string");
+            goto error;
+        }
+        struct uuid txn_uuid;
+        if (!uuid_from_string(&txn_uuid, last_id->string)) {
+            error = ovsdb_syntax_error(last_id, NULL,
+                                       "last-txn-id must be UUID format.");
+            goto error;
+        }
+        if (!uuid_is_zero(&txn_uuid)) {
+            ovsdb_monitor_get_changes_after(&txn_uuid, m->dbmon,
+                                            &m->change_set);
+        }
+    }
+    if (!m->change_set) {
+        ovsdb_monitor_get_initial(m->dbmon, &m->change_set);
+        initial = true;
+    }
+    json = ovsdb_jsonrpc_monitor_compose_update(m, initial);
     json = json ? json : json_object_create();
+
+    if (m->version == OVSDB_MONITOR_V3) {
+        struct json *json_last_id = json_string_create_nocopy(
+                xasprintf(UUID_FMT,
+                          UUID_ARGS(ovsdb_monitor_get_last_txnid(
+                                  m->dbmon))));
+
+        struct json *json_found = json_boolean_create(!initial);
+        json = json_array_create_3(json_found, json_last_id, json);
+    }
+
     return jsonrpc_create_reply(json, request_id);
 
 error:
@@ -1580,8 +1620,17 @@ ovsdb_jsonrpc_monitor_cond_change(struct ovsdb_jsonrpc_session *s,
     if (update_json) {
         struct jsonrpc_msg *msg;
         struct json *p;
-
-        p = json_array_create_2(json_clone(m->monitor_id), update_json);
+        if (m->version == OVSDB_MONITOR_V3) {
+            struct json *json_last_id = json_string_create_nocopy(
+                    xasprintf(UUID_FMT,
+                              UUID_ARGS(ovsdb_monitor_get_last_txnid(
+                                      m->dbmon))));
+
+            p = json_array_create_3(json_clone(m->monitor_id), json_last_id,
+                                    update_json);
+        } else {
+            p = json_array_create_2(json_clone(m->monitor_id), update_json);
+        }
         msg = ovsdb_jsonrpc_create_notify(m, p);
         jsonrpc_session_send(s->js, msg);
     }
@@ -1702,6 +1751,9 @@ ovsdb_jsonrpc_create_notify(const struct ovsdb_jsonrpc_monitor *m,
     case OVSDB_MONITOR_V2:
         method = "update2";
         break;
+    case OVSDB_MONITOR_V3:
+        method = "update3";
+        break;
     case OVSDB_MONITOR_VERSION_MAX:
     default:
         OVS_NOT_REACHED();
@@ -1728,8 +1780,17 @@ ovsdb_jsonrpc_monitor_flush_all(struct ovsdb_jsonrpc_session *s)
         if (json) {
             struct jsonrpc_msg *msg;
             struct json *params;
+            if (m->version == OVSDB_MONITOR_V3) {
+                struct json *json_last_id = json_string_create_nocopy(
+                        xasprintf(UUID_FMT,
+                                  UUID_ARGS(ovsdb_monitor_get_last_txnid(
+                                          m->dbmon))));
+                params = json_array_create_3(json_clone(m->monitor_id),
+                                             json_last_id, json);
+            } else {
+                params = json_array_create_2(json_clone(m->monitor_id), json);
+            }
 
-            params = json_array_create_2(json_clone(m->monitor_id), json);
             msg = ovsdb_jsonrpc_create_notify(m, params);
             jsonrpc_session_send(s->js, msg);
         }
diff --git a/ovsdb/monitor.c b/ovsdb/monitor.c
index 03be5dcaf..1c66b428e 100644
--- a/ovsdb/monitor.c
+++ b/ovsdb/monitor.c
@@ -136,6 +136,9 @@ struct ovsdb_monitor_change_set {
     struct ovs_list change_set_for_tables;
 
     int n_refs;
+
+    /* The previous txn id before this change set's start point. */
+    struct uuid prev_txn;
 };
 
 /* Contains 'struct ovsdb_monitor_row's for rows in a specific table
@@ -200,7 +203,9 @@ typedef struct json *
 
 static void ovsdb_monitor_destroy(struct ovsdb_monitor *);
 static struct ovsdb_monitor_change_set * ovsdb_monitor_add_change_set(
-        struct ovsdb_monitor *, bool init_only);
+        struct ovsdb_monitor *, bool init_only, const struct uuid *prev_txn);
+static struct ovsdb_monitor_change_set * ovsdb_monitor_find_change_set(
+        const struct ovsdb_monitor *, const struct uuid *prev_txn);
 static void ovsdb_monitor_change_set_destroy(
         struct ovsdb_monitor_change_set *);
 static void ovsdb_monitor_track_new_change_set(struct ovsdb_monitor *);
@@ -540,13 +545,14 @@ ovsdb_monitor_table_exists(struct ovsdb_monitor *m,
 
 static struct ovsdb_monitor_change_set *
 ovsdb_monitor_add_change_set(struct ovsdb_monitor *dbmon,
-                             bool init_only)
+                             bool init_only, const struct uuid *prev_txn)
 {
     struct ovsdb_monitor_change_set *change_set = xzalloc(sizeof *change_set);
     change_set->uuid = uuid_random();
     ovs_list_push_back(&(dbmon->change_sets), &change_set->list_node);
     ovs_list_init(&change_set->change_set_for_tables);
     change_set->n_refs = 1;
+    change_set->prev_txn = prev_txn ? *prev_txn : UUID_ZERO;
 
     struct shash_node *node;
     SHASH_FOR_EACH (node, &dbmon->tables) {
@@ -567,6 +573,33 @@ ovsdb_monitor_add_change_set(struct ovsdb_monitor *dbmon,
     return change_set;
 };
 
+static struct ovsdb_monitor_change_set *
+ovsdb_monitor_find_change_set(const struct ovsdb_monitor *dbmon,
+                              const struct uuid *prev_txn)
+{
+    struct ovsdb_monitor_change_set *cs;
+    LIST_FOR_EACH (cs, list_node, &dbmon->change_sets) {
+        if (uuid_equals(&cs->prev_txn, prev_txn)) {
+            /* Check n_columns for each table in dbmon, in case it is changed
+             * after the change set is populated. */
+            bool n_col_is_equal = true;
+            struct ovsdb_monitor_change_set_for_table *mcst;
+            LIST_FOR_EACH (mcst, list_in_change_set,
+                           &cs->change_set_for_tables) {
+                struct ovsdb_monitor_table *mt = mcst->mt;
+                if (mt->n_columns != mcst->n_columns) {
+                    n_col_is_equal = false;
+                    break;
+                }
+            }
+            if (n_col_is_equal) {
+                return cs;
+            }
+        }
+    }
+    return NULL;
+}
+
 static void
 ovsdb_monitor_untrack_change_set(struct ovsdb_monitor *dbmon,
                                  struct ovsdb_monitor_change_set *mcs)
@@ -591,7 +624,8 @@ ovsdb_monitor_track_new_change_set(struct ovsdb_monitor *dbmon)
     if (change_set) {
         change_set->n_refs++;
     } else {
-        change_set = ovsdb_monitor_add_change_set(dbmon, false);
+        change_set = ovsdb_monitor_add_change_set(dbmon, false,
+                                 ovsdb_monitor_get_last_txnid(dbmon));
         dbmon->new_change_set = change_set;
     }
 }
@@ -1190,12 +1224,13 @@ ovsdb_monitor_get_update(
                                             condition,
                                             ovsdb_monitor_compose_row_update);
         } else {
-            ovs_assert(version == OVSDB_MONITOR_V2);
+            ovs_assert(version == OVSDB_MONITOR_V2 ||
+                       version == OVSDB_MONITOR_V3);
+
             if (!cond_updated) {
                 json = ovsdb_monitor_compose_update(dbmon, initial, mcs,
                                             condition,
                                             ovsdb_monitor_compose_row_update2);
-
                 if (!condition || !condition->conditional) {
                     ovsdb_monitor_json_cache_insert(dbmon, version, mcs,
                                                     json);
@@ -1434,7 +1469,7 @@ ovsdb_monitor_get_initial(struct ovsdb_monitor *dbmon,
 {
     if (!dbmon->init_change_set) {
         struct ovsdb_monitor_change_set *change_set =
-            ovsdb_monitor_add_change_set(dbmon, true);
+            ovsdb_monitor_add_change_set(dbmon, true, NULL);
         dbmon->init_change_set = change_set;
 
         struct ovsdb_monitor_change_set_for_table *mcst;
@@ -1454,6 +1489,68 @@ ovsdb_monitor_get_initial(struct ovsdb_monitor *dbmon,
     *p_mcs = dbmon->init_change_set;
 }
 
+static bool
+ovsdb_monitor_history_change_cb(const struct ovsdb_row *old,
+                        const struct ovsdb_row *new,
+                        const unsigned long int *changed,
+                        void *aux)
+{
+    struct ovsdb_monitor_change_set *change_set = aux;
+    struct ovsdb_table *table = new ? new->table : old->table;
+    struct ovsdb_monitor_change_set_for_table *mcst;
+
+    enum ovsdb_monitor_selection type =
+        ovsdb_monitor_row_update_type(false, old, new);
+    LIST_FOR_EACH (mcst, list_in_change_set,
+                   &change_set->change_set_for_tables) {
+        if (mcst->mt->table == table) {
+            enum ovsdb_monitor_changes_efficacy efficacy =
+                ovsdb_monitor_changes_classify(type, mcst->mt, changed);
+            if (efficacy > OVSDB_CHANGES_NO_EFFECT) {
+                ovsdb_monitor_changes_update(old, new, mcst->mt, mcst);
+            }
+            return true;
+        }
+    }
+    return false;
+}
+
+void
+ovsdb_monitor_get_changes_after(const struct uuid *txn_uuid,
+                                struct ovsdb_monitor *dbmon,
+                                struct ovsdb_monitor_change_set **p_mcs)
+{
+    ovs_assert(*p_mcs == NULL);
+    ovs_assert(!uuid_is_zero(txn_uuid));
+    struct ovsdb_monitor_change_set *change_set =
+        ovsdb_monitor_find_change_set(dbmon, txn_uuid);
+    if (change_set) {
+        change_set->n_refs++;
+        *p_mcs = change_set;
+        return;
+    }
+
+    struct ovsdb_txn_history_node *h_node;
+    bool found = false;
+    LIST_FOR_EACH (h_node, node, &dbmon->db->txn_history) {
+        struct ovsdb_txn *txn = h_node->txn;
+        if (!found) {
+            /* find the txn with last_id in history */
+            if (uuid_equals(ovsdb_txn_get_txnid(txn), txn_uuid)) {
+                found = true;
+                change_set = ovsdb_monitor_add_change_set(dbmon, false,
+                                                          txn_uuid);
+            }
+        } else {
+            /* Already found. Add changes in each follow up transaction to
+             * the new change_set. */
+            ovsdb_txn_for_each_change(txn, ovsdb_monitor_history_change_cb,
+                                      change_set);
+        }
+    }
+    *p_mcs = change_set;
+}
+
 void
 ovsdb_monitor_remove_jsonrpc_monitor(struct ovsdb_monitor *dbmon,
                    struct ovsdb_jsonrpc_monitor *jsonrpc_monitor,
@@ -1692,3 +1789,15 @@ ovsdb_monitor_prereplace_db(struct ovsdb *db)
         }
     }
 }
+
+const struct uuid *
+ovsdb_monitor_get_last_txnid(struct ovsdb_monitor *dbmon) {
+    static struct uuid dummy = { .parts = { 0, 0, 0, 0 } };
+    if (dbmon->db->n_txn_history) {
+        struct ovsdb_txn_history_node *thn = CONTAINER_OF(
+                ovs_list_back(&dbmon->db->txn_history),
+                struct ovsdb_txn_history_node, node);
+        return ovsdb_txn_get_txnid(thn->txn);
+    }
+    return &dummy;
+}
diff --git a/ovsdb/monitor.h b/ovsdb/monitor.h
index 112c67266..1ac9aaf3d 100644
--- a/ovsdb/monitor.h
+++ b/ovsdb/monitor.h
@@ -39,6 +39,8 @@ enum ovsdb_monitor_version {
       OVSDB_MONITOR_V1,         /* RFC 7047 "monitor" method. */
       OVSDB_MONITOR_V2,         /* Extension to RFC 7047, see ovsdb-server
                                    man page for details. */
+      OVSDB_MONITOR_V3,         /* Extension to V2, see ovsdb-server man
+                                   page for details. */
 
       /* Last entry.  */
       OVSDB_MONITOR_VERSION_MAX
@@ -80,6 +82,8 @@ struct json *ovsdb_monitor_get_update(struct ovsdb_monitor *,
                                       enum ovsdb_monitor_version,
                                       struct ovsdb_monitor_change_set **p_mcs);
 
+const struct uuid *ovsdb_monitor_get_last_txnid(struct ovsdb_monitor *);
+
 void ovsdb_monitor_table_add_select(struct ovsdb_monitor *,
                                     const struct ovsdb_table *,
                                     enum ovsdb_monitor_selection);
@@ -89,6 +93,8 @@ bool ovsdb_monitor_needs_flush(struct ovsdb_monitor *,
 
 void ovsdb_monitor_get_initial(struct ovsdb_monitor *,
                                struct ovsdb_monitor_change_set **);
+void ovsdb_monitor_get_changes_after(const struct uuid *txn_uuid,
+        struct ovsdb_monitor *, struct ovsdb_monitor_change_set **);
 
 void ovsdb_monitor_get_memory_usage(struct simap *);
 
diff --git a/ovsdb/ovsdb-client.1.in b/ovsdb/ovsdb-client.1.in
index 46a3e38dd..284433bb3 100644
--- a/ovsdb/ovsdb-client.1.in
+++ b/ovsdb/ovsdb-client.1.in
@@ -44,6 +44,9 @@ ovsdb\-client \- command-line interface to \fBovsdb-server\fR(1)
 \fBovsdb\-client\fR [\fIoptions\fR] \fBmonitor\-cond\fR [\fIserver\fR] [\fIdatabase\fR] \fIconditions
 \fItable\fR [\fIcolumn\fR[\fB,\fIcolumn\fR]...]...
 .br
+\fBovsdb\-client\fR [\fIoptions\fR] \fBmonitor\-cond\-since\fR [\fIserver\fR] [\fIdatabase\fR]
+[\fIlast-id\fR] \fIconditions \fItable\fR [\fIcolumn\fR[\fB,\fIcolumn\fR]...]...
+.br
 \fBovsdb\-client \fR[\fIoptions\fR] \fBwait\fR \fR[\fIserver\fR] \fIdatabase\fR \fIstate\fR
 .IP "Testing Commands:"
 \fBovsdb\-client\fR [\fIoptions\fR] \fBlock\fR [\fIserver\fR] \fIlock\fR
@@ -243,6 +246,7 @@ with an error or succeed with surprising results.
 .
 .IP "\fBmonitor\fR [\fIserver\fR] [\fIdatabase\fR] \fItable\fR [\fIcolumn\fR[\fB,\fIcolumn\fR]...]..."
 .IQ "\fBmonitor\-cond\fR [\fIserver\fR] [\fIdatabase\fR] \fIconditions\fR \fItable\fR [\fIcolumn\fR[\fB,\fIcolumn\fR]...]..."
+.IQ "\fBmonitor\-cond\-since\fR [\fIserver\fR] [\fIdatabase\fR] [\fIlast-id\fR] \fIconditions\fR \fItable\fR [\fIcolumn\fR[\fB,\fIcolumn\fR]...]..."
 Connects to \fIserver\fR and monitors the contents of rows that match conditions in
 \fItable\fR in \fIdatabase\fR. By default, the initial contents of \fItable\fR are
 printed, followed by each change as it occurs.  If conditions empty,
@@ -269,13 +273,14 @@ line.
 with the following change: A condition can be either a 3-element JSON array
 as deescribed in the RFC or a boolean value..
 .IP
-If \fB\-\-detach\fR is used with \fBmonitor\fR or \fBmonitor\-cond\fR, then
-\fBovsdb\-client\fR detaches after it has successfully received and
-printed the initial contents of \fItable\fR.
+If \fB\-\-detach\fR is used with \fBmonitor\fR, \fBmonitor\-cond\fR or
+\fBmonitor\-cond\-since\fR, then \fBovsdb\-client\fR detaches after it has
+successfully received and printed the initial contents of \fItable\fR.
 .IP
 The \fBmonitor\fR command uses RFC 7047 "monitor" method to open a monitor
-session with the server. The \fBmonitor\-cond\fR command uses RFC 7047
-extension "monitor_cond" method. See \fBovsdb\-server\fR(1) for details.
+session with the server. The \fBmonitor\-cond\fR and \fBmonitor\-cond\-since\fR
+commandls uses RFC 7047 extension "monitor_cond" and "monitor_cond_since"
+methods. See \fBovsdb\-server\fR(1) for details.
 .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
@@ -340,10 +345,10 @@ The following options controlling output formatting:
 .so lib/table.man
 .
 .IP "\fB\-\-timestamp\fR"
-For the \fBmonitor\fR and \fBmonitor\-cond\fR commands, add a timestamp to each
-table update.  Most output formats add the timestamp on a line of its own
-just above the table.  The JSON output format puts the timestamp in a
-member of the top-level JSON object named \fBtime\fR.
+For the \fBmonitor\fR, \fBmonitor\-cond\fR and \fBmonitor\-cond\-since\fR
+commands, add a timestamp to each table update.  Most output formats add the
+timestamp on a line of its own just above the table.  The JSON output format
+puts the timestamp in a member of the top-level JSON object named \fBtime\fR.
 .
 .IP "\fB\-t\fR"
 .IQ "\fB\-\-timeout=\fIsecs\fR"
@@ -352,8 +357,9 @@ seconds.  If the timeout expires, \fBovsdb\-client\fR will exit with a
 \fBSIGALRM\fR signal.
 .
 .SS "Daemon Options"
-The daemon options apply only to the \fBmonitor\fR and \fBmonitor\-cond\fR commands.
-With any other command, they have no effect.
+The daemon options apply only to the \fBmonitor\fR, \fBmonitor\-cond\fR and
+\fBmonitor\-cond\-since\fR commands.  With any other command, they have no
+effect.
 .ds DD
 .so lib/daemon.man
 .SS "Logging Options"
diff --git a/ovsdb/ovsdb-client.c b/ovsdb/ovsdb-client.c
index 0215357fe..9ae15e557 100644
--- a/ovsdb/ovsdb-client.c
+++ b/ovsdb/ovsdb-client.c
@@ -430,6 +430,13 @@ usage(void)
            "    DATABASE on SERVER.\n"
            "    COLUMNs may include !initial, !insert, !delete, !modify\n"
            "    to avoid seeing the specified kinds of changes.\n"
+           "\n  monitor-cond-since [SERVER] [DATABASE] [LASTID] CONDITION TABLE [COLUMN,...]...\n"
+           "    monitor contents that match CONDITION of COLUMNs in TABLE in\n"
+           "    DATABASE on SERVER, since change after LASTID.\n"
+           "    LASTID specifies transaction ID after which the monitoring\n"
+           "    starts, which works only for cluster mode. If ignored, it\n"
+           "    defaults to an all-zero uuid.\n"
+           "    Other arguments are the same as in monitor-cond.\n"
            "\n  convert [SERVER] SCHEMA\n"
            "    convert database on SERVER named in SCHEMA to SCHEMA.\n"
            "\n  needs-conversion [SERVER] SCHEMA\n"
@@ -1127,6 +1134,35 @@ monitor2_print(struct json *table_updates2,
 }
 
 static void
+monitor3_print(struct json *result,
+               const struct monitored_table *mts, size_t n_mts)
+{
+    if (result->type != JSON_ARRAY) {
+        ovs_error(0, "<result> is not array");
+    }
+
+    if (result->array.n != 3) {
+        ovs_error(0, "<result> should have 3 elements, but has %"PRIuSIZE".",
+                  result->array.n);
+    }
+
+    bool found = json_boolean(result->array.elems[0]);
+    const char *last_id = json_string(result->array.elems[1]);
+    printf("found: %s, last_id: %s\n", found ? "true" : "false", last_id);
+
+    struct json *table_updates2 = result->array.elems[2];
+    monitor2_print(table_updates2, mts, n_mts);
+}
+
+static void
+monitor3_notify_print(const char *last_id, struct json *table_updates2,
+                      const struct monitored_table *mts, size_t n_mts)
+{
+    printf("\nlast_id: %s", last_id);
+    monitor2_print(table_updates2, mts, n_mts);
+}
+
+static void
 add_column(const char *server, const struct ovsdb_column *column,
            struct ovsdb_column_set *columns, struct json *columns_json)
 {
@@ -1333,7 +1369,8 @@ destroy_monitored_table(struct monitored_table *mts, size_t n)
 static void
 do_monitor__(struct jsonrpc *rpc, const char *database,
              enum ovsdb_monitor_version version,
-             int argc, char *argv[], struct json *condition)
+             int argc, char *argv[], struct json *condition,
+             const struct uuid *last_id)
 {
     const char *server = jsonrpc_get_name(rpc);
     const char *table_name = argv[0];
@@ -1411,8 +1448,24 @@ do_monitor__(struct jsonrpc *rpc, const char *database,
 
     monitor = json_array_create_3(json_string_create(database),
                                   json_null_create(), monitor_requests);
-    const char *method = version == OVSDB_MONITOR_V2 ? "monitor_cond"
-                                                     : "monitor";
+    const char *method;
+    switch (version) {
+    case OVSDB_MONITOR_V1:
+        method = "monitor";
+        break;
+    case OVSDB_MONITOR_V2:
+        method = "monitor_cond";
+        break;
+    case OVSDB_MONITOR_V3:
+        method = "monitor_cond_since";
+        struct json *json_last_id = json_string_create_nocopy(
+                xasprintf(UUID_FMT, UUID_ARGS(last_id)));
+        json_array_add(monitor, json_last_id);
+        break;
+    case OVSDB_MONITOR_VERSION_MAX:
+    default:
+        OVS_NOT_REACHED();
+    }
 
     struct jsonrpc_msg *request;
     request = jsonrpc_create_request(method, monitor, NULL);
@@ -1444,6 +1497,9 @@ do_monitor__(struct jsonrpc *rpc, const char *database,
                 case OVSDB_MONITOR_V2:
                     monitor2_print(msg->result, mts, n_mts);
                     break;
+                case OVSDB_MONITOR_V3:
+                    monitor3_print(msg->result, mts, n_mts);
+                    break;
                 case OVSDB_MONITOR_VERSION_MAX:
                 default:
                     OVS_NOT_REACHED();
@@ -1470,6 +1526,17 @@ do_monitor__(struct jsonrpc *rpc, const char *database,
                     fflush(stdout);
                 }
             } else if (msg->type == JSONRPC_NOTIFY
+                       && version == OVSDB_MONITOR_V3
+                       && !strcmp(msg->method, "update3")) {
+                struct json *params = msg->params;
+                if (params->type == JSON_ARRAY
+                    && params->array.n == 3
+                    && params->array.elems[0]->type == JSON_NULL) {
+                    monitor3_notify_print(json_string(params->array.elems[1]),
+                                          params->array.elems[2], mts, n_mts);
+                    fflush(stdout);
+                }
+            } else if (msg->type == JSONRPC_NOTIFY
                        && !strcmp(msg->method, "monitor_canceled")) {
                 ovs_fatal(0, "%s: %s database was removed",
                           server, database);
@@ -1500,12 +1567,13 @@ static void
 do_monitor(struct jsonrpc *rpc, const char *database,
            int argc, char *argv[])
 {
-    do_monitor__(rpc, database, OVSDB_MONITOR_V1, argc, argv, NULL);
+    do_monitor__(rpc, database, OVSDB_MONITOR_V1, argc, argv, NULL, NULL);
 }
 
 static void
-do_monitor_cond(struct jsonrpc *rpc, const char *database,
-           int argc, char *argv[])
+do_monitor_cond__(struct jsonrpc *rpc, const char *database,
+                  enum ovsdb_monitor_version version,
+                  struct uuid *last_id, int argc, char *argv[])
 {
     struct ovsdb_condition cnd;
     struct json *condition = NULL;
@@ -1524,10 +1592,31 @@ do_monitor_cond(struct jsonrpc *rpc, const char *database,
     check_ovsdb_error(ovsdb_condition_from_json(table, condition,
                                                     NULL, &cnd));
     ovsdb_condition_destroy(&cnd);
-    do_monitor__(rpc, database, OVSDB_MONITOR_V2, --argc, ++argv, condition);
+    do_monitor__(rpc, database, version, --argc, ++argv, condition,
+                 last_id);
     ovsdb_schema_destroy(schema);
 }
 
+static void
+do_monitor_cond(struct jsonrpc *rpc, const char *database,
+                int argc, char *argv[])
+{
+    do_monitor_cond__(rpc, database, OVSDB_MONITOR_V2, NULL, argc, argv);
+}
+
+static void
+do_monitor_cond_since(struct jsonrpc *rpc, const char *database,
+           int argc, char *argv[])
+{
+    ovs_assert(argc > 1);
+    struct uuid last_id;
+    if (uuid_from_string(&last_id, argv[0])) {
+        argc--;
+        argv++;
+    }
+    do_monitor_cond__(rpc, database, OVSDB_MONITOR_V3, &last_id, argc, argv);
+}
+
 static bool
 is_database_clustered(struct jsonrpc *rpc, const char *database)
 {
@@ -2409,6 +2498,7 @@ static const struct ovsdb_client_command all_commands[] = {
     { "query",              NEED_NONE,     1, 2,       do_query },
     { "monitor",            NEED_DATABASE, 1, INT_MAX, do_monitor },
     { "monitor-cond",       NEED_DATABASE, 2, 3,       do_monitor_cond },
+    { "monitor-cond-since", NEED_DATABASE, 2, 4,       do_monitor_cond_since },
     { "wait",               NEED_NONE,     2, 3,       do_wait },
     { "convert",            NEED_NONE,     1, 2,       do_convert },
     { "needs-conversion",   NEED_NONE,     1, 2,       do_needs_conversion },
diff --git a/ovsdb/transaction.c b/ovsdb/transaction.c
index 148d108fd..9fc1fd7f7 100644
--- a/ovsdb/transaction.c
+++ b/ovsdb/transaction.c
@@ -120,6 +120,12 @@ ovsdb_txn_set_txnid(const struct uuid *txnid, struct ovsdb_txn *txn)
     txn->txnid = *txnid;
 }
 
+const struct uuid *
+ovsdb_txn_get_txnid(const struct ovsdb_txn *txn)
+{
+    return &txn->txnid;
+}
+
 static void
 ovsdb_txn_free(struct ovsdb_txn *txn)
 {
diff --git a/ovsdb/transaction.h b/ovsdb/transaction.h
index 5e62ef06f..c81937357 100644
--- a/ovsdb/transaction.h
+++ b/ovsdb/transaction.h
@@ -26,6 +26,7 @@ struct uuid;
 
 struct ovsdb_txn *ovsdb_txn_create(struct ovsdb *);
 void ovsdb_txn_set_txnid(const struct uuid *, struct ovsdb_txn *);
+const struct uuid *ovsdb_txn_get_txnid(const struct ovsdb_txn *);
 void ovsdb_txn_abort(struct ovsdb_txn *);
 
 struct ovsdb_error *ovsdb_txn_replay_commit(struct ovsdb_txn *)