xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

Merge branch '3.0' of github.com:taosdata/TDengine into szhou/fixbugs

This commit is contained in:
shenglian zhou 2022-08-18 20:28:21 +08:00
parent 1a8fb96182 c742747428
commit 3c321c5a6c
4 changed files with 89 additions and 112 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
docs/zh/17-operation/03-tolerance.md
View File

@ -26,5 +26,3 @@ TDengine 集群中的时序数据的副本数是与数据库关联的，一个
TDengine 集群的节点数必须大于等于副本数，否则创建表时将报错。
当 TDengine 集群中的节点部署在不同的物理机上，并设置多个副本数时，就实现了系统的高可靠性，无需再使用其他软件或工具。TDengine 企业版还可以将副本部署在不同机房，从而实现异地容灾。
另外一种灾备方式是通过 `taosX` 将一个 TDengine 集群的数据同步复制到物理上位于不同数据中心的另一个 TDengine 集群。其详细使用方法请参考 [taosX 参考手册](../../reference/taosX)

6
source/common/src/tglobal.c
View File

@ -75,7 +75,7 @@ int32_t tsMonitorMaxLogs = 100;
bool tsMonitorComp = false;
// telem
bool tsEnableTelem = false;
bool tsEnableTelem = true;
int32_t tsTelemInterval = 86400;
char tsTelemServer[TSDB_FQDN_LEN] = "telemetry.taosdata.com";
uint16_t tsTelemPort = 80;
@ -166,7 +166,7 @@ int32_t tsTtlPushInterval = 86400;
int32_t tsGrantHBInterval = 60;
#ifndef _STORAGE
int32_t taosSetTfsCfg(SConfig *pCfg) {
int32_t taosSetTfsCfg(SConfig *pCfg) {
SConfigItem *pItem = cfgGetItem(pCfg, "dataDir");
memset(tsDataDir, 0, PATH_MAX);
@ -180,7 +180,7 @@ int32_t taosSetTfsCfg(SConfig *pCfg) {
uError("failed to create dataDir:%s", tsDataDir);
return -1;
}
return 0;
return 0;
}
#else
int32_t taosSetTfsCfg(SConfig *pCfg);

17
source/libs/function/src/builtinsimpl.c
View File

@ -3970,16 +3970,16 @@ int32_t elapsedFunction(SqlFunctionCtx* pCtx) {
TSKEY* ptsList = (int64_t*)colDataGetData(pCol, 0);
if (pCtx->order == TSDB_ORDER_DESC) {
if (pCtx->start.key == INT64_MIN) {
pInfo->max =
(pInfo->max < ptsList[start + pInput->numOfRows - 1]) ? ptsList[start + pInput->numOfRows - 1] : pInfo->max;
pInfo->max = (pInfo->max < ptsList[start]) ? ptsList[start] : pInfo->max;
} else {
pInfo->max = pCtx->start.key + 1;
}
if (pCtx->end.key != INT64_MIN) {
pInfo->min = pCtx->end.key;
if (pCtx->end.key == INT64_MIN) {
pInfo->min = (pInfo->min > ptsList[start + pInput->numOfRows - 1]) ?
ptsList[start + pInput->numOfRows - 1] : pInfo->min;
} else {
pInfo->min = ptsList[start];
pInfo->min = pCtx->end.key;
}
} else {
if (pCtx->start.key == INT64_MIN) {
@ -3988,10 +3988,11 @@ int32_t elapsedFunction(SqlFunctionCtx* pCtx) {
pInfo->min = pCtx->start.key;
}
if (pCtx->end.key != INT64_MIN) {
pInfo->max = pCtx->end.key + 1;
if (pCtx->end.key == INT64_MIN) {
pInfo->max = (pInfo->max < ptsList[start + pInput->numOfRows - 1]) ?
ptsList[start + pInput->numOfRows - 1] : pInfo->max;
} else {
pInfo->max = ptsList[start + pInput->numOfRows - 1];
pInfo->max = pCtx->end.key + 1;
}
}
}

176
source/libs/transport/src/thttp.c
View File

@ -14,15 +14,22 @@
*/
#define _DEFAULT_SOURCE
#ifdef USE_UV
#include <uv.h>
#endif
// clang-format off
#include <uv.h>
#include "zlib.h"
#include "thttp.h"
#include "taoserror.h"
#include "tlog.h"
typedef struct SHttpClient {
uv_connect_t conn;
uv_tcp_t tcp;
uv_write_t req;
uv_buf_t* buf;
char* addr;
uint16_t port;
} SHttpClient;
static int32_t taosBuildHttpHeader(const char* server, int32_t contLen, char* pHead, int32_t headLen,
EHttpCompFlag flag) {
if (flag == HTTP_FLAT) {
@ -45,7 +52,7 @@ static int32_t taosBuildHttpHeader(const char* server, int32_t contLen, char* pH
}
}
int32_t taosCompressHttpRport(char* pSrc, int32_t srcLen) {
static int32_t taosCompressHttpRport(char* pSrc, int32_t srcLen) {
int32_t code = -1;
int32_t destLen = srcLen;
void* pDest = taosMemoryMalloc(destLen);
@ -114,84 +121,53 @@ _OVER:
return code;
}
#ifdef USE_UV
static void clientConnCb(uv_connect_t* req, int32_t status) {
if (status < 0) {
static void destroyHttpClient(SHttpClient* cli) {
taosMemoryFree(cli->buf);
taosMemoryFree(cli->addr);
taosMemoryFree(cli);
}
static void clientCloseCb(uv_handle_t* handle) {
SHttpClient* cli = handle->data;
destroyHttpClient(cli);
}
static void clientSentCb(uv_write_t* req, int32_t status) {
SHttpClient* cli = req->data;
if (status != 0) {
terrno = TAOS_SYSTEM_ERROR(status);
uError("connection error %s", uv_strerror(status));
uv_close((uv_handle_t*)req->handle, NULL);
uError("http-report failed to send data %s", uv_strerror(status));
} else {
uInfo("http-report succ to send data");
}
uv_close((uv_handle_t*)&cli->tcp, clientCloseCb);
}
static void clientConnCb(uv_connect_t* req, int32_t status) {
SHttpClient* cli = req->data;
if (status != 0) {
terrno = TAOS_SYSTEM_ERROR(status);
uError("http-report failed to conn to server, reason:%s, dst:%s:%d", uv_strerror(status), cli->addr, cli->port);
uv_close((uv_handle_t*)&cli->tcp, clientCloseCb);
return;
}
uv_buf_t* wb = req->data;
assert(wb != NULL);
uv_write_t write_req;
uv_write(&write_req, req->handle, wb, 2, NULL);
uv_close((uv_handle_t*)req->handle, NULL);
uv_write(&cli->req, (uv_stream_t*)&cli->tcp, cli->buf, 2, clientSentCb);
}
int32_t taosSendHttpReport(const char* server, uint16_t port, char* pCont, int32_t contLen, EHttpCompFlag flag) {
uint32_t ipv4 = taosGetIpv4FromFqdn(server);
if (ipv4 == 0xffffffff) {
terrno = TAOS_SYSTEM_ERROR(errno);
uError("failed to get http server:%s ip since %s", server, terrstr());
return -1;
}
char ipv4Buf[128] = {0};
tinet_ntoa(ipv4Buf, ipv4);
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_init(loop, &socket_tcp);
uv_connect_t* connect = (uv_connect_t*)taosMemoryMalloc(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);
uv_buf_t wb[2];
wb[0] = uv_buf_init((char*)header, headLen);
wb[1] = uv_buf_init((char*)pCont, contLen);
connect->data = wb;
terrno = 0;
uv_tcp_connect(connect, &socket_tcp, (const struct sockaddr*)&dest, clientConnCb);
uv_run(loop, UV_RUN_DEFAULT);
uv_loop_close(loop);
taosMemoryFree(connect);
return terrno;
}
#else
int32_t taosSendHttpReport(const char* server, uint16_t port, char* pCont, int32_t contLen, EHttpCompFlag flag) {
int32_t code = -1;
TdSocketPtr pSocket = NULL;
static int32_t taosBuildDstAddr(const char* server, uint16_t port, struct sockaddr_in* dest) {
uint32_t ip = taosGetIpv4FromFqdn(server);
if (ip == 0xffffffff) {
terrno = TAOS_SYSTEM_ERROR(errno);
uError("failed to get http server:%s ip since %s", server, terrstr());
goto SEND_OVER;
uError("http-report failed to get http server:%s ip since %s", server, terrstr());
return -1;
}
pSocket = taosOpenTcpClientSocket(ip, port, 0);
if (pSocket == NULL) {
terrno = TAOS_SYSTEM_ERROR(errno);
uError("failed to create http socket to %s:%u since %s", server, port, terrstr());
goto SEND_OVER;
char buf[128] = {0};
tinet_ntoa(buf, ip);
uv_ip4_addr(buf, port, dest);
return 0;
}
int32_t taosSendHttpReport(const char* server, uint16_t port, char* pCont, int32_t contLen, EHttpCompFlag flag) {
struct sockaddr_in dest = {0};
if (taosBuildDstAddr(server, port, &dest) < 0) {
return -1;
}
if (flag == HTTP_GZIP) {
int32_t dstLen = taosCompressHttpRport(pCont, contLen);
if (dstLen > 0) {
@ -200,36 +176,38 @@ int32_t taosSendHttpReport(const char* server, uint16_t port, char* pCont, int32
flag = HTTP_FLAT;
}
}
terrno = 0;
char header[1024] = {0};
char header[2048] = {0};
int32_t headLen = taosBuildHttpHeader(server, contLen, header, sizeof(header), flag);
if (taosWriteMsg(pSocket, 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;
uv_buf_t* wb = taosMemoryCalloc(2, sizeof(uv_buf_t));
wb[0] = uv_buf_init((char*)header, headLen); // stack var
wb[1] = uv_buf_init((char*)pCont, contLen); // heap var
SHttpClient* cli = taosMemoryCalloc(1, sizeof(SHttpClient));
cli->conn.data = cli;
cli->tcp.data = cli;
cli->req.data = cli;
cli->buf = wb;
cli->addr = tstrdup(server);
cli->port = port;
uv_loop_t* loop = uv_default_loop();
uv_tcp_init(loop, &cli->tcp);
// set up timeout to avoid stuck;
int32_t fd = taosCreateSocketWithTimeout(5);
uv_tcp_open((uv_tcp_t*)&cli->tcp, fd);
int32_t ret = uv_tcp_connect(&cli->conn, &cli->tcp, (const struct sockaddr*)&dest, clientConnCb);
if (ret != 0) {
uError("http-report failed to connect to server, reason:%s, dst:%s:%d", uv_strerror(ret), cli->addr, cli->port);
destroyHttpClient(cli);
}
if (taosWriteMsg(pSocket, (void*)pCont, contLen) < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
uError("failed to send http content to %s:%u since %s", server, port, terrstr());
goto SEND_OVER;
}
// read something to avoid nginx error 499
if (taosWriteMsg(pSocket, header, 10) < 0) {
terrno = TAOS_SYSTEM_ERROR(errno);
uError("failed to receive response from %s:%u since %s", server, port, terrstr());
goto SEND_OVER;
}
code = 0;
SEND_OVER:
if (pSocket != NULL) {
taosCloseSocket(&pSocket);
}
return code;
uv_run(loop, UV_RUN_DEFAULT);
uv_loop_close(loop);
return terrno;
}
// clang-format on
#endif