diff --git a/include/libs/transport/thttp.h b/include/libs/transport/thttp.h index 9f635b8523..ae5e4dfc35 100644 --- a/include/libs/transport/thttp.h +++ b/include/libs/transport/thttp.h @@ -17,6 +17,7 @@ #define _TD_UTIL_HTTP_H_ #include "os.h" +#include "tdef.h" #include "tref.h" #ifdef __cplusplus @@ -35,6 +36,25 @@ int32_t taosSendHttpReportByChan(const char* server, const char* uri, uint16_t p EHttpCompFlag flag, int64_t chanId, const char* qid); void taosDestroyHttpChan(int64_t chanId); +int32_t taosAllocHttpRecvHandle(int64_t* rid); +void taosFreeHttpRecvHandle(int64_t rid); +int32_t taosGetHttpRecvBufByHandle(int64_t rid, char** pRecv, int32_t* recvLen); + +int32_t taosSendRecvHttpReportWithQID(const char* server, const char* uri, uint16_t port, char* pCont, int32_t contLen, + EHttpCompFlag flag, const char* qid, int64_t recvBufId); + +typedef struct { + char defaultAddr[256 + 1]; + char cachedAddr[256 + 1]; + int64_t recvBufRid; +} STelemAddrMgmt; +int32_t taosTelemetryMgtInit(STelemAddrMgmt* mgt, char* defaultAddr); +void taosTelemetryDestroy(STelemAddrMgmt* mgt); + +// not safe for multi-thread, should be called in the same thread +int32_t taosSendTelemReport(STelemAddrMgmt* mgt, const char* uri, uint16_t port, char* pCont, int32_t contLen, + EHttpCompFlag flag); + #ifdef __cplusplus } #endif diff --git a/source/client/src/clientEnv.c b/source/client/src/clientEnv.c index fa9df5be73..59eb2f88bc 100644 --- a/source/client/src/clientEnv.c +++ b/source/client/src/clientEnv.c @@ -166,7 +166,8 @@ static int32_t generateWriteSlowLog(STscObj *pTscObj, SRequestObj *pRequest, int ENV_JSON_FALSE_CHECK(cJSON_AddItemToObject(json, "type", cJSON_CreateNumber(reqType))); ENV_JSON_FALSE_CHECK(cJSON_AddItemToObject( json, "rows_num", cJSON_CreateNumber(pRequest->body.resInfo.numOfRows + pRequest->body.resInfo.totalRows))); - if (pRequest->sqlstr != NULL && strlen(pRequest->sqlstr) > pTscObj->pAppInfo->serverCfg.monitorParas.tsSlowLogMaxLen) { + if (pRequest->sqlstr != NULL && + strlen(pRequest->sqlstr) > pTscObj->pAppInfo->serverCfg.monitorParas.tsSlowLogMaxLen) { char tmp = pRequest->sqlstr[pTscObj->pAppInfo->serverCfg.monitorParas.tsSlowLogMaxLen]; pRequest->sqlstr[pTscObj->pAppInfo->serverCfg.monitorParas.tsSlowLogMaxLen] = '\0'; ENV_JSON_FALSE_CHECK(cJSON_AddItemToObject(json, "sql", cJSON_CreateString(pRequest->sqlstr))); @@ -816,6 +817,8 @@ static void *tscCrashReportThreadFp(void *param) { if (-1 != atomic_val_compare_exchange_32(&clientStop, -1, 0)) { return NULL; } + STelemAddrMgmt mgt; + taosTelemetryMgtInit(&mgt, tsTelemServer); while (1) { if (clientStop > 0) break; @@ -826,7 +829,7 @@ static void *tscCrashReportThreadFp(void *param) { taosReadCrashInfo(filepath, &pMsg, &msgLen, &pFile); if (pMsg && msgLen > 0) { - if (taosSendHttpReport(tsTelemServer, tsClientCrashReportUri, tsTelemPort, pMsg, msgLen, HTTP_FLAT) != 0) { + if (taosSendTelemReport(&mgt, tsClientCrashReportUri, tsTelemPort, pMsg, msgLen, HTTP_FLAT) != 0) { tscError("failed to send crash report"); if (pFile) { taosReleaseCrashLogFile(pFile, false); @@ -860,6 +863,7 @@ static void *tscCrashReportThreadFp(void *param) { taosMsleep(sleepTime); loopTimes = 0; } + taosTelemetryDestroy(&mgt); clientStop = -2; return NULL; @@ -1099,7 +1103,7 @@ int taos_options_imp(TSDB_OPTION option, const char *str) { */ uint64_t generateRequestId() { static uint32_t hashId = 0; - static int32_t requestSerialId = 0; + static int32_t requestSerialId = 0; if (hashId == 0) { int32_t code = taosGetSystemUUIDU32(&hashId); diff --git a/source/common/src/tglobal.c b/source/common/src/tglobal.c index 47d754fd04..cee2e9c6e7 100644 --- a/source/common/src/tglobal.c +++ b/source/common/src/tglobal.c @@ -127,7 +127,7 @@ bool tsEnableTelem = false; #else bool tsEnableTelem = true; #endif -int32_t tsTelemInterval = 43200; +int32_t tsTelemInterval = 60; char tsTelemServer[TSDB_FQDN_LEN] = "telemetry.tdengine.com"; uint16_t tsTelemPort = 80; char *tsTelemUri = "/report"; @@ -537,7 +537,8 @@ int32_t taosAddClientLogCfg(SConfig *pCfg) { TAOS_CHECK_RETURN(cfgAddInt32(pCfg, "jniDebugFlag", jniDebugFlag, 0, 255, CFG_SCOPE_CLIENT, CFG_DYN_CLIENT)); TAOS_CHECK_RETURN(cfgAddInt32(pCfg, "qDebugFlag", qDebugFlag, 0, 255, CFG_SCOPE_BOTH, CFG_DYN_BOTH)); TAOS_CHECK_RETURN(cfgAddInt32(pCfg, "cDebugFlag", cDebugFlag, 0, 255, CFG_SCOPE_CLIENT, CFG_DYN_CLIENT)); - TAOS_CHECK_RETURN(cfgAddInt32(pCfg, "tqClientDebugFlag", tqClientDebugFlag, 0, 255, CFG_SCOPE_CLIENT, CFG_DYN_SERVER)); + TAOS_CHECK_RETURN( + cfgAddInt32(pCfg, "tqClientDebugFlag", tqClientDebugFlag, 0, 255, CFG_SCOPE_CLIENT, CFG_DYN_SERVER)); TAOS_RETURN(TSDB_CODE_SUCCESS); } @@ -590,8 +591,10 @@ static int32_t taosAddClientCfg(SConfig *pCfg) { TAOS_CHECK_RETURN( cfgAddBool(pCfg, "queryUseNodeAllocator", tsQueryUseNodeAllocator, CFG_SCOPE_CLIENT, CFG_DYN_CLIENT)); TAOS_CHECK_RETURN(cfgAddBool(pCfg, "keepColumnName", tsKeepColumnName, CFG_SCOPE_CLIENT, CFG_DYN_CLIENT)); - TAOS_CHECK_RETURN(cfgAddInt32(pCfg, "minSlidingTime", tsMinSlidingTime, 1, 1000000, CFG_SCOPE_CLIENT, CFG_DYN_CLIENT)); - TAOS_CHECK_RETURN(cfgAddInt32(pCfg, "minIntervalTime", tsMinIntervalTime, 1, 1000000, CFG_SCOPE_CLIENT, CFG_DYN_CLIENT)); + TAOS_CHECK_RETURN( + cfgAddInt32(pCfg, "minSlidingTime", tsMinSlidingTime, 1, 1000000, CFG_SCOPE_CLIENT, CFG_DYN_CLIENT)); + TAOS_CHECK_RETURN( + cfgAddInt32(pCfg, "minIntervalTime", tsMinIntervalTime, 1, 1000000, CFG_SCOPE_CLIENT, CFG_DYN_CLIENT)); TAOS_CHECK_RETURN(cfgAddString(pCfg, "smlChildTableName", tsSmlChildTableName, CFG_SCOPE_CLIENT, CFG_DYN_CLIENT)); TAOS_CHECK_RETURN(cfgAddString(pCfg, "smlAutoChildTableNameDelimiter", tsSmlAutoChildTableNameDelimiter, CFG_SCOPE_CLIENT, CFG_DYN_CLIENT)); @@ -781,7 +784,7 @@ static int32_t taosAddServerCfg(SConfig *pCfg) { TAOS_CHECK_RETURN(cfgAddInt32(pCfg, "auditInterval", tsAuditInterval, 500, 200000, CFG_SCOPE_SERVER, CFG_DYN_NONE)); TAOS_CHECK_RETURN(cfgAddBool(pCfg, "telemetryReporting", tsEnableTelem, CFG_SCOPE_SERVER, CFG_DYN_ENT_SERVER)); - TAOS_CHECK_RETURN(cfgAddInt32(pCfg, "telemetryInterval", tsTelemInterval, 1, 200000, CFG_SCOPE_SERVER, CFG_DYN_NONE)); + TAOS_CHECK_RETURN(cfgAddInt32(pCfg, "telemetryInterval", tsTelemInterval, 59, 200000, CFG_SCOPE_SERVER, CFG_DYN_NONE)); TAOS_CHECK_RETURN(cfgAddString(pCfg, "telemetryServer", tsTelemServer, CFG_SCOPE_SERVER, CFG_DYN_BOTH)); TAOS_CHECK_RETURN(cfgAddInt32(pCfg, "telemetryPort", tsTelemPort, 1, 65056, CFG_SCOPE_SERVER, CFG_DYN_NONE)); @@ -1999,14 +2002,17 @@ static int32_t taosCfgDynamicOptionsForServer(SConfig *pCfg, const char *name) { { // 'bool/int32_t/int64_t/float/double' variables with general modification function static OptionNameAndVar debugOptions[] = { - {"dDebugFlag", &dDebugFlag}, {"vDebugFlag", &vDebugFlag}, {"mDebugFlag", &mDebugFlag}, - {"wDebugFlag", &wDebugFlag}, {"azDebugFlag", &azDebugFlag}, {"sDebugFlag", &sDebugFlag}, - {"tsdbDebugFlag", &tsdbDebugFlag}, {"tqDebugFlag", &tqDebugFlag}, {"fsDebugFlag", &fsDebugFlag}, - {"udfDebugFlag", &udfDebugFlag}, {"smaDebugFlag", &smaDebugFlag}, {"idxDebugFlag", &idxDebugFlag}, - {"tdbDebugFlag", &tdbDebugFlag}, {"tmrDebugFlag", &tmrDebugFlag}, {"uDebugFlag", &uDebugFlag}, - {"smaDebugFlag", &smaDebugFlag}, {"rpcDebugFlag", &rpcDebugFlag}, {"qDebugFlag", &qDebugFlag}, - {"metaDebugFlag", &metaDebugFlag}, {"stDebugFlag", &stDebugFlag}, {"sndDebugFlag", &sndDebugFlag}, - {"tqClientDebugFlag", &tqClientDebugFlag}, + {"dDebugFlag", &dDebugFlag}, {"vDebugFlag", &vDebugFlag}, + {"mDebugFlag", &mDebugFlag}, {"wDebugFlag", &wDebugFlag}, + {"azDebugFlag", &azDebugFlag}, {"sDebugFlag", &sDebugFlag}, + {"tsdbDebugFlag", &tsdbDebugFlag}, {"tqDebugFlag", &tqDebugFlag}, + {"fsDebugFlag", &fsDebugFlag}, {"udfDebugFlag", &udfDebugFlag}, + {"smaDebugFlag", &smaDebugFlag}, {"idxDebugFlag", &idxDebugFlag}, + {"tdbDebugFlag", &tdbDebugFlag}, {"tmrDebugFlag", &tmrDebugFlag}, + {"uDebugFlag", &uDebugFlag}, {"smaDebugFlag", &smaDebugFlag}, + {"rpcDebugFlag", &rpcDebugFlag}, {"qDebugFlag", &qDebugFlag}, + {"metaDebugFlag", &metaDebugFlag}, {"stDebugFlag", &stDebugFlag}, + {"sndDebugFlag", &sndDebugFlag}, {"tqClientDebugFlag", &tqClientDebugFlag}, }; static OptionNameAndVar options[] = {{"audit", &tsEnableAudit}, diff --git a/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c b/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c index 7f802f3837..2d6d5b59f3 100644 --- a/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c +++ b/source/dnode/mgmt/mgmt_dnode/src/dmWorker.c @@ -243,8 +243,10 @@ static void *dmCrashReportThreadFp(void *param) { bool truncateFile = false; int32_t sleepTime = 200; int32_t reportPeriodNum = 3600 * 1000 / sleepTime; - ; - int32_t loopTimes = reportPeriodNum; + int32_t loopTimes = reportPeriodNum; + + STelemAddrMgmt mgt = {0}; + taosTelemetryMgtInit(&mgt, tsTelemServer); while (1) { if (pMgmt->pData->dropped || pMgmt->pData->stopped) break; @@ -255,7 +257,7 @@ static void *dmCrashReportThreadFp(void *param) { taosReadCrashInfo(filepath, &pMsg, &msgLen, &pFile); if (pMsg && msgLen > 0) { - if (taosSendHttpReport(tsTelemServer, tsSvrCrashReportUri, tsTelemPort, pMsg, msgLen, HTTP_FLAT) != 0) { + if (taosSendTelemReport(&mgt, tsSvrCrashReportUri, tsTelemPort, pMsg, msgLen, HTTP_FLAT) != 0) { dError("failed to send crash report"); if (pFile) { taosReleaseCrashLogFile(pFile, false); @@ -289,6 +291,7 @@ static void *dmCrashReportThreadFp(void *param) { taosMsleep(sleepTime); loopTimes = 0; } + taosTelemetryDestroy(&mgt); return NULL; } diff --git a/source/dnode/mnode/impl/inc/mndInt.h b/source/dnode/mnode/impl/inc/mndInt.h index fff88e547c..982a541bb3 100644 --- a/source/dnode/mnode/impl/inc/mndInt.h +++ b/source/dnode/mnode/impl/inc/mndInt.h @@ -24,6 +24,7 @@ #include "tdatablock.h" #include "tglobal.h" #include "tgrant.h" +#include "thttp.h" #include "tqueue.h" #include "ttime.h" #include "version.h" @@ -80,8 +81,9 @@ typedef struct { } SProfileMgmt; typedef struct { - TdThreadMutex lock; - char email[TSDB_FQDN_LEN]; + TdThreadMutex lock; + char email[TSDB_FQDN_LEN]; + STelemAddrMgmt addrMgt; } STelemMgmt; typedef struct { diff --git a/source/dnode/mnode/impl/src/mndMain.c b/source/dnode/mnode/impl/src/mndMain.c index 83e8abf05e..2ee128e29d 100644 --- a/source/dnode/mnode/impl/src/mndMain.c +++ b/source/dnode/mnode/impl/src/mndMain.c @@ -410,9 +410,11 @@ void mndDoTimerPullupTask(SMnode *pMnode, int64_t sec) { mndStreamConsensusChkpt(pMnode); } +#ifndef TD_ENTERPRISE if (sec % tsTelemInterval == (TMIN(60, (tsTelemInterval - 1)))) { mndPullupTelem(pMnode); } +#endif if (sec % tsGrantHBInterval == 0) { mndPullupGrant(pMnode); diff --git a/source/dnode/mnode/impl/src/mndTelem.c b/source/dnode/mnode/impl/src/mndTelem.c index 810c71b7c5..21133ff50d 100644 --- a/source/dnode/mnode/impl/src/mndTelem.c +++ b/source/dnode/mnode/impl/src/mndTelem.c @@ -132,6 +132,8 @@ _OVER: } static int32_t mndProcessTelemTimer(SRpcMsg* pReq) { + int32_t code = 0; + int32_t line = 0; SMnode* pMnode = pReq->info.node; STelemMgmt* pMgmt = &pMnode->telemMgmt; if (!tsEnableTelem) return 0; @@ -140,15 +142,18 @@ static int32_t mndProcessTelemTimer(SRpcMsg* pReq) { char* pCont = mndBuildTelemetryReport(pMnode); (void)taosThreadMutexUnlock(&pMgmt->lock); - if (pCont != NULL) { - if (taosSendHttpReport(tsTelemServer, tsTelemUri, tsTelemPort, pCont, strlen(pCont), HTTP_FLAT) != 0) { - mError("failed to send telemetry report"); - } else { - mInfo("succeed to send telemetry report"); - } - taosMemoryFree(pCont); + if (pCont == NULL) { + return 0; } - return 0; + code = taosSendTelemReport(&pMgmt->addrMgt, tsTelemUri, tsTelemPort, pCont, strlen(pCont), HTTP_FLAT); + taosMemoryFree(pCont); + return code; +_end: + if (code != 0) { + mError("%s failed to send at line %d since %s", __func__, line, tstrerror(code)); + } + taosMemoryFree(pCont); + return code; } int32_t mndInitTelem(SMnode* pMnode) { @@ -158,6 +163,7 @@ int32_t mndInitTelem(SMnode* pMnode) { (void)taosThreadMutexInit(&pMgmt->lock, NULL); if ((code = taosGetEmail(pMgmt->email, sizeof(pMgmt->email))) != 0) mWarn("failed to get email since %s", tstrerror(code)); + taosTelemetryMgtInit(&pMgmt->addrMgt, tsTelemServer); mndSetMsgHandle(pMnode, TDMT_MND_TELEM_TIMER, mndProcessTelemTimer); return 0; @@ -165,5 +171,6 @@ int32_t mndInitTelem(SMnode* pMnode) { void mndCleanupTelem(SMnode* pMnode) { STelemMgmt* pMgmt = &pMnode->telemMgmt; + taosTelemetryDestroy(&pMgmt->addrMgt); (void)taosThreadMutexDestroy(&pMgmt->lock); } diff --git a/source/libs/transport/src/thttp.c b/source/libs/transport/src/thttp.c index bdcbfeb1cd..41347907d1 100644 --- a/source/libs/transport/src/thttp.c +++ b/source/libs/transport/src/thttp.c @@ -31,6 +31,7 @@ static int32_t FAST_FAILURE_LIMIT = 1; static int64_t httpDefaultChanId = -1; static int64_t httpSeqNum = 0; +static int32_t httpRecvRefMgt = 0; typedef struct SHttpModule { uv_loop_t* loop; @@ -54,6 +55,8 @@ typedef struct SHttpMsg { int64_t chanId; int64_t seq; char* qid; + + int64_t recvBufRid; } SHttpMsg; typedef struct SHttpClient { @@ -67,6 +70,8 @@ typedef struct SHttpClient { struct sockaddr_in dest; int64_t chanId; int64_t seq; + + int64_t recvBufRid; } SHttpClient; typedef struct SHttpConnList { @@ -74,6 +79,13 @@ typedef struct SHttpConnList { } SHttpConnList; +typedef struct { + char* pBuf; + int32_t nBuf; + SRWLatch latch; + int8_t inited; +} SHttpRecvBuf; + static TdThreadOnce transHttpInit = PTHREAD_ONCE_INIT; static void transHttpEnvInit(); @@ -94,6 +106,9 @@ static void httpModuleDestroy(SHttpModule* http); static int32_t taosSendHttpReportImplByChan(const char* server, const char* uri, uint16_t port, char* pCont, int32_t contLen, EHttpCompFlag flag, int64_t chanId, const char* qid); +static int32_t taosSendHttpReportImplByChan2(const char* server, const char* uri, uint16_t port, char* pCont, + int32_t contLen, EHttpCompFlag flag, int64_t chanId, const char* qid, + int64_t rid); static int32_t taosBuildHttpHeader(const char* server, const char* uri, int32_t contLen, const char* qid, char* pHead, int32_t headLen, @@ -394,6 +409,8 @@ static FORCE_INLINE void destroyHttpClient(SHttpClient* cli) { taosMemoryFree(cli->wbuf); taosMemoryFree(cli->rbuf); taosMemoryFree(cli->addr); + // taosFreeHttpRecvHandle(cli->recvBufRid); + taosMemoryFree(cli); } @@ -423,6 +440,28 @@ static FORCE_INLINE void clientRecvCb(uv_stream_t* handle, ssize_t nread, const tError("http-report recv error:%s, seq:%" PRId64 "", uv_strerror(nread), cli->seq); } else { tTrace("http-report succ to recv %d bytes, seq:%" PRId64 "", (int32_t)nread, cli->seq); + if (cli->recvBufRid > 0) { + SHttpRecvBuf* p = taosAcquireRef(httpRecvRefMgt, cli->recvBufRid); + if (p != NULL) { + taosWLockLatch(&p->latch); + if (p->inited != 0) { + tDebug("http-report already recv valid data"); + } else { + p->pBuf = taosMemoryCalloc(1, nread + 1); + if (p->pBuf != NULL) { + memcpy(p->pBuf, buf->base, nread); + p->nBuf = nread; + p->inited = 1; + } else { + tError("http-report failed to dump recv since %s", tstrerror(terrno)); + } + } + taosWUnLockLatch(&p->latch); + taosReleaseRef(httpRecvRefMgt, cli->recvBufRid); + } else { + tWarn("http-report failed to acquire recv buf since %s", tstrerror(terrno)); + } + } } if (!uv_is_closing((uv_handle_t*)&cli->tcp)) { uv_close((uv_handle_t*)&cli->tcp, clientCloseCb); @@ -629,6 +668,8 @@ static void httpHandleReq(SHttpMsg* msg) { cli->chanId = chanId; cli->addr = msg->server; cli->port = msg->port; + cli->recvBufRid = msg->recvBufRid; + if (msg->qid != NULL) taosMemoryFree(msg->qid); taosMemoryFree(msg->uri); taosMemoryFree(msg); @@ -714,6 +755,47 @@ void httpModuleDestroy2(SHttpModule* http) { taosMemoryFree(http); } +static int32_t taosSendHttpReportImplByChan2(const char* server, const char* uri, uint16_t port, char* pCont, + int32_t contLen, EHttpCompFlag flag, int64_t chanId, const char* qid, + int64_t recvBufRid) { + SHttpModule* load = NULL; + SHttpMsg* msg = NULL; + int32_t code = httpCreateMsg(server, uri, port, pCont, contLen, flag, chanId, qid, &msg); + if (code != 0) { + goto _ERROR; + } + + msg->recvBufRid = recvBufRid; + + load = taosAcquireRef(httpRefMgt, chanId); + if (load == NULL) { + code = terrno; + goto _ERROR; + } + + if (atomic_load_8(&load->quit)) { + code = TSDB_CODE_HTTP_MODULE_QUIT; + goto _ERROR; + } + tDebug("http-report start to report, chanId:%" PRId64 ", seq:%" PRId64 "", chanId, msg->seq); + + code = transAsyncSend(load->asyncPool, &(msg->q)); + if (code != 0) { + code = TSDB_CODE_HTTP_MODULE_QUIT; + goto _ERROR; + } + msg = NULL; + +_ERROR: + + if (code != 0) { + tError("http-report failed to report reason:%s, chanId:%" PRId64 ", seq:%" PRId64 "", tstrerror(code), chanId, + msg->seq); + } + httpDestroyMsg(msg); + if (load != NULL) taosReleaseRef(httpRefMgt, chanId); + return code; +} static int32_t taosSendHttpReportImplByChan(const char* server, const char* uri, uint16_t port, char* pCont, int32_t contLen, EHttpCompFlag flag, int64_t chanId, const char* qid) { SHttpModule* load = NULL; @@ -769,20 +851,30 @@ int32_t taosSendHttpReportWithQID(const char* server, const char* uri, uint16_t return taosSendHttpReportImplByChan(server, uri, port, pCont, contLen, flag, httpDefaultChanId, qid); } +int32_t taosSendRecvHttpReportWithQID(const char* server, const char* uri, uint16_t port, char* pCont, int32_t contLen, + EHttpCompFlag flag, const char* qid, int64_t recvBufId) { + TAOS_UNUSED(taosThreadOnce(&transHttpInit, transHttpEnvInit)); + return taosSendHttpReportImplByChan2(server, uri, port, pCont, contLen, flag, httpDefaultChanId, qid, recvBufId); +} + static void transHttpDestroyHandle(void* handle) { taosMemoryFree(handle); } +static void transHttpDestroyRecvHandle(void* handle) { taosMemoryFree(handle); } + int64_t transInitHttpChanImpl(); static void transHttpEnvInit() { httpRefMgt = taosOpenRef(64, transHttpDestroyHandle); httpDefaultChanId = transInitHttpChanImpl(); httpSeqNum = 0; + httpRecvRefMgt = taosOpenRef(8, transHttpDestroyRecvHandle); } void transHttpEnvDestroy() { // remove default chanId taosDestroyHttpChan(httpDefaultChanId); httpDefaultChanId = -1; + taosCloseRef(httpRecvRefMgt); } int64_t transInitHttpChanImpl() { @@ -868,3 +960,166 @@ void taosDestroyHttpChan(int64_t chanId) { TAOS_UNUSED(taosReleaseRef(httpRefMgt, chanId)); TAOS_UNUSED(taosRemoveRef(httpRefMgt, chanId)); } +int32_t taosAllocHttpRecvHandle(int64_t* rid) { + TAOS_UNUSED(taosThreadOnce(&transHttpInit, transHttpEnvInit)); + SHttpRecvBuf* p = taosMemoryCalloc(1, sizeof(SHttpRecvBuf)); + if (p == NULL) { + return terrno; + } + taosInitRWLatch(&p->latch); + + int64_t id = taosAddRef(httpRecvRefMgt, p); + if (taosAcquireRef(httpRecvRefMgt, id) == NULL) { + taosMemoryFree(p); + return terrno; + } + *rid = id; + return 0; +} +void taosFreeHttpRecvHandle(int64_t rid) { + if (rid <= 0) { + return; + } + taosReleaseRef(httpRecvRefMgt, rid); + taosRemoveRef(httpRecvRefMgt, rid); +} +int32_t taosGetHttpRecvById(int64_t rid, char** pRecv, int32_t* len) { + int32_t code = 0; + SHttpRecvBuf* p = taosAcquireRef(httpRecvRefMgt, rid); + if (p == NULL) { + return TSDB_CODE_INVALID_PARA; + } + taosWLockLatch(&p->latch); + if (p->inited) { + *pRecv = p->pBuf; + *len = p->nBuf; + p->pBuf = 0; + p->nBuf = 0; + } else { + code = TSDB_CODE_INVALID_PARA; + } + taosWUnLockLatch(&p->latch); + taosReleaseRef(httpRecvRefMgt, rid); + + return code; +} + +int32_t taosTelemetryMgtInit(STelemAddrMgmt* mgt, char* defaultAddr) { + tstrncpy(mgt->defaultAddr, defaultAddr, sizeof(mgt->defaultAddr)); + mgt->cachedAddr[0] = 0; + mgt->recvBufRid = 0; + return 0; +} +int32_t taosTelemetryMgtGetAddr(STelemAddrMgmt* mgt, char* addr) { + if (mgt->cachedAddr[0] == 0) { + tstrncpy(addr, mgt->defaultAddr, TD_FQDN_LEN); + } else { + tstrncpy(addr, mgt->cachedAddr, TD_FQDN_LEN); + } + return 0; +} +void taosTelemetryDestroy(STelemAddrMgmt* mgt) { + if (mgt == NULL || mgt->recvBufRid <= 0) { + return; + } + taosFreeHttpRecvHandle(mgt->recvBufRid); + mgt->recvBufRid = 0; +} +// TODO: parse http response head By LIB +int32_t taosGetValueFromHttpResp(const char* response, const char* key, char* val) { + if (key == NULL || val == NULL) { + return TSDB_CODE_INVALID_PARA; + } + + int32_t code = 0, line = 0; + int32_t len = strlen(key); + int32_t cap = len + 16; + // const char* reportUrlKey = "\"report_url\":\""; + + char* buf = taosMemoryCalloc(1, cap); + if (buf == NULL) { + return terrno; + } + int32_t nBytes = snprintf(buf, cap, "\"%s\":\"", key); + if ((uint32_t)nBytes >= cap) { + TAOS_CHECK_GOTO(TSDB_CODE_INVALID_PARA, &line, _end); + } + + char* start = strstr(response, buf); + if (start) { + start += strlen(buf); + char* end = strchr(start, '"'); + if (end && end - start > 0) { + tstrncpy(val, start, end - start + 1); + } else { + TAOS_CHECK_GOTO(TSDB_CODE_INVALID_PARA, &line, _end); + } + } else { + TAOS_CHECK_GOTO(TSDB_CODE_INVALID_PARA, &line, _end); + } +_end: + if (code != 0) { + tError("failed to get value from http response since %s", tstrerror(code)); + } + taosMemoryFree(buf); + return code; +} +static int32_t taosGetAddrFromHttpResp(int64_t recvBufRid, char* telemAddr) { + int32_t code = 0; + char* pRecv = NULL; + int32_t nRecv = 0; + + code = taosGetHttpRecvById(recvBufRid, &pRecv, &nRecv); + if (code != 0) { + tError("failed to get http recv buf since %s", tstrerror(code)); + return code; + } + + code = taosGetValueFromHttpResp(pRecv, "report_url", telemAddr); + taosMemoryFree(pRecv); + return code; +} +int32_t taosSendTelemReport(STelemAddrMgmt* mgt, const char* uri, uint16_t port, char* pCont, int32_t contLen, + EHttpCompFlag flag) { + int32_t code = 0; + int32_t line = 0; + int32_t sendMsg = 0; + char* addr = mgt->defaultAddr; + if (mgt->cachedAddr[0] == 0) { + if (mgt->recvBufRid == 0) { + code = taosAllocHttpRecvHandle(&mgt->recvBufRid); + TAOS_CHECK_GOTO(code, &line, _end); + code = taosSendRecvHttpReportWithQID(mgt->defaultAddr, uri, port, pCont, contLen, flag, NULL, mgt->recvBufRid); + TAOS_CHECK_GOTO(code, &line, _end); + } else { + code = taosGetAddrFromHttpResp(mgt->recvBufRid, mgt->cachedAddr); + if (code != 0) { + tError("failed to get cache addr from http response since %s", tstrerror(code)); + addr = mgt->defaultAddr; + } else { + addr = mgt->cachedAddr; + } + + if (mgt->recvBufRid > 0) { + taosFreeHttpRecvHandle(mgt->recvBufRid); + mgt->recvBufRid = 0; + } + sendMsg = 1; + } + } else { + addr = mgt->cachedAddr; + sendMsg = 1; + } + if (sendMsg == 1) { + code = taosSendHttpReport(addr, uri, port, pCont, contLen, flag); + } + tDebug("send telemetry port oldAddr:[%s], newAddr:[%s]", mgt->defaultAddr, mgt->cachedAddr); + return code; + +_end: + if (code != 0) { + tError("failed to send telemetry since %s, default addr:%s, cachedAddr:%s", tstrerror(code), mgt->defaultAddr, + mgt->cachedAddr); + } + return code; +} diff --git a/source/libs/transport/test/transUT2.cpp b/source/libs/transport/test/transUT2.cpp index 54d23b1f64..f8448a3b67 100644 --- a/source/libs/transport/test/transUT2.cpp +++ b/source/libs/transport/test/transUT2.cpp @@ -22,6 +22,7 @@ #include "transLog.h" #include "trpc.h" #include "tversion.h" +#include "thttp.h" using namespace std; const char *label = "APP"; @@ -527,3 +528,12 @@ TEST_F(TransEnv, noResp) { taosMsleep(10000); // no resp } + +TEST_F(TransEnv, http) { + STelemAddrMgmt mgt; + taosTelemetryMgtInit(&mgt, "telemetry.tdengine.com"); + int32_t code = taosSendTelemReport(&mgt,tsTelemUri, tsTelemPort, "test", strlen("test"),HTTP_FLAT); + taosMsleep(10000); + + taosTelemetryDestroy(&mgt); +}