diff options
Diffstat (limited to 'common/src/tfe_resource.cpp')
| -rw-r--r-- | common/src/tfe_resource.cpp | 136 |
1 files changed, 105 insertions, 31 deletions
diff --git a/common/src/tfe_resource.cpp b/common/src/tfe_resource.cpp index 9caae9a..2a4e73f 100644 --- a/common/src/tfe_resource.cpp +++ b/common/src/tfe_resource.cpp @@ -10,12 +10,7 @@ #define MAAT_INPUT_REDIS 1 #define MAAT_INPUT_FILE 2 -struct maat_table_info -{ - int id; - const char *name; -}; - +static int scan_table_id[TABLE_TYPE_MAX]; static struct maat *static_maat = NULL; static tfe_kafka_logger_t *kafka_logger = NULL; static struct tfe_fieldstat_metric_t *dynamic_fieldstat = NULL; @@ -184,7 +179,8 @@ static tfe_kafka_logger_t *create_kafka_logger(const char *profile, const char * int enable = 0, vsystem_id = 0; char nic_name[TFE_SYMBOL_MAX] = {0}; char brokerlist[TFE_STRING_MAX] = {0}; - char topic_name[TFE_STRING_MAX] = {0}; + char logger_topic[TFE_STRING_MAX] = {0}; + char bucket_topic[TFE_STRING_MAX] = {0}; char sasl_username[TFE_STRING_MAX] = {0}; char sasl_passwd[TFE_STRING_MAX] = {0}; tfe_kafka_logger_t *kafka_logger = NULL; @@ -193,7 +189,8 @@ static tfe_kafka_logger_t *create_kafka_logger(const char *profile, const char * MESA_load_profile_int_def(profile, section, "VSYSTEM_ID", &vsystem_id, 1); MESA_load_profile_string_def(profile, section, "NIC_NAME", nic_name, sizeof(nic_name), "eth0"); MESA_load_profile_string_def(profile, section, "KAFKA_BROKERLIST", brokerlist, sizeof(brokerlist), ""); - MESA_load_profile_string_def(profile, section, "KAFKA_TOPIC", topic_name, sizeof(topic_name), "POLICY-EVENT-LOG"); + MESA_load_profile_string_def(profile, section, "LOGGER_SEND_TOPIC", logger_topic, sizeof(logger_topic), "PROXY-EVENT"); + MESA_load_profile_string_def(profile, section, "FILE_BUCKET_TOPIC", bucket_topic, sizeof(bucket_topic), "TRAFFIC-FILE-STREAM-RECORD"); MESA_load_profile_string_def(profile, section, "SASL_USERNAME", sasl_username, sizeof(sasl_username), ""); MESA_load_profile_string_def(profile, section, "SASL_PASSWD", sasl_passwd, sizeof(sasl_passwd), ""); @@ -203,18 +200,24 @@ static tfe_kafka_logger_t *create_kafka_logger(const char *profile, const char * return NULL; } - kafka_logger = tfe_kafka_logger_create(enable, nic_name, brokerlist, topic_name, sasl_username, sasl_passwd, logger); + kafka_logger = tfe_kafka_logger_create(enable, nic_name, brokerlist, logger_topic, sasl_username, sasl_passwd, logger); if (kafka_logger == NULL) { TFE_LOG_ERROR(logger, "tfe kafka init failed, error to create kafka logger."); return NULL; } - kafka_logger->t_vsys_id=vsystem_id; + int ret = tfe_kafka_logger_topic_new(kafka_logger, bucket_topic, logger); + if(ret == 0) + { + return NULL; + } + kafka_logger->t_vsys_id=vsystem_id; - TFE_LOG_INFO(logger, "tfe kafka logger : %s", enable ? "ENABLE" : "DISABLE"); - TFE_LOG_INFO(logger, "tfe kafka vsystem id : %d", vsystem_id); - TFE_LOG_INFO(logger, "tfe kafka topic : %s", topic_name); - TFE_LOG_INFO(logger, "tfe kafka brokerlist : %s", brokerlist); + TFE_LOG_INFO(logger, "tfe kafka logger : %s", enable ? "ENABLE" : "DISABLE"); + TFE_LOG_INFO(logger, "tfe kafka vsystem id : %d", vsystem_id); + TFE_LOG_INFO(logger, "tfe logger kafka topic : %s", logger_topic); + TFE_LOG_INFO(logger, "tfe bucket kafka topic : %s", bucket_topic); + TFE_LOG_INFO(logger, "tfe kafka brokerlist : %s", brokerlist); if (strlen(sasl_username) > 0 && strlen(sasl_passwd) > 0) { @@ -302,25 +305,96 @@ static char* create_effective_device_tag(const char *profile, const char *sectio return effective_device_tag; } -static struct maat_table_info maat_pub_tables[TABLE_TYPE_MAX] = { - {0, "ATTR_SOURCE_ASN"}, - {0, "ATTR_DESTINATION_ASN"}, - {0, "ATTR_SOURCE_LOCATION"}, - {0, "ATTR_DESTINATION_LOCATION"}, - {0, "ATTR_SUBSCRIBER_ID"}}; +void app_dict_table_new_cb(const char *table_name, int table_id, const char* key, const char* table_line, void **ad, long argl, void* argp) +{ + int ret=0; + size_t offset=0, len=0; + char *app_id_str=NULL, *group_id_str=NULL; + struct app_id_dict *app_dict=ALLOC(struct app_id_dict, 1); -static int register_maat_table() + ret = maat_helper_read_column(table_line, 1, &offset, &len); + if(ret >= 0) + { + app_id_str=ALLOC(char, len+1); + memcpy(app_id_str, table_line+offset, len); + app_dict->app_id=atoi(app_id_str); + FREE(&app_id_str); + } + + ret = maat_helper_read_column(table_line, 18, &offset, &len); + if(ret >= 0) + { + group_id_str=ALLOC(char, len+1); + memcpy(group_id_str, table_line+offset, len); + app_dict->group_id=atoll(group_id_str); + FREE(&group_id_str); + } + + app_dict->ref_cnt=1; + pthread_mutex_init(&(app_dict->lock), NULL); + *ad=app_dict; + return; +} + +void app_dict_table_free_cb(int table_id, void **ad, long argl, void* argp) { + if(*ad==NULL) + { + return; + } + + struct app_id_dict *app_dict=(struct app_id_dict *)(*ad); + pthread_mutex_lock(&(app_dict->lock)); + app_dict->ref_cnt--; + if(app_dict->ref_cnt>0) + { + pthread_mutex_unlock(&(app_dict->lock)); + return; + } + pthread_mutex_unlock(&(app_dict->lock)); + pthread_mutex_destroy(&(app_dict->lock)); + + FREE(&app_dict); + *ad=NULL; + return; +} + +void app_id_dict_free(struct app_id_dict *app_dict) +{ + app_dict_table_free_cb(0, (void **)&app_dict, 0, NULL); +} + +void app_dict_table_dup_cb(int table_id, void **to, void **from, long argl, void* argp) +{ + struct app_id_dict *app_dict=(struct app_id_dict *)(*from); + pthread_mutex_lock(&(app_dict->lock)); + app_dict->ref_cnt++; + pthread_mutex_unlock(&(app_dict->lock)); + *to=app_dict; + + return; +} + +static int maat_common_table_init() +{ + const char * table_name[TABLE_TYPE_MAX]; + table_name[TABLE_SECURITY_SOURCE_ASN] = "ATTR_SOURCE_ASN"; + table_name[TABLE_SECURITY_DESTINATION_ASN]="ATTR_DESTINATION_ASN"; + table_name[TABLE_SECURITY_SOURCE_LOCATION] = "ATTR_SOURCE_LOCATION"; + table_name[TABLE_SECURITY_DESTINATION_LOCATION] = "ATTR_DESTINATION_LOCATION"; + table_name[TABLE_OBJ_SUBSCRIBER_ID] = "ATTR_SUBSCRIBER_ID"; + table_name[TABLE_OBJ_APP_ID_DICT] = "APP_ID_DICT"; + for (int i = 0; i < TABLE_TYPE_MAX; i++) - { - maat_pub_tables[i].id = maat_get_table_id(static_maat, maat_pub_tables[i].name); - if (maat_pub_tables[i].id < 0) - { - TFE_LOG_ERROR(g_default_logger, "Maat table %s register failed.", maat_pub_tables[i].name); + { + scan_table_id[i] = maat_get_table_id(static_maat, table_name[i]); + if (scan_table_id[i] < 0) + { + TFE_LOG_ERROR(g_default_logger, "Maat table %s register failed.", table_name[i]); return -1; - } - } - + } + } + maat_plugin_table_ex_schema_register(static_maat, "APP_ID_DICT", app_dict_table_new_cb, app_dict_table_free_cb, app_dict_table_dup_cb, 0, NULL); return 0; } @@ -350,7 +424,7 @@ int tfe_bussiness_resouce_init() effective_device_tag = create_effective_device_tag(profile_path, "MAAT", g_default_logger); - if (register_maat_table()) + if (maat_common_table_init()) { return -1; } @@ -379,5 +453,5 @@ void *tfe_bussiness_resouce_get(enum RESOURCE_TYPE type) int tfe_bussiness_tableid_get(enum TABLE_TYPE type) { - return maat_pub_tables[type].id; + return scan_table_id[type]; }
\ No newline at end of file |