1 files changed, 515 insertions, 0 deletions

diff --git a/xlators/features/changelog/lib/src/gf-changelog.c b/xlators/features/changelog/lib/src/gf-changelog.c
new file mode 100644
index 000000000..ca8e373e7
--- /dev/null
+++ b/xlators/features/changelog/lib/src/gf-changelog.c
@@ -0,0 +1,515 @@
+/*
+   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 <errno.h>
+#include <dirent.h>
+#include <stddef.h>
+#include <sys/types.h>
+#include <sys/socket.h>
+#include <sys/un.h>
+
+#ifndef _GNU_SOURCE
+#define _GNU_SOURCE
+#endif
+#include <string.h>
+
+#include "globals.h"
+#include "glusterfs.h"
+#include "logging.h"
+
+#include "gf-changelog-helpers.h"
+
+/* from the changelog translator */
+#include "changelog-misc.h"
+#include "changelog-mem-types.h"
+
+int byebye = 0;
+
+static void
+gf_changelog_cleanup (gf_changelog_t *gfc)
+{
+        /* socket */
+        if (gfc->gfc_sockfd != -1)
+                close (gfc->gfc_sockfd);
+        /* tracker fd */
+        if (gfc->gfc_fd != -1)
+                close (gfc->gfc_fd);
+        /* processing dir */
+        if (gfc->gfc_dir)
+                closedir (gfc->gfc_dir);
+
+        if (gfc->gfc_working_dir)
+                free (gfc->gfc_working_dir); /* allocated by realpath */
+}
+
+void
+__attribute__ ((constructor)) gf_changelog_ctor (void)
+{
+        glusterfs_ctx_t *ctx = NULL;
+
+        ctx = glusterfs_ctx_new ();
+        if (!ctx)
+                return;
+
+        if (glusterfs_globals_init (ctx)) {
+                free (ctx);
+                ctx = NULL;
+                return;
+        }
+
+        THIS->ctx = ctx;
+}
+
+void
+__attribute__ ((destructor)) gf_changelog_dtor (void)
+{
+        xlator_t        *this = NULL;
+        glusterfs_ctx_t *ctx  = NULL;
+        gf_changelog_t  *gfc  = NULL;
+
+        this = THIS;
+        if (!this)
+                return;
+
+        ctx = this->ctx;
+        gfc = this->private;
+
+        if (gfc) {
+                gf_changelog_cleanup (gfc);
+                GF_FREE (gfc);
+        }
+
+        if (ctx) {
+                pthread_mutex_destroy (&ctx->lock);
+                free (ctx);
+                ctx = NULL;
+        }
+}
+
+
+static int
+gf_changelog_open_dirs (gf_changelog_t *gfc)
+{
+        int  ret                    = -1;
+        DIR *dir                    = NULL;
+        int  tracker_fd             = 0;
+        char tracker_path[PATH_MAX] = {0,};
+
+        (void) snprintf (gfc->gfc_current_dir, PATH_MAX,
+                         "%s/"GF_CHANGELOG_CURRENT_DIR"/",
+                         gfc->gfc_working_dir);
+        ret = mkdir_p (gfc->gfc_current_dir, 0600, _gf_false);
+        if (ret)
+                goto out;
+
+        (void) snprintf (gfc->gfc_processed_dir, PATH_MAX,
+                         "%s/"GF_CHANGELOG_PROCESSED_DIR"/",
+                         gfc->gfc_working_dir);
+        ret = mkdir_p (gfc->gfc_processed_dir, 0600, _gf_false);
+        if (ret)
+                goto out;
+
+        (void) snprintf (gfc->gfc_processing_dir, PATH_MAX,
+                         "%s/"GF_CHANGELOG_PROCESSING_DIR"/",
+                         gfc->gfc_working_dir);
+        ret = mkdir_p (gfc->gfc_processing_dir, 0600, _gf_false);
+        if (ret)
+                goto out;
+
+        dir = opendir (gfc->gfc_processing_dir);
+        if (!dir) {
+                gf_log ("", GF_LOG_ERROR,
+                        "opendir() error [reason: %s]", strerror (errno));
+                goto out;
+        }
+
+        gfc->gfc_dir = dir;
+
+        (void) snprintf (tracker_path, PATH_MAX,
+                         "%s/"GF_CHANGELOG_TRACKER, gfc->gfc_working_dir);
+
+        tracker_fd = open (tracker_path, O_CREAT | O_APPEND | O_RDWR,
+                           S_IRUSR | S_IWUSR | S_IRGRP | S_IROTH);
+        if (tracker_fd < 0) {
+                closedir (gfc->gfc_dir);
+                ret = -1;
+                goto out;
+        }
+
+        gfc->gfc_fd = tracker_fd;
+        ret = 0;
+ out:
+        return ret;
+}
+
+int
+gf_changelog_notification_init (xlator_t *this, gf_changelog_t *gfc)
+{
+        int                ret    = 0;
+        int                len    = 0;
+        int                tries  = 0;
+        int                sockfd = 0;
+        struct sockaddr_un remote;
+
+        this = gfc->this;
+
+        if (gfc->gfc_sockfd != -1) {
+                gf_log (this->name, GF_LOG_INFO,
+                        "Reconnecting...");
+                close (gfc->gfc_sockfd);
+        }
+
+        sockfd = socket (AF_UNIX, SOCK_STREAM, 0);
+        if (sockfd < 0) {
+                ret = -1;
+                goto out;
+        }
+
+        CHANGELOG_MAKE_SOCKET_PATH (gfc->gfc_brickpath,
+                                    gfc->gfc_sockpath, PATH_MAX);
+        gf_log (this->name, GF_LOG_INFO,
+                "connecting to changelog socket: %s (brick: %s)",
+                gfc->gfc_sockpath, gfc->gfc_brickpath);
+
+        remote.sun_family = AF_UNIX;
+        strcpy (remote.sun_path, gfc->gfc_sockpath);
+
+        len = strlen (remote.sun_path) + sizeof (remote.sun_family);
+
+        while (tries < gfc->gfc_connretries) {
+                gf_log (this->name, GF_LOG_WARNING,
+                        "connection attempt %d/%d...",
+                        tries + 1, gfc->gfc_connretries);
+
+                /* initiate a connect */
+                if (connect (sockfd, (struct sockaddr *) &remote, len) == 0) {
+                        gfc->gfc_sockfd = sockfd;
+                        break;
+                }
+
+                tries++;
+                sleep (2);
+        }
+
+        if (tries == gfc->gfc_connretries) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "could not connect to changelog socket!"
+                        " bailing out...");
+                ret = -1;
+        } else
+                gf_log (this->name, GF_LOG_INFO,
+                        "connection successful");
+
+ out:
+        return ret;
+}
+
+int
+gf_changelog_done (char *file)
+{
+        int             ret    = -1;
+        char           *buffer = NULL;
+        xlator_t       *this   = NULL;
+        gf_changelog_t *gfc    = NULL;
+        char to_path[PATH_MAX] = {0,};
+
+        errno = EINVAL;
+
+        this = THIS;
+        if (!this)
+                goto out;
+
+        gfc = (gf_changelog_t *) this->private;
+        if (!gfc)
+                goto out;
+
+        if (!file || !strlen (file))
+                goto out;
+
+        /* make sure 'file' is inside ->gfc_working_dir */
+        buffer = realpath (file, NULL);
+        if (!buffer)
+                goto out;
+
+        if (strncmp (gfc->gfc_working_dir,
+                     buffer, strlen (gfc->gfc_working_dir)))
+                goto out;
+
+        (void) snprintf (to_path, PATH_MAX, "%s%s",
+                         gfc->gfc_processed_dir, basename (buffer));
+        gf_log (this->name, GF_LOG_DEBUG,
+                "moving %s to processed directory", file);
+        ret = rename (buffer, to_path);
+        if (ret) {
+                gf_log (this->name, GF_LOG_ERROR,
+                        "cannot move %s to %s (reason: %s)",
+                        file, to_path, strerror (errno));
+                goto out;
+        }
+
+        ret = 0;
+
+ out:
+        if (buffer)
+                free (buffer); /* allocated by realpath() */
+        return ret;
+}
+
+/**
+ * @API
+ *  for a set of changelogs, start from the begining
+ */
+int
+gf_changelog_start_fresh ()
+{
+        xlator_t *this = NULL;
+        gf_changelog_t *gfc = NULL;
+
+        this = THIS;
+        if (!this)
+                goto out;
+
+        errno = EINVAL;
+
+        gfc = (gf_changelog_t *) this->private;
+        if (!gfc)
+                goto out;
+
+        if (gf_ftruncate (gfc->gfc_fd, 0))
+                goto out;
+
+        return 0;
+
+ out:
+        return -1;
+}
+
+/**
+ * @API
+ * return the next changelog file entry. zero means all chanelogs
+ * consumed.
+ */
+ssize_t
+gf_changelog_next_change (char *bufptr, size_t maxlen)
+{
+        ssize_t         size       = 0;
+        int             tracker_fd = 0;
+        xlator_t       *this       = NULL;
+        gf_changelog_t *gfc        = NULL;
+        char buffer[PATH_MAX]      = {0,};
+
+        errno = EINVAL;
+
+        this = THIS;
+        if (!this)
+                goto out;
+
+        gfc = (gf_changelog_t *) this->private;
+        if (!gfc)
+                goto out;
+
+        tracker_fd = gfc->gfc_fd;
+
+        size = gf_readline (tracker_fd, buffer, maxlen);
+        if (size < 0)
+                goto out;
+        if (size == 0)
+                return 0;
+
+        memcpy (bufptr, buffer, size - 1);
+        *(buffer + size) = '\0';
+
+        return size;
+
+ out:
+        return -1;
+}
+
+/**
+ * @API
+ *  gf_changelog_scan() - scan and generate a list of change entries
+ *
+ * calling this api multiple times (without calling gf_changlog_done())
+ * would result new changelogs(s) being refreshed in the tracker file.
+ * This call also acts as a cancellation point for the consumer.
+ */
+ssize_t
+gf_changelog_scan ()
+{
+        int             ret        = 0;
+        int             tracker_fd = 0;
+        size_t          len        = 0;
+        size_t          off        = 0;
+        xlator_t       *this       = NULL;
+        size_t          nr_entries = 0;
+        gf_changelog_t *gfc        = NULL;
+        struct dirent  *entryp     = NULL;
+        struct dirent  *result     = NULL;
+        char buffer[PATH_MAX]      = {0,};
+
+        this = THIS;
+        if (!this)
+                goto out;
+
+        gfc = (gf_changelog_t *) this->private;
+        if (!gfc)
+                got