From bef117d7db3485ea146eead2f55e01a1492d2f37 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Mon, 7 Mar 2022 13:52:49 +0800 Subject: [PATCH] enable monitor gzip --- include/common/tglobal.h | 1 + include/libs/monitor/monitor.h | 1 + include/util/thttp.h | 4 ++- source/common/src/tglobal.c | 3 ++ source/dnode/mgmt/impl/src/dndEnv.c | 2 +- source/dnode/mnode/impl/src/mndTelem.c | 2 +- source/libs/monitor/inc/monInt.h | 1 + source/libs/monitor/src/monitor.c | 3 +- source/util/src/thttp.c | 49 ++++++++++++++++---------- 9 files changed, 44 insertions(+), 22 deletions(-) diff --git a/include/common/tglobal.h b/include/common/tglobal.h index 2261170e63..ba41cb0292 100644 --- a/include/common/tglobal.h +++ b/include/common/tglobal.h @@ -58,6 +58,7 @@ extern int32_t tsMonitorInterval; extern char tsMonitorFqdn[]; extern uint16_t tsMonitorPort; extern int32_t tsMonitorMaxLogs; +extern bool tsMonitorComp; // query buffer management extern int32_t tsQueryBufferSize; // maximum allowed usage buffer size in MB for each data node during query processing diff --git a/include/libs/monitor/monitor.h b/include/libs/monitor/monitor.h index 0c832f802b..a3e049b00c 100644 --- a/include/libs/monitor/monitor.h +++ b/include/libs/monitor/monitor.h @@ -130,6 +130,7 @@ typedef struct { const char *server; uint16_t port; int32_t maxLogs; + bool comp; } SMonCfg; int32_t monInit(const SMonCfg *pCfg); diff --git a/include/util/thttp.h b/include/util/thttp.h index f211b2615d..31d84850e0 100644 --- a/include/util/thttp.h +++ b/include/util/thttp.h @@ -22,7 +22,9 @@ extern "C" { #endif -int32_t taosSendHttpReport(const char* server, uint16_t port, const char* pCont, int32_t contLen); +typedef enum { HTTP_GZIP, HTTP_FLAT } EHttpCompFlag; + +int32_t taosSendHttpReport(const char* server, uint16_t port, const char* pCont, int32_t contLen, EHttpCompFlag flag); #ifdef __cplusplus } diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index a521e10cc7..f62f8e5d7a 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -52,6 +52,7 @@ int32_t tsMonitorInterval = 5; char tsMonitorFqdn[TSDB_FQDN_LEN] = {0}; uint16_t tsMonitorPort = 6043; int32_t tsMonitorMaxLogs = 100; +bool tsMonitorComp = false; /* * denote if the server needs to compress response message at the application layer to client, including query rsp, @@ -346,6 +347,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) { if (cfgAddString(pCfg, "monitorFqdn", tsMonitorFqdn, 0) != 0) return -1; if (cfgAddInt32(pCfg, "monitorPort", tsMonitorPort, 1, 65056, 0) != 0) return -1; if (cfgAddInt32(pCfg, "monitorMaxLogs", tsMonitorMaxLogs, 1, 1000000, 0) != 0) return -1; + if (cfgAddBool(pCfg, "monitorComp", tsMonitorComp, 0) != 0) return -1; return 0; } @@ -462,6 +464,7 @@ static int32_t taosSetServerCfg(SConfig *pCfg) { tstrncpy(tsMonitorFqdn, cfgGetItem(pCfg, "monitorFqdn")->str, TSDB_FQDN_LEN); tsMonitorPort = (uint16_t)cfgGetItem(pCfg, "monitorPort")->i32; tsMonitorMaxLogs = cfgGetItem(pCfg, "monitorMaxLogs")->i32; + tsMonitorComp = cfgGetItem(pCfg, "monitorComp")->bval; if (tsQueryBufferSize >= 0) { tsQueryBufferSizeBytes = tsQueryBufferSize * 1048576UL; diff --git a/source/dnode/mgmt/impl/src/dndEnv.c b/source/dnode/mgmt/impl/src/dndEnv.c index 1cbdf0c1dc..84b2dca326 100644 --- a/source/dnode/mgmt/impl/src/dndEnv.c +++ b/source/dnode/mgmt/impl/src/dndEnv.c @@ -298,7 +298,7 @@ int32_t dndInit() { return -1; } - SMonCfg monCfg = {.maxLogs = tsMonitorMaxLogs, .port = tsMonitorPort, .server = tsMonitorFqdn}; + SMonCfg monCfg = {.maxLogs = tsMonitorMaxLogs, .port = tsMonitorPort, .server = tsMonitorFqdn, .comp = tsMonitorComp}; if (monInit(&monCfg) != 0) { dError("failed to init monitor since %s", terrstr()); dndCleanup(); diff --git a/source/dnode/mnode/impl/src/mndTelem.c b/source/dnode/mnode/impl/src/mndTelem.c index 0e2141a4d7..453535c6e7 100644 --- a/source/dnode/mnode/impl/src/mndTelem.c +++ b/source/dnode/mnode/impl/src/mndTelem.c @@ -87,7 +87,7 @@ static int32_t mndProcessTelemTimer(SMnodeMsg* pReq) { taosWLockLatch(&pMgmt->lock); char* pCont = mndBuildTelemetryReport(pMnode); if (pCont != NULL) { - taosSendHttpReport(TELEMETRY_SERVER, TELEMETRY_PORT, pCont, strlen(pCont)); + taosSendHttpReport(TELEMETRY_SERVER, TELEMETRY_PORT, pCont, strlen(pCont), HTTP_FLAT); free(pCont); } taosWUnLockLatch(&pMgmt->lock); diff --git a/source/libs/monitor/inc/monInt.h b/source/libs/monitor/inc/monInt.h index bfb73af034..c3b6569555 100644 --- a/source/libs/monitor/inc/monInt.h +++ b/source/libs/monitor/inc/monInt.h @@ -54,6 +54,7 @@ typedef struct { int32_t maxLogs; const char *server; uint16_t port; + bool comp; SMonState state; } SMonitor; diff --git a/source/libs/monitor/src/monitor.c b/source/libs/monitor/src/monitor.c index 354989a7a1..905ff8d6da 100644 --- a/source/libs/monitor/src/monitor.c +++ b/source/libs/monitor/src/monitor.c @@ -45,6 +45,7 @@ int32_t monInit(const SMonCfg *pCfg) { tsMonitor.maxLogs = pCfg->maxLogs; tsMonitor.server = pCfg->server; tsMonitor.port = pCfg->port; + tsMonitor.comp = pCfg->comp; tsLogFp = monRecordLog; tsMonitor.state.time = taosGetTimestampMs(); pthread_mutex_init(&tsMonitor.lock, NULL); @@ -375,7 +376,7 @@ void monSendReport(SMonInfo *pMonitor) { char *pCont = tjsonToString(pMonitor->pJson); if (pCont != NULL) { - taosSendHttpReport(tsMonitor.server, tsMonitor.port, pCont, strlen(pCont)); + taosSendHttpReport(tsMonitor.server, tsMonitor.port, pCont, strlen(pCont), tsMonitor.comp); free(pCont); } } diff --git a/source/util/src/thttp.c b/source/util/src/thttp.c index b8d73478ea..c8a52a4735 100644 --- a/source/util/src/thttp.c +++ b/source/util/src/thttp.c @@ -18,6 +18,28 @@ #include "taoserror.h" #include "tlog.h" +static int32_t taosBuildHttpHeader(const char* server, int32_t contLen, char* pHead, int32_t headLen, + EHttpCompFlag flag) { + if (flag == HTTP_FLAT) { + return snprintf(pHead, headLen, + "POST /report HTTP/1.1\n" + "Host: %s\n" + "Content-Type: application/json\n" + "Content-Length: %d\n\n", + server, contLen); + } else if (flag == HTTP_GZIP) { + return snprintf(pHead, headLen, + "POST /report HTTP/1.1\n" + "Host: %s\n" + "Content-Type: application/json\n" + "Content-Encoding: gzip\n" + "Content-Length: %d\n\n", + server, contLen); + } else { + return -1; + } +} + #ifdef USE_UV static void clientConnCb(uv_connect_t* req, int32_t status) { if(status < 0) { @@ -36,7 +58,7 @@ static void clientConnCb(uv_connect_t* req, int32_t status) { uv_close((uv_handle_t *)req->handle,NULL); } -int32_t taosSendHttpReport(const char* server, uint16_t port, const char* pCont, int32_t contLen) { +int32_t taosSendHttpReport(const char* server, uint16_t port, const char* pCont, int32_t contLen, EHttpCompFlag flag) { uint32_t ipv4 = taosGetIpv4FromFqdn(server); if (ipv4 == 0xffffffff) { terrno = TAOS_SYSTEM_ERROR(errno); @@ -50,18 +72,14 @@ int32_t taosSendHttpReport(const char* server, uint16_t port, const char* pCont, struct sockaddr_in dest = {0}; uv_ip4_addr(ipv4Buf, port, &dest); - uv_tcp_t socket_tcp = {0}; - uv_loop_t *loop = uv_default_loop(); + uv_tcp_t socket_tcp = {0}; + uv_loop_t* loop = uv_default_loop(); uv_tcp_init(loop, &socket_tcp); uv_connect_t* connect = (uv_connect_t*)malloc(sizeof(uv_connect_t)); char header[1024] = {0}; - int32_t headLen = snprintf(header, sizeof(header), - "POST /report HTTP/1.1\n" - "Host: %s\n" - "Content-Type: application/json\n" - "Content-Length: %d\n\n", - server, contLen); + int32_t headLen = taosBuildHttpHeader(server, contLen, header, sizeof(header), flag); + uv_buf_t wb[2]; wb[0] = uv_buf_init((char*)header, headLen); wb[1] = uv_buf_init((char*)pCont, contLen); @@ -76,7 +94,7 @@ int32_t taosSendHttpReport(const char* server, uint16_t port, const char* pCont, } #else -int32_t taosSendHttpReport(const char* server, uint16_t port, const char* pCont, int32_t contLen) { +int32_t taosSendHttpReport(const char* server, uint16_t port, const char* pCont, int32_t contLen, EHttpCompFlag flag) { int32_t code = -1; SOCKET fd = 0; @@ -94,15 +112,10 @@ int32_t taosSendHttpReport(const char* server, uint16_t port, const char* pCont, goto SEND_OVER; } - char header[4096] = {0}; - int32_t headLen = snprintf(header, sizeof(header), - "POST /report HTTP/1.1\n" - "Host: %s\n" - "Content-Type: application/json\n" - "Content-Length: %d\n\n", - server, contLen); + char header[1024] = {0}; + int32_t headLen = taosBuildHttpHeader(server, contLen, header, sizeof(header), flag); - if (taosWriteSocket(fd, (void*)header, headLen) < 0) { + if (taosWriteSocket(fd, header, headLen) < 0) { terrno = TAOS_SYSTEM_ERROR(errno); uError("failed to send http header to %s:%u since %s", server, port, terrstr()); goto SEND_OVER; -- GitLab