mbtk/mbtk_logd/syslog_read.c - T108_Public - Gitiles

 /*
  * Copyright (C) 2013 Felix Fietkau <nbd@openwrt.org>
  * Copyright (C) 2013 John Crispin <blogic@openwrt.org>
  *
  * This program is free software; you can redistribute it and/or modify
  * it under the terms of the GNU Lesser General Public License version 2.1
  * as published by the Free Software Foundation
  *
  * This program is distributed in the hope that it will be useful,
  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  * GNU General Public License for more details.
  */

 #include <sys/types.h>
 #include <sys/stat.h>

 #include <fcntl.h>
 #include <time.h>
 #include <stdio.h>
 #include <unistd.h>
 #include <sys/types.h>
 #include <sys/socket.h>

 #define SYSLOG_NAMES
 #include <syslog.h>

 #include <libubox/ustream.h>
 #include <libubox/blobmsg_json.h>
 #include <libubox/usock.h>
 #include <libubox/uloop.h>
 #include "libubus.h"
 #include "syslog.h"
 #include "log_config.h"

 enum {
 	LOG_STDOUT,
 	LOG_FILE,
 	LOG_NET,
 };

 enum {
 	LOG_MSG,
 	LOG_ID,
 	LOG_PRIO,
 	LOG_SOURCE,
 	LOG_TIME,
 	__LOG_MAX
 };

 static const struct blobmsg_policy log_policy[] = {
 	[LOG_MSG] = { .name = "msg", .type = BLOBMSG_TYPE_STRING },
 	[LOG_ID] = { .name = "id", .type = BLOBMSG_TYPE_INT32 },
 	[LOG_PRIO] = { .name = "priority", .type = BLOBMSG_TYPE_INT32 },
 	[LOG_SOURCE] = { .name = "source", .type = BLOBMSG_TYPE_INT32 },
 	[LOG_TIME] = { .name = "time", .type = BLOBMSG_TYPE_INT64 },
 };

 static struct uloop_timeout retry;
 static struct uloop_fd sender;
 static const char *log_file, *log_ip, *log_port, *log_prefix, *pid_file, *hostname;
 static int log_type = LOG_STDOUT;
 static int log_size = 1 * 1024 * 1024, log_udp, log_follow = 0;
 static struct file_list_t file_list;
 static struct filter_list_t *filter_log = NULL;
 static char tmp_log[48] = {0};

 static const char* getcodetext(int value, CODE *codetable) {
 	CODE *i;

 	if (value >= 0)
 		for (i = codetable; i->c_val != -1; i++)
 			if (i->c_val == value)
 				return (i->c_name);
 	return "<unknown>";
 };

 static void log_handle_reconnect(struct uloop_timeout *timeout)
 {
 	sender.fd = usock((log_udp) ? (USOCK_UDP) : (USOCK_TCP), log_ip, log_port);
 	if (sender.fd < 0) {
 		fprintf(stderr, "failed to connect: %s\n", strerror(errno));
 		uloop_timeout_set(&retry, 1000);
 	} else {
 		uloop_fd_add(&sender, ULOOP_READ);
 		syslog(0, "Logread connected to %s:%s\n", log_ip, log_port);
 	}
 }

 static void log_handle_fd(struct uloop_fd *u, unsigned int events)
 {
 	if (u->eof) {
 		uloop_fd_delete(u);
 		close(sender.fd);
 		sender.fd = -1;
 		uloop_timeout_set(&retry, 1000);
 	}
 }

 static int filter_char_to_pri(char c)
 {
     switch (c) {
         case 'v':
             return 8;
         case 'd':
             return LOG_DEBUG;
         case 'i':
             return LOG_INFO;
         case 'w':
             return LOG_WARNING;
         case 'e':
             return LOG_ERR;
         case 'f':
             return LOG_ALERT;
         case '*':
         default:
             return 8;
     }
 }

 static int syslog_fileter_log(int pri, char *tag, struct filter_list_t *filter)
 {
     struct filter_list_t *_filter = filter;

     while(_filter)
     {
         int p = filter_char_to_pri(_filter->priority);
         if(_filter->tag)
         {
             int len = strlen(_filter->tag);
             // tag and priority
             if(0 == memcmp(_filter->tag, tag, len) && ((pri < p) || (pri == p)))
                 return 0;
         }else{ // have no tag
             if(pri > p)
                 return -1;
             else
                 return 0;
         }
         _filter = _filter->next;
     }

     return -1;
 }
 static int log_notify(struct blob_attr *msg)
 {
 	struct blob_attr *tb[__LOG_MAX];
 	struct stat s;
 	char buf[512];
 	uint32_t p;
 	char *str;
 	time_t t;
 	char *c, *m;

 	if (sender.fd < 0)
 		return 0;

 	blobmsg_parse(log_policy, ARRAY_SIZE(log_policy), tb, blob_data(msg), blob_len(msg));
 	if (!tb[LOG_ID] || !tb[LOG_PRIO] || !tb[LOG_SOURCE] || !tb[LOG_TIME] || !tb[LOG_MSG])
 		return 1;

 	if ((log_type == LOG_FILE) && log_size && (!stat(tmp_log, &s)) && (s.st_size > log_size)) {
         sender.fd = get_rotate_file(sender.fd, log_file, &file_list);
 		if (sender.fd < 0) {
 			fprintf(stderr, "failed to open %s: %s\n", tmp_log, strerror(errno));
 			exit(-1);
         }
 	}

 	m = blobmsg_get_string(tb[LOG_MSG]);
 	t = blobmsg_get_u64(tb[LOG_TIME]) / 1000;
 	c = ctime(&t);
 	p = blobmsg_get_u32(tb[LOG_PRIO]);
 	c[strlen(c) - 1] = '\0';
 	str = blobmsg_format_json(msg, true);

     if(filter_log && syslog_fileter_log(LOG_PRI(p), m, filter_log))
     {
         // printf("%s %d: fileter pri:%d tag:%s!\n", __FUNCTION__, __LINE__, p, m);
         exit(-1);
     }
 	if (log_type == LOG_NET) {
 		int err;

 		snprintf(buf, sizeof(buf), "<%u>", p);
 		strncat(buf, c + 4, 16);
 		if (hostname) {
 			strncat(buf, hostname, sizeof(buf));
 			strncat(buf, " ", sizeof(buf));
 		}
 		if (log_prefix) {
 			strncat(buf, log_prefix, sizeof(buf));
 			strncat(buf, ": ", sizeof(buf));
 		}
 		if (blobmsg_get_u32(tb[LOG_SOURCE]) == SOURCE_KLOG)
 			strncat(buf, "kernel: ", sizeof(buf));
 		strncat(buf, m, sizeof(buf));
 		if (log_udp)
 			err = write(sender.fd, buf, strlen(buf));
 		else
 			err = send(sender.fd, buf, strlen(buf), 0);

 		if (err < 0) {
 			syslog(0, "failed to send log data to %s:%s via %s\n",
 				log_ip, log_port, (log_udp) ? ("udp") : ("tcp"));
 			uloop_fd_delete(&sender);
 			close(sender.fd);
 			sender.fd = -1;
 			uloop_timeout_set(&retry, 1000);
 		}
 	} else {
 		snprintf(buf, sizeof(buf), "%s %s.%s%s %s\n",
 			c, getcodetext(LOG_FAC(p) << 3, facilitynames), getcodetext(LOG_PRI(p), prioritynames),
 			(blobmsg_get_u32(tb[LOG_SOURCE])) ? ("") : (" kernel:"), m);
 		write(sender.fd, buf, strlen(buf));
 	}

 	free(str);
 	if (log_type == LOG_FILE)
 		fsync(sender.fd);

 	return 0;
 }

 static void logread_fd_data_cb(struct ustream *s, int bytes)
 {
 	while (true) {
 		int len;
 		struct blob_attr *a;

 		a = (void*) ustream_get_read_buf(s, &len);
 		if (len < sizeof(*a) || len < blob_len(a) + sizeof(*a))
 			break;
 		log_notify(a);
 		ustream_consume(s, blob_len(a) + sizeof(*a));
 	}
 	if (!log_follow)
 		uloop_end();
 }

 static void logread_fd_cb(struct ubus_request *req, int fd)
 {
 	static struct ustream_fd test_fd;

 	test_fd.stream.notify_read = logread_fd_data_cb;
 	ustream_fd_init(&test_fd, fd);
 }

 static void logread_complete_cb(struct ubus_request *req, int ret)
 {
 }

 void* syslog_main(void* argv)
 {
 	static struct ubus_request req;
 	struct ubus_context *ctx;
 	uint32_t id;
 	const char *ubus_socket = NULL;
 	int ch, ret, lines = 0;
 	static struct blob_buf b;
 	int tries = 5;
     log_config_entry *config = (log_config_entry *)argv;

     pthread_detach(pthread_self());

     if (NULL == argv)
         return NULL;

 	signal(SIGPIPE, SIG_IGN);
 	uloop_init();

     log_file = config->out_path;
     memset(&file_list, 0, sizeof(struct file_list_t));
     file_list.total = config->rotate_file_count;
     if(config->rotate_file_size)
         log_size = config->rotate_file_size;
     if(config->ip)
     {
         printf("%s %d : %s:%s\n", __FUNCTION__, __LINE__, config->ip, config->port);
         log_ip = config->ip;
         log_port = config->port;
     }
     filter_log = config->filter_list;
     // Follow log messages
     log_follow = 1;
 	ctx = ubus_connect(ubus_socket);
 	if (!ctx) {
 		fprintf(stderr, "Failed to connect to ubus\n");
 		return -1;
 	}
 	ubus_add_uloop(ctx);

     printf("syslog log start...\n");
 	/* ugly ugly ugly ... we need a real reconnect logic */
 	do {
 		ret = ubus_lookup_id(ctx, "log", &id);
 		if (ret) {
 			fprintf(stderr, "Failed to find log object: %s\n", ubus_strerror(ret));
 			sleep(1);
 			continue;
 		}

 		blob_buf_init(&b, 0);
 		if (lines)
 			blobmsg_add_u32(&b, "lines", lines);
 		else if (log_follow)
 			blobmsg_add_u32(&b, "lines", 0);
 		if (log_follow) {
 			if (pid_file) {
 				FILE *fp = fopen(pid_file, "w+");
 				if (fp) {
 					fprintf(fp, "%d", getpid());
 					fclose(fp);
 				}
 			}
 		}

 		if (log_ip && log_port) {
 			openlog("logread", LOG_PID, LOG_DAEMON);
 			log_type = LOG_NET;
 			sender.cb = log_handle_fd;
 			retry.cb = log_handle_reconnect;
 			uloop_timeout_set(&retry, 1000);
 		} else if (log_file) {
 			log_type = LOG_FILE;
             // 先将文件保存到 /tmp/log/ 目录下，后面到达 rotate_file_size 后，转移到out_path
             sprintf(tmp_log, "/tmp/log%s", strstr_tail(log_file, "/"));
 			sender.fd = open(tmp_log, O_CREAT | O_WRONLY| O_APPEND, 0600);
 			if (sender.fd < 0) {
 				fprintf(stderr, "failed to open %s: %s\n", tmp_log, strerror(errno));
 				exit(-1);
 			}
 		} else {
 			sender.fd = STDOUT_FILENO;
 		}

 		ubus_invoke_async(ctx, id, "read", b.head, &req);
 		req.fd_cb = logread_fd_cb;
 		req.complete_cb = logread_complete_cb;
 		ubus_complete_request_async(ctx, &req);

 		uloop_run();
 		ubus_free(ctx);
 		uloop_done();

 	} while (ret && tries--);

     pthread_exit(NULL);
 	return NULL;
 }
	/*
	* Copyright (C) 2013 Felix Fietkau <nbd@openwrt.org>
	* Copyright (C) 2013 John Crispin <blogic@openwrt.org>
	*
	* This program is free software; you can redistribute it and/or modify
	* it under the terms of the GNU Lesser General Public License version 2.1
	* as published by the Free Software Foundation
	*
	* This program is distributed in the hope that it will be useful,
	* but WITHOUT ANY WARRANTY; without even the implied warranty of
	* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	* GNU General Public License for more details.
	*/

	#include <sys/types.h>
	#include <sys/stat.h>

	#include <fcntl.h>
	#include <time.h>
	#include <stdio.h>
	#include <unistd.h>
	#include <sys/types.h>
	#include <sys/socket.h>

	#define SYSLOG_NAMES
	#include <syslog.h>

	#include <libubox/ustream.h>
	#include <libubox/blobmsg_json.h>
	#include <libubox/usock.h>
	#include <libubox/uloop.h>
	#include "libubus.h"
	#include "syslog.h"
	#include "log_config.h"

	enum {
	LOG_STDOUT,
	LOG_FILE,
	LOG_NET,
	};

	enum {
	LOG_MSG,
	LOG_ID,
	LOG_PRIO,
	LOG_SOURCE,
	LOG_TIME,
	__LOG_MAX
	};

	static const struct blobmsg_policy log_policy[] = {
	[LOG_MSG] = { .name = "msg", .type = BLOBMSG_TYPE_STRING },
	[LOG_ID] = { .name = "id", .type = BLOBMSG_TYPE_INT32 },
	[LOG_PRIO] = { .name = "priority", .type = BLOBMSG_TYPE_INT32 },
	[LOG_SOURCE] = { .name = "source", .type = BLOBMSG_TYPE_INT32 },
	[LOG_TIME] = { .name = "time", .type = BLOBMSG_TYPE_INT64 },
	};

	static struct uloop_timeout retry;
	static struct uloop_fd sender;
	static const char log_file, log_ip, log_port, log_prefix, pid_file, hostname;
	static int log_type = LOG_STDOUT;
	static int log_size = 1 * 1024 * 1024, log_udp, log_follow = 0;
	static struct file_list_t file_list;
	static struct filter_list_t *filter_log = NULL;
	static char tmp_log[48] = {0};

	static const char* getcodetext(int value, CODE *codetable) {
	CODE *i;

	if (value >= 0)
	for (i = codetable; i->c_val != -1; i++)
	if (i->c_val == value)
	return (i->c_name);
	return "<unknown>";
	};

	static void log_handle_reconnect(struct uloop_timeout *timeout)
	{
	sender.fd = usock((log_udp) ? (USOCK_UDP) : (USOCK_TCP), log_ip, log_port);
	if (sender.fd < 0) {
	fprintf(stderr, "failed to connect: %s\n", strerror(errno));
	uloop_timeout_set(&retry, 1000);
	} else {
	uloop_fd_add(&sender, ULOOP_READ);
	syslog(0, "Logread connected to %s:%s\n", log_ip, log_port);
	}
	}

	static void log_handle_fd(struct uloop_fd *u, unsigned int events)
	{
	if (u->eof) {
	uloop_fd_delete(u);
	close(sender.fd);
	sender.fd = -1;
	uloop_timeout_set(&retry, 1000);
	}
	}

	static int filter_char_to_pri(char c)
	{
	switch (c) {
	case 'v':
	return 8;
	case 'd':
	return LOG_DEBUG;
	case 'i':
	return LOG_INFO;
	case 'w':
	return LOG_WARNING;
	case 'e':
	return LOG_ERR;
	case 'f':
	return LOG_ALERT;
	case '*':
	default:
	return 8;
	}
	}

	static int syslog_fileter_log(int pri, char tag, struct filter_list_t filter)
	{
	struct filter_list_t *_filter = filter;

	while(_filter)
	{
	int p = filter_char_to_pri(_filter->priority);
	if(_filter->tag)
	{
	int len = strlen(_filter->tag);
	// tag and priority
	if(0 == memcmp(_filter->tag, tag, len) && ((pri < p) \|\| (pri == p)))
	return 0;
	}else{ // have no tag
	if(pri > p)
	return -1;
	else
	return 0;
	}
	_filter = _filter->next;
	}

	return -1;
	}
	static int log_notify(struct blob_attr *msg)
	{
	struct blob_attr *tb[__LOG_MAX];
	struct stat s;
	char buf[512];
	uint32_t p;
	char *str;
	time_t t;
	char c, m;

	if (sender.fd < 0)
	return 0;

	blobmsg_parse(log_policy, ARRAY_SIZE(log_policy), tb, blob_data(msg), blob_len(msg));
	if (!tb[LOG_ID] \|\| !tb[LOG_PRIO] \|\| !tb[LOG_SOURCE] \|\| !tb[LOG_TIME] \|\| !tb[LOG_MSG])
	return 1;

	if ((log_type == LOG_FILE) && log_size && (!stat(tmp_log, &s)) && (s.st_size > log_size)) {
	sender.fd = get_rotate_file(sender.fd, log_file, &file_list);
	if (sender.fd < 0) {
	fprintf(stderr, "failed to open %s: %s\n", tmp_log, strerror(errno));
	exit(-1);
	}
	}

	m = blobmsg_get_string(tb[LOG_MSG]);
	t = blobmsg_get_u64(tb[LOG_TIME]) / 1000;
	c = ctime(&t);
	p = blobmsg_get_u32(tb[LOG_PRIO]);
	c[strlen(c) - 1] = '\0';
	str = blobmsg_format_json(msg, true);

	if(filter_log && syslog_fileter_log(LOG_PRI(p), m, filter_log))
	{
	// printf("%s %d: fileter pri:%d tag:%s!\n", __FUNCTION__, __LINE__, p, m);
	exit(-1);
	}
	if (log_type == LOG_NET) {
	int err;

	snprintf(buf, sizeof(buf), "<%u>", p);
	strncat(buf, c + 4, 16);
	if (hostname) {
	strncat(buf, hostname, sizeof(buf));
	strncat(buf, " ", sizeof(buf));
	}
	if (log_prefix) {
	strncat(buf, log_prefix, sizeof(buf));
	strncat(buf, ": ", sizeof(buf));
	}
	if (blobmsg_get_u32(tb[LOG_SOURCE]) == SOURCE_KLOG)
	strncat(buf, "kernel: ", sizeof(buf));
	strncat(buf, m, sizeof(buf));
	if (log_udp)
	err = write(sender.fd, buf, strlen(buf));
	else
	err = send(sender.fd, buf, strlen(buf), 0);

	if (err < 0) {
	syslog(0, "failed to send log data to %s:%s via %s\n",
	log_ip, log_port, (log_udp) ? ("udp") : ("tcp"));
	uloop_fd_delete(&sender);
	close(sender.fd);
	sender.fd = -1;
	uloop_timeout_set(&retry, 1000);
	}
	} else {
	snprintf(buf, sizeof(buf), "%s %s.%s%s %s\n",
	c, getcodetext(LOG_FAC(p) << 3, facilitynames), getcodetext(LOG_PRI(p), prioritynames),
	(blobmsg_get_u32(tb[LOG_SOURCE])) ? ("") : (" kernel:"), m);
	write(sender.fd, buf, strlen(buf));
	}

	free(str);
	if (log_type == LOG_FILE)
	fsync(sender.fd);

	return 0;
	}

	static void logread_fd_data_cb(struct ustream *s, int bytes)
	{
	while (true) {
	int len;
	struct blob_attr *a;

	a = (void*) ustream_get_read_buf(s, &len);
	if (len < sizeof(a) \|\| len < blob_len(a) + sizeof(a))
	break;
	log_notify(a);
	ustream_consume(s, blob_len(a) + sizeof(*a));
	}
	if (!log_follow)
	uloop_end();
	}

	static void logread_fd_cb(struct ubus_request *req, int fd)
	{
	static struct ustream_fd test_fd;

	test_fd.stream.notify_read = logread_fd_data_cb;
	ustream_fd_init(&test_fd, fd);
	}

	static void logread_complete_cb(struct ubus_request *req, int ret)
	{
	}

	void* syslog_main(void* argv)
	{
	static struct ubus_request req;
	struct ubus_context *ctx;
	uint32_t id;
	const char *ubus_socket = NULL;
	int ch, ret, lines = 0;
	static struct blob_buf b;
	int tries = 5;
	log_config_entry config = (log_config_entry )argv;

	pthread_detach(pthread_self());

	if (NULL == argv)
	return NULL;

	signal(SIGPIPE, SIG_IGN);
	uloop_init();

	log_file = config->out_path;
	memset(&file_list, 0, sizeof(struct file_list_t));
	file_list.total = config->rotate_file_count;
	if(config->rotate_file_size)
	log_size = config->rotate_file_size;
	if(config->ip)
	{
	printf("%s %d : %s:%s\n", __FUNCTION__, __LINE__, config->ip, config->port);
	log_ip = config->ip;
	log_port = config->port;
	}
	filter_log = config->filter_list;
	// Follow log messages
	log_follow = 1;
	ctx = ubus_connect(ubus_socket);
	if (!ctx) {
	fprintf(stderr, "Failed to connect to ubus\n");
	return -1;
	}
	ubus_add_uloop(ctx);

	printf("syslog log start...\n");
	/* ugly ugly ugly ... we need a real reconnect logic */
	do {
	ret = ubus_lookup_id(ctx, "log", &id);
	if (ret) {
	fprintf(stderr, "Failed to find log object: %s\n", ubus_strerror(ret));
	sleep(1);
	continue;
	}

	blob_buf_init(&b, 0);
	if (lines)
	blobmsg_add_u32(&b, "lines", lines);
	else if (log_follow)
	blobmsg_add_u32(&b, "lines", 0);
	if (log_follow) {
	if (pid_file) {
	FILE *fp = fopen(pid_file, "w+");
	if (fp) {
	fprintf(fp, "%d", getpid());
	fclose(fp);
	}
	}
	}

	if (log_ip && log_port) {
	openlog("logread", LOG_PID, LOG_DAEMON);
	log_type = LOG_NET;
	sender.cb = log_handle_fd;
	retry.cb = log_handle_reconnect;
	uloop_timeout_set(&retry, 1000);
	} else if (log_file) {
	log_type = LOG_FILE;
	// 先将文件保存到 /tmp/log/ 目录下，后面到达 rotate_file_size 后，转移到out_path
	sprintf(tmp_log, "/tmp/log%s", strstr_tail(log_file, "/"));
	sender.fd = open(tmp_log, O_CREAT \| O_WRONLY\| O_APPEND, 0600);
	if (sender.fd < 0) {
	fprintf(stderr, "failed to open %s: %s\n", tmp_log, strerror(errno));
	exit(-1);
	}
	} else {
	sender.fd = STDOUT_FILENO;
	}

	ubus_invoke_async(ctx, id, "read", b.head, &req);
	req.fd_cb = logread_fd_cb;
	req.complete_cb = logread_complete_cb;
	ubus_complete_request_async(ctx, &req);

	uloop_run();
	ubus_free(ctx);
	uloop_done();

	} while (ret && tries--);

	pthread_exit(NULL);
	return NULL;
	}