From e1b8711a017d5823375f7c973150446abcc92d19 Mon Sep 17 00:00:00 2001 From: Philipp Moritz Date: Fri, 23 Sep 2016 17:10:15 -0700 Subject: [PATCH] Redis logging (#17) * Redis logging * Rearrange logging interfaces * Fix test case * Changes to logging interface and test case for logging * Fixes * Fix memory leaks * Add interface method to destroy logger * is_local -> is_direct * Merge fix --- Makefile | 4 +-- io.c | 15 ++++++++ io.h | 1 + logging.c | 78 ++++++++++++++++++++++++++++++++++++++++ logging.h | 39 ++++++++++++++++++++ state/redis.c | 20 ----------- state/redis.h | 5 ++- test/redis_tests.c | 89 ++++++++++++++++++++++++++++++++++++++++------ 8 files changed, 217 insertions(+), 34 deletions(-) create mode 100644 logging.c create mode 100644 logging.h diff --git a/Makefile b/Makefile index 36793877f..aa5ee292e 100644 --- a/Makefile +++ b/Makefile @@ -19,8 +19,8 @@ $(BUILD)/io_tests: test/io_tests.c $(BUILD)/libcommon.a $(BUILD)/task_tests: test/task_tests.c $(BUILD)/libcommon.a $(CC) -o $@ $^ $(CFLAGS) -$(BUILD)/redis_tests: hiredis test/redis_tests.c $(BUILD)/libcommon.a - $(CC) -o $@ test/redis_tests.c $(BUILD)/libcommon.a thirdparty/hiredis/libhiredis.a $(CFLAGS) +$(BUILD)/redis_tests: hiredis test/redis_tests.c $(BUILD)/libcommon.a logging.h + $(CC) -o $@ test/redis_tests.c logging.c $(BUILD)/libcommon.a thirdparty/hiredis/libhiredis.a $(CFLAGS) clean: rm -f *.o state/*.o test/*.o diff --git a/io.c b/io.c index 99295512b..d4a89a4a4 100644 --- a/io.c +++ b/io.c @@ -7,6 +7,8 @@ #include #include #include +#include +#include #include "common.h" @@ -139,3 +141,16 @@ char *read_string(int fd) { read_bytes(fd, &bytes, &length); return (char *) bytes; } + +void write_formatted_string(int socket_fd, const char *format, ...) { + UT_string *cmd; + va_list ap; + + utstring_new(cmd); + va_start(ap, format); + utstring_printf_va(cmd, format, ap); + va_end(ap); + + write_string(socket_fd, utstring_body(cmd)); + utstring_free(cmd); +} diff --git a/io.h b/io.h index c6dd3bb30..e6f227c98 100644 --- a/io.h +++ b/io.h @@ -16,6 +16,7 @@ void write_bytes(int fd, uint8_t *bytes, int64_t length); void read_bytes(int fd, uint8_t **bytes, int64_t *length); void write_string(int fd, char *message); +void write_formatted_string(int fd, const char *format, ...); char *read_string(int fd); #endif diff --git a/logging.c b/logging.c new file mode 100644 index 000000000..38abc7bf9 --- /dev/null +++ b/logging.c @@ -0,0 +1,78 @@ +#include "logging.h" + +#include +#include + +#include "state/redis.h" +#include "io.h" + +static const char *log_levels[5] = {"DEBUG", "INFO", "WARN", "ERROR", "FATAL"}; +static const char *log_fmt = + "HMSET log:%s:%s:%s log_level %s event_type %s message %s timestamp %s"; + +struct ray_logger_impl { + /* String that identifies this client type. */ + char *client_type; + /* Suppress all log messages below this level. */ + int log_level; + /* Whether or not we have a direct connection to Redis. */ + int is_direct; + /* Either a db_conn or a socket to a process with a db_conn, + * depending on the is_direct flag. */ + void *conn; +}; + +ray_logger *init_ray_logger(const char *client_type, + int log_level, + int is_direct, + void *conn) { + ray_logger *logger = malloc(sizeof(ray_logger)); + logger->client_type = client_type; + logger->log_level = log_level; + logger->is_direct = is_direct; + logger->conn = conn; + return logger; +} + +void free_ray_logger(ray_logger *logger) { + free(logger); +} + +void ray_log(ray_logger *logger, + int log_level, + const char *event_type, + const char *message) { + if (log_level < logger->log_level) { + return; + } + if (log_level < RAY_DEBUG || log_level > RAY_FATAL) { + return; + } + struct timeval tv; + UT_string *timestamp; + utstring_new(timestamp); + gettimeofday(&tv, NULL); + utstring_printf(timestamp, "%ld.%ld", tv.tv_sec, tv.tv_usec); + + UT_string *origin_id; + utstring_new(origin_id); + if (logger->is_direct) { + db_conn *db = (db_conn *) logger->conn; + utstring_printf(origin_id, "%ld:%s", db->client_id, ""); + redisAsyncCommand(db->context, NULL, NULL, log_fmt, + utstring_body(timestamp), logger->client_type, + utstring_body(origin_id), log_levels[log_level], + event_type, message, utstring_body(timestamp)); + } else { + /* If we don't own a Redis connection, we leave our client + * ID to be filled in by someone else. */ + utstring_printf(origin_id, "%s:%s", "%ld", "%ld"); + int *socket_fd = (int *) logger->conn; + write_formatted_string(*socket_fd, log_fmt, utstring_body(timestamp), + logger->client_type, utstring_body(origin_id), + log_levels[log_level], event_type, message, + utstring_body(timestamp)); + } + utstring_free(origin_id); + utstring_free(timestamp); +} diff --git a/logging.h b/logging.h new file mode 100644 index 000000000..4ef7c8fcb --- /dev/null +++ b/logging.h @@ -0,0 +1,39 @@ +#ifndef LOGGING_H +#define LOGGING_H + +#define RAY_VERBOSE -1 +#define RAY_DEBUG 0 +#define RAY_INFO 1 +#define RAY_WARNING 2 +#define RAY_ERROR 3 +#define RAY_FATAL 4 + +/* Entity types. */ +#define RAY_FUNCTION "FUNCTION" +#define RAY_OBJECT "OBJECT" +#define RAY_TASK "TASK" + +typedef struct ray_logger_impl ray_logger; + +/* Initialize a Ray logger for the given client type and logging level. If the + * is_direct flag is set, the logger will treat the given connection as a + * direct connection to the log. Otherwise, it will treat it as a socket to + * another process with a connection to the log. + * NOTE: User is responsible for freeing the returned logger. */ +ray_logger *init_ray_logger(const char *client_type, + int log_level, + int is_direct, + void *conn); + +/* Free the logger. This does not free the connection to the log. */ +void free_ray_logger(ray_logger *logger); + +/* Log an event at the given log level with the given event_type. + * NOTE: message cannot contain spaces! JSON format is recommended. + * TODO: Support spaces in messages. */ +void ray_log(ray_logger *logger, + int log_level, + const char *event_type, + const char *message); + +#endif diff --git a/state/redis.c b/state/redis.c index c2e01df5e..ac56206e8 100644 --- a/state/redis.c +++ b/state/redis.c @@ -232,23 +232,3 @@ void task_queue_submit_task(db_conn *db, task_iid task_iid, task_spec *task) { } utstring_free(command); } - -void send_redis_command(int socket_fd, const char *format, ...) { - char *cmd; - va_list ap; - int len; - - va_start(ap, format); - len = redisvFormatCommand(&cmd, format, ap); - va_end(ap); - if (len == -1) { - LOG_ERR("Out of memory while formatting Redis command."); - return; - } else if (len == -2) { - LOG_ERR("Invalid Redis format string."); - return; - } - - write_string(socket_fd, cmd); - free(cmd); -} diff --git a/state/redis.h b/state/redis.h index 132724e67..09d66bea6 100644 --- a/state/redis.h +++ b/state/redis.h @@ -1,3 +1,6 @@ +#ifndef REDIS_H +#define REDIS_H + #include "db.h" #include "object_table.h" @@ -47,4 +50,4 @@ void object_table_lookup_callback(redisAsyncContext *c, void *r, void *privdata); -void send_redis_command(int socket_fd, const char *format, ...); +#endif diff --git a/test/redis_tests.c b/test/redis_tests.c index 2f0f09110..d8df538e8 100644 --- a/test/redis_tests.c +++ b/test/redis_tests.c @@ -7,6 +7,7 @@ #include "state/db.h" #include "state/redis.h" #include "io.h" +#include "logging.h" SUITE(redis_tests); @@ -31,6 +32,16 @@ void async_redis_socket_test_callback(redisAsyncContext *ac, lookup_successful = 1; } +void logging_test_callback(redisAsyncContext *ac, void *r, void *privdata) { + redisContext *context = redisConnect("127.0.0.1", 6379); + redisReply *reply = redisCommand(context, "KEYS %s", "log:*"); + redisFree(context); + assert(reply != NULL); + assert(reply->elements > 0); + freeReplyObject(reply); + lookup_successful = 1; +} + TEST redis_socket_test(void) { const char *socket_pathname = "redis-test-socket"; redisContext *context = redisConnect("127.0.0.1", 6379); @@ -40,8 +51,7 @@ TEST redis_socket_test(void) { int client_fd = connect_ipc_sock(socket_pathname); ASSERT(client_fd >= 0); - - send_redis_command(client_fd, test_set_format, test_key, test_value); + write_formatted_string(client_fd, test_set_format, test_key, test_value); int server_fd = accept_client(socket_fd); char *cmd = read_string(server_fd); @@ -50,11 +60,10 @@ TEST redis_socket_test(void) { close(socket_fd); unlink(socket_pathname); - redisAppendFormattedCommand(context, cmd, strlen(cmd)); - redisReply *tmp; - redisGetReply(context, &tmp); - freeReplyObject(tmp); - redisReply *reply = redisCommand(context, "GET %s", test_key); + redisReply *reply; + reply = redisCommand(context, cmd, 0, 0); + freeReplyObject(reply); + reply = redisCommand(context, "GET %s", test_key); ASSERT(reply != NULL); ASSERT_STR_EQ(reply->str, test_value); freeReplyObject(reply); @@ -82,7 +91,7 @@ TEST async_redis_socket_test(void) { /* Send a command to the Redis process. */ client_fd = connect_ipc_sock(socket_pathname); ASSERT(client_fd >= 0); - send_redis_command(client_fd, test_set_format, test_key, test_value); + write_formatted_string(client_fd, test_set_format, test_key, test_value); while (!lookup_successful) { int num_ready = event_loop_poll(&loop, -1); @@ -104,9 +113,8 @@ TEST async_redis_socket_test(void) { event_loop_attach(&loop, 1, NULL, server_fd, POLLIN); } else { char *cmd = read_string(waiting->fd); - redisAsyncFormattedCommand(conn.context, - async_redis_socket_test_callback, NULL, cmd, - strlen(cmd)); + redisAsyncCommand(conn.context, async_redis_socket_test_callback, NULL, + cmd, conn.client_id, 0); free(cmd); } } @@ -121,11 +129,70 @@ TEST async_redis_socket_test(void) { PASS(); } +TEST logging_test(void) { + int socket_fd, server_fd, client_fd; + event_loop loop; + event_loop_init(&loop); + /* Start IPC channel. */ + const char *socket_pathname = "logging-test-socket"; + socket_fd = bind_ipc_sock(socket_pathname); + ASSERT(socket_fd >= 0); + int64_t ipc_index = event_loop_attach(&loop, 1, NULL, socket_fd, POLLIN); + + /* Start connection to Redis. */ + db_conn conn; + db_connect("127.0.0.1", 6379, "", "", 0, &conn); + int64_t db_index = db_attach(&conn, &loop, 0); + + /* Send a command to the Redis process. */ + client_fd = connect_ipc_sock(socket_pathname); + ASSERT(client_fd >= 0); + ray_logger *logger = init_ray_logger("worker", RAY_INFO, 0, &client_fd); + ray_log(logger, RAY_INFO, "TEST", "Message"); + + while (!lookup_successful) { + int num_ready = event_loop_poll(&loop, -1); + if (num_ready < 0) { + exit(-1); + } + for (int i = 0; i < event_loop_size(&loop); ++i) { + struct pollfd *waiting = event_loop_get(&loop, i); + if (waiting->revents == 0) + continue; + if (i == db_index) { + db_event(&conn); + } else if (i == ipc_index) { + /* For some reason, this check is necessary for Travis + * to pass these tests. */ + ASSERT(waiting->revents & POLLIN); + server_fd = accept_client(socket_fd); + ASSERT(server_fd >= 0); + event_loop_attach(&loop, 1, NULL, server_fd, POLLIN); + } else { + char *cmd = read_string(waiting->fd); + redisAsyncCommand(conn.context, logging_test_callback, NULL, cmd, + conn.client_id, 0); + free(cmd); + } + } + } + free_ray_logger(logger); + db_disconnect(&conn); + event_loop_free(&loop); + close(server_fd); + close(client_fd); + close(socket_fd); + unlink(socket_pathname); + lookup_successful = 0; + PASS(); +} + SUITE(redis_tests) { redisContext *context = redisConnect("127.0.0.1", 6379); freeReplyObject(redisCommand(context, "FLUSHALL")); RUN_REDIS_TEST(context, redis_socket_test); RUN_REDIS_TEST(context, async_redis_socket_test); + RUN_REDIS_TEST(context, logging_test); redisFree(context); }