Synchronize data between PostgreSQL and files

Databases are great for data processing and storage. However, in many cases it is better or easier to work with data in files on a file system, some tools even cannot access the data in any other way. When a database (DB) is created in a database management system (DBMS) using a file system as its data storage, it of course uses files on the given file system to store the data. But working with those files outside of the DBMS, even for read-only access to the data stored in the DB, is practically impossible. So what can be done if some setup requires data in files while at the same time, the data processing and storage requires a use of a DB(MS)? The answer is synchronization between two storage places – a DB and files. It can either be from the DB to the files where the files are then treated as read-only for the parties working with the data, or with modifications of the files being synchronized to the DB. In the former setup, the DB is the single source of truth – the data in the files may be out of sync, but the DB has the up to date version. In the latter setup, the DB provides a backup or alternative read-only access to the data that is primarily stored in the files or the files provide an alternative write-only access to the DB. A two-way synchronization and thus a combination of read and write access in both places, the DB and the files, should be avoided because it's very hard (one could even say impossible) to properly implement mechanisms ensuring data consistency. Both between the two storages, but even in each of them alone.

Synchronizing data to JSON files

In CFEngine Enterprise, we needed to provide file-based access to host-specific data stored in our PostgreSQL database. Users define the host-specific data in our web user interface (Mission Portal), but the individual hosts expect their specific host-specific.json files on the CFEngine hub's file system. This could have been achieved by extending the web UI code to write the data to both places, but we enforce privilege separation and don't want the web UI code to have access to CFEngine policy, the part of which the host-specific data is. And because we already needed a daemon reacting to changes and events in the PostgreSQL database, we used techniques described in a previous blog post to implement a synchronization of the host-specific data between the DB and the respective JSON files. The daemon is called cf-reactor and it is written in C. Let's take a look at how it works and what it does.

Listening to notifications from PostgreSQL

The cf-reactor daemon needs to know when some JSON file with host-specific data needs to be updated with the new data from the PostgreSQL database. The data in question is in the following __cmdb table:

CREATE TABLE IF NOT EXISTS __cmdb (
  hostkey     text        PRIMARY KEY,
  value       jsonb       NOT NULL,
  updated_at  timestamptz NOT NULL DEFAULT now(),
  epoch       bigint      NOT NULL DEFAULT 0
);

cf-reactor could check and compare the last modification timestamp of each of the JSON files with the last modification timestamp of the respective row (updated_at) in the table storing the host-specific data. Another thing that it would need to do is to check if there is some new data, for hosts that had no host-specific data before. And this is actually how things used to be implemented before cf-reactor was introduced – CFEngine policy on the hub periodically compared the timestamps, updated the files that were out of date, and created new files for hosts which the user defined host-specific data for since the last check. It was a working solution, but since an agent run evaluating the policy only happens every 5 minutes (by default), it took up to 5 minutes for the changes made in the __cmdb table to propagate to the respective host-specific JSON files.

Instead of using the same approach as described above, potentially with a shorter interval for the periodic checks, cf-reactor uses the technique described in a previous blog post – asynchronous notifications sent by a trigger. Whenever a new row is added to the __cmdb table (INSERT) or an existing row is updated (UPDATE), it receives a notification through the cmdb_refresh channel and knows it needs to synchronize the host-specific.json file for the particular host (identified by the hostkey column) with the data in the database. This is done by the following SQL commands in our schema:

CREATE FUNCTION update_cmdb_data_file()
RETURNS TRIGGER AS $$
BEGIN
  EXECUTE 'NOTIFY cmdb_refresh, ''' || NEW.hostkey || '''';
  RETURN NULL; -- result is ignored since this is an AFTER trigger
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER update_cmdb_data_file
  AFTER INSERT OR UPDATE ON __cmdb
  FOR EACH ROW EXECUTE FUNCTION update_cmdb_data_file();

To be able to listen to the notifications sent by the trigger defined above and react on them, cf-reactor does this: (in pseudo-code)

conn = PQ_Connection(db="cfdb")
while not terminate:
  wait_for_new_notifications()
  while PQ_have_notifications():
    notif = PQ_next_notification()
    hostkey = notif.payload
    UpdateHostSpecificJSONFile(hostkey)

C code to work with notifications

The wait_for_new_notifications() function call above may look like cf-reactor is constantly polling PostgreSQL for new notifications creating a heavy load on the system. However, the real implementation using C actually uses the poll() function (system call) which, contradictory to its name, provides an efficient way of waiting for resources being available. Simply put, the poll() function takes a set of file descriptors plus a timeout specification and only returns when one or more of the given file descriptors are ready for the specified operation (read, write), if an error occurs, or if the given timeout expires. While waiting, the process making the poll() system call is put out of the queue of processes waiting to be executed on the CPUs and so such processes consume little to no resources.

The real code then looks similar to the following snippet (with error and result checking omitted):

#define TIMEOUT (60 * 1000) /* milliseconds */

bool terminate = false;
PGconn *conn = PQconnectdb("dbname=cfdb");
int sock = PQsocket(conn);

PGresult *res = PQexec(conn, "LISTEN cmdb_refresh");
PQclear(res);

struct pollfd sock_poll = { .fd = sock, .events = POLLIN, .revents = 0 };
int ret;
while (!terminate) {
  ret = poll(&sock_poll, 1, TIMEOUT);
  if (ret > 0) {
    if ((sock_poll.revents & (POLLERR|POLLHUP|POLLNVAL)) != 0) {
      /* some error on the socket, try to reconnect */
      PQfinish(conn);
      conn = PQconnectdb("dbname=cfdb");
      sock = PQsocket(conn);
      sock_poll.fd = sock;
      continue;
    }
    PQconsumeInput(conn);
    PGNotify *notify;
    while ((notify = PQnotifies(conn)) != NULL) {
      assert(StringEqual(notify->relname, "cmdb_refresh");
      printf("CMDB refresh request received from PID %d: '%s'\n",
              notify->be_pid, notify->extra ? notify->extra : "");
      PQfreemem(notify);
      PQconsumeInput(conn);
    }
  } else if (ret == 0) {
    /* handle timeout */
  }
  else {
    /* handle error (can be EINTR because of signal received) */
  }
}
PQfinish(conn);

Of course, instead of just using printf() to inform about the received CMDB refresh request, the real code extracts the hostkey from the notification's payload (notify->extra), executes an SQL query to get the JSON data (value) for the given host and stores the data in the respective file and directory. The data is already stored as JSON in the DB so it’s ready to be written, but this code could easily be expanded to transform the data or even combine data from multiple tables.

Timing of the events

Our previous blog post describes in which cases and when notifications are sent by triggers like the one defined above and emphasizes that the notifications are asynchronous. So synchronization of the data from the DB to files implemented using such notifications is not synchronous. What does this beautiful sentence mean in practice? First of all, there's no guarantee that when something reads the respective JSON file after a change of the data in the DB, it will get the same data. In other words, there are periods of time when the data in the DB differs from the data in the respective files. And when the data for a host is added to the table (or deleted from it), even the existence of the respective file is not guaranteed. On the other hand, as we can see in the following screencast, the timing is in reality very fast, just not guaranteed to be fast enough for real-time restrictions. For our use case in CFEngine Enterprise it is good enough because we only need the future agent runs on the respective hosts to have their host-specific data files up to date and available on the CFEngine hub.

The above screencast shows a couple things. First, when cf-reactor starts, it checks whether the data in the JSON files is up to date because some modifications of the data in the DB could be done (and were done, in this case) while it was not running. It sees that the JSON file's last modification timestamp is older than the updated_at timestamp in the DB so it updates the file contents. Then, when another modification of the data in the DB is done by changing the value of a CMDB variable in Mission Portal (web UI), it gets a notification and updates the contents of the file again. As the timestamps demonstrate, the reaction is very fast. The Updating CMDB data file… log message has the same timestamp in seconds as the updated_at timestamp in the DB and as the last modification timestamp of the respective JSON file. So the update of the file happens in the same second in which the user submits the new value. And if we look closer at the updated_at and the file's last modification timestamps we can see that the update of the file's contents happens in less than 20 ms after the change in the DB. The update is asynchronous and window of time when the file contents are out of date, is less than 20 ms, which is much more than good enough for our use case.

Timeouts

The C code example above has the following block:

} else if (ret == 0) {
  /* handle timeout */
}

A timeout from the poll() function/system call should be handled. It actually doesn't have to be handled and the timeout interval given to the poll() function can be a negative number meaning infinite timeout. However, this timeout is actually a very useful mechanism for throttling – controlling when and how often things happen. For example, if we knew that there would be many changes of the data in the DB in a short sequence, it would make more sense to only synchronize the data into the respective file after a quiet period – a period of time with no changes of the data. Or if we knew that nothing will read the data file until a specific time, we could avoid updating the file on every change of the data in the DB and only do it right before the data file is read. If the reactions to the events (notifications) are resource-intensive, they could even be held off until the system load drops below a certain level. And so on, there are many options. And since the implementation of the code handling the notifications can use an arbitrary language and run in an arbitrary environment, there are basically no limits. However, all these approaches require that the code knows that there were no notifications for some time and that's what the timeout argument given to the poll() function ensures. The value of it doesn't have to be a static hard-coded one, it can be a dynamic value set by the code processing the notifications because that code knows when it will need to continue doing things due to past notifications.

Conclusions

This blog post describes how PostgreSQL's asynchronous notifications sent by triggers can be used to synchronize data from a database into a set of JSON files. In CFEngine Enterprise, this is a job of the recently introduced cf-reactor daemon which is also responsible for other tasks, reacting to other types of events and requests. Generally, asynchronous notifications can be used for replacing mechanisms using periodic checks and updates with more event-driven implementations saving resources while making changes propagate faster, more interactively and more efficiently. We definitely plan to extend the area of tasks and mechanisms in CFEngine Enterprise using this approach to make our product better, both regarding the improved more interactive user experience, and regarding lower system resource usage.