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/un.h>
 #include <sys/socket.h>
 #include <stdlib.h>
 #include <pthread.h>

 #define SYSLOG_NAMES
 #include <syslog.h>
 #include "json-c/json.h"
 #include "json-c/printbuf.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"

 #include "mbtk_utils.h"
 //#include "lynq/liblog.h"

 enum {
 	SOURCE_KLOG = 0,
 	SOURCE_SYSLOG = 1,
 	SOURCE_INTERNAL = 2,
 	SOURCE_ANY = 0xff,
 };

 #define LOG_CONFIG_LEN 50
 enum {
 	LOG_STDOUT,
 	LOG_FILE,
 	LOG_NET,
 };

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

 #define    SYSLOG_BUFF_SIZE (4*1024)
 #define    MAX_BUFFER_SIZE (8*1024)

 extern int tmp_syslog_fd;
 extern char syslog_buff[MAX_BUFFER_SIZE];

 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 char log_file[LOG_CONFIG_LEN], log_ip[LOG_CONFIG_LEN], log_port[LOG_CONFIG_LEN], log_prefix[LOG_CONFIG_LEN], pid_file[LOG_CONFIG_LEN], hostname[LOG_CONFIG_LEN];
 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[100] = {0};
 pthread_t attr = -1;

 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 LOG_CRIT;
         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;
 	struct filter_list_t *_filter_common = _filter;

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

 	return 0;
 }
 static int log_notify(struct blob_attr *msg)
 {
         int index = 0;
         int len = 0;
 	struct blob_attr *tb[__LOG_MAX];
 	struct stat s;
 	char buf[1024 * 2] = {'\0'};
 	char tmp_buf[1024] = {0};
 	uint32_t p;
 	char *str;
 	time_t t;
 	char *c, *m;

     static char buffer[MAX_BUFFER_SIZE] = {0};
     static int buffer_index = 0;

 	//
 	//sprintf(tmp_buf, "/tmp/log%s", strstr_tail(log_file, "/"));


     snprintf(tmp_buf,sizeof(tmp_buf), "%s", log_file);

     if(fcntl(sender.fd, F_GETFL) == -1 || access(tmp_buf, W_OK) != 0)
     {
         sender.fd = open(tmp_buf, O_CREAT | O_WRONLY | O_APPEND, 0600);
         if(sender.fd < 0)
         {
             perror("Failed to open file ");
             return -1;
         }
         tmp_syslog_fd = sender.fd;
     }


 	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(log_file, &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", log_file, strerror(errno));
                 exit(-1);
             }
             tmp_syslog_fd = sender.fd;
     }

 	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);
         return 0;
         //exit(-1);
     }
 	if (log_type == LOG_NET)
 	{
 		int err;

 		snprintf(buf, sizeof(buf), "<%u>", p);
 		strncat(buf, c + 4, 16);
 		if (strlen(hostname) > 0) {
 			strncat(buf, hostname, strlen(hostname));
 			strncat(buf, " ", 1);
 		}
 		if (strlen(log_prefix) > 0) {
 			strncat(buf, log_prefix, strlen(log_prefix));
 			strncat(buf, ": ", 2);
 		}
 		if (blobmsg_get_u32(tb[LOG_SOURCE]) == SOURCE_KLOG)
 			strncat(buf, "kernel: ", strlen("kernel: "));
 		strncat(buf, m, strlen(m));
 		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);

                 len = strlen(buf);
                 if(access("/etc/syslog_encrypt_flag", F_OK) == 0)
                 {
                     for(index = 0; index < len; index++)
                     {
                         buf[index] ^= 1;
                     }
                 }

         if(buffer_index + len >= SYSLOG_BUFF_SIZE && buffer_index < SYSLOG_BUFF_SIZE)
         {
             memcpy(buffer + buffer_index, buf, len);
             buffer_index += len;
             if (write(sender.fd, buffer, buffer_index) < 0)
             {
                 perror("write error");
                 close(sender.fd);
                 return -1;
             }
             buffer_index = 0;
         }
         else
         {
             memcpy(buffer + buffer_index, buf, len);
             buffer_index += len;
             memcpy(syslog_buff,buffer,buffer_index);
         }


 	}

 	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)
 {
 }

 int lynq_update_log_level()
 {
     json_object* jsonobj = NULL;
     json_object* tmpjson = NULL;
     json_object* datajson = NULL;
     json_object* listjson = NULL;
     json_object* fileterjson = NULL;
     json_object* fileter_listjson = NULL;
 	struct filter_list_t* filter_list_head = NULL;
 	struct filter_list_t* tmp_filter_list = NULL;
 	struct filter_list_t* _filter_list = NULL;

     int n;
 	int array_length;
 //    char* tmp_string = NULL;

     jsonobj = json_object_from_file(LOG_CONFIG_PATH);
     if (NULL == jsonobj) {
         printf("Can't open config file: %s\n", LOG_CONFIG_PATH);
         return -1;
     }
     /***获取data***/
     json_object_object_get_ex(jsonobj, "buffer_list", &tmpjson);
     datajson = json_object_array_get_idx(tmpjson, 0);//syslog index is 0
     if (NULL == datajson) {
         json_object_put(jsonobj);
         return -1;
     }
     json_object_object_get_ex(datajson, "filter_list", &listjson);
     if (NULL == listjson) {
         printf("%s %d: object failure!\n", __FUNCTION__, __LINE__);
         json_object_put(listjson);
         return -1;
     }
 	filter_list_head = (struct filter_list_t*)malloc(sizeof(struct filter_list_t));
 	_filter_list = filter_list_head;

 	array_length = json_object_array_length(listjson);
     for (n = 0 ; n < array_length; n++) {
             fileterjson = json_object_array_get_idx(listjson, n);
             if (NULL == fileterjson) {
                 printf("the fileterjson exit\n");
                 free(tmp_filter_list->next);
                 tmp_filter_list->next = NULL;
                 break;
             }
             memset(_filter_list, 0, sizeof(struct filter_list_t));
             json_object_object_get_ex(fileterjson, "priority", &fileter_listjson);
             const char* str = json_object_get_string(fileter_listjson);
             if (str) {
                 _filter_list->priority = str[0];
                 printf("fileter_listjson: %c\n", _filter_list->priority);
             }

             json_object_object_get_ex(fileterjson, "tag", &fileter_listjson);

             str = json_object_get_string(fileter_listjson);
             if (str) {
                 _filter_list->tag = strdup(str);
                 printf("fileter_listjson: %s\n", _filter_list->tag);
             }
             else
             {
                 _filter_list->tag = "\0";
             }
             //json_object_put(fileter_listjson);
             _filter_list->next = (struct filter_list_t*)malloc(sizeof(struct filter_list_t));
             if (NULL == _filter_list->next) {
                 printf("%s %d: malloc failure!\n", __FUNCTION__, __LINE__);
                 break;
             }
             tmp_filter_list = _filter_list;
             _filter_list = _filter_list->next;
 			_filter_list->next = NULL;
     }
     /***释放json对象***/
     json_object_put(jsonobj);

 	tmp_filter_list = filter_log;
 	filter_log = filter_list_head;

     while(tmp_filter_list != NULL) {
 		_filter_list = tmp_filter_list;
         free(tmp_filter_list);
         tmp_filter_list = _filter_list->next;
     }

     return 0;
 }


 void* wait_update_log_level(void *arg)
 {
 //	int i = 0;
 	char recvBuff[100];
 	int serverFd,clientFd;
     socklen_t addrLen;
     struct sockaddr_un serverAddr,clientAddr;

 	pthread_detach(pthread_self());
 	printf("MBTK: in wait_update_log_level\n");

     memset(&serverAddr,0,sizeof(serverAddr));
     serverAddr.sun_family = AF_UNIX;
     sprintf(serverAddr.sun_path,"%s","/var/log_server.socket");

     unlink("/var/log_server.socket");   /* in case it already exists */

     if ((serverFd = socket(AF_UNIX,SOCK_STREAM,0)) < 0)
     {
         printf("err -1\n");
         return NULL;
     }

     if (bind(serverFd,(struct sockaddr *)&serverAddr,sizeof(serverAddr)) < 0)
     {
         printf("err -2\n");
         close(serverFd);
         return NULL;
     }

     if(listen(serverFd,10) < 0)
     {
         printf("err -3\n");
         return NULL;
     }

     while(1)
     {
         addrLen = sizeof(clientAddr);
         memset(&clientAddr,0,sizeof(clientAddr));
         memset(&recvBuff,0,100);

         if((clientFd = accept(serverFd,(struct sockaddr*)&clientAddr,&addrLen)) < 0)
         {
             printf("err -4\n");
             continue;
         }
 		printf("MBTK: wait recv\n");
         if(recv(clientFd,recvBuff,100,0) < 0)
         {
             printf("err -5");
             close(clientFd);
             continue;
         }
         if(strncmp(recvBuff, "update", strlen("update")) == 0)
         {
             lynq_update_log_level();
         }

         close(clientFd);
     }
     close(serverFd);

 	return NULL;
 }

 int syslog_pthread_create()
 {
     int ret;

 	ret = pthread_create(&attr, NULL, wait_update_log_level, NULL);

     if (ret < 0)
     {
         printf("MBTK:pthread create fail");
         return -1;
     }

     return -1;
 }

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

     log_config_entry *config = (log_config_entry *)argv;

     pthread_detach(pthread_self());

 //    if (NULL == argv)
 //        return NULL;

 	signal(SIGPIPE, SIG_IGN);
 	uloop_init();


 	syslog_pthread_create();
     //log_file = config->out_path;
 	memset(log_file, 0, sizeof(log_file));
 	memset(log_ip, 0, sizeof(log_ip));
 	memset(log_port, 0, sizeof(log_port));
 	memset(log_prefix, 0, sizeof(log_prefix));
 	memset(pid_file, 0, sizeof(pid_file));
 	memset(hostname, 0, sizeof(hostname));

 	if(config->out_path != NULL)
 	{
 		strncpy(log_file, config->out_path, LOG_CONFIG_LEN - 1);
 	}

     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;
         strncpy(log_ip, config->ip, LOG_CONFIG_LEN - 1);
         //log_port = config->port;
         if(config->port != NULL)
         {
             strncpy(log_port, config->port, LOG_CONFIG_LEN - 1);
         }
     }
     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 NULL;
 	}
 	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 (strlen(pid_file) > 0) {
 				FILE *fp = fopen(pid_file, "w+");
 				if (fp) {
 					fprintf(fp, "%d", getpid());
 					fclose(fp);
 				}
 			}
 		}

 		if (strlen(log_ip) > 0 && strlen(log_port) > 0) {
 			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 (strlen(log_file) > 0) {
 			log_type = LOG_FILE;
             // 先将文件保存到 /tmp/log/ 目录下，后面到达 rotate_file_size 后，转移到out_path

 			//sprintf(tmp_log, "/tmp/log%s", strstr_tail(log_file, "/"));

 			//直接将log文件存储在不会掉电丢失的路径
 			snprintf(tmp_log,sizeof(tmp_log), "%s",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);
 			}
             tmp_syslog_fd = sender.fd;
 		} 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--);

 	if (attr) {
 		if (pthread_join(attr, &tret) != 0) {
 			printf("MBTK:Join thread: %ld error!\n", attr);
 			exit(1);
 		}
 	}

     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/un.h>
	#include <sys/socket.h>
	#include <stdlib.h>
	#include <pthread.h>

	#define SYSLOG_NAMES
	#include <syslog.h>
	#include "json-c/json.h"
	#include "json-c/printbuf.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"

	#include "mbtk_utils.h"
	//#include "lynq/liblog.h"

	enum {
	SOURCE_KLOG = 0,
	SOURCE_SYSLOG = 1,
	SOURCE_INTERNAL = 2,
	SOURCE_ANY = 0xff,
	};

	#define LOG_CONFIG_LEN 50
	enum {
	LOG_STDOUT,
	LOG_FILE,
	LOG_NET,
	};

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

	#define SYSLOG_BUFF_SIZE (4*1024)
	#define MAX_BUFFER_SIZE (8*1024)

	extern int tmp_syslog_fd;
	extern char syslog_buff[MAX_BUFFER_SIZE];

	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 char log_file[LOG_CONFIG_LEN], log_ip[LOG_CONFIG_LEN], log_port[LOG_CONFIG_LEN], log_prefix[LOG_CONFIG_LEN], pid_file[LOG_CONFIG_LEN], hostname[LOG_CONFIG_LEN];
	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[100] = {0};
	pthread_t attr = -1;

	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 LOG_CRIT;
	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;
	struct filter_list_t *_filter_common = _filter;

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

	return 0;
	}
	static int log_notify(struct blob_attr *msg)
	{
	int index = 0;
	int len = 0;
	struct blob_attr *tb[__LOG_MAX];
	struct stat s;
	char buf[1024 * 2] = {'\0'};
	char tmp_buf[1024] = {0};
	uint32_t p;
	char *str;
	time_t t;
	char c, m;

	static char buffer[MAX_BUFFER_SIZE] = {0};
	static int buffer_index = 0;

	//
	//sprintf(tmp_buf, "/tmp/log%s", strstr_tail(log_file, "/"));


	snprintf(tmp_buf,sizeof(tmp_buf), "%s", log_file);

	if(fcntl(sender.fd, F_GETFL) == -1 \|\| access(tmp_buf, W_OK) != 0)
	{
	sender.fd = open(tmp_buf, O_CREAT \| O_WRONLY \| O_APPEND, 0600);
	if(sender.fd < 0)
	{
	perror("Failed to open file ");
	return -1;
	}
	tmp_syslog_fd = sender.fd;
	}



	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(log_file, &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", log_file, strerror(errno));
	exit(-1);
	}
	tmp_syslog_fd = sender.fd;
	}

	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);
	return 0;
	//exit(-1);
	}
	if (log_type == LOG_NET)
	{
	int err;

	snprintf(buf, sizeof(buf), "<%u>", p);
	strncat(buf, c + 4, 16);
	if (strlen(hostname) > 0) {
	strncat(buf, hostname, strlen(hostname));
	strncat(buf, " ", 1);
	}
	if (strlen(log_prefix) > 0) {
	strncat(buf, log_prefix, strlen(log_prefix));
	strncat(buf, ": ", 2);
	}
	if (blobmsg_get_u32(tb[LOG_SOURCE]) == SOURCE_KLOG)
	strncat(buf, "kernel: ", strlen("kernel: "));
	strncat(buf, m, strlen(m));
	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);

	len = strlen(buf);
	if(access("/etc/syslog_encrypt_flag", F_OK) == 0)
	{
	for(index = 0; index < len; index++)
	{
	buf[index] ^= 1;
	}
	}

	if(buffer_index + len >= SYSLOG_BUFF_SIZE && buffer_index < SYSLOG_BUFF_SIZE)
	{
	memcpy(buffer + buffer_index, buf, len);
	buffer_index += len;
	if (write(sender.fd, buffer, buffer_index) < 0)
	{
	perror("write error");
	close(sender.fd);
	return -1;
	}
	buffer_index = 0;
	}
	else
	{
	memcpy(buffer + buffer_index, buf, len);
	buffer_index += len;
	memcpy(syslog_buff,buffer,buffer_index);
	}


	}

	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)
	{
	}

	int lynq_update_log_level()
	{
	json_object* jsonobj = NULL;
	json_object* tmpjson = NULL;
	json_object* datajson = NULL;
	json_object* listjson = NULL;
	json_object* fileterjson = NULL;
	json_object* fileter_listjson = NULL;
	struct filter_list_t* filter_list_head = NULL;
	struct filter_list_t* tmp_filter_list = NULL;
	struct filter_list_t* _filter_list = NULL;

	int n;
	int array_length;
	// char* tmp_string = NULL;

	jsonobj = json_object_from_file(LOG_CONFIG_PATH);
	if (NULL == jsonobj) {
	printf("Can't open config file: %s\n", LOG_CONFIG_PATH);
	return -1;
	}
	/*获取data*/
	json_object_object_get_ex(jsonobj, "buffer_list", &tmpjson);
	datajson = json_object_array_get_idx(tmpjson, 0);//syslog index is 0
	if (NULL == datajson) {
	json_object_put(jsonobj);
	return -1;
	}
	json_object_object_get_ex(datajson, "filter_list", &listjson);
	if (NULL == listjson) {
	printf("%s %d: object failure!\n", __FUNCTION__, __LINE__);
	json_object_put(listjson);
	return -1;
	}
	filter_list_head = (struct filter_list_t*)malloc(sizeof(struct filter_list_t));
	_filter_list = filter_list_head;

	array_length = json_object_array_length(listjson);
	for (n = 0 ; n < array_length; n++) {
	fileterjson = json_object_array_get_idx(listjson, n);
	if (NULL == fileterjson) {
	printf("the fileterjson exit\n");
	free(tmp_filter_list->next);
	tmp_filter_list->next = NULL;
	break;
	}
	memset(_filter_list, 0, sizeof(struct filter_list_t));
	json_object_object_get_ex(fileterjson, "priority", &fileter_listjson);
	const char* str = json_object_get_string(fileter_listjson);
	if (str) {
	_filter_list->priority = str[0];
	printf("fileter_listjson: %c\n", _filter_list->priority);
	}

	json_object_object_get_ex(fileterjson, "tag", &fileter_listjson);

	str = json_object_get_string(fileter_listjson);
	if (str) {
	_filter_list->tag = strdup(str);
	printf("fileter_listjson: %s\n", _filter_list->tag);
	}
	else
	{
	_filter_list->tag = "\0";
	}
	//json_object_put(fileter_listjson);
	_filter_list->next = (struct filter_list_t*)malloc(sizeof(struct filter_list_t));
	if (NULL == _filter_list->next) {
	printf("%s %d: malloc failure!\n", __FUNCTION__, __LINE__);
	break;
	}
	tmp_filter_list = _filter_list;
	_filter_list = _filter_list->next;
	_filter_list->next = NULL;
	}
	/*释放json对象*/
	json_object_put(jsonobj);

	tmp_filter_list = filter_log;
	filter_log = filter_list_head;

	while(tmp_filter_list != NULL) {
	_filter_list = tmp_filter_list;
	free(tmp_filter_list);
	tmp_filter_list = _filter_list->next;
	}

	return 0;
	}


	void* wait_update_log_level(void *arg)
	{
	// int i = 0;
	char recvBuff[100];
	int serverFd,clientFd;
	socklen_t addrLen;
	struct sockaddr_un serverAddr,clientAddr;

	pthread_detach(pthread_self());
	printf("MBTK: in wait_update_log_level\n");

	memset(&serverAddr,0,sizeof(serverAddr));
	serverAddr.sun_family = AF_UNIX;
	sprintf(serverAddr.sun_path,"%s","/var/log_server.socket");

	unlink("/var/log_server.socket"); /* in case it already exists */

	if ((serverFd = socket(AF_UNIX,SOCK_STREAM,0)) < 0)
	{
	printf("err -1\n");
	return NULL;
	}

	if (bind(serverFd,(struct sockaddr *)&serverAddr,sizeof(serverAddr)) < 0)
	{
	printf("err -2\n");
	close(serverFd);
	return NULL;
	}

	if(listen(serverFd,10) < 0)
	{
	printf("err -3\n");
	return NULL;
	}

	while(1)
	{
	addrLen = sizeof(clientAddr);
	memset(&clientAddr,0,sizeof(clientAddr));
	memset(&recvBuff,0,100);

	if((clientFd = accept(serverFd,(struct sockaddr*)&clientAddr,&addrLen)) < 0)
	{
	printf("err -4\n");
	continue;
	}
	printf("MBTK: wait recv\n");
	if(recv(clientFd,recvBuff,100,0) < 0)
	{
	printf("err -5");
	close(clientFd);
	continue;
	}
	if(strncmp(recvBuff, "update", strlen("update")) == 0)
	{
	lynq_update_log_level();
	}

	close(clientFd);
	}
	close(serverFd);

	return NULL;
	}

	int syslog_pthread_create()
	{
	int ret;

	ret = pthread_create(&attr, NULL, wait_update_log_level, NULL);

	if (ret < 0)
	{
	printf("MBTK:pthread create fail");
	return -1;
	}

	return -1;
	}

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

	log_config_entry config = (log_config_entry )argv;

	pthread_detach(pthread_self());

	// if (NULL == argv)
	// return NULL;

	signal(SIGPIPE, SIG_IGN);
	uloop_init();


	syslog_pthread_create();
	//log_file = config->out_path;
	memset(log_file, 0, sizeof(log_file));
	memset(log_ip, 0, sizeof(log_ip));
	memset(log_port, 0, sizeof(log_port));
	memset(log_prefix, 0, sizeof(log_prefix));
	memset(pid_file, 0, sizeof(pid_file));
	memset(hostname, 0, sizeof(hostname));

	if(config->out_path != NULL)
	{
	strncpy(log_file, config->out_path, LOG_CONFIG_LEN - 1);
	}

	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;
	strncpy(log_ip, config->ip, LOG_CONFIG_LEN - 1);
	//log_port = config->port;
	if(config->port != NULL)
	{
	strncpy(log_port, config->port, LOG_CONFIG_LEN - 1);
	}
	}
	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 NULL;
	}
	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 (strlen(pid_file) > 0) {
	FILE *fp = fopen(pid_file, "w+");
	if (fp) {
	fprintf(fp, "%d", getpid());
	fclose(fp);
	}
	}
	}

	if (strlen(log_ip) > 0 && strlen(log_port) > 0) {
	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 (strlen(log_file) > 0) {
	log_type = LOG_FILE;
	// 先将文件保存到 /tmp/log/ 目录下，后面到达 rotate_file_size 后，转移到out_path

	//sprintf(tmp_log, "/tmp/log%s", strstr_tail(log_file, "/"));

	//直接将log文件存储在不会掉电丢失的路径
	snprintf(tmp_log,sizeof(tmp_log), "%s",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);
	}
	tmp_syslog_fd = sender.fd;
	} 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--);

	if (attr) {
	if (pthread_join(attr, &tret) != 0) {
	printf("MBTK:Join thread: %ld error!\n", attr);
	exit(1);
	}
	}

	pthread_exit(NULL);
	return NULL;
	}