mmstatd.c

mmstatd包含一个 C库和服务器

/* $Id: mmstatd.c,v 1.5 2003/01/10 03:52:36 mmondor Exp $ */

/*
 * Copyright (C) 2002-2003, Matthew Mondor
 * All rights reserved.
 *
 * Redistribution and use in source and binary forms, with or without
 * modification, are permitted provided that the following conditions
 * are met:
 * 1. Redistributions of source code must retain the above copyright
 *    notice, this list of conditions and the following disclaimer.
 * 2. Redistributions in binary form must reproduce the above copyright
 *    notice, this list of conditions and the following disclaimer in the
 *    documentation and/or other materials provided with the distribution.
 * 3. All advertising materials mentioning features or use of this software
 *    must display the following acknowledgement:
 *      This product includes software written by Matthew Mondor.
 * 4. The name of Matthew Mondor may not be used to endorse or promote
 *    products derived from this software without specific prior written
 *    permission.
 *
 * THIS SOFTWARE IS PROVIDED BY MATTHEW MONDOR ``AS IS'' AND ANY EXPRESS OR
 * IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES
 * OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED.
 * IN NO EVENT SHALL MATTHEW MONDOR BE LIABLE FOR ANY DIRECT, INDIRECT,
 * INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING,
 * BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF
 * USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
 * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
 * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF
 * THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 */

/* TODO
 * - Finish client packet uid credential checking, credentials have to be
 *   passed through the unix domain socket
 * - Have endian-independant logfile and database formats
 */




/* HEADERS */

#include <sys/types.h>
#include <sys/stat.h>
#include <sys/wait.h>
#include <sys/socket.h>
#include <sys/un.h>
#include <sys/uio.h>
#include <sys/file.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <arpa/nameser.h>
#include <resolv.h>
#include <poll.h>
#include <syslog.h>
#include <signal.h>
#include <stddef.h>
#include <unistd.h>
#include <fcntl.h>
#include <stdlib.h>
#include <stdio.h>
#include <time.h>
#include <dirent.h>
#include <pwd.h>
#include <grp.h>

#include <mmtypes.h>
#include <mmlist.h>
#include <mmstring.h>
#include <mmstat.h>
#include <mmstatd.h>
#include <mmreadcfg.h>




MMCOPYRIGHT("@(#) Copyright (c) 2002-2003\n\
\tMatthew Mondor. All rights reserved.\n");
MMRCSID("$Id: mmstatd.c,v 1.5 2003/01/10 03:52:36 mmondor Exp $");




/* PROTOTYPES */

static pid_t spawn_process(int (*)(void *), void *, bool);
static void sighandler(int);
static bool checklock(char *);
static int unix_init(char *, gid_t, mode_t, int, bool, bool);
static bool log_match(const char *, const char *);
static bool writelogentries(int, struct log_entry *, int, int *, long *,
       	off_t *);
static bool readlogentry(int, struct log_entry *, int *, long *, off_t *);
static int readlogentries(int, struct log_entry *, int, int *, long *, off_t *);
static bool processlogentry(struct log_entry *, bool);
static bool processlogentries(struct log_entry *, int, bool);
/* static void debuglogentry(char, struct log_entry *); */
static void load_db(long *, off_t *);
static bool load_db_v0_0_1(FILE *, long *, long *);
static bool load_db_v0_0_2(FILE *, long *, off_t *);
static bool load_db_v0_0_3(FILE *, long *, off_t *);
static void sync_db(long, off_t, bool);
static void free_db(void);
static void recover_db(void);
static void writestats(int, char *);
static void rotatestats(char *, char *);

int main(int, char **);

static int librarian_init(void *);
static void librarian_main(int, int, int, long *, off_t *);
static int logger_init(void *);
static void logger_main(int, int, int, long *, off_t *);




/* GLOBALS */

static struct mmstat_config CONF;
static pid_t librarian_pid = -1, logger_pid = -1;
static list *key_list, *data_list;
static int pipefds[2], syncbytes = 0;
static bool run = TRUE, pipesend = TRUE, kdirection = TRUE, ddirection = TRUE;




/* FUNCTIONS */

static pid_t
spawn_process(int (*function)(void *), void *params, bool leader)
{
    pid_t pid = -1;
    int fd;

    /* Create new process */
    if (!(pid = fork())) {
	struct sigaction act;

	/* Child */
	if (leader) setsid();
	chdir("/");
	umask(0);

	/* Make sure that stdin, stdout and stderr are safe */
	if ((fd = open("/dev/null", O_RDWR)) != -1) {
	    dup2(fd, 0);
	    dup2(fd, 1);
	    dup2(fd, 2);
	    if (fd > 2)
		close(fd);
	}

	/* Setup our break handler */
	act.sa_handler = sighandler;
	act.sa_flags = SA_NOCLDWAIT;
	sigemptyset(&act.sa_mask);
	sigaction(SIGTERM, &act, NULL);
	sigaction(SIGSEGV, &act, NULL);
	sigaction(SIGPIPE, &act, NULL);

	/* Signals we want to ignore */
	signal(SIGTTOU, SIG_IGN);
	signal(SIGTTIN, SIG_IGN);
	signal(SIGTSTP, SIG_IGN);

	/* Simply call the wanted child function */
	exit(function(params));

    }

    /* Parent */
    return (pid);
}


static void
sighandler(int sig)
{
    pid_t pid = getpid();

    switch (sig) {
    case SIGTERM:
	syslog(LOG_NOTICE, "Received SIGTERM, cleaning up");
	run = FALSE;
	if (librarian_pid != -1 && librarian_pid != pid)
	    kill(librarian_pid, SIGTERM);
	if (logger_pid != -1 && logger_pid != pid)
	    kill(logger_pid, SIGTERM);
	break;
    case SIGSEGV:
	syslog(LOG_NOTICE, "Received SIGSEGV! Cleaning up");
	kill(0, SIGTERM);
	run = FALSE;
	break;
    case SIGPIPE:
	pipesend = FALSE;
	break;
    default:
	syslog(LOG_NOTICE, "Signal handler catched unexpected signal");
	break;
    }
}


/* This uses an fd which remains open in child processes, if they close it the
 * lock seems to be released automatically.
 */
static bool
checklock(char *file)
{
    int fd;

    if ((fd = open(file, O_CREAT | O_TRUNC | O_WRONLY, 0600)) != -1) {
	if (!(flock(fd, LOCK_EX | LOCK_NB))) return (TRUE);
	close(fd);
    }

    return (FALSE);
}


static int
unix_init(char *name, gid_t group, mode_t mode, int backlog, bool stream,
	bool del)
{
    int sock;
    struct sockaddr_un sau;

    if (del) unlink(name);

    /* Open public UNIX domain socket */
    if (stream) {
	if ((sock = socket(AF_LOCAL, SOCK_STREAM, 0)) != -1) {
	    mm_strncpy(sau.sun_path, name, 100);
	    sau.sun_family = AF_UNIX;
	    if (bind(sock, (struct sockaddr *)&sau, sizeof(struct sockaddr_un))
		    != -1) {
		if (!chmod(name, mode)) {
		    chown(name, -1, group);
		    if (!(listen(sock, backlog)))
			return (sock);
		    else
			syslog(LOG_NOTICE, "unix_init() - listen()");
		} else
		    syslog(LOG_NOTICE, "unix_init() - chmod()");
	    } else
		syslog(LOG_NOTICE, "unix_init() - bind()");
	    close(sock);
	} else
	    syslog(LOG_NOTICE, "unix_init() - socket()");
    } else {
	if ((sock = socket(AF_LOCAL, SOCK_DGRAM, 0)) != -1) {
	    mm_strncpy(sau.sun_path, name, 100);
	    sau.sun_family = AF_UNIX;
	    if (bind(sock, (struct sockaddr *)&sau, sizeof(struct sockaddr_un))
		    != -1) {
		if (!chmod(name, mode)) {
		    chown(name, -1, group);
		    return (sock);
		} else
		    syslog(LOG_NOTICE, "unix_init() - chmod()");
	    } else
		syslog(LOG_NOTICE, "unix_init() - bind()");
	    close(sock);
	} else
	    syslog(LOG_NOTICE, "unix_init() - socket()");
    }

    return (-1);
}


static bool
log_match(const char *str, const char *pat)
{           
    for (; *pat != '*'; pat++, str++) {
	if (!(*str)) {
	    if (*pat) return (FALSE);
	    else return (TRUE);
	}
	if(*str != *pat && *pat != '?') return (FALSE);
    }
    while (pat[1] == '*') pat++;
    do
	if (log_match(str, pat + 1)) return (TRUE);
    while (*str++);

    return (FALSE);
}


/* Writes requested log entries to specified fd, and if required automatically
 * perform logfile rotation, of course putting a mark at the end of the logfile
 * pointing to the next one continueing it. Files are rotated when reaching
 * approximately one megabyte in length. Returns TRUE on success or FALSE on
 * fatal error (eg: disk full).
 */
static bool
writelogentries(int pfd, struct log_entry *entries, int len, int *fd,
	long *lognum, off_t *logpos)
{
    ssize_t l;
    bool ok;
    char filename[256], dat[MAX_TRANSACT];

    ok = TRUE;

    l = len * sizeof(struct log_entry);

    /* First perform logfile rotation if required */
    if (*logpos + l > CONF.max_logsize) {
	struct log_entry newentry;
	int newfd;

	(*lognum)++;
	if (*lognum > 99999999) *lognum = 0;
	*logpos = 0;
	snprintf(filename, 255, "%s/%08ld.log", CONF.ENV_DIR, *lognum);
	if ((newfd = open(filename, O_WRONLY | O_CREAT | O_TRUNC, 0600))
		!= -1) {
	    fsync(newfd);
	    mm_memclr(&newentry, sizeof(struct log_entry));
	    newentry.type = STAT_NEWFILE;
	    newentry.un.newfile.lognum = *lognum;
	    if ((write(*fd, &newentry, sizeof(struct log_entry))) !=
		    sizeof(struct log_entry))
		syslog(LOG_NOTICE,
		       "writelogentry() - write(STAT_NEWFILE)");
	    fsync(*fd);
	    close(*fd);
	    *fd = newfd;
	    write(pfd, dat, 1);
	} else {
	    syslog(LOG_NOTICE, "writelogentry() - open()");
	    ok = FALSE;
	}
    }

    /* Write our new log entry */
    if (ok) {
	if (write(*fd, entries, l) == l) {
	    *logpos += l;
	    /* A trick to keep resonable physical disk sync rate */
	    syncbytes += l;
	    if (syncbytes >= CONF.SYNC_BYTES) {
		syncbytes = 0;
		fdatasync(*fd);
	    }
	    write(pfd, dat, len);
	} else {
	    syslog(LOG_NOTICE, "writelogentry() - write(STAT_*)");
	    ok = FALSE;
	}
    }

    return (ok);
}


/* This function attempts to read a log entry from specified fd, transparently
 * rotating to the new logfile when required, and updates fd and lognum via
 * provided pointers. If pfd is -1, we attempt to read an entry, and return
 * FALSE if none could be read. Otherwise we attempt to read for an entry
 * monitoring pfd for new data notification, until at least a full entry
 * could be read. We return TRUE if an entry could be read.
 */
static bool
readlogentry(int pfd, struct log_entry *entry, int *fd, long *lognum,
	off_t *logpos)
{
    char filename[256], c;
    int newfd;
    bool ok, redo;
    struct pollfd fds[] = {
	{pfd, POLLIN, 0}
    };

    redo = TRUE;
    while (redo) {
	redo = FALSE;

	ok = FALSE;
	if (pfd == -1) {
	    if ((read(*fd, entry, sizeof(struct log_entry))) ==
		    sizeof(struct log_entry))
		ok = TRUE;
	} else {
	    while (run) {
		if ((poll(fds, 1, 5000)) > 0) {
		    if (fds[0].revents & POLLIN) {
			if ((read(pfd, &c, 1)) == 1) {
			    if ((read(*fd, entry, sizeof(struct log_entry)))
				    == sizeof(struct log_entry))
				ok = TRUE;
			    else
				syslog(LOG_NOTICE,
				       "readlogentry() - partial read");
			    break;
			}
		    }
		}
	    }
	}

	if (ok) {
	    /* debuglogentry('|', entry); */
	    *logpos += sizeof(struct log_entry);
	    /* If required switch to next logfile */
	    if (entry->type == STAT_NEWFILE) {
		snprintf(filename, 255, "%s/%08ld.log", CONF.ENV_DIR,
			entry->un.newfile.lognum);
		if ((newfd = open(filename, O_RDONLY)) != -1) {
		    close(*fd);
		    *fd = newfd;
		    *logpos = 0;
		    *lognum = entry->un.newfile.lognum;
		    redo = TRUE;
		} else {
		    syslog(LOG_NOTICE, "* readlogentry() - open()");
		    ok = FALSE;
		}
	    }
	}

    }

    return (ok);
}


/* This function attempts to read at least one entry, but all entries of a
 * transaction if any is detected. If pfd is -1, we return FALSE if we
 * cannot read an entry or the whole transaction, otherwise we wait for
 * more data to be available using fdb with poll(). We only return TRUE
 * if a single entry or whole transaction could be read.
 * If STAT_NEWFILE entry is encoutered, auto rotation to the new logfile
 * is performed and fd,lognum,logpos are updated via the provided pointers,
 * transparently.
 * The STAT_TRANSACT header and footer entries are never put in the buffer,
 * this way their persistant flag is ignored.
 * IMPORTANT: Because of the way this function works to prevent additionnal
 * memory copy operations, the supplied buffer size should at least have one
 * additionnal entry than specified maximum size.
 */
static int
readlogentries(int pfd, struct log_entry *entries, int max, int *fd,
	long *lognum, off_t *logpos)
{
    struct log_entry *ptr;
    int len;
    bool transact;

    transact = FALSE;
    ptr = entries;
    len = 0;

    if (readlogentry(pfd, entries, fd, lognum, logpos)) {
	if (entries->type == STAT_TRANSACT) {
	    if (entries->un.transact.begin) transact = TRUE;
	    else {
		/* Mismatch */
		len = 0;
		syslog(LOG_NOTICE,
			"readlogentries() - Mismatched transaction start");
	    }
	} else
	    len++;
    }

    /* If we fill the buffer of max entries or if we reach EOF before end of
     * transaction marker can be found, we simply drop the whole transaction
     * and return 0. That is where we need a read-ahead buffer, and require
     * an additionnal entry.
     */
    while (transact) {
	if (len > max) {
	    len = 0;
	    syslog(LOG_NOTICE, "readlogentries() - MAX_TRANSACT exceeded");
	    break;
	}
	if (!readlogentry(pfd, ptr, fd, lognum, logpos)) {
	    len = 1;
	    break;
	} else {
	    if (ptr->type == STAT_TRANSACT) {
		if (ptr->un.transact.begin) {
		    /* Mismatch */
		    len = 0;
		    syslog(LOG_NOTICE,
			   "readlogentries() - Mismatched transaction end");
		}
		break;
	    } else {
		ptr++;
		len++;
	    }
	}
    }

    return (len);
}


/* Apply requested log entry to the live db.
 * STAT_NEWFILE or STAT_TRANSACT entries are never processed through this
 * function.
 */
static bool
processlogentry(struct log_entry *entry, bool tmp)
{
    bool ok;
    u_int64_t hash;
    enum stat_type type;

    /* debuglogentry(':', entry); */

    ok = TRUE;
    type = entry->type;

    if (tmp || entry->persistant) {
	register char *ptr;

	/* Verify if we should perform wildcard matching and perform an atomic
	 * operation on all matching keys
	 */
	for (ptr = entry->key; *ptr; ptr++)
	    if (*ptr == '?' || *ptr == '*') break;

	if (!(*ptr)) {
	    register struct key_node *knod;

	    /* Operating on a single key.
	     * Locate corresponding key in db, if any
	     */
	    hash = mm_strhash64(entry->key);
	    if (kdirection) {
		for (knod = (struct key_node *)key_list->top; knod;
			knod = (struct key_node *)knod->nod.next)
		    if (knod->hash == hash) break;
	    } else {