ovsdb: Allow conversion records with no data in a clustered storage.

If the schema with no data was read from the clustered storage, it
should mean a database conversion request.  In general, we can get:

1. Just data --> Transaction record.
2. Schema + Data --> Database conversion or raft snapshot install.
3. Just schema --> New.  Database conversion request.

We cannot distinguish between conversion and snapshot installation
request in the current implementation, so we will keep handling
conversion with data in the same way as before, i.e. if data is
provided, we should use it.

ovsdb-tool is updated to handle this record type as well while
converting cluster to standalone.

This change doesn't introduce a way for such records to appear in
the database.  That will be added in the future commits targeting
conversion speed increase.

Reviewed-by: Simon Horman <simon.horman@corigine.com>
Acked-by: Dumitru Ceara <dceara@redhat.com>
Signed-off-by: Ilya Maximets <i.maximets@ovn.org>

1 files changed, 18 insertions, 4 deletions

diff --git a/ovsdb/relay.c b/ovsdb/relay.c
index 9ff6ed8f3..94ffe01e5 100644
--- a/ovsdb/relay.c
+++ b/ovsdb/relay.c
@@ -301,6 +301,8 @@ static void
 ovsdb_relay_parse_update(struct relay_ctx *ctx,
                          const struct ovsdb_cs_update_event *update)
 {
+    struct ovsdb_error *error = NULL;
+
     if (!ctx->db) {
         return;
     }
@@ -308,15 +310,27 @@ ovsdb_relay_parse_update(struct relay_ctx *ctx,
     if (update->monitor_reply && ctx->new_schema) {
         /* There was a schema change.  Updating a database with a new schema
          * before processing monitor reply with the new data. */
-        ctx->schema_change_cb(ctx->db, ctx->new_schema,
-                              ctx->schema_change_aux);
+        error = ctx->schema_change_cb(ctx->db, ctx->new_schema, false,
+                                      ctx->schema_change_aux);
+        if (error) {
+            /* Should never happen, but handle this case anyway. */
+            static struct vlog_rate_limit rl = VLOG_RATE_LIMIT_INIT(1, 5);
+            char *s = ovsdb_error_to_string_free(error);
+
+            VLOG_ERR_RL(&rl, "%s", s);
+            free(s);
+
+            ovsdb_cs_flag_inconsistency(ctx->cs);
+            return;
+        }
         ovsdb_schema_destroy(ctx->new_schema);
         ctx->new_schema = NULL;
     }
 
     struct ovsdb_cs_db_update *du;
-    struct ovsdb_error *error = ovsdb_cs_parse_db_update(update->table_updates,
-                                                         update->version, &du);
+
+    error = ovsdb_cs_parse_db_update(update->table_updates,
+                                     update->version, &du);
     if (!error) {
         if (update->clear) {
             error = ovsdb_relay_clear(ctx->db);