summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
authorliuxueli <[email protected]>2023-03-22 16:59:56 +0800
committerliuxueli <[email protected]>2023-03-22 16:59:56 +0800
commit6e499f73b343c764e561ba0c806df12ee30b9377 (patch)
tree7eca404ecefb942f3309f981905a19d28be2e2ca
parent3f04cf081b9a6b3a0152ddebccb6b0366ec48dd1 (diff)
TSG-14382: 功能端支持对发送kafka的日志内容进行压缩v5.8.12
-rw-r--r--src/tsg_send_log.cpp6
-rw-r--r--src/tsg_send_log_internal.h1
2 files changed, 7 insertions, 0 deletions
diff --git a/src/tsg_send_log.cpp b/src/tsg_send_log.cpp
index 93181a0..f5999bc 100644
--- a/src/tsg_send_log.cpp
+++ b/src/tsg_send_log.cpp
@@ -1906,6 +1906,7 @@ struct tsg_log_instance_t *tsg_sendlog_init(const char *conffile, screen_stat_ha
MESA_load_profile_string_def(conffile, "TSG_LOG", "BROKER_LIST", _instance->broker_list, sizeof(_instance->broker_list), NULL);
MESA_load_profile_string_def(conffile, "TSG_LOG", "SASL_USERNAME", _instance->sasl_username, sizeof(_instance->sasl_username), ""); //admin
MESA_load_profile_string_def(conffile, "TSG_LOG", "SASL_PASSWD", _instance->sasl_passwd, sizeof(_instance->sasl_passwd), "");
+ MESA_load_profile_string_def(conffile, "TSG_LOG", "COMPRESSION_TYPE", _instance->compression, sizeof(_instance->compression), ""); //snappy
MESA_load_profile_string_def(conffile, "TSG_LOG", "SEND_QUEUE_MAX_MESSAGE", _instance->send_queue_max_msg, sizeof(_instance->send_queue_max_msg), "1000000");
MESA_load_profile_string_def(conffile, "TSG_LOG", "REFRESH_INTERVAL_MS", _instance->refresh_interval_ms, sizeof(_instance->refresh_interval_ms), "600000");
@@ -1956,6 +1957,11 @@ struct tsg_log_instance_t *tsg_sendlog_init(const char *conffile, screen_stat_ha
rd_kafka_conf_set(rdkafka_conf, "request.required.acks", _instance->require_ack, kafka_errstr, sizeof(kafka_errstr));
rd_kafka_conf_set(rdkafka_conf, "socket.keepalive.enable", "true", kafka_errstr, sizeof(kafka_errstr));
rd_kafka_conf_set(rdkafka_conf, "bootstrap.servers", _instance->broker_list, kafka_errstr, sizeof(kafka_errstr));
+
+ if(strlen(_instance->compression)>0)
+ {
+ rd_kafka_conf_set(rdkafka_conf, "compression.codec", _instance->compression, kafka_errstr, sizeof(kafka_errstr));
+ }
if(strlen(_instance->sasl_username)> 0 && strlen(_instance->sasl_passwd)>0)
{
diff --git a/src/tsg_send_log_internal.h b/src/tsg_send_log_internal.h
index cddf3af..b0dea70 100644
--- a/src/tsg_send_log_internal.h
+++ b/src/tsg_send_log_internal.h
@@ -217,6 +217,7 @@ struct tsg_log_instance_t
char common_field_file[MAX_STRING_LEN*4];
char broker_list[MAX_STRING_LEN*8];
char sasl_username[MAX_STRING_LEN];
+ char compression[MAX_STRING_LEN];
char sasl_passwd[MAX_STRING_LEN];
char send_queue_max_msg[MAX_STRING_LEN];
char require_ack[MAX_STRING_LEN];