ovsdb: Make OVSDB backup sever read only

When ovsdb-sever is running in the backup state, it would be nice to
make sure there is no un-intended changes to the backup database.

This patch makes the ovsdb server only accepts 'read' transactions as
a backup server. When the server role is changed into an active server,
all existing client connections will be reset. After reconnect, all
clinet transactions will then be accepted.

Signed-off-by: Andy Zhou <azhou@ovn.org>
Acked-by: Ben Pfaff <blp@ovn.org>

ovsdb/execution.c

@@ -61,24 +61,25 @@ static ovsdb_operation_executor ovsdb_execute_comment;
 static ovsdb_operation_executor ovsdb_execute_assert;
 
 static ovsdb_operation_executor *
-lookup_executor(const char *name)
+lookup_executor(const char *name, bool *read_only)
 {
     struct ovsdb_operation {
         const char *name;
+        bool read_only;
         ovsdb_operation_executor *executor;
     };
 
     static const struct ovsdb_operation operations[] = {
-        { "insert", ovsdb_execute_insert },
+        { "insert", false, ovsdb_execute_insert },
-        { "select", ovsdb_execute_select },
+        { "select", true, ovsdb_execute_select },
-        { "update", ovsdb_execute_update },
+        { "update", false, ovsdb_execute_update },
-        { "mutate", ovsdb_execute_mutate },
+        { "mutate", false, ovsdb_execute_mutate },
-        { "delete", ovsdb_execute_delete },
+        { "delete", false, ovsdb_execute_delete },
-        { "wait", ovsdb_execute_wait },
+        { "wait", true, ovsdb_execute_wait },
-        { "commit", ovsdb_execute_commit },
+        { "commit", false, ovsdb_execute_commit },
-        { "abort", ovsdb_execute_abort },
+        { "abort", true, ovsdb_execute_abort },
-        { "comment", ovsdb_execute_comment },
+        { "comment", true, ovsdb_execute_comment },
-        { "assert", ovsdb_execute_assert },
+        { "assert", true, ovsdb_execute_assert },
     };
 
     size_t i;
@@ -86,6 +87,7 @@ lookup_executor(const char *name)
     for (i = 0; i < ARRAY_SIZE(operations); i++) {
         const struct ovsdb_operation *c = &operations[i];
         if (!strcmp(c->name, name)) {
+            *read_only = c->read_only;
             return c->executor;
         }
     }
@@ -94,7 +96,7 @@ lookup_executor(const char *name)
 
 struct json *
 ovsdb_execute(struct ovsdb *db, const struct ovsdb_session *session,
-              const struct json *params,
+              const struct json *params, bool read_only,
               long long int elapsed_msec, long long int *timeout_msec)
 {
     struct ovsdb_execution x;
@@ -137,15 +139,18 @@ ovsdb_execute(struct ovsdb *db, const struct ovsdb_session *session,
         struct ovsdb_parser parser;
         struct json *result;
         const struct json *op;
+        const char *op_name = NULL;
+        bool ro = false;
 
         /* Parse and execute operation. */
         ovsdb_parser_init(&parser, operation,
-                          "ovsdb operation %"PRIuSIZE" of %"PRIuSIZE, i, n_operations);
+                          "ovsdb operation %"PRIuSIZE" of %"PRIuSIZE, i,
+                          n_operations);
         op = ovsdb_parser_member(&parser, "op", OP_ID);
         result = json_object_create();
         if (op) {
-            const char *op_name = json_string(op);
+            op_name = json_string(op);
-            ovsdb_operation_executor *executor = lookup_executor(op_name);
+            ovsdb_operation_executor *executor = lookup_executor(op_name, &ro);
             if (executor) {
                 error = executor(&x, &parser, result);
             } else {
@@ -163,6 +168,13 @@ ovsdb_execute(struct ovsdb *db, const struct ovsdb_session *session,
             ovsdb_error_destroy(error);
             error = parse_error;
         }
+        /* Create read-only violation error if there is one. */
+        if (!error && read_only && !ro) {
+            error = ovsdb_error("not allowed",
+                                "%s operation not allowed when "
+                                "database server is in read only mode",
+                                op_name);
+        }
         if (error) {
             json_destroy(result);
             result = ovsdb_error_to_json(error);

ovsdb/jsonrpc-server.c

@@ -56,7 +56,7 @@ static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5);
 
 /* Sessions. */
 static struct ovsdb_jsonrpc_session *ovsdb_jsonrpc_session_create(
-    struct ovsdb_jsonrpc_remote *, struct jsonrpc_session *);
+    struct ovsdb_jsonrpc_remote *, struct jsonrpc_session *, bool);
 static void ovsdb_jsonrpc_session_run_all(struct ovsdb_jsonrpc_remote *);
 static void ovsdb_jsonrpc_session_wait_all(struct ovsdb_jsonrpc_remote *);
 static void ovsdb_jsonrpc_session_get_memory_usage_all(
@@ -114,6 +114,8 @@ static struct jsonrpc_msg * ovsdb_jsonrpc_create_notify(
 struct ovsdb_jsonrpc_server {
     struct ovsdb_server up;
     unsigned int n_sessions;
+    bool read_only;            /* This server is does not accept any
+                                  transactions that can modify the database. */
     struct shash remotes;      /* Contains "struct ovsdb_jsonrpc_remote *"s. */
 };
 
@@ -138,11 +140,12 @@ static void ovsdb_jsonrpc_server_del_remote(struct shash_node *);
  * The caller must call ovsdb_jsonrpc_server_add_db() for each database to
  * which 'server' should provide access. */
 struct ovsdb_jsonrpc_server *
-ovsdb_jsonrpc_server_create(void)
+ovsdb_jsonrpc_server_create(bool read_only)
 {
     struct ovsdb_jsonrpc_server *server = xzalloc(sizeof *server);
     ovsdb_server_init(&server->up);
     shash_init(&server->remotes);
+    server->read_only = read_only;
     return server;
 }
 
@@ -160,7 +163,7 @@ ovsdb_jsonrpc_server_add_db(struct ovsdb_jsonrpc_server *svr, struct ovsdb *db)
      * If this is too big of a hammer in practice, we could be more selective,
      * e.g. disconnect only connections that actually tried to use a database
      * with 'db''s name. */
-    ovsdb_jsonrpc_server_reconnect(svr);
+    ovsdb_jsonrpc_server_reconnect(svr, svr->read_only);
 
     return ovsdb_server_add_db(&svr->up, db);
 }
@@ -177,7 +180,7 @@ ovsdb_jsonrpc_server_remove_db(struct ovsdb_jsonrpc_server *svr,
      *
      * If this is too big of a hammer in practice, we could be more selective,
      * e.g. disconnect only connections that actually reference 'db'. */
-    ovsdb_jsonrpc_server_reconnect(svr);
+    ovsdb_jsonrpc_server_reconnect(svr, svr->read_only);
 
     return ovsdb_server_remove_db(&svr->up, db);
 }
@@ -268,7 +271,8 @@ ovsdb_jsonrpc_server_add_remote(struct ovsdb_jsonrpc_server *svr,
     shash_add(&svr->remotes, name, remote);
 
     if (!listener) {
-        ovsdb_jsonrpc_session_create(remote, jsonrpc_session_open(name, true));
+        ovsdb_jsonrpc_session_create(remote, jsonrpc_session_open(name, true),
+                                      svr->read_only);
     }
     return remote;
 }
@@ -327,10 +331,11 @@ ovsdb_jsonrpc_server_free_remote_status(
 /* Forces all of the JSON-RPC sessions managed by 'svr' to disconnect and
  * reconnect. */
 void
-ovsdb_jsonrpc_server_reconnect(struct ovsdb_jsonrpc_server *svr)
+ovsdb_jsonrpc_server_reconnect(struct ovsdb_jsonrpc_server *svr, bool read_only)
 {
     struct shash_node *node;
 
+    svr->read_only = read_only;
     SHASH_FOR_EACH (node, &svr->remotes) {
         struct ovsdb_jsonrpc_remote *remote = node->data;
 
@@ -355,7 +360,7 @@ ovsdb_jsonrpc_server_run(struct ovsdb_jsonrpc_server *svr)
                 struct jsonrpc_session *js;
                 js = jsonrpc_session_open_unreliably(jsonrpc_open(stream),
                                                      remote->dscp);
-                ovsdb_jsonrpc_session_create(remote, js);
+                ovsdb_jsonrpc_session_create(remote, js, svr->read_only);
             } else if (error != EAGAIN) {
                 VLOG_WARN_RL(&rl, "%s: accept failed: %s",
                              pstream_get_name(remote->listener),
@@ -415,6 +420,10 @@ struct ovsdb_jsonrpc_session {
     /* Network connectivity. */
     struct jsonrpc_session *js;  /* JSON-RPC session. */
     unsigned int js_seqno;       /* Last jsonrpc_session_get_seqno() value. */
+
+    /* Read only. */
+    bool read_only;             /*  When true, not allow to modify the
+                                    database. */
 };
 
 static void ovsdb_jsonrpc_session_close(struct ovsdb_jsonrpc_session *);
@@ -429,7 +438,7 @@ static void ovsdb_jsonrpc_session_got_notify(struct ovsdb_jsonrpc_session *,
 
 static struct ovsdb_jsonrpc_session *
 ovsdb_jsonrpc_session_create(struct ovsdb_jsonrpc_remote *remote,
-                             struct jsonrpc_session *js)
+                             struct jsonrpc_session *js, bool read_only)
 {
     struct ovsdb_jsonrpc_session *s;
 
@@ -441,6 +450,7 @@ ovsdb_jsonrpc_session_create(struct ovsdb_jsonrpc_remote *remote,
     hmap_init(&s->monitors);
     s->js = js;
     s->js_seqno = jsonrpc_session_get_seqno(js);
+    s->read_only = read_only;
 
     remote->server->n_sessions++;
 
@@ -745,6 +755,14 @@ ovsdb_jsonrpc_session_notify(struct ovsdb_session *session,
     ovsdb_jsonrpc_session_send(s, jsonrpc_create_notify(method, params));
 }
 
+static struct jsonrpc_msg *
+jsonrpc_create_readonly_lock_error(const struct json *id)
+{
+    return jsonrpc_create_error(json_string_create(
+            "lock and unlock methods not allowed,"
+            " DB server is read only."), id);
+}
+
 static struct jsonrpc_msg *
 ovsdb_jsonrpc_session_lock(struct ovsdb_jsonrpc_session *s,
                            struct jsonrpc_msg *request,
@@ -757,6 +775,10 @@ ovsdb_jsonrpc_session_lock(struct ovsdb_jsonrpc_session *s,
     const char *lock_name;
     struct json *result;
 
+    if (s->read_only) {
+        return jsonrpc_create_readonly_lock_error(request->id);
+    }
+
     error = ovsdb_jsonrpc_session_parse_lock_name(request, &lock_name);
     if (error) {
         goto error;
@@ -827,6 +849,10 @@ ovsdb_jsonrpc_session_unlock(struct ovsdb_jsonrpc_session *s,
     struct ovsdb_error *error;
     const char *lock_name;
 
+    if (s->read_only) {
+        return jsonrpc_create_readonly_lock_error(request->id);
+    }
+
     error = ovsdb_jsonrpc_session_parse_lock_name(request, &lock_name);
     if (error) {
         goto error;
@@ -995,7 +1021,8 @@ ovsdb_jsonrpc_trigger_create(struct ovsdb_jsonrpc_session *s, struct ovsdb *db,
 
     /* Insert into trigger table. */
     t = xmalloc(sizeof *t);
-    ovsdb_trigger_init(&s->up, db, &t->trigger, params, time_msec());
+    ovsdb_trigger_init(&s->up, db, &t->trigger, params, time_msec(),
+                       s->read_only);
     t->id = id;
     hmap_insert(&s->triggers, &t->hmap_node, hash);
 

ovsdb/jsonrpc-server.h

@@ -23,7 +23,7 @@ struct ovsdb;
 struct shash;
 struct simap;
 
-struct ovsdb_jsonrpc_server *ovsdb_jsonrpc_server_create(void);
+struct ovsdb_jsonrpc_server *ovsdb_jsonrpc_server_create(bool read_only);
 bool ovsdb_jsonrpc_server_add_db(struct ovsdb_jsonrpc_server *,
                                  struct ovsdb *);
 bool ovsdb_jsonrpc_server_remove_db(struct ovsdb_jsonrpc_server *,
@@ -61,7 +61,7 @@ bool ovsdb_jsonrpc_server_get_remote_status(
 void ovsdb_jsonrpc_server_free_remote_status(
     struct ovsdb_jsonrpc_remote_status *);
 
-void ovsdb_jsonrpc_server_reconnect(struct ovsdb_jsonrpc_server *);
+void ovsdb_jsonrpc_server_reconnect(struct ovsdb_jsonrpc_server *, bool read_only);
 
 void ovsdb_jsonrpc_server_run(struct ovsdb_jsonrpc_server *);
 void ovsdb_jsonrpc_server_wait(struct ovsdb_jsonrpc_server *);

ovsdb/ovsdb-server.1.in

@@ -36,6 +36,23 @@ If none is specified, the default is \fB@DBDIR@/conf.db\fR.  The database
 files must already have been created and initialized using, for
 example, \fBovsdb\-tool create\fR.
 .
+.SH "ACTIVE and BACKUP "
+\fBovsdb\-server\fR runs either as a backup server, or as an active server.
+When  \fBovsdb\-server\fR is running as a backup server, all transactions that
+can modify the database content, including the lock commands are rejected.
+Active server, on the other hand, accepts all ovsdb server transactions.
+When \fBovsdb\-server\fR role changes, all existing client connection are
+reset, requiring clients to reconnect to the server.
+.PP
+By default, \fBovsdb\-server\fR runs as an active server, except when the
+\fB\-\-sync\-from=\fIserver\fR command line option is specified.  During
+runtime, \fBovsdb\-server\fR role can be switch by using appctl commands.
+.PP
+\fBovsdb-server/connect\-active\-ovsdb\-server\fR switches
+\fBovsdb\-server\fR role into a backup server, Conversely,
+\fBovsdb-server/disconnect\-active\-ovsdb\-server\fR changes server into
+an active one.
+.
 .SH OPTIONS
 .
 .IP "\fB\-\-remote=\fIremote\fR"

ovsdb/ovsdb-server.c

@@ -153,8 +153,15 @@ main_loop(struct ovsdb_jsonrpc_server *jsonrpc, struct shash *all_dbs,
         /* Run unixctl_server_run() before reconfigure_remotes() because
          * ovsdb-server/add-remote and ovsdb-server/remove-remote can change
          * the set of remotes that reconfigure_remotes() uses. */
+        bool last_role = is_backup_server;
         unixctl_server_run(unixctl);
 
+        /* In case unixctl commands change the role of ovsdb-server,
+         *  from active to backup or vise versa, recoonect jsonrpc server.  */
+        if (last_role != is_backup_server) {
+            ovsdb_jsonrpc_server_reconnect(jsonrpc, is_backup_server);
+        }
+
         report_error_if_changed(
             reconfigure_remotes(jsonrpc, all_dbs, remotes),
             &remotes_error);
@@ -267,7 +274,7 @@ main(int argc, char *argv[])
 
     /* Load the saved config. */
     load_config(config_tmpfile, &remotes, &db_filenames);
-    jsonrpc = ovsdb_jsonrpc_server_create();
+    jsonrpc = ovsdb_jsonrpc_server_create(is_backup_server);
 
     shash_init(&all_dbs);
     server_config.all_dbs = &all_dbs;
@@ -348,14 +355,18 @@ main(int argc, char *argv[])
                               ovsdb_server_set_active_ovsdb_server, NULL);
     unixctl_command_register("ovsdb-server/get-active-ovsdb-server", "", 0, 0,
                               ovsdb_server_get_active_ovsdb_server, NULL);
-    unixctl_command_register("ovsdb-server/connect-active-ovsdb-server", "", 0, 0,
+    unixctl_command_register("ovsdb-server/connect-active-ovsdb-server", "",
-                              ovsdb_server_connect_active_ovsdb_server, NULL);
+                             0, 0, ovsdb_server_connect_active_ovsdb_server,
-    unixctl_command_register("ovsdb-server/disconnect-active-ovsdb-server", "", 0, 0,
+                             NULL);
-                              ovsdb_server_disconnect_active_ovsdb_server, NULL);
+    unixctl_command_register("ovsdb-server/disconnect-active-ovsdb-server", "",
-    unixctl_command_register("ovsdb-server/set-sync-excluded-tables", "", 0, 1,
+                             0, 0, ovsdb_server_disconnect_active_ovsdb_server,
-                              ovsdb_server_set_sync_excluded_tables, NULL);
+                             NULL);
-    unixctl_command_register("ovsdb-server/get-sync-excluded-tables", "", 0, 0,
+    unixctl_command_register("ovsdb-server/set-sync-excluded-tables", "",
-                              ovsdb_server_get_sync_excluded_tables, NULL);
+                             0, 1, ovsdb_server_set_sync_excluded_tables,
+                             NULL);
+    unixctl_command_register("ovsdb-server/get-sync-excluded-tables", "",
+                             0, 0, ovsdb_server_get_sync_excluded_tables,
+                             NULL);
 
     /* Simulate the behavior of OVS release prior to version 2.5 that
      * does not support the monitor_cond method.  */
@@ -1048,6 +1059,7 @@ ovsdb_server_set_active_ovsdb_server(struct unixctl_conn *conn,
 {
     set_active_ovsdb_server(argv[1]);
     is_backup_server = true;
+    VLOG_INFO("become a backup server");
     unixctl_command_reply(conn, NULL);
 }
 
@@ -1087,6 +1099,7 @@ ovsdb_server_disconnect_active_ovsdb_server(struct unixctl_conn *conn,
 {
     disconnect_active_server();
     is_backup_server = false;
+    VLOG_INFO("become an active server");
     unixctl_command_reply(conn, NULL);
 }
 
@@ -1161,7 +1174,7 @@ ovsdb_server_disable_monitor_cond(struct unixctl_conn *conn,
     struct ovsdb_jsonrpc_server *jsonrpc = jsonrpc_;
 
     ovsdb_jsonrpc_disable_monitor_cond();
-    ovsdb_jsonrpc_server_reconnect(jsonrpc);
+    ovsdb_jsonrpc_server_reconnect(jsonrpc, is_backup_server);
     unixctl_command_reply(conn, NULL);
 }
 
@@ -1217,7 +1230,7 @@ ovsdb_server_reconnect(struct unixctl_conn *conn, int argc OVS_UNUSED,
 {
     struct ovsdb_jsonrpc_server *jsonrpc = jsonrpc_;
 
-    ovsdb_jsonrpc_server_reconnect(jsonrpc);
+    ovsdb_jsonrpc_server_reconnect(jsonrpc, is_backup_server);
     unixctl_command_reply(conn, NULL);
 }
 

ovsdb/ovsdb-tool.c

@@ -366,7 +366,7 @@ transact(bool read_only, int argc, char *argv[])
     check_ovsdb_error(ovsdb_file_open(db_file_name, read_only, &db, NULL));
 
     request = parse_json(transaction);
-    result = ovsdb_execute(db, NULL, request, 0, NULL);
+    result = ovsdb_execute(db, NULL, request, false, 0, NULL);
     json_destroy(request);
 
     print_and_free_json(result);

ovsdb/ovsdb.h

@@ -72,7 +72,7 @@ void ovsdb_get_memory_usage(const struct ovsdb *, struct simap *usage);
 struct ovsdb_table *ovsdb_get_table(const struct ovsdb *, const char *);
 
 struct json *ovsdb_execute(struct ovsdb *, const struct ovsdb_session *,
-                           const struct json *params,
+                           const struct json *params, bool read_only,
                            long long int elapsed_msec,
                            long long int *timeout_msec);
 

ovsdb/trigger.c

@@ -31,7 +31,8 @@ static void ovsdb_trigger_complete(struct ovsdb_trigger *);
 void
 ovsdb_trigger_init(struct ovsdb_session *session, struct ovsdb *db,
                    struct ovsdb_trigger *trigger,
-                   struct json *request, long long int now)
+                   struct json *request, long long int now,
+                   bool read_only)
 {
     trigger->session = session;
     trigger->db = db;
@@ -40,6 +41,7 @@ ovsdb_trigger_init(struct ovsdb_session *session, struct ovsdb *db,
     trigger->result = NULL;
     trigger->created = now;
     trigger->timeout_msec = LLONG_MAX;
+    trigger->read_only = read_only;
     ovsdb_trigger_try(trigger, now);
 }
 
@@ -111,7 +113,8 @@ static bool
 ovsdb_trigger_try(struct ovsdb_trigger *t, long long int now)
 {
     t->result = ovsdb_execute(t->db, t->session,
-                              t->request, now - t->created, &t->timeout_msec);
+                              t->request, t->read_only,
+                              now - t->created, &t->timeout_msec);
     if (t->result) {
         ovsdb_trigger_complete(t);
         return true;

ovsdb/trigger.h

@@ -29,11 +29,13 @@ struct ovsdb_trigger {
     struct json *result;        /* Result (null if none yet). */
     long long int created;      /* Time created. */
     long long int timeout_msec; /* Max wait duration. */
+    bool read_only;             /* Database is in read only mode. */
 };
 
 void ovsdb_trigger_init(struct ovsdb_session *, struct ovsdb *,
                         struct ovsdb_trigger *,
-                        struct json *request, long long int now);
+                        struct json *request, long long int now,
+                        bool read_only);
 void ovsdb_trigger_destroy(struct ovsdb_trigger *);
 
 bool ovsdb_trigger_is_complete(const struct ovsdb_trigger *);

tests/ovsdb-execution.at

@@ -123,6 +123,45 @@ EOF
 ]
 m4_divert_pop([PREPARE_TESTS])
 
+#
+# OVSDB_CHECK_EXECUTION_RO(TITLE, SCHEMA, TRANSACTIONS, OUTPUT, [KEYWORDS])
+#
+# Runs "test-ovsdb execute-readonly" with the given SCHEMA and each of the
+# TRANSACTIONS (which should be a quoted list of quoted strings).
+#
+# Checks that the overall output is OUTPUT, but UUIDs in the output
+# are replaced by markers of the form <N> where N is a number.  The
+# first unique UUID is replaced by <0>, the next by <1>, and so on.
+# If a given UUID appears more than once it is always replaced by the
+# same marker.
+#
+# TITLE is provided to AT_SETUP and KEYWORDS to AT_KEYWORDS.
+m4_define([OVSDB_CHECK_EXECUTION_RO],
+  [AT_SETUP([$1])
+   AT_KEYWORDS([ovsdb execute execution positive $5])
+   AT_CHECK([test-ovsdb execute-readonly "`$2`" m4_foreach([txn], [$3],
+                                         [ 'txn'])], [0], [stdout], [])
+   AT_CHECK([${PERL} $srcdir/uuidfilt.pl stdout], [0], [$4])
+   AT_CLEANUP])
+
+OVSDB_CHECK_EXECUTION_RO([block insert on read only DB],
+  [ordinal_schema],
+  [[[["ordinals",
+      {"op": "insert",
+       "table": "ordinals",
+       "row": {}}]]]],
+  [[[{"details":"insert operation not allowed when database server is in read only mode","error":"not allowed"}]
+]])
+
+OVSDB_CHECK_EXECUTION_RO([allow select on read only DB],
+  [ordinal_schema],
+  [[[["ordinals",
+      {"op": "select",
+       "table": "ordinals",
+       "where": []}]]]],
+  [[[{"rows":[]}]
+]])
+
 # OVSDB_CHECK_EXECUTION(TITLE, SCHEMA, TRANSACTIONS, OUTPUT, [KEYWORDS])
 #
 # Runs "test-ovsdb execute" with the given SCHEMA and each of the

tests/ovsdb-server.at

@@ -1222,8 +1222,19 @@ AT_CHECK([ovsdb-client transact unix:db.sock \
    "row": {"number": 0, "name": "zero"}}]]'], [0], [stdout], [ignore],
 [test ! -e pid || kill `cat pid`; test ! -e pid2 || kill `cat pid2`])
 
+dnl Make sure the transaction shows up in db2. This also tests the back up server
+dnl can be read.
 OVS_WAIT_UNTIL([ovsdb-client dump unix:db2.sock | grep zero])
 
+dnl The backup server does not accept any write transaction
+AT_CHECK([ovsdb-client transact unix:db2.sock \
+'[["mydb",
+  {"op": "insert",
+   "table": "b",
+   "row": {"number": 1, "name": "one"}}]]'], [0],
+   [[[{"details":"insert operation not allowed when database server is in read only mode","error":"not allowed"}]]
+])
+
 AT_CHECK([ovs-appctl -t "`pwd`"/unixctl2 ovsdb-server/disconnect-active-ovsdb-server], [0], [ignore], [ignore],
         [test ! -e pid || kill `cat pid`; test ! -e pid2 || kill `cat pid2`])
 
@@ -1255,6 +1266,29 @@ AT_CHECK([${PERL} $srcdir/uuidfilt.pl output], [0], [7,9c7,8
 > ----- ---- ------
 ], [ignore], [test ! -e pid || kill `cat pid`; test ! -e pid2 || kill `cat pid2`])
 
+
+dnl The backup server now become active, and can accept write transactions.
+AT_CHECK([ovsdb-client transact unix:db2.sock \
+'[["mydb",
+  {"op": "insert",
+   "table": "b",
+   "row": {"number": 1, "name": "one"}}]]'], [0], [stdout], [ignore],
+[test ! -e pid || kill `cat pid`; test ! -e pid2 || kill `cat pid2`])
+
+AT_CHECK([ovsdb-client dump unix:db2.sock], [0], [stdout])
+cat stdout > output
+
+AT_CHECK([${PERL} $srcdir/uuidfilt.pl output], [0], [a table
+_uuid                                name number
+------------------------------------ ---- ------
+<0> zero 0     @&t@
+
+b table
+_uuid                                name number
+------------------------------------ ---- ------
+<1> one  1     @&t@
+])
+
 OVSDB_SERVER_SHUTDOWN
 OVSDB_SERVER_SHUTDOWN2
 AT_CLEANUP

tests/test-ovsdb.c

@@ -195,6 +195,9 @@ usage(void)
            "  execute SCHEMA TRANSACTION...\n"
            "    executes each TRANSACTION on an initially empty database\n"
            "    the specified SCHEMA\n"
+           "  execute-readonly SCHEMA TRANSACTION...\n"
+           "    same as execute, except the TRANSACTION will be executed\n"
+           "    against the database server that is in read only mode\n"
            "  trigger SCHEMA TRANSACTION...\n"
            "    executes each TRANSACTION on an initially empty database\n"
            "    the specified SCHEMA.   A TRANSACTION of the form\n"
@@ -1402,7 +1405,7 @@ do_parse_schema(struct ovs_cmdl_context *ctx)
 }
 
 static void
-do_execute(struct ovs_cmdl_context *ctx)
+do_execute__(struct ovs_cmdl_context *ctx, bool ro)
 {
     struct ovsdb_schema *schema;
     struct json *json;
@@ -1420,7 +1423,7 @@ do_execute(struct ovs_cmdl_context *ctx)
         char *s;
 
         params = parse_json(ctx->argv[i]);
-        result = ovsdb_execute(db, NULL, params, 0, NULL);
+        result = ovsdb_execute(db, NULL, params, ro,  0, NULL);
         s = json_to_string(result, JSSF_SORT);
         printf("%s\n", s);
         free(s);
@@ -1431,6 +1434,18 @@ do_execute(struct ovs_cmdl_context *ctx)
     ovsdb_destroy(db);
 }
 
+static void
+do_execute_ro(struct ovs_cmdl_context *ctx)
+{
+    do_execute__(ctx, true);
+}
+
+static void
+do_execute(struct ovs_cmdl_context *ctx)
+{
+    do_execute__(ctx, false);
+}
+
 struct test_trigger {
     struct ovsdb_trigger trigger;
     int number;
@@ -1486,7 +1501,7 @@ do_trigger(struct ovs_cmdl_context *ctx)
             json_destroy(params);
         } else {
             struct test_trigger *t = xmalloc(sizeof *t);
-            ovsdb_trigger_init(&session, db, &t->trigger, params, now);
+            ovsdb_trigger_init(&session, db, &t->trigger, params, now, false);
             t->number = number++;
             if (ovsdb_trigger_is_complete(&t->trigger)) {
                 do_trigger_dump(t, now, "immediate");
@@ -2729,6 +2744,7 @@ static struct ovs_cmdl_command all_commands[] = {
     { "transact", NULL, 1, INT_MAX, do_transact },
     { "parse-schema", NULL, 1, 1, do_parse_schema },
     { "execute", NULL, 2, INT_MAX, do_execute },
+    { "execute-readonly", NULL, 2, INT_MAX, do_execute_ro },
     { "trigger", NULL, 2, INT_MAX, do_trigger },
     { "idl", NULL, 1, INT_MAX, do_idl },
     { "idl-partial-update-map-column", NULL, 1, INT_MAX,