mmstatd.c

mmstatd包含一个 C库和服务器

    if (key[KEY_SIZE - 1] == '\0') {
	/* Verify if this consists of: '*' and/or '?' pattern, absolute
	 * key name or full report request
	 */
	if (*key == '\0') {
	    /* Full statistics report request */
	    for (dnod = (struct data_node *)data_list->top; dnod && pipesend;
		    dnod = (struct data_node *)dnod->nod.next)
		write(fd, &dnod->entry, sizeof(MMSTATENT));
	    ddirection = 0;
	} else {
	    for (ptr = key; *ptr; ptr++)
		if (*ptr == '*' || *ptr == '?') break;
	    if (*ptr) {
		/* Key pattern matching report request */
		for (dnod = (struct data_node *)data_list->top;
			dnod && pipesend;
			dnod = (struct data_node *)dnod->nod.next) {
		    if (log_match(dnod->entry.key, key))
			write(fd, &dnod->entry, sizeof(MMSTATENT));
		}
		ddirection = 0;
	    } else {
		/* Absolute key report request */
		hash = mm_strhash64(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 {
		    for (knod = (struct key_node *)key_list->bottom; knod;
			    knod = (struct key_node *)knod->nod.prev)
			if (knod->hash == hash) break;
		}
		kdirection = !kdirection;
		if (knod && pipesend) {
		    dnod = knod->data;
		    write(fd, &dnod->entry, sizeof(MMSTATENT));
		}
	    }
	}
    }
}


static void
rotatestats(char *pattern, char *prefix)
{
    struct key_node *knod;
    char key[KEY_SIZE + 1];

    /* Sanity check */
    if (pattern[KEY_SIZE - 1] == '\0' && prefix[KEY_SIZE - 1] == '\0') {
	for (knod = (struct key_node *)key_list->top; knod;
		knod = (struct key_node *)knod->nod.next) {
	    if (knod->data->entry.persistant &&
		    log_match(knod->data->entry.key, pattern)) {
		snprintf(key, KEY_SIZE - 1, "%s%s", prefix,
			knod->data->entry.key);
		mm_strncpy(knod->data->entry.key, key, KEY_SIZE - 1);
		knod->hash = mm_strhash64(knod->data->entry.key);
	    }
	}
    }
}


int
main(int argc, char **argv)
{
    char *conf_file = "/etc/mmstatd.conf", *tmp;
    int ret = 0, ngids;
    uid_t uid;
    gid_t *gids;
    LONG facility;
    CRES cres;
    CARG *cargp;
    CARG cargs[] = {
	{CAT_STR, 1, 31, CAS_UNTOUCHED, "USER", CONF.USER},
	{CAT_STR, 1, 255, CAS_UNTOUCHED, "GROUPS", CONF.GROUPS},
	{CAT_STR, 1, 31, CAS_UNTOUCHED, "LOG_FACILITY", CONF.LOG_FACILITY},
	{CAT_STR, 1, 255, CAS_UNTOUCHED, "PID_FILE", CONF.PID_FILE},
	{CAT_STR, 1, 255, CAS_UNTOUCHED, "LOCK_FILE", CONF.LOCK_FILE},
	{CAT_STR, 1, 255, CAS_UNTOUCHED, "LOG_SOCKET", CONF.LOG_SOCKET},
	{CAT_STR, 1, 255, CAS_UNTOUCHED, "STAT_SOCKET", CONF.STAT_SOCKET},
	{CAT_STR, 1, 127, CAS_UNTOUCHED, "ENV_DIR", CONF.ENV_DIR},
	{CAT_STR, 1, 31, CAS_UNTOUCHED, "LOG_GROUP", CONF.LOG_GROUP},
	{CAT_STR, 1, 31, CAS_UNTOUCHED, "STAT_GROUP", CONF.STAT_GROUP},
	{CAT_VAL, 1, 99999999, CAS_UNTOUCHED, "SYNC_INTERVAL",
	    &CONF.SYNC_INTERVAL},
	{CAT_VAL, 0, 99999999, CAS_UNTOUCHED, "SYNC_BYTES",
	    &CONF.SYNC_BYTES},
	{CAT_VAL, 1, 99999999, CAS_UNTOUCHED, "MAX_LOGSIZE",
	    &CONF.MAX_LOGSIZE},
	{CAT_VAL, 1, 9999, CAS_UNTOUCHED, "STATS_RATE", &CONF.STATS_RATE},
	{CAT_VAL, 1, 9999, CAS_UNTOUCHED, "STATS_TIME", &CONF.STATS_TIME},
	{CAT_END, 0, 0, 0, NULL, NULL}
    };
    CMAP cmap[] = {
	{"LOG_AUTH", LOG_AUTH},
	{"LOG_AUTHPRIV", LOG_AUTHPRIV},
	{"LOG_CRON", LOG_CRON},
	{"LOG_DAEMON", LOG_DAEMON},
	{"LOG_FTP", LOG_FTP},
	{"LOG_LPR", LOG_LPR},
	{"LOG_MAIL", LOG_MAIL},
	{"LOG_NEWS", LOG_NEWS},
	{"LOG_SYSLOG", LOG_SYSLOG},
	{"LOG_USER", LOG_USER},
	{"LOG_UUCP", LOG_UUCP},
	{NULL, 0}
    };

    /* Set defaults */
    mm_strcpy(CONF.USER, "mmstatd");
    mm_strcpy(CONF.GROUPS, "mmstat staff");
    mm_strcpy(CONF.LOG_FACILITY, "LOG_AUTHPRIV");
    mm_strcpy(CONF.PID_FILE, "/var/mmstatd/mmstatd.pid");
    mm_strcpy(CONF.LOCK_FILE, "/var/mmstatd/mmstatd.lock");
    mm_strcpy(CONF.LOG_SOCKET, "/var/mmstatd/mmstatd_log.sock");
    mm_strcpy(CONF.STAT_SOCKET, "/var/mmstatd/mmstatd_stat.sock");
    mm_strcpy(CONF.ENV_DIR, "/var/mmstatd");
    mm_strcpy(CONF.LOG_GROUP, "mmstat");
    mm_strcpy(CONF.STAT_GROUP, "staff");
    CONF.SYNC_INTERVAL = 1800;
    CONF.SYNC_BYTES = 4096;
    CONF.MAX_LOGSIZE = 1048576;
    CONF.STATS_RATE = 5;
    CONF.STATS_TIME = 10;

    /* Read config file */
    if ((tmp = getenv("MMSTATCONF")))
	conf_file = tmp;
    if (argc == 2)
	conf_file = argv[1];
    if (!mmreadcfg(conf_file, cargs, &cres)) {
	/* Error parsing configuration file, report which */
	printf("\nError parsing /etc/mmstatd.conf\n");
	printf("Error  : %s\n", mmreadcfg_strerr(cres.CR_Err));
	if (*(cres.CR_Data)) printf("Data   : %s\n", cres.CR_Data);
	if (cres.CR_Number != -1) {
	    cargp = &cargs[cres.CR_Number];
	    printf("Keyword: %s\n", cargp->CA_KW);
	    printf("Minimum: %ld\n", cargp->CA_Min);
	    printf("Maximum: %ld\n", cargp->CA_Max);
	}
	printf("\n");
	exit(-1);
    }

    if (!mmmapstring(cmap, CONF.LOG_FACILITY, &facility)) {
	printf("\nUnknown syslog facility %s\n\n", CONF.LOG_FACILITY);
	exit(-1);
    }
    openlog(DAEMON_NAME, LOG_PID | LOG_NDELAY, facility);

#ifndef NODROPPRIVS
    if ((getuid())) {
	printf("\nOnly the super user may start this daemon\n\n");
	syslog(LOG_NOTICE, "* Only superuser can start me");
	exit(-1);
    }
#else /* NODROPPRIVS */
    if ((getuid()) == 0) {
	printf("\nCompiled with NODROPPRIVS, refusing to run as uid 0\n\n");
	syslog(LOG_NOTICE, "* NODROPPRIVS, refusing to run as uid 0");
	exit(-1);
    }
#endif /* !NODROPPRIVS */

    if (!checklock(CONF.LOCK_FILE)) {
	printf("\nmmstatd already running\n\n");
	exit(-1);
    }
    /* Post parsing */
    if ((uid = mmgetuid(CONF.USER)) == -1) {
	printf("\nUnknown USER '%s'\n\n", CONF.USER);
	exit(-1);
    }
    if (!(gids = mmgetgidarray(&ngids, CONF.GROUPS))) {
	printf("\nOne or more of following GROUPS unknown: '%s'\n\n",
		CONF.GROUPS);
	exit(-1);
    }
    if ((CONF.log_group = mmgetgid(CONF.LOG_GROUP)) == -1) {
	printf("\nUnknown LOG_GROUP '%s'\n\n", CONF.LOG_GROUP);
	exit(-1);
    }
    if ((CONF.stat_group = mmgetgid(CONF.STAT_GROUP)) == -1) {
	printf("\nUnknown STAT_GROUP '%s'\n\n", CONF.STAT_GROUP);
	exit(-1);
    }
    CONF.max_logsize = (off_t)CONF.MAX_LOGSIZE;

    /* Initialization */
    librarian_pid = logger_pid = -1;
    key_list = data_list = NULL;
    pipefds[0] = pipefds[1] = -1;

    printf("\r\n+++ %s (%s)\r\n\r\n", DAEMON_NAME, DAEMON_VERSION);

    /* Drop root privileges */
    if (mmdropprivs(uid, gids, ngids)) {
	mmfreegidarray(gids);
	/* Launch the librarian process */
	if ((librarian_pid = spawn_process(librarian_init, NULL, TRUE))
		== -1) {
	    syslog(LOG_NOTICE, "* main() - spawn_process(librarian)");
	    ret = -1;
	}
    } else {
	ret = -1;
	syslog(LOG_NOTICE, "* main() - mmdropprivs()");
    }

    closelog();

    return (ret);
}


static int
librarian_init(void *args)
{
    int fd;

    syslog(LOG_NOTICE, "Librarian process started");

    /* Write PID file */
    if ((fd = open(CONF.PID_FILE, O_CREAT | O_TRUNC | O_WRONLY, 0600)) != -1) {
	char str[16];
	snprintf(str, 15, "%d", getpid());
	write(fd, str, mm_strlen(str));
	close(fd);
    } else
	syslog(LOG_NOTICE, "* librarian_init() - Can't write pid file");

    /* Perform recovery */
    recover_db();

    /* Prepare our notification pipe */
    if (!(pipe(pipefds))) {

	/* Start the logger process */
	if ((logger_pid = spawn_process(logger_init, NULL, FALSE)) != -1) {
	    char filename[256];
	    long lognum;
	    off_t logpos;
	    int ufd, lfd, max;

	    close(pipefds[1]);
	    pipefds[1] = -1;

	    if ((ufd = unix_init(CONF.STAT_SOCKET, CONF.stat_group, 0660, 16,
			    TRUE, TRUE)) != -1) {
		syslog(LOG_NOTICE, "Loading database");
		/* Because we recovered, lognum and logpos will be 0 */
		load_db(&lognum, &logpos);
		/* Make sure that we open for reading the file that the logger
		 * opened for writing
		 */
		snprintf(filename, 255, "%s/00000000.log", CONF.ENV_DIR);
		max = 10;
		while ((lfd = open(filename, O_RDONLY)) == -1 && max) {
		    sleep(1);
		    max--;
		}
		if (lfd != -1) {
		    /* Finally start main loop */
		    librarian_main(pipefds[0], ufd, lfd, &lognum, &logpos);
		    close(lfd);
		} else
		    syslog(LOG_NOTICE, "* librarian_init() - open(%s)",
			    filename);
		sync_db(lognum, logpos, FALSE);
		free_db();
		close(ufd);
		unlink(CONF.STAT_SOCKET);
	    } else
		syslog(LOG_NOTICE, "* librarian_init() - unix_init(%s)",
			CONF.STAT_SOCKET);
	} else
	    syslog(LOG_NOTICE,
		    "* librarian_init() - spawn_process(librarian)");

	close(pipefds[0]);
	pipefds[0] = -1;
    } else
	syslog(LOG_NOTICE, "* librarian_init() - pipe()");

    unlink(CONF.PID_FILE);

    /* Kill logger daemon */
    if (logger_pid != -1) {
	int status;
	if (!kill(logger_pid, SIGTERM)) waitpid(logger_pid, &status, 0);
    }

    syslog(LOG_NOTICE, "Exiting librarian");
    return (0);
}


/* Here consists of the main librarian server process. It's function consists
 * in following the logs created by the logger process in an ASYNC manner,
 * and process them, managing the database. It also performs total SYNC of the
 * database to disk at fixed intervals, cleaning up obsolete recovery logs.
 */
static void
librarian_main(int pfd, int ufd, int lfd, long *lognum, off_t *logpos)
{
    time_t otim, ttim;
    long secs, cnt;
    int len;
    struct log_entry entries[MAX_TRANSACT + 1];
    struct pollfd fds[] = {
	{pfd, POLLIN, 0},
	{ufd, POLLIN, 0},
    };

    ttim = time(NULL);
    secs = cnt = 0;
    while (run) {
	otim = time(NULL);
	if (poll(fds, 2, 60000) > 0) {
	    /* Process more log entries if any */
	    if (fds[0].revents & POLLIN) {
		if ((len = readlogentries(pfd, entries, MAX_TRANSACT, &lfd,
				lognum, logpos)))
		    processlogentries(entries, len, TRUE);
	    }
	    /* Verify if we obtain a report request connection */
	    if (fds[1].revents & POLLIN) {
		socklen_t addrl;
		struct sockaddr addr;
		int sfd;
		time_t t;
		char key[KEY_SIZE + 1], key2[KEY_SIZE + 1], c;

		/* Accept connection and send status report */
		addrl = sizeof(struct sockaddr);
		pipesend = TRUE;
		if ((sfd = accept(ufd, &addr, &addrl)) != -1) {
		    struct pollfd fds2[] = {
			{sfd, POLLIN, 0}
		    };

		    /* Make sure to drop connection immediately if rate
		     * was exceeded
		     */
		    if ((t = time(NULL)) > ttim + CONF.STATS_TIME) {
			ttim = t;
			cnt = 0;
		    }
		    if (cnt < CONF.STATS_RATE) {
			cnt++;
			if (pipesend) write(sfd, "+", 1);
			if (poll(fds2, 1, 250) == 1) {
			    if (fds2[0].revents & POLLIN) {
				if (read(sfd, &c, 1) == 1) {
				    if (c == 's') {
					if (read(sfd, key, KEY_SIZE) ==
						KEY_SIZE) {
					    shutdown(sfd, SHUT_RD);
					    if (pipesend) writestats(sfd, key);
					}
				    } else if (c == 'r') {
					if (read(sfd, key, KEY_SIZE) ==
						KEY_SIZE &&
						read(sfd, key2, KEY_SIZE) ==
						KEY_SIZE) {
					    rotatestats(key, key2);
					    /* Force immediate db sync */
					    secs += CONF.SYNC_INTERVAL;
					}
				    } else
					syslog(LOG_NOTICE,
					"librarian_main() - invalid req");
				} else
				    syslog(LOG_NOTICE,
					   "librarian_main() - libreq-read()");
			    }
			}
		    } else if (pipesend) write(sfd, "-", 1);
		    close(sfd);
		}
	    }
	}
	/* Verify if it's time for a sync */
	secs += (time(NULL) - otim);
	if (secs > CONF.SYNC_INTERVAL) {
	    secs = 0;
	    sync_db(*lognum, *logpos, FALSE);
	}
    }
}


static int
logger_init(void *args)
{
    char filename[256];
    long lognum;
    off_t logpos;
    int ufd, lfd;

    syslog(LOG_NOTICE, "Logger process started");

    close(pipefds[0]);
    pipefds[0] = -1;
    lognum = 0;
    logpos = 0;

    if ((ufd = unix_init(CONF.LOG_SOCKET, CONF.log_group, 0220, 0, FALSE,
		    TRUE)) != -1) {
	snprintf(filename, 255, "%s/00000000.log", CONF.ENV_DIR);
	if ((lfd = open(filename, O_WRONLY | O_CREAT | O_TRUNC, 0600)) != -1) {
	    fsync(lfd);
	    logger_main(pipefds[1], ufd, lfd, &lognum, &logpos);
	    close(lfd);
	} else
	    syslog(LOG_NOTICE, "logger_init() - open(%s)", filename);
	close(ufd);
	unlink(CONF.LOG_SOCKET);
    } else
	syslog(LOG_NOTICE, "logger_init() - unix_init(%s)", CONF.LOG_SOCKET);
    close(pipefds[1]);
    pipefds[1] = -1;

    syslog(LOG_NOTICE, "Exiting logger");
    return (0);
}


/* Here consists of the logger process server, using unix domain sockets.
 * It's function is to obtain single as well as transaction locked requests
 * from the various processes, and to generate recovery logs and sync them
 * to disk as soon as possible. It is of major importance that this process
 * perform all required sanity checking on the input datagrams, thus preventing
 * packet attacks resulting in undefined behavior. We do not allow the client
 * to generate STAT_NEWFILE or STAT_TRANSACT control packets, and we ensure
 * that they only send valid packet lengths, particularly valid key string
 * as well. Morover, we make sure to set the euid of the datagram originator.
 * XXX BUG: There seems to be a pretty serious problem, I seem to misunderstand
 *     the man pages about recvmsg(), cmsg, etc to obtain user credentials...
 *     even the data seems to not be right, but the size is.
 */
static void
logger_main(int pfd, int ufd, int lfd, long *lognum, off_t *logpos)
{
    uid_t uid, euid;
    int len, l, i, i2;
    enum stat_type t;
    char *tmp;
    struct log_entry aentries[MAX_TRANSACT + 2], *entries;
    struct pollfd fds[] = {
	{ufd, POLLIN, 0}
    };

    /* Prepare transaction header and footer entries, the footer will need
     * to be copied earlier in the buffer when the transaction packet is
     * smaller than MAX_TRANSACT, to allow using one write() only.
     */
    entries = &(aentries[1]);
    mm_memclr(aentries, sizeof(struct log_entry));
    aentries->type = STAT_TRANSACT;
    mm_memcpy(&(aentries[MAX_TRANSACT + 1]), aentries,
	    sizeof(struct log_entry));
    aentries->un.transact.begin = TRUE;

    while (run) {
	if (poll(fds, 1, -1) > 0) {
	    if (fds[0].revents & POLLIN) {
		/* New packet to log, may consist of a single operation or
		 * of transaction-protected atomic operations array.
		 */
		if ((len = recvfrom(ufd, entries,
				sizeof(struct log_entry) * MAX_TRANSACT,
				MSG_WAITALL, NULL, NULL)) > 0) {
		    /* XXX Eventually obtain packet sender credentials */
		    uid = euid = 0;
		    /* Perform some sanity checking, first verify packet size */
		    l = 0;
		    i2 = 1;
		    if ((len >= sizeof(struct log_entry)) &&
    			    (len <= sizeof(struct log_entry) * MAX_TRANSACT) &&
    			    ((len % sizeof(struct log_entry)) == 0)) {
			/* Now verify packet type and key C string validity
			 * XXX Should eventually use cleaner code here
			 */
			l = len / sizeof(struct log_entry);
			for (i2 = 0; i2 < l; i2++) {
			    t = entries[i2].type;
			    if (t != STAT_UPDATE && t != STAT_RESET &&
				    t != STAT_DELETE)
				break;
			    tmp = entries[i2].key;
			    for (i = 0; i < KEY_SIZE; i++)
				if (!tmp[i] || tmp[i] < 33 || tmp[i] == '%')
				    break;
			    if (i < 1 || i > KEY_SIZE - 1 ||
				    (tmp[i] && (tmp[i] < 33 || tmp[i] == '%')))
				break;
			    /* Make sure that packet originator uid cannot be
			     * spoofed
			     */
			    entries[i2].uid = euid;
			}
		    }
		    if (i2 == l) {
			/* This packet at least won't crash us, it will
			 * simply be ignored if invalid to this point.
			 * Perform some magic before writing the entry if
			 * it consists of a transaction.
			 */
			if (len == sizeof(struct log_entry)) {
			    if (!writelogentries(pfd, entries, 1, &lfd, lognum,
					logpos))
				syslog(LOG_NOTICE,
					"logger_main() - Error writing logs!");
			} else {
			    t = len / sizeof(struct log_entry);
			    if (t < MAX_TRANSACT)
				mm_memcpy(&entries[t],
					&aentries[MAX_TRANSACT + 1],
					sizeof(struct log_entry));
			    t += 2;
			    if (!writelogentries(pfd, aentries, t, &lfd,
					lognum, logpos))
				syslog(LOG_NOTICE,
				       "logger_main() - Error writing logs!");
			}
		    } else
			syslog(LOG_NOTICE,
			       "Illegal packet from uid %d, euid %d, %d bytes",
			       uid, euid, len);
		} else
		    syslog(LOG_NOTICE,
			   "logger_main() - Error reading packet!");
	    }
	}
    }
}