From 514f9f6626de43b4657b12733ae1d2d89a52a620 Mon Sep 17 00:00:00 2001 From: Shengliang Guan Date: Mon, 7 Mar 2022 14:54:57 +0800 Subject: [PATCH] enable monitor gzip --- include/util/taoserror.h | 1 + include/util/thttp.h | 2 +- source/libs/monitor/src/monitor.c | 3 +- source/util/CMakeLists.txt | 2 +- source/util/src/terror.c | 1 + source/util/src/thttp.c | 101 +++++++++++++++++++++++++++--- 6 files changed, 100 insertions(+), 10 deletions(-) diff --git a/include/util/taoserror.h b/include/util/taoserror.h index 5fe69284a2..6ad34eb0c0 100644 --- a/include/util/taoserror.h +++ b/include/util/taoserror.h @@ -84,6 +84,7 @@ int32_t* taosGetErrno(); #define TSDB_CODE_INVALID_VERSION_NUMBER TAOS_DEF_ERROR_CODE(0, 0x0120) #define TSDB_CODE_INVALID_VERSION_STRING TAOS_DEF_ERROR_CODE(0, 0x0121) #define TSDB_CODE_VERSION_NOT_COMPATIBLE TAOS_DEF_ERROR_CODE(0, 0x0122) +#define TSDB_CODE_COMPRESS_ERROR TAOS_DEF_ERROR_CODE(0, 0x0123) //client #define TSDB_CODE_TSC_INVALID_OPERATION TAOS_DEF_ERROR_CODE(0, 0x0200) diff --git a/include/util/thttp.h b/include/util/thttp.h index 31d84850e0..7d8c588bfc 100644 --- a/include/util/thttp.h +++ b/include/util/thttp.h @@ -24,7 +24,7 @@ extern "C" { typedef enum { HTTP_GZIP, HTTP_FLAT } EHttpCompFlag; -int32_t taosSendHttpReport(const char* server, uint16_t port, const char* pCont, int32_t contLen, EHttpCompFlag flag); +int32_t taosSendHttpReport(const char* server, uint16_t port, char* pCont, int32_t contLen, EHttpCompFlag flag); #ifdef __cplusplus } diff --git a/source/libs/monitor/src/monitor.c b/source/libs/monitor/src/monitor.c index 905ff8d6da..0760d7ae9e 100644 --- a/source/libs/monitor/src/monitor.c +++ b/source/libs/monitor/src/monitor.c @@ -376,7 +376,8 @@ void monSendReport(SMonInfo *pMonitor) { char *pCont = tjsonToString(pMonitor->pJson); if (pCont != NULL) { - taosSendHttpReport(tsMonitor.server, tsMonitor.port, pCont, strlen(pCont), tsMonitor.comp); + EHttpCompFlag flag = tsMonitor.comp ? HTTP_GZIP : HTTP_FLAT; + taosSendHttpReport(tsMonitor.server, tsMonitor.port, pCont, strlen(pCont), flag); free(pCont); } } diff --git a/source/util/CMakeLists.txt b/source/util/CMakeLists.txt index 6effdff712..42950b2284 100644 --- a/source/util/CMakeLists.txt +++ b/source/util/CMakeLists.txt @@ -10,7 +10,7 @@ target_link_libraries( util PRIVATE os PUBLIC lz4_static - PUBLIC api cjson + PUBLIC api cjson zlib ) if(${BUILD_WITH_UV}) target_link_libraries( diff --git a/source/util/src/terror.c b/source/util/src/terror.c index 9c2fef5b47..6cf6eeb371 100644 --- a/source/util/src/terror.c +++ b/source/util/src/terror.c @@ -68,6 +68,7 @@ TAOS_DEFINE_ERROR(TSDB_CODE_RPC_INVALID_TIME_STAMP, "Client and server's t TAOS_DEFINE_ERROR(TSDB_CODE_APP_NOT_READY, "Database not ready") TAOS_DEFINE_ERROR(TSDB_CODE_RPC_FQDN_ERROR, "Unable to resolve FQDN") TAOS_DEFINE_ERROR(TSDB_CODE_RPC_INVALID_VERSION, "Invalid app version") +TAOS_DEFINE_ERROR(TSDB_CODE_COMPRESS_ERROR, "Failed to compress msg") //common & util TAOS_DEFINE_ERROR(TSDB_CODE_OPS_NOT_SUPPORT, "Operation not supported") diff --git a/source/util/src/thttp.c b/source/util/src/thttp.c index c8a52a4735..9d5df20337 100644 --- a/source/util/src/thttp.c +++ b/source/util/src/thttp.c @@ -17,6 +17,7 @@ #include "thttp.h" #include "taoserror.h" #include "tlog.h" +#include "zlib.h" static int32_t taosBuildHttpHeader(const char* server, int32_t contLen, char* pHead, int32_t headLen, EHttpCompFlag flag) { @@ -40,22 +41,91 @@ static int32_t taosBuildHttpHeader(const char* server, int32_t contLen, char* pH } } +int32_t taosCompressHttpRport(char* pSrc, int32_t srcLen) { + int32_t code = -1; + int32_t destLen = srcLen; + void* pDest = malloc(destLen); + + if (pDest == NULL) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + goto _OVER; + } + + z_stream gzipStream = {0}; + gzipStream.zalloc = (alloc_func)0; + gzipStream.zfree = (free_func)0; + gzipStream.opaque = (voidpf)0; + if (deflateInit2(&gzipStream, Z_DEFAULT_COMPRESSION, Z_DEFLATED, MAX_WBITS + 16, 8, Z_DEFAULT_STRATEGY) != Z_OK) { + terrno = TSDB_CODE_OUT_OF_MEMORY; + goto _OVER; + } + + gzipStream.next_in = (Bytef*)pSrc; + gzipStream.avail_in = (uLong)srcLen; + gzipStream.next_out = (Bytef*)pDest; + gzipStream.avail_out = (uLong)(destLen); + + while (gzipStream.avail_in != 0 && gzipStream.total_out < (uLong)(destLen)) { + if (deflate(&gzipStream, Z_FULL_FLUSH) != Z_OK) { + terrno = TSDB_CODE_COMPRESS_ERROR; + goto _OVER; + } + } + + if (gzipStream.avail_in != 0) { + terrno = TSDB_CODE_COMPRESS_ERROR; + goto _OVER; + } + + int32_t err = 0; + while (1) { + if ((err = deflate(&gzipStream, Z_FINISH)) == Z_STREAM_END) { + break; + } + if (err != Z_OK) { + terrno = TSDB_CODE_COMPRESS_ERROR; + goto _OVER; + } + } + + if (deflateEnd(&gzipStream) != Z_OK) { + terrno = TSDB_CODE_COMPRESS_ERROR; + goto _OVER; + } + + if (gzipStream.total_out >= srcLen) { + terrno = TSDB_CODE_COMPRESS_ERROR; + goto _OVER; + } + + code = 0; + +_OVER: + if (code == 0) { + memcpy(pSrc, pDest, gzipStream.total_out); + code = gzipStream.total_out; + } + + free(pDest); + return code; +} + #ifdef USE_UV static void clientConnCb(uv_connect_t* req, int32_t status) { - if(status < 0) { + if (status < 0) { terrno = TAOS_SYSTEM_ERROR(status); - uError("Connection error %s\n",uv_strerror(status)); + uError("Connection error %s\n", uv_strerror(status)); return; } // impl later uv_buf_t* wb = req->data; if (wb == NULL) { - uv_close((uv_handle_t *)req->handle,NULL); + uv_close((uv_handle_t*)req->handle, NULL); } uv_write_t write_req; uv_write(&write_req, req->handle, wb, 2, NULL); - uv_close((uv_handle_t *)req->handle,NULL); + uv_close((uv_handle_t*)req->handle, NULL); } int32_t taosSendHttpReport(const char* server, uint16_t port, const char* pCont, int32_t contLen, EHttpCompFlag flag) { @@ -77,6 +147,15 @@ int32_t taosSendHttpReport(const char* server, uint16_t port, const char* pCont, uv_tcp_init(loop, &socket_tcp); uv_connect_t* connect = (uv_connect_t*)malloc(sizeof(uv_connect_t)); + if (flag == HTTP_GZIP) { + int32_t dstLen = taosCompressHttpRport(pCont, contLen); + if (dstLen > 0) { + contLen = dstLen; + } else { + flag = HTTP_FLAT; + } + } + char header[1024] = {0}; int32_t headLen = taosBuildHttpHeader(server, contLen, header, sizeof(header), flag); @@ -87,14 +166,14 @@ int32_t taosSendHttpReport(const char* server, uint16_t port, const char* pCont, connect->data = wb; uv_tcp_connect(connect, &socket_tcp, (const struct sockaddr*)&dest, clientConnCb); terrno = 0; - uv_run(loop,UV_RUN_DEFAULT); + uv_run(loop, UV_RUN_DEFAULT); uv_loop_close(loop); free(connect); return terrno; } #else -int32_t taosSendHttpReport(const char* server, uint16_t port, const char* pCont, int32_t contLen, EHttpCompFlag flag) { +int32_t taosSendHttpReport(const char* server, uint16_t port, char* pCont, int32_t contLen, EHttpCompFlag flag) { int32_t code = -1; SOCKET fd = 0; @@ -112,6 +191,15 @@ int32_t taosSendHttpReport(const char* server, uint16_t port, const char* pCont, goto SEND_OVER; } + if (flag == HTTP_GZIP) { + int32_t dstLen = taosCompressHttpRport(pCont, contLen); + if (dstLen > 0) { + contLen = dstLen; + } else { + flag = HTTP_FLAT; + } + } + char header[1024] = {0}; int32_t headLen = taosBuildHttpHeader(server, contLen, header, sizeof(header), flag); @@ -134,7 +222,6 @@ int32_t taosSendHttpReport(const char* server, uint16_t port, const char* pCont, goto SEND_OVER; } - uTrace("send http to %s:%u, len:%d content: %s", server, port, contLen, pCont); code = 0; SEND_OVER: -- GitLab