/* ********************************************************************************************** * File: maat_command.c * Description: * Authors: Liu WenTan * Date: 2022-10-31 * Copyright: (c) Since 2022 Geedge Networks, Ltd. All rights reserved. *********************************************************************************************** */ #include #include #include #include #include "maat_utils.h" #include "maat_command.h" #include "maat_rule.h" #include "hiredis/hiredis.h" #include "maat_config_monitor.h" #include "maat_redis_monitor.h" #define MODULE_MAAT_COMMAND module_name_str("maat.command") extern const char *foreign_source_prefix; extern const char *foreign_key_prefix; extern const char *mr_key_prefix; extern const char *mr_expire_lock; extern const long mr_expire_lock_time; extern const char *mr_status_sset; extern const char *mr_version_sset; extern const char *mr_label_sset; extern const int MAAT_REDIS_SYNC_TIME; redisReply *maat_wrap_redis_command(redisContext *c, const char *format, ...) { va_list ap; void *reply = NULL; int ret = REDIS_ERR; int retry = 0; while (reply == NULL && retry < 2 && ret != REDIS_OK) { va_start(ap,format); reply = redisvCommand(c,format,ap); va_end(ap); if (NULL == reply) { ret = redisReconnect(c); retry++; } } return (redisReply *)reply; } redisContext *maat_connect_redis(const char *redis_ip, int redis_port, int redis_db, struct log_handle *logger) { struct timeval connect_timeout; connect_timeout.tv_sec = 0; connect_timeout.tv_usec = 100 * 1000; // 100 ms redisContext *c = redisConnectWithTimeout(redis_ip, redis_port, connect_timeout); if (NULL == c || c->err) { if (NULL == logger) { printf("Unable to connect redis server %s:%d db%d, error: %s", redis_ip, redis_port, redis_db, c == NULL ? "Unknown" : c->errstr); } else { log_fatal(logger, MODULE_MAAT_COMMAND, "[%s:%d] Unable to connect redis server %s:%d db%d, error: %s", __FUNCTION__, __LINE__, redis_ip, redis_port, redis_db, c == NULL ? "Unknown" : c->errstr); } if (c != NULL) { redisFree(c); } return NULL; } redisEnableKeepAlive(c); redisReply *reply = maat_wrap_redis_command(c, "select %d", redis_db); freeReplyObject(reply); reply = NULL; return c; } long long maat_read_redis_integer(const redisReply *reply) { switch (reply->type) { case REDIS_REPLY_INTEGER: return reply->integer; break; case REDIS_REPLY_ARRAY: assert(reply->element[0]->type == REDIS_REPLY_INTEGER); return reply->element[0]->integer; break; case REDIS_REPLY_STRING: return atoll(reply->str); break; default: return -1; break; } return 0; } static int redis_flushDB(redisContext *ctx, int db_index, struct log_handle *logger) { long long maat_redis_version = 0; redisReply *data_reply = maat_wrap_redis_command(ctx, "WATCH MAAT_VERSION"); freeReplyObject(data_reply); data_reply = NULL; data_reply = maat_wrap_redis_command(ctx, "GET MAAT_VERSION"); if (data_reply->type == REDIS_REPLY_NIL) { maat_redis_version = 0; } else { maat_redis_version = maat_read_redis_integer(data_reply); maat_redis_version++; freeReplyObject(data_reply); data_reply = NULL; } data_reply = maat_wrap_redis_command(ctx, "DBSIZE"); long long dbsize = maat_read_redis_integer(data_reply); freeReplyObject(data_reply); data_reply = NULL; data_reply = maat_wrap_redis_command(ctx, "MULTI"); freeReplyObject(data_reply); data_reply = NULL; int append_cmd_cnt = 0; redisAppendCommand(ctx, "FLUSHDB"); append_cmd_cnt++; redisAppendCommand(ctx, "SET MAAT_VERSION %lld", maat_redis_version); append_cmd_cnt++; redisAppendCommand(ctx, "SET MAAT_PRE_VER %lld", maat_redis_version); append_cmd_cnt++; redisAppendCommand(ctx, "SET %s 1", mr_region_id_var); append_cmd_cnt++; redisAppendCommand(ctx, "SET %s 1", mr_group_id_var); append_cmd_cnt++; redisAppendCommand(ctx, "EXEC"); append_cmd_cnt++; int ret = 0; int redis_transaction_success = 1; for (int i = 0; i < append_cmd_cnt; i++) { ret = maat_wrap_redis_get_reply(ctx, &data_reply); if (ret == REDIS_OK) { if (data_reply->type == REDIS_REPLY_NIL) { redis_transaction_success = 0; } freeReplyObject(data_reply); data_reply = NULL; } } if (redis_transaction_success == 1) { log_info(logger, MODULE_MAAT_COMMAND, "FlushDB %d, MAAT_VERSION:%llu, DBSize:%llu.", db_index, (maat_redis_version == 0)?0:(maat_redis_version-1),dbsize); } return redis_transaction_success; } static int connect_redis_for_write(struct source_redis_ctx *redis_ctx, struct log_handle *logger) { assert(redis_ctx->write_ctx == NULL); redis_ctx->write_ctx = maat_connect_redis(redis_ctx->redis_ip, redis_ctx->redis_port, redis_ctx->redis_db, logger); if (NULL == redis_ctx->write_ctx) { return -1; } else { return 0; } } static redisContext *get_redis_ctx_for_write(struct maat *maat_inst) { if (NULL == maat_inst->opts.redis_ctx.write_ctx) { int ret = connect_redis_for_write(&(maat_inst->opts.redis_ctx), maat_inst->logger); if (ret != 0) { return NULL; } } return maat_inst->opts.redis_ctx.write_ctx; } int maat_cmd_flushDB(struct maat *maat_inst) { int ret = 0; redisContext *write_ctx = get_redis_ctx_for_write(maat_inst); if (NULL == write_ctx) { return -1; } do { ret = redis_flushDB(maat_inst->opts.redis_ctx.write_ctx, maat_inst->opts.redis_ctx.redis_db, maat_inst->logger); } while(0 == ret); return 0; } int maat_get_valid_flag_offset(const char *line, int column_seq) { size_t offset = 0; size_t len = 0; int ret = get_column_pos(line, column_seq, &offset, &len); // 0 is also a valid value for some non-MAAT producer. if (ret < 0 || offset >= strlen(line) || (line[offset] != '1' && line[offset] != '0')) { return -1; } return offset; } long long maat_redis_server_time_s(redisContext *c) { long long server_time = 0; redisReply *data_reply = maat_wrap_redis_command(c, "TIME"); if (data_reply->type == REDIS_REPLY_ARRAY) { server_time = atoll(data_reply->element[0]->str); freeReplyObject(data_reply); data_reply = NULL; } return server_time; } int maat_wrap_redis_get_reply(redisContext *c, redisReply **reply) { return redisGetReply(c, (void **)reply); } int maat_cmd_set_line(struct maat *maat_inst, const struct maat_cmd_line *line_rule) { int ret = 0; long long absolute_expire_time = 0; redisContext *write_ctx = get_redis_ctx_for_write(maat_inst); if (NULL == write_ctx) { return -1; } long long server_time = maat_redis_server_time_s(write_ctx); if(!server_time) { return -1; } struct serial_rule *s_rule = ALLOC(struct serial_rule, 1); int table_id = table_manager_get_table_id(maat_inst->tbl_mgr, line_rule->table_name); if (table_id < 0) { log_fatal(maat_inst->logger, MODULE_MAAT_COMMAND, "[%s:%d] Command set line id %lld failed: unknown table %s", __FUNCTION__, __LINE__, line_rule->rule_id, line_rule->table_name); FREE(s_rule); return -1; } int valid_column = table_manager_get_valid_column(maat_inst->tbl_mgr, table_id); if (valid_column < 0) { log_fatal(maat_inst->logger, MODULE_MAAT_COMMAND, "[%s:%d] Command set line id %lld failed: table %s is not a plugin or ip_plugin table", __FUNCTION__, __LINE__, line_rule->rule_id, line_rule->table_name); FREE(s_rule); return -1; } int valid_offset = maat_get_valid_flag_offset(line_rule->table_line, valid_column); if (valid_offset < 0) { log_fatal(maat_inst->logger, MODULE_MAAT_COMMAND, "[%s:%d] Command set line id %lld failed: table %s valid_offset error", __FUNCTION__, __LINE__, line_rule->rule_id, line_rule->table_name); FREE(s_rule); return -1; } int is_valid = atoi(line_rule->table_line + valid_offset); if (line_rule->expire_after > 0) { absolute_expire_time = server_time + line_rule->expire_after; } maat_set_serial_rule(s_rule, (enum maat_operation)is_valid, line_rule->rule_id, line_rule->table_name, line_rule->table_line, absolute_expire_time); int success_cnt = maat_cmd_write_rule(write_ctx, s_rule, 1, server_time, maat_inst->logger); if (success_cnt != 1) { ret = -1; goto error_out; } ret = success_cnt; maat_inst->stat->line_cmd_acc_num += success_cnt; error_out: maat_clear_rule_cache(s_rule); FREE(s_rule); return ret; } int maat_cmd_set_file(struct maat *maat_inst, const char *key, const char *value, size_t size, enum maat_operation op) { redisContext *ctx = maat_inst->opts.redis_ctx.write_ctx; if (NULL == ctx) { log_fatal(maat_inst->logger, MODULE_MAAT_COMMAND, "[%s:%d] failed: Redis is not connected.", __FUNCTION__, __LINE__); return -1; } const char *arg_vec[3]; size_t len_vec[3]; arg_vec[0] = "SET"; len_vec[0] = strlen("SET"); arg_vec[1] = key; len_vec[1] = strlen(key); arg_vec[2] = value; len_vec[2] = size; redisReply *reply = NULL; if (0 != strncmp(key, foreign_key_prefix, strlen(foreign_key_prefix))) { log_fatal(maat_inst->logger, MODULE_MAAT_COMMAND, "Invalid File key, prefix %s is mandatory.", foreign_key_prefix); return -1; } switch (op) { case MAAT_OP_ADD: reply = (redisReply *)redisCommandArgv(ctx, sizeof(arg_vec) / sizeof(arg_vec[0]), arg_vec, len_vec); break; case MAAT_OP_DEL: reply = maat_wrap_redis_command(ctx, "EXPIRE %s %d", key, MAAT_REDIS_SYNC_TIME); break; default: return -1; break; } if (NULL == reply || reply->type == REDIS_REPLY_NIL || reply->type == REDIS_REPLY_ERROR) { log_fatal(maat_inst->logger, MODULE_MAAT_COMMAND, "Set file failed, maybe Redis is busy."); freeReplyObject(reply); reply = NULL; return -1; } freeReplyObject(reply); reply = NULL; return 1; } long long maat_cmd_incrby(struct maat *maat_inst, const char *key, int increment) { long long result = 0; redisContext *write_ctx = get_redis_ctx_for_write(maat_inst); if (NULL == write_ctx) { return -1; } redisReply *data_reply = maat_wrap_redis_command(write_ctx, "INCRBY %s %d", key, increment); if (data_reply->type == REDIS_REPLY_INTEGER) { result = data_reply->integer; } else { result = -1; } freeReplyObject(data_reply); data_reply = NULL; return result; }