GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-26 03:40:01 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #612 from eeff/eeff/log

Browse Source 
feat(restful): support log file downloading
This commit is contained in:
fengzeroz
2022-07-04 17:22:02 +08:00
committed by GitHub
parent a7d5270b65 a5e6512d74
commit 3299eab989
3 changed files with 167 additions and 336 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/errcodes.h
+1
View File
@@ -41,6 +41,7 @@ typedef enum {
NEU_ERR_INVALID_TOKEN = 1008,
NEU_ERR_INVALID_USER_OR_PASSWORD = 1009,
NEU_ERR_IS_BUSY = 1010,
NEU_ERR_FILE_NOT_EXIST = 1011,
NEU_ERR_NODE_EXIST = 2002,
NEU_ERR_NODE_NOT_EXIST = 2003,
plugins/restful/http.c
+1
View File
@@ -312,6 +312,7 @@ int http_response(nng_aio *aio, neu_err_code_e code, char *content)
case NEU_ERR_GROUP_NAME_TOO_LONG:
status = NNG_HTTP_STATUS_BAD_REQUEST;
break;
case NEU_ERR_FILE_NOT_EXIST:
case NEU_ERR_LIBRARY_NOT_FOUND:
case NEU_ERR_NODE_NOT_EXIST:
case NEU_ERR_TAG_NOT_EXIST:
plugins/restful/log_handle.c
+165 -336
View File
@@ -17,379 +17,208 @@
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#define _XOPEN_SOURCE
#define _POSIX_C_SOURCE 200809L
#include <dirent.h>
#include <errno.h>
#include <fcntl.h>
#include <jwt.h>
#include <stdint.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <strings.h>
#include <sys/stat.h>
#include <sys/types.h>
#include <time.h>
#include <unistd.h>
#include "parser/neu_json_log.h"
#include "json/neu_json_fn.h"
#include <nng/nng.h>
#include <nng/supplemental/http/http.h>
#include "define.h"
#include "errcodes.h"
#include "handle.h"
#include "http.h"
#include "utils/log.h"
#include "utils/neu_jwt.h"
#include "json/neu_json_fn.h"
#include "log_handle.h"
#include "utils/utarray.h"
#ifdef __GNUC__
#define thread_local __thread
#elif __STDC_VERSION__ >= 201112L
#define thread_local _Thread_local
#elif defined(_MSC_VER)
#define thread_local __declspec(thread)
#else
#error Cannot define thread_local
#endif
UT_array *collect_log_files();
void handle_get_log_files(nng_aio *aio);
int read_log_file(const char *log_file, void **datap, size_t *lenp);
#define LOG_TIME_FMT "%Y-%m-%d %H:%M:%S"
#define LOG_FILE "./logs/neuron.log"
#define LOG_PAGE_SIZE_MIN 200
#define LOG_PAGE_SIZE_MAX 10000
// static size_t binary_search(FILE *fp, size_t lo, size_t hi, uint64_t ts);
// static int get_log_level(const char *line);
// static int get_log(const char *fname, uint64_t since, uint64_t until, int
// level, uint32_t page, uint32_t page_size, vector_t *lines, size_t *line_tot);
// static int string_to_log_level(const char *s, size_t n, int *level);
enum {
NEU_LOG_TRACE,
NEU_LOG_DEBUG,
NEU_LOG_INFO,
NEU_LOG_WARN,
NEU_LOG_ERROR,
NEU_LOG_FATAL
};
void handle_get_log(nng_aio *aio)
{
(void) aio;
// neu_json_get_log_resp_t resp = {};
// uint64_t since = 0;
// uint64_t until = 0;
// const char * level_str = NULL;
// size_t len = 0;
// int level = NEU_LOG_TRACE;
// uint32_t page = 0;
// uint32_t page_size = 0;
// size_t line_tot = 0;
int rv = 0;
char log_file[NEU_NODE_NAME_LEN + 32] = { 0 };
// VALIDATE_JWT(aio);
VALIDATE_JWT(aio);
//// required params
// if (0 != http_get_param_uint64(aio, "since", &since) ||
// 0 != http_get_param_uint64(aio, "until", &until) || since > until ||
// 0 != http_get_param_uint32(aio, "page", &page) ||
// 0 != http_get_param_uint32(aio, "page_size", &page_size) ||
// page_size < LOG_PAGE_SIZE_MIN || LOG_PAGE_SIZE_MAX < page_size) {
// NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
// http_response(aio, error_code.error, result_error);
//});
// return;
//}
//// optional param
// level_str = http_get_param(aio, "level", &len);
// if (NULL != level_str && 0 != string_to_log_level(level_str, len,
// &level)) { NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
// http_response(aio, error_code.error, result_error);
//});
// return;
//}
// vector_t lines;
// vector_init(&lines, 1, sizeof(char *));
// if (0 !=
// get_log(LOG_FILE, since, until, level, page, page_size, &lines,
//&line_tot)) {
// resp.error = NEU_ERR_EINTERNAL;
//}
// resp.n_row = lines.size;
// resp.rows = lines.data;
// resp.page_count = (line_tot + page_size - 1) / page_size;
// char *result = NULL;
// neu_json_encode_by_fn(&resp, neu_json_encode_get_log_resp, &result);
// http_ok(aio, result);
// VECTOR_FOR_EACH(&lines, iter)
//{
// char *line = *(char **) iterator_get(&iter);
// free(line);
//}
// vector_uninit(&lines);
// free(result);
}
static thread_local struct {
int level; // This three indicates a same get log query,
uint64_t since; // without taking page and page size into account.
uint64_t until; //
time_t time; // time stamp in last query result line
size_t line_start; // last query result start line number
size_t line_end; // last query result end line number
size_t line_tot; // total number of result log lines
size_t offset_start; // file offset corresponding to line_start
size_t offset_end; // file offset corresponding to line_end
} g_log_cache_;
static inline bool validate_cache(FILE *fp, int level, uint64_t since,
uint64_t until)
{
char buf[20] = { 0 };
struct tm tm = { .tm_isdst = -1 };
time_t ts = 0;
if (since == g_log_cache_.since && until == g_log_cache_.until &&
level == g_log_cache_.level) {
// check against file modification, maybe due to log rotation
fseek(fp, g_log_cache_.offset_end, SEEK_SET);
char *ret = fgets(buf, sizeof(buf), fp);
if (NULL == ret) {
return false;
}
if (NULL != strptime(buf, LOG_TIME_FMT, &tm)) {
ts = mktime(&tm);
if (ts == g_log_cache_.time) {
return true;
}
}
// clear cache
g_log_cache_.since = 1;
g_log_cache_.until = 0;
g_log_cache_.time = 0;
g_log_cache_.line_start = 0;
g_log_cache_.line_end = 0;
g_log_cache_.offset_start = 0;
g_log_cache_.offset_end = 0;
// optional params
rv = http_get_param_str(aio, "log_file", log_file, sizeof(log_file));
if (-1 == rv) {
nlog_error("parse query param `log_file` fail");
NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
http_response(aio, error_code.error, result_error);
});
return;
} else if (-2 == rv) {
// handle no param
handle_get_log_files(aio);
return;
} else if ((size_t) rv >= sizeof(log_file)) {
nlog_error("query param overflow, `log_file`:%s", log_file);
NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
http_response(aio, error_code.error, result_error);
});
return;
}
return false;
void * data = NULL;
size_t len = 0;
nng_http_res *res = NULL;
rv = read_log_file(log_file, &data, &len);
if (0 != rv) {
NEU_JSON_RESPONSE_ERROR(
rv, { http_response(aio, error_code.error, result_error); });
return;
}
if (((rv = nng_http_res_alloc(&res)) != 0) ||
((rv = nng_http_res_set_status(res, NNG_HTTP_STATUS_OK)) != 0) ||
((rv = nng_http_res_set_header(res, "Content-Type",
"application/octet-stream")) != 0) ||
((rv = nng_http_res_set_header(res, "Access-Control-Allow-Origin",
"*")) != 0) ||
((rv = nng_http_res_set_header(res, "Access-Control-Allow-Methods",
"POST,GET,PUT,DELETE,OPTIONS")) != 0) ||
((rv = nng_http_res_set_header(res, "Access-Control-Allow-Headers",
"*")) != 0) ||
((rv = nng_http_res_copy_data(res, data, len)) != 0)) {
nng_http_res_free(res);
free(data);
nng_aio_finish(aio, rv);
return;
}
free(data);
nng_aio_set_output(aio, 0, res);
nng_aio_finish(aio, 0);
}
// Find log lines within time range.
// static int get_log(const char *fname, uint64_t since, uint64_t until, int
// level, uint32_t page, uint32_t page_size, vector_t *lines, size_t *line_tot)
//{
// FILE *fp = fopen(fname, "r");
void handle_get_log_files(nng_aio *aio)
{
UT_array *log_files = collect_log_files();
if (NULL == log_files) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
http_response(aio, error_code.error, result_error);
});
return;
}
// if (NULL == fp) {
// return NEU_ERR_ENOFILES;
//}
size_t len = 1 + strlen("{log_files:[ ]}");
utarray_foreach(log_files, char **, log_file)
{
// one byte for the comma,
// the last one will be for the terminating NULL byte
len += strlen(*log_file) + 1;
}
// int rv = 0;
// size_t i = 0;
// size_t size = 0;
// size_t n = 1024;
// struct tm tm = { .tm_isdst = -1 };
// char * lineptr = calloc(n, sizeof(char));
// size_t nline = 0;
// size_t start = page * page_size;
// size_t end = start + page_size;
char *res = malloc(len);
if (NULL == res) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_EINTERNAL, {
http_response(aio, error_code.error, result_error);
});
utarray_free(log_files);
return;
}
// if (NULL == lineptr) {
// return NEU_ERR_ENOMEM;
//}
// with the space char to simplify code
int i = snprintf(res, len, "{log_files:[ ");
utarray_foreach(log_files, char **, log_file)
{
i += snprintf(res + i, len - i, "%s,", *log_file);
}
// will write over the last comma/space
snprintf(res + i - 1, len, "]}");
http_ok(aio, res);
free(res);
utarray_free(log_files);
}
//// get file size
// fseek(fp, 0, SEEK_END);
// size = ftell(fp);
// fseek(fp, 0, SEEK_SET);
static inline bool ends_with(const char *str, const char *suffix)
{
size_t m = strlen(str);
size_t n = strlen(suffix);
return m >= n && !strcmp(str + m - n, suffix);
}
// bool cached = validate_cache(fp, level, since, until);
// if (cached && g_log_cache_.line_start <= start &&
// start < g_log_cache_.line_end + LOG_PAGE_SIZE_MAX) {
//// continue from last file offset
// if (g_log_cache_.line_end <= start) {
// i = g_log_cache_.offset_end;
// nline = g_log_cache_.line_end;
//} else {
// i = g_log_cache_.offset_start;
// nline = g_log_cache_.line_start;
//}
//} else {
// i = binary_search(fp, 0, size, since);
//}
UT_array *collect_log_files()
{
DIR *dp = opendir("./logs");
if (NULL == dp) {
nlog_error("open logs directory fail");
return NULL;
}
// fseek(fp, i, SEEK_SET);
UT_array *log_files = NULL;
utarray_new(log_files, &ut_str_icd);
// while (i < size) {
// int len = getline(&lineptr, &n, fp);
// if (len <= 0) {
// break;
//}
// i += len;
struct dirent *de = NULL;
while (NULL != (de = readdir(dp))) {
char *name = de->d_name;
if (ends_with(de->d_name, ".log")) {
utarray_push_back(log_files, &name);
}
}
// if (get_log_level(lineptr) < level) {
// continue;
//}
closedir(dp);
// if (NULL == strptime(lineptr, LOG_TIME_FMT, &tm)) {
// continue;
//}
// uint64_t ts = mktime(&tm);
// if (ts < since) {
// continue;
//}
// if (ts >= until) {
// break;
//}
// if (nline < start) {
//++nline;
// continue;
//} else if (nline == start) {
// g_log_cache_.line_start = nline;
// g_log_cache_.offset_start = i - len;
//}
return log_files;
}
// if (nline < end) {
// char *line = strdup(lineptr);
// if (NULL == line) {
// rv = NEU_ERR_ENOMEM;
// break;
//}
// if (0 != vector_push_back(lines, &line)) {
// rv = NEU_ERR_ENOMEM;
// break;
//}
// reads the entire named file, allocating storage to receive the data and
// returning the data and the size in the reference arguments.
int read_log_file(const char *log_file, void **datap, size_t *lenp)
{
int rv = 0;
FILE * f;
struct stat st;
size_t len;
void * data;
char name[128] = { 0 };
// g_log_cache_.line_end = nline;
// g_log_cache_.offset_end = i - len;
// g_log_cache_.time = ts;
//}
if ((size_t) snprintf(name, sizeof(name), "./logs/%s", log_file) >=
sizeof(name)) {
nlog_error("log_file path overflow:%s", name);
return NEU_ERR_EINTERNAL;
}
//++nline;
if (stat(name, &st) != 0) {
return NEU_ERR_FILE_NOT_EXIST;
}
// if (nline == end) {
// if (cached && nline < g_log_cache_.line_tot) {
//// no need to count total number of lines.
// break;
//} else {
// cached = false;
//}
//}
//}
if ((f = fopen(name, "rb")) == NULL) {
nlog_error("open fail: %s", name);
return NEU_ERR_EINTERNAL;
}
// if (0 != rv || !cached) {
// g_log_cache_.level = level;
// g_log_cache_.since = since;
// g_log_cache_.until = until;
// g_log_cache_.line_tot = nline;
//}
len = st.st_size;
if (len > 0) {
if ((data = malloc(len)) == NULL) {
rv = NEU_ERR_EINTERNAL;
goto done;
}
if (fread(data, 1, len, f) != len) {
rv = errno;
free(data);
goto done;
}
} else {
data = NULL;
}
// if (NULL != line_tot) {
//*line_tot = g_log_cache_.line_tot;
//}
// free(lineptr);
// fclose(fp);
// return rv;
//}
// Look for file offset, starting at which logs with target time stamp follows.
// The returned offset is not a strict lower bound, just enough for filtering.
// static size_t binary_search(FILE *fp, size_t lo, size_t hi, uint64_t ts)
//{
// size_t n = 1024;
// char * lineptr = calloc(n, sizeof(char));
// struct tm tm = { .tm_isdst = -1 };
// while (lo < hi) {
// int len1 = 0, len2 = 0;
// size_t mid = (lo + hi) / 2;
// fseek(fp, mid, SEEK_SET);
//// scan to a log line following mid
// len1 = getline(&lineptr, &n, fp);
// if (len1 < 0 || mid + len1 >= hi) {
//// outside offset range
// break;
//}
// while (NULL == strptime(lineptr, LOG_TIME_FMT, &tm)) {
// len2 = getline(&lineptr, &n, fp);
// if (len2 < 0 || mid + len1 + len2 >= hi) {
//// outside offset range
// break;
//}
// len1 += len2;
//}
// if (len2 < 0) {
//// outside offset range
// break;
//}
// uint64_t t = mktime(&tm);
// if (ts <= t) {
// hi = mid + len1;
//} else {
// lo = mid + len1;
//}
//}
// free(lineptr);
// return lo;
//}
// static int get_log_level(const char *line)
//{
// int level = NEU_LOG_TRACE;
// line += 19; // skip time string
// while (*line && 'T' != *line && 'D' != *line && 'I' != *line &&
//'W' != *line && 'E' != *line && 'F' != *line) {
//++line;
//}
// if ('T' == *line && 0 == strncmp(line, "TRACE", 5)) {
// level = NEU_LOG_TRACE;
//} else if ('D' == *line && 0 == strncmp(line, "DEBUG", 5)) {
// level = NEU_LOG_DEBUG;
//} else if ('I' == *line && 0 == strncmp(line, "INFO", 4)) {
// level = NEU_LOG_INFO;
//} else if ('W' == *line && 0 == strncmp(line, "WARN", 4)) {
// level = NEU_LOG_WARN;
//} else if ('E' == *line && 0 == strncmp(line, "ERROR", 5)) {
// level = NEU_LOG_ERROR;
//} else if ('F' == *line && 0 == strncmp(line, "FATAL", 5)) {
// level = NEU_LOG_FATAL;
//}
// return level;
//}
// static int string_to_log_level(const char *s, size_t n, int *level)
//{
//(void) s;
//(void) n;
//(void) level;
// return -1;
// if (0 == strncasecmp(s, "trace", n)) {
//*level = NEU_LOG_TRACE;
//} else if (0 == strncasecmp(s, "debug", n)) {
//*level = NEU_LOG_DEBUG;
//} else if (0 == strncasecmp(s, "info", n)) {
//*level = NEU_LOG_INFO;
//} else if (0 == strncasecmp(s, "warn", n)) {
//*level = NEU_LOG_WARN;
//} else if (0 == strncasecmp(s, "error", n)) {
//*level = NEU_LOG_ERROR;
//} else if (0 == strncasecmp(s, "fatal", n)) {
//*level = NEU_LOG_FATAL;
//} else {
// return -1;
//}
// return 0;
//}
*datap = data;
*lenp = len;
done:
fclose(f);
return (rv);
}