/* $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_tspawn_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 voidsighandler(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 boolchecklock(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 intunix_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 boollog_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 boolwritelogentries(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 boolreadlogentry(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 intreadlogentries(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 boolprocesslogentry(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 {