ddb443a1121b1a399b8885c535bd45f5c7f96440
[openvswitch] / ovsdb / file.c
1 /* Copyright (c) 2009, 2010 Nicira Networks
2  *
3  * Licensed under the Apache License, Version 2.0 (the "License");
4  * you may not use this file except in compliance with the License.
5  * You may obtain a copy of the License at:
6  *
7  *     http://www.apache.org/licenses/LICENSE-2.0
8  *
9  * Unless required by applicable law or agreed to in writing, software
10  * distributed under the License is distributed on an "AS IS" BASIS,
11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12  * See the License for the specific language governing permissions and
13  * limitations under the License.
14  */
15
16 #include <config.h>
17
18 #include "file.h"
19
20 #include <assert.h>
21 #include <errno.h>
22 #include <fcntl.h>
23 #include <unistd.h>
24
25 #include "bitmap.h"
26 #include "column.h"
27 #include "log.h"
28 #include "json.h"
29 #include "lockfile.h"
30 #include "ovsdb.h"
31 #include "ovsdb-error.h"
32 #include "row.h"
33 #include "socket-util.h"
34 #include "table.h"
35 #include "timeval.h"
36 #include "transaction.h"
37 #include "uuid.h"
38 #include "util.h"
39 #include "vlog.h"
40
41 VLOG_DEFINE_THIS_MODULE(ovsdb_file)
42
43 /* Minimum number of milliseconds between database compactions. */
44 #define COMPACT_MIN_MSEC        (10 * 60 * 1000) /* 10 minutes. */
45
46 /* Minimum number of milliseconds between trying to compact the database if
47  * compacting fails. */
48 #define COMPACT_RETRY_MSEC      (60 * 1000)      /* 1 minute. */
49
50 /* A transaction being converted to JSON for writing to a file. */
51 struct ovsdb_file_txn {
52     struct json *json;          /* JSON for the whole transaction. */
53     struct json *table_json;    /* JSON for 'table''s transaction. */
54     struct ovsdb_table *table;  /* Table described in 'table_json'.  */
55 };
56
57 static void ovsdb_file_txn_init(struct ovsdb_file_txn *);
58 static void ovsdb_file_txn_add_row(struct ovsdb_file_txn *,
59                                    const struct ovsdb_row *old,
60                                    const struct ovsdb_row *new,
61                                    const unsigned long int *changed);
62 static struct ovsdb_error *ovsdb_file_txn_commit(struct json *,
63                                                  const char *comment,
64                                                  bool durable,
65                                                  struct ovsdb_log *);
66
67 static struct ovsdb_error *ovsdb_file_open__(const char *file_name,
68                                              const struct ovsdb_schema *,
69                                              bool read_only, struct ovsdb **,
70                                              struct ovsdb_file **);
71 static struct ovsdb_error *ovsdb_file_txn_from_json(
72     struct ovsdb *, const struct json *, bool converting,
73     long long int *date, struct ovsdb_txn **);
74 static struct ovsdb_error *ovsdb_file_create(struct ovsdb *,
75                                              struct ovsdb_log *,
76                                              const char *file_name,
77                                              long long int oldest_commit,
78                                              unsigned int n_transactions,
79                                              struct ovsdb_file **filep);
80
81 /* Opens database 'file_name' and stores a pointer to the new database in
82  * '*dbp'.  If 'read_only' is false, then the database will be locked and
83  * changes to the database will be written to disk.  If 'read_only' is true,
84  * the database will not be locked and changes to the database will persist
85  * only as long as the "struct ovsdb".
86  *
87  * If 'filep' is nonnull and 'read_only' is false, then on success sets
88  * '*filep' to an ovsdb_file that represents the open file.  This ovsdb_file
89  * persists until '*dbp' is destroyed.
90  *
91  * On success, returns NULL.  On failure, returns an ovsdb_error (which the
92  * caller must destroy) and sets '*dbp' and '*filep' to NULL. */
93 struct ovsdb_error *
94 ovsdb_file_open(const char *file_name, bool read_only,
95                 struct ovsdb **dbp, struct ovsdb_file **filep)
96 {
97     return ovsdb_file_open__(file_name, NULL, read_only, dbp, filep);
98 }
99
100 /* Opens database 'file_name' with an alternate schema.  The specified 'schema'
101  * is used to interpret the data in 'file_name', ignoring the schema actually
102  * stored in the file.  Data in the file for tables or columns that do not
103  * exist in 'schema' are ignored, but the ovsdb file format must otherwise be
104  * observed, including column constraints.
105  *
106  * This function can be useful for upgrading or downgrading databases to
107  * "almost-compatible" formats.
108  *
109  * The database will not be locked.  Changes to the database will persist only
110  * as long as the "struct ovsdb".
111  *
112  * On success, stores a pointer to the new database in '*dbp' and returns a
113  * null pointer.  On failure, returns an ovsdb_error (which the caller must
114  * destroy) and sets '*dbp' to NULL. */
115 struct ovsdb_error *
116 ovsdb_file_open_as_schema(const char *file_name,
117                           const struct ovsdb_schema *schema,
118                           struct ovsdb **dbp)
119 {
120     return ovsdb_file_open__(file_name, schema, true, dbp, NULL);
121 }
122
123 static struct ovsdb_error *
124 ovsdb_file_open__(const char *file_name,
125                   const struct ovsdb_schema *alternate_schema,
126                   bool read_only, struct ovsdb **dbp,
127                   struct ovsdb_file **filep)
128 {
129     enum ovsdb_log_open_mode open_mode;
130     long long int oldest_commit;
131     unsigned int n_transactions;
132     struct ovsdb_schema *schema = NULL;
133     struct ovsdb_error *error;
134     struct ovsdb_log *log = NULL;
135     struct json *json;
136     struct ovsdb *db = NULL;
137
138     /* In read-only mode there is no ovsdb_file so 'filep' must be null. */
139     assert(!(read_only && filep));
140
141     open_mode = read_only ? OVSDB_LOG_READ_ONLY : OVSDB_LOG_READ_WRITE;
142     error = ovsdb_log_open(file_name, open_mode, -1, &log);
143     if (error) {
144         goto error;
145     }
146
147     error = ovsdb_log_read(log, &json);
148     if (error) {
149         goto error;
150     } else if (!json) {
151         error = ovsdb_io_error(EOF, "%s: database file contains no schema",
152                                file_name);
153         goto error;
154     }
155
156     if (alternate_schema) {
157         schema = ovsdb_schema_clone(alternate_schema);
158     } else {
159         error = ovsdb_schema_from_json(json, &schema);
160         if (error) {
161             json_destroy(json);
162             error = ovsdb_wrap_error(error,
163                                      "failed to parse \"%s\" as ovsdb schema",
164                                      file_name);
165             goto error;
166         }
167     }
168     json_destroy(json);
169
170     db = ovsdb_create(schema);
171     schema = NULL;
172
173     oldest_commit = LLONG_MAX;
174     n_transactions = 0;
175     while ((error = ovsdb_log_read(log, &json)) == NULL && json) {
176         struct ovsdb_txn *txn;
177         long long int date;
178
179         error = ovsdb_file_txn_from_json(db, json, alternate_schema != NULL,
180                                          &date, &txn);
181         json_destroy(json);
182         if (error) {
183             break;
184         }
185
186         n_transactions++;
187         if (date < oldest_commit) {
188             oldest_commit = date;
189         }
190
191         ovsdb_txn_commit(txn, false);
192     }
193     if (error) {
194         /* Log error but otherwise ignore it.  Probably the database just got
195          * truncated due to power failure etc. and we should use its current
196          * contents. */
197         char *msg = ovsdb_error_to_string(error);
198         VLOG_WARN("%s", msg);
199         free(msg);
200
201         ovsdb_error_destroy(error);
202     }
203
204     if (!read_only) {
205         struct ovsdb_file *file;
206
207         error = ovsdb_file_create(db, log, file_name, oldest_commit,
208                                   n_transactions, &file);
209         if (error) {
210             goto error;
211         }
212         if (filep) {
213             *filep = file;
214         }
215     } else {
216         ovsdb_log_close(log);
217     }
218
219     *dbp = db;
220     return NULL;
221
222 error:
223     *dbp = NULL;
224     if (filep) {
225         *filep = NULL;
226     }
227     ovsdb_destroy(db);
228     ovsdb_schema_destroy(schema);
229     ovsdb_log_close(log);
230     return error;
231 }
232
233 static struct ovsdb_error *
234 ovsdb_file_update_row_from_json(struct ovsdb_row *row, bool converting,
235                                 const struct json *json)
236 {
237     struct ovsdb_table_schema *schema = row->table->schema;
238     struct ovsdb_error *error;
239     struct shash_node *node;
240
241     if (json->type != JSON_OBJECT) {
242         return ovsdb_syntax_error(json, NULL, "row must be JSON object");
243     }
244
245     SHASH_FOR_EACH (node, json_object(json)) {
246         const char *column_name = node->name;
247         const struct ovsdb_column *column;
248         struct ovsdb_datum datum;
249
250         column = ovsdb_table_schema_get_column(schema, column_name);
251         if (!column) {
252             if (converting) {
253                 continue;
254             }
255             return ovsdb_syntax_error(json, "unknown column",
256                                       "No column %s in table %s.",
257                                       column_name, schema->name);
258         }
259
260         error = ovsdb_datum_from_json(&datum, &column->type, node->data, NULL);
261         if (error) {
262             return error;
263         }
264         ovsdb_datum_swap(&row->fields[column->index], &datum);
265         ovsdb_datum_destroy(&datum, &column->type);
266     }
267
268     return NULL;
269 }
270
271 static struct ovsdb_error *
272 ovsdb_file_txn_row_from_json(struct ovsdb_txn *txn, struct ovsdb_table *table,
273                              bool converting,
274                              const struct uuid *row_uuid, struct json *json)
275 {
276     const struct ovsdb_row *row = ovsdb_table_get_row(table, row_uuid);
277     if (json->type == JSON_NULL) {
278         if (!row) {
279             return ovsdb_syntax_error(NULL, NULL, "transaction deletes "
280                                       "row "UUID_FMT" that does not exist",
281                                       UUID_ARGS(row_uuid));
282         }
283         ovsdb_txn_row_delete(txn, row);
284         return NULL;
285     } else if (row) {
286         return ovsdb_file_update_row_from_json(ovsdb_txn_row_modify(txn, row),
287                                                converting, json);
288     } else {
289         struct ovsdb_error *error;
290         struct ovsdb_row *new;
291
292         new = ovsdb_row_create(table);
293         *ovsdb_row_get_uuid_rw(new) = *row_uuid;
294         error = ovsdb_file_update_row_from_json(new, converting, json);
295         if (error) {
296             ovsdb_row_destroy(new);
297         }
298
299         ovsdb_txn_row_insert(txn, new);
300
301         return error;
302     }
303 }
304
305 static struct ovsdb_error *
306 ovsdb_file_txn_table_from_json(struct ovsdb_txn *txn,
307                                struct ovsdb_table *table,
308                                bool converting, struct json *json)
309 {
310     struct shash_node *node;
311
312     if (json->type != JSON_OBJECT) {
313         return ovsdb_syntax_error(json, NULL, "object expected");
314     }
315
316     SHASH_FOR_EACH (node, json->u.object) {
317         const char *uuid_string = node->name;
318         struct json *txn_row_json = node->data;
319         struct ovsdb_error *error;
320         struct uuid row_uuid;
321
322         if (!uuid_from_string(&row_uuid, uuid_string)) {
323             return ovsdb_syntax_error(json, NULL, "\"%s\" is not a valid UUID",
324                                       uuid_string);
325         }
326
327         error = ovsdb_file_txn_row_from_json(txn, table, converting,
328                                              &row_uuid, txn_row_json);
329         if (error) {
330             return error;
331         }
332     }
333
334     return NULL;
335 }
336
337 /* Converts 'json' to an ovsdb_txn for 'db', storing the new transaction in
338  * '*txnp'.  Returns NULL if successful, otherwise an error.
339  *
340  * If 'converting' is true, then unknown table and column names are ignored
341  * (which can ease upgrading and downgrading schemas); otherwise, they are
342  * treated as errors.
343  *
344  * If successful, the date associated with the transaction, as the number of
345  * milliseconds since the epoch, is stored in '*date'.  If the transaction does
346  * not include a date, LLONG_MAX is stored. */
347 static struct ovsdb_error *
348 ovsdb_file_txn_from_json(struct ovsdb *db, const struct json *json,
349                          bool converting, long long int *date,
350                          struct ovsdb_txn **txnp)
351 {
352     struct ovsdb_error *error;
353     struct shash_node *node;
354     struct ovsdb_txn *txn;
355
356     *txnp = NULL;
357     *date = LLONG_MAX;
358
359     if (json->type != JSON_OBJECT) {
360         return ovsdb_syntax_error(json, NULL, "object expected");
361     }
362
363     txn = ovsdb_txn_create(db);
364     SHASH_FOR_EACH (node, json->u.object) {
365         const char *table_name = node->name;
366         struct json *node_json = node->data;
367         struct ovsdb_table *table;
368
369         table = shash_find_data(&db->tables, table_name);
370         if (!table) {
371             if (!strcmp(table_name, "_date")
372                 && node_json->type == JSON_INTEGER) {
373                 if (date) {
374                     *date = json_integer(node_json);
375                 }
376                 continue;
377             } else if (!strcmp(table_name, "_comment") || converting) {
378                 continue;
379             }
380
381             error = ovsdb_syntax_error(json, "unknown table",
382                                        "No table named %s.", table_name);
383             goto error;
384         }
385
386         error = ovsdb_file_txn_table_from_json(txn, table, converting,
387                                                node_json);
388         if (error) {
389             goto error;
390         }
391     }
392     *txnp = txn;
393     return NULL;
394
395 error:
396     ovsdb_txn_abort(txn);
397     return error;
398 }
399
400 static struct ovsdb_error *
401 ovsdb_file_save_copy__(const char *file_name, int locking,
402                        const char *comment, const struct ovsdb *db,
403                        struct ovsdb_log **logp)
404 {
405     const struct shash_node *node;
406     struct ovsdb_file_txn ftxn;
407     struct ovsdb_error *error;
408     struct ovsdb_log *log;
409     struct json *json;
410
411     error = ovsdb_log_open(file_name, OVSDB_LOG_CREATE, locking, &log);
412     if (error) {
413         return error;
414     }
415
416     /* Write schema. */
417     json = ovsdb_schema_to_json(db->schema);
418     error = ovsdb_log_write(log, json);
419     json_destroy(json);
420     if (error) {
421         goto exit;
422     }
423
424     /* Write data. */
425     ovsdb_file_txn_init(&ftxn);
426     SHASH_FOR_EACH (node, &db->tables) {
427         const struct ovsdb_table *table = node->data;
428         const struct ovsdb_row *row;
429
430         HMAP_FOR_EACH (row, hmap_node, &table->rows) {
431             ovsdb_file_txn_add_row(&ftxn, NULL, row, NULL);
432         }
433     }
434     error = ovsdb_file_txn_commit(ftxn.json, comment, true, log);
435
436 exit:
437     if (logp) {
438         if (!error) {
439             *logp = log;
440             log = NULL;
441         } else {
442             *logp = NULL;
443         }
444     }
445     ovsdb_log_close(log);
446     if (error) {
447         remove(file_name);
448     }
449     return error;
450 }
451
452 /* Saves a snapshot of 'db''s current contents as 'file_name'.  If 'comment' is
453  * nonnull, then it is added along with the data contents and can be viewed
454  * with "ovsdb-tool show-log".
455  *
456  * 'locking' is passed along to ovsdb_log_open() untouched. */
457 struct ovsdb_error *
458 ovsdb_file_save_copy(const char *file_name, int locking,
459                      const char *comment, const struct ovsdb *db)
460 {
461     return ovsdb_file_save_copy__(file_name, locking, comment, db, NULL);
462 }
463 \f
464 /* Replica implementation. */
465
466 struct ovsdb_file {
467     struct ovsdb_replica replica;
468     struct ovsdb *db;
469     struct ovsdb_log *log;
470     char *file_name;
471     long long int oldest_commit;
472     long long int next_compact;
473     unsigned int n_transactions;
474 };
475
476 static const struct ovsdb_replica_class ovsdb_file_class;
477
478 static struct ovsdb_error *
479 ovsdb_file_create(struct ovsdb *db, struct ovsdb_log *log,
480                   const char *file_name,
481                   long long int oldest_commit,
482                   unsigned int n_transactions,
483                   struct ovsdb_file **filep)
484 {
485     long long int now = time_msec();
486     struct ovsdb_file *file;
487     char *abs_name;
488
489     /* Use the absolute name of the file because ovsdb-server opens its
490      * database before daemonize() chdirs to "/". */
491     abs_name = abs_file_name(NULL, file_name);
492     if (!abs_name) {
493         *filep = NULL;
494         return ovsdb_io_error(0, "could not determine current "
495                               "working directory");
496     }
497
498     file = xmalloc(sizeof *file);
499     ovsdb_replica_init(&file->replica, &ovsdb_file_class);
500     file->db = db;
501     file->log = log;
502     file->file_name = abs_name;
503     file->oldest_commit = MIN(oldest_commit, now);
504     file->next_compact = file->oldest_commit + COMPACT_MIN_MSEC;
505     file->n_transactions = n_transactions;
506     ovsdb_add_replica(db, &file->replica);
507
508     *filep = file;
509     return NULL;
510 }
511
512 static struct ovsdb_file *
513 ovsdb_file_cast(struct ovsdb_replica *replica)
514 {
515     assert(replica->class == &ovsdb_file_class);
516     return CONTAINER_OF(replica, struct ovsdb_file, replica);
517 }
518
519 static bool
520 ovsdb_file_change_cb(const struct ovsdb_row *old,
521                      const struct ovsdb_row *new,
522                      const unsigned long int *changed,
523                      void *ftxn_)
524 {
525     struct ovsdb_file_txn *ftxn = ftxn_;
526     ovsdb_file_txn_add_row(ftxn, old, new, changed);
527     return true;
528 }
529
530 static struct ovsdb_error *
531 ovsdb_file_commit(struct ovsdb_replica *replica,
532                   const struct ovsdb_txn *txn, bool durable)
533 {
534     struct ovsdb_file *file = ovsdb_file_cast(replica);
535     struct ovsdb_file_txn ftxn;
536     struct ovsdb_error *error;
537
538     ovsdb_file_txn_init(&ftxn);
539     ovsdb_txn_for_each_change(txn, ovsdb_file_change_cb, &ftxn);
540     if (!ftxn.json) {
541         /* Nothing to commit. */
542         return NULL;
543     }
544
545     error = ovsdb_file_txn_commit(ftxn.json, ovsdb_txn_get_comment(txn),
546                                   durable, file->log);
547     if (error) {
548         return error;
549     }
550     file->n_transactions++;
551
552     /* If it has been at least COMPACT_MIN_MSEC millseconds since the last time
553      * we compacted (or at least COMPACT_RETRY_MSEC since the last time we
554      * tried), and if there are at least 100 transactions in the database, and
555      * if the database is at least 1 MB, then compact the database. */
556     if (time_msec() >= file->next_compact
557         && file->n_transactions >= 100
558         && ovsdb_log_get_offset(file->log) >= 10 * 1024 * 1024)
559     {
560         error = ovsdb_file_compact(file);
561         if (error) {
562             char *s = ovsdb_error_to_string(error);
563             ovsdb_error_destroy(error);
564             VLOG_WARN("%s: compacting database failed (%s), retrying in "
565                       "60 seconds", file->file_name, s);
566             free(s);
567
568             file->next_compact = time_msec() + COMPACT_RETRY_MSEC;
569         }
570     }
571
572     return NULL;
573 }
574
575 struct ovsdb_error *
576 ovsdb_file_compact(struct ovsdb_file *file)
577 {
578     struct ovsdb_log *new_log = NULL;
579     struct lockfile *tmp_lock = NULL;
580     struct ovsdb_error *error;
581     char *tmp_name = NULL;
582     char *comment = NULL;
583     int retval;
584
585     comment = xasprintf("compacting database online "
586                         "(%.3f seconds old, %u transactions, %llu bytes)",
587                         (time_msec() - file->oldest_commit) / 1000.0,
588                         file->n_transactions,
589                         (unsigned long long) ovsdb_log_get_offset(file->log));
590     VLOG_INFO("%s: %s", file->file_name, comment);
591
592     /* Commit the old version, so that we can be assured that we'll eventually
593      * have either the old or the new version. */
594     error = ovsdb_log_commit(file->log);
595     if (error) {
596         goto exit;
597     }
598
599     /* Lock temporary file. */
600     tmp_name = xasprintf("%s.tmp", file->file_name);
601     retval = lockfile_lock(tmp_name, 0, &tmp_lock);
602     if (retval) {
603         error = ovsdb_io_error(retval, "could not get lock on %s", tmp_name);
604         goto exit;
605     }
606
607     /* Remove temporary file.  (It might not exist.) */
608     if (unlink(tmp_name) < 0 && errno != ENOENT) {
609         error = ovsdb_io_error(errno, "failed to remove %s", tmp_name);
610         goto exit;
611     }
612
613     /* Save a copy. */
614     error = ovsdb_file_save_copy__(tmp_name, false, comment, file->db,
615                                    &new_log);
616     if (error) {
617         goto exit;
618     }
619
620     /* Replace original by temporary. */
621     if (rename(tmp_name, file->file_name)) {
622         error = ovsdb_io_error(errno, "failed to rename \"%s\" to \"%s\"",
623                                tmp_name, file->file_name);
624         goto exit;
625     }
626     fsync_parent_dir(file->file_name);
627
628 exit:
629     if (!error) {
630         ovsdb_log_close(file->log);
631         file->log = new_log;
632         file->oldest_commit = time_msec();
633         file->next_compact = file->oldest_commit + COMPACT_MIN_MSEC;
634         file->n_transactions = 1;
635     } else {
636         ovsdb_log_close(new_log);
637         if (tmp_lock) {
638             unlink(tmp_name);
639         }
640     }
641
642     lockfile_unlock(tmp_lock);
643     free(tmp_name);
644     free(comment);
645
646     return error;
647 }
648
649 static void
650 ovsdb_file_destroy(struct ovsdb_replica *replica)
651 {
652     struct ovsdb_file *file = ovsdb_file_cast(replica);
653
654     ovsdb_log_close(file->log);
655     free(file->file_name);
656     free(file);
657 }
658
659 static const struct ovsdb_replica_class ovsdb_file_class = {
660     ovsdb_file_commit,
661     ovsdb_file_destroy
662 };
663 \f
664 static void
665 ovsdb_file_txn_init(struct ovsdb_file_txn *ftxn)
666 {
667     ftxn->json = NULL;
668     ftxn->table_json = NULL;
669     ftxn->table = NULL;
670 }
671
672 static void
673 ovsdb_file_txn_add_row(struct ovsdb_file_txn *ftxn,
674                        const struct ovsdb_row *old,
675                        const struct ovsdb_row *new,
676                        const unsigned long int *changed)
677 {
678     struct json *row;
679
680     if (!new) {
681         row = json_null_create();
682     } else {
683         struct shash_node *node;
684
685         row = old ? NULL : json_object_create();
686         SHASH_FOR_EACH (node, &new->table->schema->columns) {
687             const struct ovsdb_column *column = node->data;
688             const struct ovsdb_type *type = &column->type;
689             unsigned int idx = column->index;
690
691             if (idx != OVSDB_COL_UUID && column->persistent
692                 && (old
693                     ? bitmap_is_set(changed, idx)
694                     : !ovsdb_datum_is_default(&new->fields[idx], type)))
695             {
696                 if (!row) {
697                     row = json_object_create();
698                 }
699                 json_object_put(row, column->name,
700                                 ovsdb_datum_to_json(&new->fields[idx], type));
701             }
702         }
703     }
704
705     if (row) {
706         struct ovsdb_table *table = new ? new->table : old->table;
707         char uuid[UUID_LEN + 1];
708
709         if (table != ftxn->table) {
710             /* Create JSON object for transaction overall. */
711             if (!ftxn->json) {
712                 ftxn->json = json_object_create();
713             }
714
715             /* Create JSON object for transaction on this table. */
716             ftxn->table_json = json_object_create();
717             ftxn->table = table;
718             json_object_put(ftxn->json, table->schema->name, ftxn->table_json);
719         }
720
721         /* Add row to transaction for this table. */
722         snprintf(uuid, sizeof uuid,
723                  UUID_FMT, UUID_ARGS(ovsdb_row_get_uuid(new ? new : old)));
724         json_object_put(ftxn->table_json, uuid, row);
725     }
726 }
727
728 static struct ovsdb_error *
729 ovsdb_file_txn_commit(struct json *json, const char *comment,
730                       bool durable, struct ovsdb_log *log)
731 {
732     struct ovsdb_error *error;
733
734     if (!json) {
735         json = json_object_create();
736     }
737     if (comment) {
738         json_object_put_string(json, "_comment", comment);
739     }
740     json_object_put(json, "_date", json_integer_create(time_wall()));
741
742     error = ovsdb_log_write(log, json);
743     json_destroy(json);
744     if (error) {
745         return ovsdb_wrap_error(error, "writing transaction failed");
746     }
747
748     if (durable) {
749         error = ovsdb_log_commit(log);
750         if (error) {
751             return ovsdb_wrap_error(error, "committing transaction failed");
752         }
753     }
754
755     return NULL;
756 }