features/changelog: changelog translator

This is the initial version of the Changelog Translator.

What is it
-----------
Goal is to capture changes performed on a GlusterFS volume.
The translator needs to be loaded on the server (bricks) and
captures changes in a plain text file inside a configured
directory path (controlled by "changelog-dir", should be
somewhere in <export>/.glusterfs/changelog by default).

Changes are classified into 3 types:
    - Data:     : TYPE-I
    - Metadata  : TYPE-II
    - Entry     : TYPE-III

Changelog file is rolled over after a certain time interval
(defauls to 60 seconds) after which a changelog is started.
The thing to be noted here is that for a time interval
(time slice) multiple changes for an inode are recorded only
once (ie. say for 100+ writes on an inode that happens within
the time slice has only a single corresponding entry in the
changelog file). That way we do not bloat up the changelog
and also save lots of writes.

Changelog Format
-----------------
TYPE-I and TYPE-II changes have the gfid on the entity on
which the operation happened. TYPE-III being a entry op
requires the parent gfid and the basename. Changelog format
has been kept to a minimal and it's upto the consumers to
do the heavy loading of figuring out deletes, renames etc..
A single changelog file records all three types of changes,
with each change starting with an identifier ("D": DATA,
"M": METADATA and "E": ENTRY). Option is provided for the
encoding type (See TUNABLES).

Consumers
----------
The only consumer as of today would be geo-replication, although
backup utilities, self-heal, bit-rot detection could be possible
consumers in the future.

CLI
----
By default, change-logging is disabled (the translator is present
in the server graph but does nothing). When enabled (via cli) each
brick starts to log the changes. There are a set of tunable that
can be used to change the translators behaviour:

- enable/disable changelog (disabled by default)
  gluster volume set <volume> changelog {on|off}

- set the logging directory (<brick>/.glusterfs/changelogs is the
  default)
  gluster volume set <volume> changelog-dir /path/to/dir

- select encoding type (binary (default) or ascii)
  gluster volume set <volume> encoding {binary|ascii}

- change the rollover time for the logs (60 secs by default)
  gluster volume set <volume> rollover-time <secs>

- when secs > 0, changelog file is not open()'d with O_SYNC flag
- and fsync is trigerred periodically every <secs> seconds.
  gluster volume set <volume> fsync-interval <secs>

features/changelog: changelog consumer library (libgfchangelog)

A shared library is provided for the consumer of the changelogs
for easy acess via APIs. Application can link against this library
and request for changelog updates. Conversion of binary logs to
human-readable ascii format is also taken care by the library which
keeps a copy of the changelog in application provided working
directory.

Change-Id: I75575fb7f1c53d2bec3dba1a329ea7bb3c628497
BUG: 847839
Original Author: Venky Shankar <vshankar@redhat.com>
Signed-off-by: Avra Sengupta <asengupt@redhat.com>
Reviewed-on: http://review.gluster.org/5127
Tested-by: Gluster Build System <jenkins@build.gluster.com>
Reviewed-by: Vijay Bellur <vbellur@redhat.com>

1 files changed, 571 insertions, 0 deletions

diff --git a/xlators/features/changelog/lib/src/gf-changelog-process.c b/xlators/features/changelog/lib/src/gf-changelog-process.c
new file mode 100644
index 000000000..df7204931
--- /dev/null
+++ b/xlators/features/changelog/lib/src/gf-changelog-process.c
@@ -0,0 +1,571 @@
+/*
+   Copyright (c) 2013 Red Hat, Inc. <http://www.redhat.com>
+   This file is part of GlusterFS.
+
+   This file is licensed to you under your choice of the GNU Lesser
+   General Public License, version 3 or any later version (LGPLv3 or
+   later), or the GNU General Public License, version 2 (GPLv2), in all
+   cases as published by the Free Software Foundation.
+*/
+
+#include <unistd.h>
+#include <pthread.h>
+
+#include "uuid.h"
+#include "globals.h"
+#include "glusterfs.h"
+
+#include "gf-changelog-helpers.h"
+
+/* from the changelog translator */
+#include "changelog-misc.h"
+
+extern int byebye;
+
+/**
+ * number of gfid records after fop number
+ */
+int nr_gfids[] = {
+        [GF_FOP_MKNOD]   = 1,
+        [GF_FOP_MKDIR]   = 1,
+        [GF_FOP_UNLINK]  = 1,
+        [GF_FOP_RMDIR]   = 1,
+        [GF_FOP_SYMLINK] = 1,
+        [GF_FOP_RENAME]  = 2,
+        [GF_FOP_LINK]    = 1,
+        [GF_FOP_CREATE]  = 1,
+};
+
+static char *
+binary_to_ascii (uuid_t uuid)
+{
+        return uuid_utoa (uuid);
+}
+
+static char *
+conv_noop (char *ptr) { return ptr; }
+
+#define VERIFY_SEPARATOR(ptr, plen, perr)       \
+        {                                       \
+                if (*(ptr + plen) != '\0') {    \
+                        perr = 1;               \
+                        break;                  \
+                }                               \
+        }
+
+#define MOVER_MOVE(mover, nleft, bytes)         \
+        {                                       \
+                mover += bytes;                 \
+                nleft -= bytes;                 \
+        }                                       \
+
+#define PARSE_GFID(mov, ptr, le, fn, perr)                      \
+        {                                                       \
+                VERIFY_SEPARATOR (mov, le, perr);               \
+                ptr = fn (mov);                                 \
+                if (!ptr) {                                     \
+                        perr = 1;                               \
+                        break;                                  \
+                }                                               \
+        }
+
+#define FILL_AND_MOVE(pt, buf, of, mo, nl, le)                          \
+        {                                                               \
+                GF_CHANGELOG_FILL_BUFFER (pt, buf, of, strlen (pt));    \
+                MOVER_MOVE (mo, nl, le);                                \
+        }
+
+
+#define PARSE_GFID_MOVE(ptr, uuid, mover, nleft, perr)          \
+        {                                                       \
+                memcpy (uuid, mover, sizeof (uuid_t));          \
+                ptr = binary_to_ascii (uuid);                   \
+                if (!ptr) {                                     \
+                        perr = 1;                               \
+                        break;                                  \
+                }                                               \
+                MOVER_MOVE (mover, nleft, sizeof (uuid_t));     \
+        }                                                       \
+
+#define LINE_BUFSIZE  3*PATH_MAX /* enough buffer for extra chars too */
+
+/**
+ * using mmap() makes parsing easy. fgets() cannot be used here as
+ * the binary gfid could contain a line-feed (0x0A), in that case fgets()
+ * would read an incomplete line and parsing would fail. using POSIX fds
+ * would result is additional code to maintain state in case of partial
+ * reads of data (where multiple entries do not fit extirely in the buffer).
+ *
+ * mmap() gives the flexibility of pointing to an offset in the file
+ * without us worrying about reading it in memory (VM does that for us for
+ * free).
+ */
+
+static int
+gf_changelog_parse_binary (xlator_t *this,
+                           gf_changelog_t *gfc, int from_fd, int to_fd,
+                           size_t start_offset, struct stat *stbuf)
+
+{
+        int     ret              = -1;
+        off_t   off              = 0;
+        off_t   nleft            = 0;
+        uuid_t  uuid             = {0,};
+        char   *ptr              = NULL;
+        char   *bname_start      = NULL;
+        char   *bname_end        = NULL;
+        char   *mover            = NULL;
+        char   *start            = NULL;
+        char    current_mover    = ' ';
+        size_t  blen             = 0;
+        int     parse_err        = 0;
+        char ascii[LINE_BUFSIZE] = {0,};
+
+        nleft = stbuf->st_size;
+
+        start = (char *) mmap (NULL, nleft,
+                               PROT_READ, MAP_PRIVATE, from_fd, 0);
+        if (!start) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "mmap() error (reason: %s)", strerror (errno));
+                goto out;
+        }
+
+        mover = start;
+
+        MOVER_MOVE (mover, nleft, start_offset);
+
+        while (nleft > 0) {
+
+                off = blen = 0;
+                ptr = bname_start = bname_end = NULL;
+
+                current_mover = *mover;
+
+                switch (current_mover) {
+                case 'D':
+                case 'M':
+                        MOVER_MOVE (mover, nleft, 1);
+                        PARSE_GFID_MOVE (ptr, uuid, mover, nleft, parse_err);
+
+                        break;
+
+                case 'E':
+                        MOVER_MOVE (mover, nleft, 1);
+                        PARSE_GFID_MOVE (ptr, uuid, mover, nleft, parse_err);
+
+                        bname_start = mover;
+                        if ( (bname_end = strchr (mover, '\n')) == NULL ) {
+                                parse_err = 1;
+                                break;
+                        }
+
+                        blen = bname_end - bname_start;
+                        MOVER_MOVE (mover, nleft, blen);
+
+                        break;
+
+                default:
+                        parse_err = 1;
+                }
+
+                if (parse_err)
+                        break;
+
+                GF_CHANGELOG_FILL_BUFFER (&current_mover, ascii, off, 1);
+                GF_CHANGELOG_FILL_BUFFER (" ", ascii, off, 1);
+                GF_CHANGELOG_FILL_BUFFER (ptr, ascii, off, strlen (ptr));
+                if (blen)
+                        GF_CHANGELOG_FILL_BUFFER (bname_start,
+                                                  ascii, off, blen);
+                GF_CHANGELOG_FILL_BUFFER ("\n", ascii, off, 1);
+
+                if (gf_changelog_write (to_fd, ascii, off) != off) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "processing binary changelog failed due to "
+                                " error in writing ascii change (reason: %s)",
+                                strerror (errno));
+                        break;
+                }
+
+                MOVER_MOVE (mover, nleft, 1);
+        }
+
+        if ( (nleft == 0) && (!parse_err))
+                ret = 0;
+
+        if (munmap (start, stbuf->st_size))
+                gf_log (this->name, GF_LOG_ERROR,
+                        "munmap() error (reason: %s)", strerror (errno));
+ out:
+        return ret;
+}
+
+/**
+ * ascii decoder:
+ *  - separate out one entry from another
+ *  - use fop name rather than fop number
+ */
+static int
+gf_changelog_parse_ascii (xlator_t *this,
+                          gf_changelog_t *gfc, int from_fd, int to_fd,
+                          size_t start_offset, struct stat *stbuf)
+{
+        int         ng            = 0;
+        int         ret           = -1;
+        int         fop           = 0;
+        int         len           = 0;
+        off_t       off           = 0;
+        off_t       nleft         = 0;
+        char       *ptr           = NULL;
+        char       *eptr          = NULL;
+        char       *start         = NULL;
+        char       *mover         = NULL;
+        int         parse_err     = 0;
+        char        current_mover = ' ';
+        char ascii[LINE_BUFSIZE]  = {0,};
+        const char *fopname       = NULL;
+
+        nleft = stbuf->st_size;
+
+        start = (char *) mmap (NULL, nleft,
+                               PROT_READ, MAP_PRIVATE, from_fd, 0);
+        if (!start) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "mmap() error (reason: %s)", strerror (errno));
+                goto out;
+        }
+
+        mover = start;
+
+        MOVER_MOVE (mover, nleft, start_offset);
+
+        while (nleft > 0) {
+                off = 0;
+                current_mover = *mover;
+
+                GF_CHANGELOG_FILL_BUFFER (&current_mover, ascii, off, 1);
+                GF_CHANGELOG_FILL_BUFFER (" ", ascii, off, 1);
+
+                switch (current_mover) {
+                case 'D':
+                case 'M':
+                        MOVER_MOVE (mover, nleft, 1);
+
+                        /* target gfid */
+                        PARSE_GFID (mover, ptr, UUID_CANONICAL_FORM_LEN,
+                                    conv_noop, parse_err);
+                        FILL_AND_MOVE(ptr, ascii, off,
+                                      mover, nleft, UUID_CANONICAL_FORM_LEN);
+                        break;
+
+                case 'E':
+                        MOVER_MOVE (mover, nleft, 1);
+
+                        /* target gfid */
+                        PARSE_GFID (mover, ptr, UUID_CANONICAL_FORM_LEN,
+                                    conv_noop, parse_err);
+                        FILL_AND_MOVE (ptr, ascii, off,
+                                       mover, nleft, UUID_CANONICAL_FORM_LEN);
+                        FILL_AND_MOVE (" ", ascii, off,
+                                       mover, nleft, 1);
+
+                        /* fop */
+                        len = strlen (mover);
+                        VERIFY_SEPARATOR (mover, len, parse_err);
+
+                        fop = atoi (mover);
+                        if ( (fopname = gf_fop_list[fop]) == NULL) {
+                                parse_err = 1;
+                                break;
+                        }
+
+                        MOVER_MOVE (mover, nleft, len);
+
+                        len = strlen (fopname);
+                        GF_CHANGELOG_FILL_BUFFER (fopname, ascii, off, len);
+
+                        /* pargfid + bname */
+                        ng = nr_gfids[fop];
+                        while (ng-- > 0) {
+                                MOVER_MOVE (mover, nleft, 1);
+                                len = strlen (mover);
+                                GF_CHANGELOG_FILL_BUFFER (" ", ascii, off, 1);
+
+                                PARSE_GFID (mover, ptr, len,
+                                            conv_noop, parse_err);
+                                eptr = calloc (3, strlen (ptr));
+                                if (!eptr) {
+                                        parse_err = 1;
+                                        break;
+                                }
+
+                                gf_rfc3986_encode ((unsigned char *) ptr,
+                                                   eptr, gfc->rfc3986);
+                                FILL_AND_MOVE (eptr, ascii, off,
+                                               mover, nleft, len);
+                                free (eptr);
+                        }
+
+                        break;
+                default:
+                        parse_err = 1;
+                }
+
+                if (parse_err)
+                        break;
+
+                GF_CHANGELOG_FILL_BUFFER ("\n", ascii, off, 1);
+
+                if (gf_changelog_write (to_fd, ascii, off) != off) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "processing ascii changelog failed due to "
+                                " wrror in writing change (reason: %s)",
+                                strerror (errno));
+                        break;
+                }
+
+                MOVER_MOVE (mover, nleft, 1);
+
+        }
+
+        if ( (nleft == 0) && (!parse_err))
+                ret = 0;
+
+        if (munmap (start, stbuf->st_size))
+                gf_log (this->name, GF_LOG_ERROR,
+                        "munmap() error (reason: %s)", strerror (errno));
+
+ out:
+        return ret;
+}
+
+#define COPY_BUFSIZE  8192
+static int
+gf_changelog_copy (xlator_t *this, int from_fd, int to_fd)
+{
+        ssize_t size                  = 0;
+        char   buffer[COPY_BUFSIZE+1] = {0,};
+
+        while (1) {
+                size = read (from_fd, buffer, COPY_BUFSIZE);
+                if (size <= 0)
+                        break;
+
+                if (gf_changelog_write (to_fd,
+                                        buffer, size) != size) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "error processing ascii changlog");
+                        size = -1;
+                        break;
+                }
+        }
+
+        return (size < 0 ? -1 : 0);
+}
+
+static int
+gf_changelog_decode (xlator_t *this, gf_changelog_t *gfc, int from_fd,
+                     int to_fd, struct stat *stbuf, int *zerob)
+{
+        int    ret        = -1;
+        int    encoding   = -1;
+        size_t elen       = 0;
+        char buffer[1024] = {0,};
+
+        CHANGELOG_GET_ENCODING (from_fd, buffer, 1024, encoding, elen);
+        if (encoding == -1) /* unknown encoding */
+                goto out;
+
+        if (!CHANGELOG_VALID_ENCODING (encoding))
+                goto out;
+
+        if (elen == stbuf->st_size) {
+                *zerob = 1;
+                goto out;
+        }
+
+        /**
+         * start processing after the header
+         */
+        lseek (from_fd, elen, SEEK_SET);
+
+        switch (encoding) {
+        case CHANGELOG_ENCODE_BINARY:
+                /**
+                 * this ideally should have been a part of changelog-encoders.c
+                 * (ie. part of the changelog translator).
+                 */
+                ret = gf_changelog_parse_binary (this, gfc, from_fd,
+                                                 to_fd, elen, stbuf);
+                break;
+
+        case CHANGELOG_ENCODE_ASCII:
+                ret = gf_changelog_parse_ascii (this, gfc, from_fd,
+                                                to_fd, elen, stbuf);
+                break;
+        default:
+                ret = gf_changelog_copy (this, from_fd, to_fd);
+        }
+
+ out:
+        return ret;
+}
+
+static int
+gf_changelog_consume (xlator_t *this, gf_changelog_t *gfc, char *from_path)
+{
+        int         ret        = -1;
+        int         fd1        = 0;
+        int         fd2        = 0;
+        int         zerob      = 0;
+        struct stat stbuf      = {0,};
+        char dest[PATH_MAX]    = {0,};
+        char to_path[PATH_MAX] = {0,};
+
+        ret = stat (from_path, &stbuf);
+        if (ret || !S_ISREG(stbuf.st_mode)) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "stat failed on changelog file: %s", from_path);
+                goto out;
+        }
+
+        fd1 = open (from_path, O_RDONLY);
+        if (fd1 < 0) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "cannot open changelog file: %s (reason: %s)",
+                        from_path, strerror (errno));
+                goto out;
+        }
+
+        (void) snprintf (to_path, PATH_MAX, "%s%s",
+                         gfc->gfc_current_dir, basename (from_path));
+        (void) snprintf (dest, PATH_MAX, "%s%s",
+                         gfc->gfc_processing_dir, basename (from_path));
+
+        fd2 = open (to_path, O_CREAT | O_TRUNC | O_RDWR,
+                    S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH);
+        if (fd2 < 0) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "cannot create ascii changelog file %s (reason %s)",
+                        to_path, strerror (errno));
+                goto close_fd;
+        } else {
+                ret = gf_changelog_decode (this, gfc, fd1,
+                                           fd2, &stbuf, &zerob);
+
+                close (fd2);
+
+                if (!ret) {
+                        /* move it to processing on a successfull
+                           decode */
+                        ret = rename (to_path, dest);
+                        if (ret)
+                                gf_log (this->name, GF_LOG_ERROR,
+                                        "error moving %s to processing dir"
+                                        " (reason: %s)", to_path,
+                                        strerror (errno));
+                }
+
+                /* remove it from .current if it's an empty file */
+                if (zerob) {
+                        ret = unlink (to_path);
+                        if (ret)
+                                gf_log (this->name, GF_LOG_ERROR,
+                                        "could not unlink %s (reason: %s",
+                                        to_path, strerror (errno));
+                }
+        }
+
+ close_fd:
+        close (fd1);
+
+ out:
+        return ret;
+}
+
+static char *
+gf_changelog_ext_change (xlator_t *this,
+                         gf_changelog_t *gfc, char *path, size_t readlen)
+{
+        int     alo = 0;
+        int     ret = 0;
+        size_t  len = 0;
+        char   *buf = NULL;
+
+        buf = path;
+        while (len < readlen) {
+                if (*buf == '\0') {
+                        alo = 1;
+                        gf_log (this->name, GF_LOG_DEBUG,
+                                "processing changelog: %s", path);
+                        ret = gf_changelog_consume (this, gfc, path);
+                }
+
+                if (ret)
+                        break;
+
+                len++; buf++;
+                if (alo) {
+                        alo = 0;
+                        path = buf;
+                }
+        }
+
+        return (ret) ? NULL : path;
+}
+
+void *
+gf_changelog_process (void *data)
+{
+        ssize_t         len      = 0;
+        ssize_t         offlen   = 0;
+        xlator_t       *this     = NULL;
+        char           *sbuf     = NULL;
+        gf_changelog_t *gfc      = NULL;
+        char from_path[PATH_MAX] = {0,};
+
+        gfc = (gf_changelog_t *) data;
+        this = gfc->this;
+
+        pthread_detach (pthread_self());
+
+        for (;;) {
+                len = gf_changelog_read_path (gfc->gfc_sockfd,
+                                              from_path + offlen,
+                                              PATH_MAX - offlen);
+                if (len < 0)
+                        continue; /* ignore it for now */
+
+                if (len == 0) { /* close() from the changelog translator */
+                        gf_log (this->name, GF_LOG_INFO, "close from changelog"
+                                " notification translator.");
+
+                        if (gfc->gfc_connretries != 1) {
+                                if (!gf_changelog_notification_init(this, gfc))
+                                        continue;
+                        }
+
+                        byebye = 1;
+                        break;
+                }
+
+                len += offlen;
+                sbuf = gf_changelog_ext_change (this, gfc, from_path, len);
+                if (!sbuf) {
+                        gf_log (this->name, GF_LOG_ERROR,
+                                "could not extract changelog filename");
+                        continue;
+                }
+
+                offlen = 0;
+                if (sbuf != (from_path + len)) {
+                        offlen = from_path + len - sbuf;
+                        memmove (from_path, sbuf, offlen);
+                }
+        }
+
+        gf_log (this->name, GF_LOG_DEBUG,
+                "byebye (%d) from processing thread...", byebye);
+        return NULL;
+}