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

[td-225] merge develop

This commit is contained in:
Haojun Liao 2020-09-07 15:22:46 +08:00
parent 75ddb1ff0c 5e5750bde3
commit 93967816ea
108 changed files with 1252 additions and 782 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
CMakeLists.txt
View File

@ -13,6 +13,9 @@ ENDIF ()
SET(TD_ACCOUNT FALSE)
SET(TD_ADMIN FALSE)
SET(TD_GRANT FALSE)
SET(TD_SYNC TRUE)
SET(TD_MQTT TRUE)
SET(TD_TSDB_PLUGINS FALSE)
SET(TD_COVER FALSE)
SET(TD_MEM_CHECK FALSE)

1
README.md
View File

@ -3,6 +3,7 @@
[![Coverage Status](https://coveralls.io/repos/github/taosdata/TDengine/badge.svg?branch=develop)](https://coveralls.io/github/taosdata/TDengine?branch=develop)
[![CII Best Practices](https://bestpractices.coreinfrastructure.org/projects/4201/badge)](https://bestpractices.coreinfrastructure.org/projects/4201)
[![Docker Pulls](https://img.shields.io/docker/pulls/tdengine/tdengine)](https://hub.docker.com/repository/docker/tdengine/tdengine)
[![tdengine](https://snapcraft.io//tdengine/badge.svg)](https://snapcraft.io/tdengine)
[![TDengine](TDenginelogo.png)](https://www.taosdata.com)

12
cmake/define.inc
View File

@ -13,6 +13,18 @@ IF (TD_GRANT)
ADD_DEFINITIONS(-D_GRANT)
ENDIF ()
IF (TD_SYNC)
ADD_DEFINITIONS(-D_SYNC)
ENDIF ()
IF (TD_MQTT)
ADD_DEFINITIONS(-D_MQTT)
ENDIF ()
IF (TD_TSDB_PLUGINS)
ADD_DEFINITIONS(-D_TSDB_PLUGINS)
ENDIF ()
IF (TD_GODLL)
ADD_DEFINITIONS(-D_TD_GO_DLL_)
ENDIF ()

10
cmake/input.inc
View File

@ -42,6 +42,16 @@ IF (${MEM_CHECK} MATCHES "true")
MESSAGE(STATUS "build with memory check")
ENDIF ()
IF (${MQTT} MATCHES "false")
SET(TD_MQTT FALSE)
MESSAGE(STATUS "build without mqtt module")
ENDIF ()
IF (${SYNC} MATCHES "false")
SET(TD_SYNC FALSE)
MESSAGE(STATUS "build without sync module")
ENDIF ()
IF (${RANDOM_FILE_FAIL} MATCHES "true")
SET(TD_RANDOM_FILE_FAIL TRUE)
MESSAGE(STATUS "build with random-file-fail enabled")

5
cmake/platform.inc
View File

@ -114,6 +114,9 @@ ELSEIF (${OSTYPE} MATCHES "Ningsi80")
MESSAGE(STATUS "input osType: Ningsi80")
ELSEIF (${OSTYPE} MATCHES "Linux")
MESSAGE(STATUS "input osType: Linux")
ELSEIF (${OSTYPE} MATCHES "Alpine")
MESSAGE(STATUS "input osType: Alpine")
SET(TD_APLHINE TRUE)
ELSE ()
MESSAGE(STATUS "input osType unknown: " ${OSTYPE})
ENDIF ()
ENDIF ()

2
cmake/version.inc
View File

@ -4,7 +4,7 @@ PROJECT(TDengine)
IF (DEFINED VERNUMBER)
SET(TD_VER_NUMBER ${VERNUMBER})
ELSE ()
SET(TD_VER_NUMBER "2.0.2.0")
SET(TD_VER_NUMBER "2.0.3.0")
ENDIF ()
IF (DEFINED VERCOMPATIBLE)

2
deps/CMakeLists.txt vendored
View File

@ -10,6 +10,6 @@ ADD_SUBDIRECTORY(cJson)
ADD_SUBDIRECTORY(wepoll)
ADD_SUBDIRECTORY(MsvcLibX)
IF (TD_LINUX)
IF (TD_LINUX AND TD_MQTT)
ADD_SUBDIRECTORY(MQTT-C)
ENDIF ()

3
deps/MQTT-C/CMakeLists.txt vendored
View File

@ -1,5 +1,4 @@
cmake_minimum_required(VERSION 3.5)
project(MQTT-C VERSION 1.1.2 LANGUAGES C)
CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
# MQTT-C build options
option(MQTT_C_OpenSSL_SUPPORT "Build MQTT-C with OpenSSL support?" OFF)

12
documentation20/webdocs/markdowndocs/TAOS SQL-ch.md
View File

@ -82,13 +82,23 @@ TDengine缺省的时间戳是毫秒精度，但通过修改配置参数enableMic
```
删除数据库。所包含的全部数据表将被删除，谨慎使用
- **修改数据库参数**
```mysql
ALTER DATABASE db_name COMP 2;
```
修改数据库文件压缩标志位，有效数字为0，1，2. 0表示不压缩，1表示一阶段压缩，2表示两阶段压缩。修改后可以使用show databases命令查看是否修改成功
```mysql
ALTER DATABASE db_name REPLICA 2;
```
修改数据库副本数，有效副本数为1到3。在集群中使用，副本数必须小于dnode的数目。修改后可以使用show databases命令查看是否修改成功
- **显示系统所有数据库**
```mysql
SHOW DATABASES;
```
## 表管理
- **创建数据表**

3
documentation20/webdocs/markdowndocs/connector-java-ch.md
View File

@ -228,7 +228,8 @@ resultSet.close();
stmt.close();
conn.close();
```
> `注意务必要将 connection 进行关闭`，否则会出现连接泄露。
> `注意务必要将 connection 进行关闭`，否则会出现连接泄露。
## 与连接池使用
**HikariCP**

11
packaging/cfg/taos.cfg
View File

@ -122,11 +122,14 @@
# number of replications, for cluster only
# replica 1
# mqtt uri
# mqttBrokerAddress mqtt://username:password@hostname:1883/taos/
# mqtt hostname
# mqttHostName test.mosquitto.org
# mqtt client name
# mqttBrokerClientId taos_mqtt
# mqtt port
# mqttPort 1883
# mqtt topic
# mqttTopic /weather/loop
# the compressed rpc message, option:
# -1 (no compression)

1
packaging/docker/dockerbuild.sh
View File

@ -1,6 +1,5 @@
#!/bin/bash
set -x
$1
docker build --rm -f "Dockerfile" -t tdengine/tdengine:$1 "."
docker login -u tdengine -p $2 #replace the docker registry username and password
docker push tdengine/tdengine:$1

34
snap/snapcraft.yaml
View File

@ -1,6 +1,6 @@
name: tdengine
base: core18 # the base snap is the execution environment for this snap
version: '2.0.2.0' # just for humans, typically '1.2+git' or '1.3.2'
base: core18
version: 'RELEASE_VERSION'
icon: snap/gui/t-dengine.svg
summary: an open-source big data platform designed and optimized for IoT.
description: |
@ -18,41 +18,18 @@ apps:
- network
- network-bind
- system-observe
- systemfiles
taos:
command: taoswrapper.sh
plugs:
- network
- system-observe
- systemfiles
- historyfile
taosdemo:
command: usr/bin/taosdemo
plugs:
- network
plugs:
historyfile:
interface: personal-files
read:
- $HOME/.taos_history
write:
- $HOME/.taos_history
systemfiles:
interface: system-files
read:
- /etc/taos
- /var/lib/taos
- /var/log/taos
- /tmp
write:
- /var/log/taos
- /var/lib/taos
- /tmp
parts:
script:
plugin: dump
@ -95,7 +72,7 @@ parts:
- usr/bin/taosd
- usr/bin/taos
- usr/bin/taosdemo
- usr/lib/libtaos.so.2.0.2.0
- usr/lib/libtaos.so.RELEASE_VERSION
- usr/lib/libtaos.so.1
- usr/lib/libtaos.so
@ -115,8 +92,3 @@ layout:
bind: $SNAP_DATA/var/log/taos
/etc/taos:
bind: $SNAP_DATA/etc/taos
hooks:
install:
plugs: [systemfiles, historyfile]

4
src/CMakeLists.txt
View File

@ -10,7 +10,9 @@ ADD_SUBDIRECTORY(client)
ADD_SUBDIRECTORY(query)
ADD_SUBDIRECTORY(kit)
ADD_SUBDIRECTORY(plugins)
ADD_SUBDIRECTORY(sync)
IF (TD_SYNC)
ADD_SUBDIRECTORY(sync)
ENDIF ()
ADD_SUBDIRECTORY(balance)
ADD_SUBDIRECTORY(mnode)
ADD_SUBDIRECTORY(vnode)

1
src/client/inc/tsclient.h
View File

@ -348,6 +348,7 @@ typedef struct SSqlObj {
void * pStream;
void * pSubscription;
char * sqlstr;
char parseRetry;
char retry;
char maxRetry;
SRpcEpSet epSet;

1
src/client/src/tscAsync.c
View File

@ -43,6 +43,7 @@ void doAsyncQuery(STscObj* pObj, SSqlObj* pSql, void (*fp)(), void* param, const
pSql->signature = pSql;
pSql->param = param;
pSql->pTscObj = pObj;
pSql->parseRetry= 0;
pSql->maxRetry = TSDB_MAX_REPLICA;
pSql->fp = fp;
pSql->fetchFp = fp;

12
src/client/src/tscParseInsert.c
View File

@ -1335,13 +1335,13 @@ int tsParseSql(SSqlObj *pSql, bool initial) {
// make a backup as tsParseInsertSql may modify the string
char* sqlstr = strdup(pSql->sqlstr);
ret = tsParseInsertSql(pSql);
if (sqlstr == NULL || pSql->retry >= 1 || ret != TSDB_CODE_TSC_INVALID_SQL) {
if (sqlstr == NULL || pSql->parseRetry >= 1 || ret != TSDB_CODE_TSC_INVALID_SQL) {
free(sqlstr);
} else {
tscResetSqlCmdObj(pCmd, true);
free(pSql->sqlstr);
pSql->sqlstr = sqlstr;
pSql->retry++;
pSql->parseRetry++;
if ((ret = tsInsertInitialCheck(pSql)) == TSDB_CODE_SUCCESS) {
ret = tsParseInsertSql(pSql);
}
@ -1349,18 +1349,14 @@ int tsParseSql(SSqlObj *pSql, bool initial) {
} else {
SSqlInfo SQLInfo = qSQLParse(pSql->sqlstr);
ret = tscToSQLCmd(pSql, &SQLInfo);
if (ret == TSDB_CODE_TSC_INVALID_SQL && pSql->retry == 0 && SQLInfo.type == TSDB_SQL_NULL) {
if (ret == TSDB_CODE_TSC_INVALID_SQL && pSql->parseRetry == 0 && SQLInfo.type == TSDB_SQL_NULL) {
tscResetSqlCmdObj(pCmd, true);
pSql->retry++;
pSql->parseRetry++;
ret = tscToSQLCmd(pSql, &SQLInfo);
}
SQLInfoDestroy(&SQLInfo);
}
if (ret == TSDB_CODE_SUCCESS) {
pSql->retry = 0;
}
/*
* the pRes->code may be modified or released by another thread in tscTableMetaCallBack function,
* so do NOT use pRes->code to determine if the getTableMeta function

8
src/common/inc/tglobal.h
View File

@ -104,8 +104,12 @@ extern int32_t tsTelegrafUseFieldNum;
// mqtt
extern int32_t tsEnableMqttModule;
extern char tsMqttBrokerAddress[];
extern char tsMqttBrokerClientId[];
extern char tsMqttHostName[];
extern char tsMqttPort[];
extern char tsMqttUser[];
extern char tsMqttPass[];
extern char tsMqttClientId[];
extern char tsMqttTopic[];
// monitor
extern int32_t tsEnableMonitorModule;

52
src/common/src/tglobal.c
View File

@ -137,8 +137,12 @@ int32_t tsTelegrafUseFieldNum = 0;
// mqtt
int32_t tsEnableMqttModule = 0; // not finished yet, not started it by default
char tsMqttBrokerAddress[128] = {0};
char tsMqttBrokerClientId[128] = {0};
char tsMqttHostName[TSDB_MQTT_HOSTNAME_LEN] = "test.mosquitto.org";
char tsMqttPort[TSDB_MQTT_PORT_LEN] = "1883";
char tsMqttUser[TSDB_MQTT_USER_LEN] = {0};
char tsMqttPass[TSDB_MQTT_PASS_LEN] = {0};
char tsMqttClientId[TSDB_MQTT_CLIENT_ID_LEN] = "TDengineMqttSubscriber";
char tsMqttTopic[TSDB_MQTT_TOPIC_LEN] = "/test"; // #
// monitor
int32_t tsEnableMonitorModule = 1;
@ -247,8 +251,11 @@ bool taosCfgDynamicOptions(char *msg) {
for (int32_t i = 0; i < tsGlobalConfigNum; ++i) {
SGlobalCfg *cfg = tsGlobalConfig + i;
if (!(cfg->cfgType & TSDB_CFG_CTYPE_B_LOG)) continue;
//if (!(cfg->cfgType & TSDB_CFG_CTYPE_B_LOG)) continue;
if (cfg->valType != TAOS_CFG_VTYPE_INT32) continue;
int32_t cfgLen = strlen(cfg->option);
if (cfgLen != olen) continue;
if (strncasecmp(option, cfg->option, olen) != 0) continue;
*((int32_t *)cfg->ptr) = vint;
@ -767,26 +774,36 @@ static void doInitGlobalConfig(void) {
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosInitConfigOption(cfg);
cfg.option = "mqttBrokerAddress";
cfg.ptr = tsMqttBrokerAddress;
cfg.option = "mqttHostName";
cfg.ptr = tsMqttHostName;
cfg.valType = TAOS_CFG_VTYPE_STRING;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_CLIENT | TSDB_CFG_CTYPE_B_NOT_PRINT;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_NOT_PRINT;
cfg.minValue = 0;
cfg.maxValue = 0;
cfg.ptrLength = 126;
cfg.ptrLength = TSDB_MQTT_HOSTNAME_LEN;
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosInitConfigOption(cfg);
cfg.option = "mqttBrokerClientId";
cfg.ptr = tsMqttBrokerClientId;
cfg.option = "mqttPort";
cfg.ptr = tsMqttPort;
cfg.valType = TAOS_CFG_VTYPE_STRING;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_CLIENT | TSDB_CFG_CTYPE_B_NOT_PRINT;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_NOT_PRINT;
cfg.minValue = 0;
cfg.maxValue = 0;
cfg.ptrLength = 126;
cfg.ptrLength = TSDB_MQTT_PORT_LEN;
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosInitConfigOption(cfg);
cfg.option = "mqttTopic";
cfg.ptr = tsMqttTopic;
cfg.valType = TAOS_CFG_VTYPE_STRING;
cfg.cfgType = TSDB_CFG_CTYPE_B_CONFIG | TSDB_CFG_CTYPE_B_NOT_PRINT;
cfg.minValue = 0;
cfg.maxValue = 0;
cfg.ptrLength = TSDB_MQTT_TOPIC_LEN;
cfg.unitType = TAOS_CFG_UTYPE_NONE;
taosInitConfigOption(cfg);
cfg.option = "compressMsgSize";
cfg.ptr = &tsCompressMsgSize;
cfg.valType = TAOS_CFG_VTYPE_INT32;
@ -1270,6 +1287,9 @@ void taosInitGlobalCfg() {
}
bool taosCheckGlobalCfg() {
char fqdn[TSDB_FQDN_LEN];
uint16_t port;
if (debugFlag & DEBUG_TRACE || debugFlag & DEBUG_DEBUG || debugFlag & DEBUG_DUMP) {
taosSetAllDebugFlag();
}
@ -1283,12 +1303,18 @@ bool taosCheckGlobalCfg() {
if (tsFirst[0] == 0) {
strcpy(tsFirst, tsLocalEp);
} else {
taosGetFqdnPortFromEp(tsFirst, fqdn, &port);
snprintf(tsFirst, sizeof(tsFirst), "%s:%d", fqdn, port);
}
if (tsSecond[0] == 0) {
strcpy(tsSecond, tsLocalEp);
} else {
taosGetFqdnPortFromEp(tsSecond, fqdn, &port);
snprintf(tsSecond, sizeof(tsSecond), "%s:%d", fqdn, port);
}
taosGetSystemInfo();
tsSetLocale();

14
src/dnode/CMakeLists.txt
View File

@ -11,10 +11,12 @@ AUX_SOURCE_DIRECTORY(src SRC)
IF (TD_LINUX)
ADD_EXECUTABLE(taosd ${SRC})
TARGET_LINK_LIBRARIES(taosd mnode monitor http tsdb twal vnode cJson lz4)
IF (TD_SOMODE_STATIC)
TARGET_LINK_LIBRARIES(taosd mnode taos_static monitor http mqtt tsdb twal vnode cJson lz4 balance sync)
TARGET_LINK_LIBRARIES(taosd taos_static)
ELSE ()
TARGET_LINK_LIBRARIES(taosd mnode taos monitor http mqtt tsdb twal vnode cJson lz4 balance sync)
TARGET_LINK_LIBRARIES(taosd taos)
ENDIF ()
IF (TD_ACCOUNT)
@ -25,6 +27,14 @@ IF (TD_LINUX)
TARGET_LINK_LIBRARIES(taosd grant)
ENDIF ()
IF (TD_MQTT)
TARGET_LINK_LIBRARIES(taosd mqtt)
ENDIF ()
IF (TD_SYNC)
TARGET_LINK_LIBRARIES(taosd balance sync)
ENDIF ()
SET(PREPARE_ENV_CMD "prepare_env_cmd")
SET(PREPARE_ENV_TARGET "prepare_env_target")
ADD_CUSTOM_COMMAND(OUTPUT ${PREPARE_ENV_CMD}

4
src/dnode/src/dnodeMgmt.c
View File

@ -611,7 +611,7 @@ static bool dnodeReadMnodeInfos() {
}
for (int i = 0; i < size; ++i) {
cJSON* nodeInfo = cJSON_GetArrayItem(nodeInfos, i);
cJSON *nodeInfo = cJSON_GetArrayItem(nodeInfos, i);
if (nodeInfo == NULL) continue;
cJSON *nodeId = cJSON_GetObjectItem(nodeInfo, "nodeId");
@ -627,7 +627,7 @@ static bool dnodeReadMnodeInfos() {
goto PARSE_OVER;
}
strncpy(tsDMnodeInfos.nodeInfos[i].nodeEp, nodeEp->valuestring, TSDB_EP_LEN);
}
}
ret = true;

2
src/dnode/src/dnodeModule.c
View File

@ -62,6 +62,7 @@ static void dnodeAllocModules() {
dnodeSetModuleStatus(TSDB_MOD_HTTP);
}
#ifdef _MQTT
tsModule[TSDB_MOD_MQTT].enable = (tsEnableMqttModule == 1);
tsModule[TSDB_MOD_MQTT].name = "mqtt";
tsModule[TSDB_MOD_MQTT].initFp = mqttInitSystem;
@ -71,6 +72,7 @@ static void dnodeAllocModules() {
if (tsEnableMqttModule) {
dnodeSetModuleStatus(TSDB_MOD_MQTT);
}
#endif
tsModule[TSDB_MOD_MONITOR].enable = (tsEnableMonitorModule == 1);
tsModule[TSDB_MOD_MONITOR].name = "monitor";

7
src/inc/taosdef.h
View File

@ -272,6 +272,13 @@ void tsDataSwap(void *pLeft, void *pRight, int32_t type, int32_t size);
#define TSDB_SHOW_SQL_LEN 64
#define TSDB_SLOW_QUERY_SQL_LEN 512
#define TSDB_MQTT_HOSTNAME_LEN 64
#define TSDB_MQTT_PORT_LEN 8
#define TSDB_MQTT_USER_LEN 24
#define TSDB_MQTT_PASS_LEN 24
#define TSDB_MQTT_TOPIC_LEN 64
#define TSDB_MQTT_CLIENT_ID_LEN 32
#define TSDB_METER_STATE_OFFLINE 0
#define TSDB_METER_STATE_ONLLINE 1

6
src/inc/tmqtt.h
View File

@ -19,11 +19,11 @@
#ifdef __cplusplus
extern "C" {
#endif
#include <stdint.h>
int32_t mqttInitSystem();
int32_t mqttStartSystem();
void mqttStopSystem();
void mqttCleanUpSystem();
void mqttStopSystem();
void mqttCleanUpSystem();
#ifdef __cplusplus
}

5
src/inc/tsdb.h
View File

@ -38,6 +38,10 @@ extern "C" {
#define TSDB_STATUS_COMMIT_START 1
#define TSDB_STATUS_COMMIT_OVER 2
// TSDB STATE DEFINITION
#define TSDB_STATE_OK 0x0
#define TSDB_STATE_BAD_FILE 0x1
// --------- TSDB APPLICATION HANDLE DEFINITION
typedef struct {
void *appH;
@ -80,6 +84,7 @@ int32_t tsdbDropRepo(char *rootDir);
TSDB_REPO_T *tsdbOpenRepo(char *rootDir, STsdbAppH *pAppH);
void tsdbCloseRepo(TSDB_REPO_T *repo, int toCommit);
int32_t tsdbConfigRepo(TSDB_REPO_T *repo, STsdbCfg *pCfg);
int tsdbGetState(TSDB_REPO_T *repo);
// --------- TSDB TABLE DEFINITION
typedef struct {

2
src/kit/shell/src/shellLinux.c
View File

@ -409,7 +409,7 @@ void set_terminal_mode() {
}
}
void get_history_path(char *history) { sprintf(history, "%s/%s", getpwuid(getuid())->pw_dir, HISTORY_FILE); }
void get_history_path(char *history) { sprintf(history, "%s/%s", getenv("HOME"), HISTORY_FILE); }
void clearScreen(int ecmd_pos, int cursor_pos) {
struct winsize w;

7
src/mnode/src/mnodeDb.c
View File

@ -310,6 +310,13 @@ static int32_t mnodeCheckDbCfg(SDbCfg *pCfg) {
return TSDB_CODE_MND_INVALID_DB_OPTION;
}
#ifndef _SYNC
if (pCfg->replications != 1) {
mError("invalid db option replications:%d can only be 1 in this version", pCfg->replications);
return TSDB_CODE_MND_INVALID_DB_OPTION;
}
#endif
return TSDB_CODE_SUCCESS;
}

63
src/mnode/src/mnodeDnode.c
View File

@ -78,6 +78,9 @@ static int32_t mnodeDnodeActionInsert(SSdbOper *pOper) {
static int32_t mnodeDnodeActionDelete(SSdbOper *pOper) {
SDnodeObj *pDnode = pOper->pObj;
#ifndef _SYNC
mnodeDropAllDnodeVgroups(pDnode);
#endif
mnodeDropMnodeLocal(pDnode->dnodeId);
balanceAsyncNotify();
@ -585,7 +588,11 @@ static int32_t mnodeDropDnodeByEp(char *ep, SMnodeMsg *pMsg) {
mInfo("dnode:%d, start to drop it", pDnode->dnodeId);
#ifndef _SYNC
int32_t code = mnodeDropDnode(pDnode, pMsg);
#else
int32_t code = balanceDropDnode(pDnode);
#endif
mnodeDecDnodeRef(pDnode);
return code;
}
@ -1043,3 +1050,59 @@ static char* mnodeGetDnodeAlternativeRoleStr(int32_t alternativeRole) {
}
}
#ifndef _SYNC
int32_t balanceInit() { return TSDB_CODE_SUCCESS; }
void balanceCleanUp() {}
void balanceAsyncNotify() {}
void balanceSyncNotify() {}
void balanceReset() {}
int32_t balanceAlterDnode(struct SDnodeObj *pDnode, int32_t vnodeId, int32_t dnodeId) { return TSDB_CODE_SYN_NOT_ENABLED; }
char* syncRole[] = {
"offline",
"unsynced",
"syncing",
"slave",
"master"
};
int32_t balanceAllocVnodes(SVgObj *pVgroup) {
void * pIter = NULL;
SDnodeObj *pDnode = NULL;
SDnodeObj *pSelDnode = NULL;
float vnodeUsage = 1000.0;
while (1) {
pIter = mnodeGetNextDnode(pIter, &pDnode);
if (pDnode == NULL) break;
if (pDnode->numOfCores > 0 && pDnode->openVnodes < TSDB_MAX_VNODES) {
float openVnodes = pDnode->openVnodes;
if (pDnode->isMgmt) openVnodes += tsMnodeEqualVnodeNum;
float usage = openVnodes / pDnode->numOfCores;
if (usage <= vnodeUsage) {
pSelDnode = pDnode;
vnodeUsage = usage;
}
}
mnodeDecDnodeRef(pDnode);
}
sdbFreeIter(pIter);
if (pSelDnode == NULL) {
mError("failed to alloc vnode to vgroup");
return TSDB_CODE_MND_NO_ENOUGH_DNODES;
}
pVgroup->vnodeGid[0].dnodeId = pSelDnode->dnodeId;
pVgroup->vnodeGid[0].pDnode = pSelDnode;
mDebug("dnode:%d, alloc one vnode to vgroup, openVnodes:%d", pSelDnode->dnodeId, pSelDnode->openVnodes);
return TSDB_CODE_SUCCESS;
}
#endif

5
src/mnode/src/mnodeProfile.c
View File

@ -98,8 +98,10 @@ SConnObj *mnodeCreateConn(char *user, uint32_t ip, uint16_t port) {
.connId = connId,
.stime = taosGetTimestampMs()
};
tstrncpy(connObj.user, user, sizeof(connObj.user));
connObj.lastAccess = connObj.stime;
SConnObj *pConn = taosCachePut(tsMnodeConnCache, &connId, sizeof(int32_t), &connObj, sizeof(connObj), CONN_KEEP_TIME * 1000);
mDebug("connId:%d, is created, user:%s ip:%s:%u", connId, user, taosIpStr(ip), port);
@ -244,6 +246,7 @@ static int32_t mnodeRetrieveConns(SShowObj *pShow, char *data, int32_t rows, voi
cols++;
pWrite = data + pShow->offset[cols] * rows + pShow->bytes[cols] * numOfRows;
if (pConnObj->lastAccess < pConnObj->stime) pConnObj->lastAccess = pConnObj->stime;
*(int64_t *)pWrite = pConnObj->lastAccess;
cols++;

16
src/mnode/src/mnodeTable.c
View File

@ -1711,14 +1711,20 @@ static int32_t mnodeDoCreateChildTable(SMnodeMsg *pMsg, int32_t tid) {
mnodeDestroyChildTable(pTable);
return TSDB_CODE_MND_INVALID_TABLE_NAME;
}
pTable->suid = pMsg->pSTable->uid;
pTable->uid = (((uint64_t)pTable->vgId) << 40) + ((((uint64_t)pTable->sid) & ((1ul << 24) - 1ul)) << 16) +
(sdbGetVersion() & ((1ul << 16) - 1ul));
pTable->uid = (((uint64_t)pTable->vgId) << 48) + ((((uint64_t)pTable->sid) & ((1ul << 24) - 1ul)) << 24) +
((sdbGetVersion() & ((1ul << 16) - 1ul)) << 8) + (taosRand() & ((1ul << 8) - 1ul));
pTable->superTable = pMsg->pSTable;
} else {
pTable->uid = (((uint64_t)pTable->vgId) << 40) + ((((uint64_t)pTable->sid) & ((1ul << 24) - 1ul)) << 16) +
(sdbGetVersion() & ((1ul << 16) - 1ul));
if (pTable->info.type == TSDB_SUPER_TABLE) {
int64_t us = taosGetTimestampUs();
pTable->uid = (us << 24) + ((sdbGetVersion() & ((1ul << 16) - 1ul)) << 8) + (taosRand() & ((1ul << 8) - 1ul));
} else {
pTable->uid = (((uint64_t)pTable->vgId) << 48) + ((((uint64_t)pTable->sid) & ((1ul << 24) - 1ul)) << 24) +
((sdbGetVersion() & ((1ul << 16) - 1ul)) << 8) + (taosRand() & ((1ul << 8) - 1ul));
}
pTable->sversion = 0;
pTable->numOfColumns = htons(pCreate->numOfColumns);
pTable->sqlLen = htons(pCreate->sqlLen);

2
src/mnode/src/mnodeUser.c
View File

@ -581,7 +581,7 @@ void mnodeDropAllUsers(SAcctObj *pAcct) {
int32_t mnodeRetriveAuth(char *user, char *spi, char *encrypt, char *secret, char *ckey) {
if (!sdbIsMaster()) {
*secret = 0;
mDebug("user:%s, failed to auth user, reason:%s", user, tstrerror(TSDB_CODE_APP_NOT_READY));
mDebug("user:%s, failed to auth user, mnode is not master", user);
return TSDB_CODE_APP_NOT_READY;
}

2
src/os/inc/os.h
View File

@ -36,7 +36,7 @@ extern "C" {
#include "osLinux32.h"
#endif
#ifdef _TD_ALPINE
#ifdef _ALPINE
#include "osAlpine.h"
#endif

4
src/os/inc/osAlpine.h
View File

@ -13,8 +13,8 @@
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef TDENGINE_OS_LINUX64_H
#define TDENGINE_OS_LINUX64_H
#ifndef TDENGINE_OS_ALPINE_H
#define TDENGINE_OS_ALPINE_H
#ifdef __cplusplus
extern "C" {

2
src/os/inc/osLinux64.h
View File

@ -75,7 +75,9 @@ extern "C" {
#include <fcntl.h>
#include <sys/utsname.h>
#include <sys/resource.h>
#ifndef _ALPINE
#include <error.h>
#endif
#include <linux/sysctl.h>
#ifdef __cplusplus

4
src/os/inc/osWindows.h
View File

@ -179,9 +179,9 @@ int gettimeofday(struct timeval *ptv, void *pTimeZone);
#endif
#ifdef _MSC_VER
#if _MSC_VER >= 1900
//#if _MSC_VER >= 1900
#define TAOS_OS_FUNC_SOCKET_INET
#endif
//#endif
#endif
#define SHUT_RDWR SD_BOTH

10
src/os/src/detail/osSysinfo.c
View File

@ -579,7 +579,11 @@ void taosSetCoreDump() {
struct rlimit rlim;
struct rlimit rlim_new;
if (getrlimit(RLIMIT_CORE, &rlim) == 0) {
#ifndef _ALPINE
uInfo("the old unlimited para: rlim_cur=%" PRIu64 ", rlim_max=%" PRIu64, rlim.rlim_cur, rlim.rlim_max);
#else
uInfo("the old unlimited para: rlim_cur=%llu, rlim_max=%llu", rlim.rlim_cur, rlim.rlim_max);
#endif
rlim_new.rlim_cur = RLIM_INFINITY;
rlim_new.rlim_max = RLIM_INFINITY;
if (setrlimit(RLIMIT_CORE, &rlim_new) != 0) {
@ -591,7 +595,11 @@ void taosSetCoreDump() {
}
if (getrlimit(RLIMIT_CORE, &rlim) == 0) {
#ifndef _ALPINE
uInfo("the new unlimited para: rlim_cur=%" PRIu64 ", rlim_max=%" PRIu64, rlim.rlim_cur, rlim.rlim_max);
#else
uInfo("the new unlimited para: rlim_cur=%llu, rlim_max=%llu", rlim.rlim_cur, rlim.rlim_max);
#endif
}
#ifndef _TD_ARM_
@ -659,4 +667,4 @@ bool taosGetSystemUid(char *uid) {
return false;
}
#endif
#endif

4
src/os/src/windows/wSocket.c
View File

@ -61,6 +61,10 @@ int taosSetSockOpt(SOCKET socketfd, int level, int optname, void *optval, int op
return 0;
}
if (level == SOL_TCP && optname == TCP_KEEPCNT) {
return 0;
}
return setsockopt(socketfd, level, optname, optval, optlen);
}

4
src/plugins/CMakeLists.txt
View File

@ -3,4 +3,6 @@ PROJECT(TDengine)
ADD_SUBDIRECTORY(monitor)
ADD_SUBDIRECTORY(http)
ADD_SUBDIRECTORY(mqtt)
IF (TD_MQTT)
ADD_SUBDIRECTORY(mqtt)
ENDIF ()

5
src/plugins/http/CMakeLists.txt
View File

@ -11,11 +11,12 @@ AUX_SOURCE_DIRECTORY(src SRC)
IF (TD_LINUX)
ADD_LIBRARY(http ${SRC})
TARGET_LINK_LIBRARIES(http z)
IF (TD_SOMODE_STATIC)
TARGET_LINK_LIBRARIES(http taos_static z)
TARGET_LINK_LIBRARIES(http taos_static)
ELSE ()
TARGET_LINK_LIBRARIES(http taos z)
TARGET_LINK_LIBRARIES(http taos)
ENDIF ()
IF (TD_ADMIN)

6
src/plugins/monitor/CMakeLists.txt
View File

@ -2,11 +2,11 @@ CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/client/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/query/inc)
AUX_SOURCE_DIRECTORY(./src SRC)
IF (TD_LINUX)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/client/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/query/inc)
ADD_LIBRARY(monitor ${SRC})
IF (TD_SOMODE_STATIC)

348
src/plugins/monitor/src/monitorMain.c
View File

@ -25,6 +25,7 @@
#include "tsclient.h"
#include "dnode.h"
#include "monitor.h"
#include "taoserror.h"
#define monitorFatal(...) { if (monitorDebugFlag & DEBUG_FATAL) { taosPrintLog("MON FATAL ", 255, __VA_ARGS__); }}
#define monitorError(...) { if (monitorDebugFlag & DEBUG_ERROR) { taosPrintLog("MON ERROR ", 255, __VA_ARGS__); }}
@ -33,129 +34,159 @@
#define monitorDebug(...) { if (monitorDebugFlag & DEBUG_DEBUG) { taosPrintLog("MON ", monitorDebugFlag, __VA_ARGS__); }}
#define monitorTrace(...) { if (monitorDebugFlag & DEBUG_TRACE) { taosPrintLog("MON ", monitorDebugFlag, __VA_ARGS__); }}
#define SQL_LENGTH 1024
#define SQL_LENGTH 1030
#define LOG_LEN_STR 100
#define IP_LEN_STR TSDB_EP_LEN
#define CHECK_INTERVAL 1000
typedef enum {
MONITOR_CMD_CREATE_DB,
MONITOR_CMD_CREATE_TB_LOG,
MONITOR_CMD_CREATE_MT_DN,
MONITOR_CMD_CREATE_MT_ACCT,
MONITOR_CMD_CREATE_TB_DN,
MONITOR_CMD_CREATE_TB_ACCT_ROOT,
MONITOR_CMD_CREATE_TB_SLOWQUERY,
MONITOR_CMD_MAX
MON_CMD_CREATE_DB,
MON_CMD_CREATE_TB_LOG,
MON_CMD_CREATE_MT_DN,
MON_CMD_CREATE_MT_ACCT,
MON_CMD_CREATE_TB_DN,
MON_CMD_CREATE_TB_ACCT_ROOT,
MON_CMD_CREATE_TB_SLOWQUERY,
MON_CMD_MAX
} EMonitorCommand;
typedef enum {
MONITOR_STATE_UN_INIT,
MONITOR_STATE_INITIALIZING,
MONITOR_STATE_INITIALIZED,
MONITOR_STATE_STOPPED
MON_STATE_NOT_INIT,
MON_STATE_INITED
} EMonitorState;
typedef struct {
void * conn;
void * timer;
char ep[TSDB_EP_LEN];
int8_t cmdIndex;
int8_t state;
char sql[SQL_LENGTH + 1];
void * initTimer;
void * diskTimer;
pthread_t thread;
void * conn;
char ep[TSDB_EP_LEN];
int8_t cmdIndex;
int8_t state;
int8_t start; // enable/disable by mnode
int8_t quiting; // taosd is quiting
char sql[SQL_LENGTH + 1];
} SMonitorConn;
static SMonitorConn tsMonitorConn;
static void monitorInitConn(void *para, void *unused);
static void monitorInitConnCb(void *param, TAOS_RES *result, int32_t code);
static void monitorInitDatabase();
static void monitorInitDatabaseCb(void *param, TAOS_RES *result, int32_t code);
static void monitorStartTimer();
static void monitorSaveSystemInfo();
static SMonitorConn tsMonitor = {0};
static void monitorSaveSystemInfo();
static void *monitorThreadFunc(void *param);
static void monitorBuildMonitorSql(char *sql, int32_t cmd);
extern int32_t (*monitorStartSystemFp)();
extern void (*monitorStopSystemFp)();
extern void (*monitorExecuteSQLFp)(char *sql);
static void monitorCheckDiskUsage(void *para, void *unused) {
taosGetDisk();
taosTmrReset(monitorCheckDiskUsage, CHECK_INTERVAL, NULL, tscTmr, &tsMonitorConn.diskTimer);
}
extern void (*monitorStopSystemFp)();
extern void (*monitorExecuteSQLFp)(char *sql);
int32_t monitorInitSystem() {
taos_init();
taosTmrReset(monitorCheckDiskUsage, CHECK_INTERVAL, NULL, tscTmr, &tsMonitorConn.diskTimer);
if (tsMonitor.ep[0] == 0) {
strcpy(tsMonitor.ep, tsLocalEp);
}
int len = strlen(tsMonitor.ep);
for (int i = 0; i < len; ++i) {
if (tsMonitor.ep[i] == ':' || tsMonitor.ep[i] == '-' || tsMonitor.ep[i] == '.') {
tsMonitor.ep[i] = '_';
}
}
pthread_attr_t thAttr;
pthread_attr_init(&thAttr);
pthread_attr_setdetachstate(&thAttr, PTHREAD_CREATE_JOINABLE);
if (pthread_create(&tsMonitor.thread, &thAttr, monitorThreadFunc, NULL)) {
monitorError("failed to create thread to for monitor module, reason:%s", strerror(errno));
return -1;
}
pthread_attr_destroy(&thAttr);
monitorDebug("monitor thread is launched");
monitorStartSystemFp = monitorStartSystem;
monitorStopSystemFp = monitorStopSystem;
return 0;
}
int32_t monitorStartSystem() {
monitorInfo("start monitor module");
monitorInitSystem();
taosTmrReset(monitorInitConn, 10, NULL, tscTmr, &tsMonitorConn.initTimer);
taos_init();
tsMonitor.start = 1;
monitorExecuteSQLFp = monitorExecuteSQL;
monitorInfo("monitor module start");
return 0;
}
static void monitorStartSystemRetry() {
if (tsMonitorConn.initTimer != NULL) {
taosTmrReset(monitorInitConn, 3000, NULL, tscTmr, &tsMonitorConn.initTimer);
}
}
static void *monitorThreadFunc(void *param) {
monitorDebug("starting to initialize monitor module ...");
static void monitorInitConn(void *para, void *unused) {
if (dnodeGetDnodeId() <= 0) {
monitorStartSystemRetry();
return;
}
monitorInfo("starting to initialize monitor service ..");
tsMonitorConn.state = MONITOR_STATE_INITIALIZING;
while (1) {
if (tsMonitor.quiting) {
tsMonitor.state = MON_STATE_NOT_INIT;
monitorInfo("monitor thread will quit, for taosd is quiting");
break;
} else {
taosGetDisk();
}
if (tsMonitorConn.ep[0] == 0)
strcpy(tsMonitorConn.ep, tsLocalEp);
if (tsMonitor.start == 0) {
continue;
}
int len = strlen(tsMonitorConn.ep);
for (int i = 0; i < len; ++i) {
if (tsMonitorConn.ep[i] == ':' || tsMonitorConn.ep[i] == '-') {
tsMonitorConn.ep[i] = '_';
static int32_t accessTimes = 0;
accessTimes++;
taosMsleep(1000);
if (dnodeGetDnodeId() <= 0) {
monitorDebug("dnode not initialized, waiting for 3000 ms to start monitor module");
continue;
}
if (tsMonitor.conn == NULL) {
tsMonitor.state = MON_STATE_NOT_INIT;
tsMonitor.conn = taos_connect(NULL, "monitor", tsInternalPass, "", 0);
if (tsMonitor.conn == NULL) {
monitorError("failed to connect to database, reason:%s", tstrerror(terrno));
continue;
} else {
monitorDebug("connect to database success");
}
}
if (tsMonitor.state == MON_STATE_NOT_INIT) {
for (; tsMonitor.cmdIndex < MON_CMD_MAX; ++tsMonitor.cmdIndex) {
monitorBuildMonitorSql(tsMonitor.sql, tsMonitor.cmdIndex);
void *res = taos_query(tsMonitor.conn, tsMonitor.sql);
int code = taos_errno(res);
taos_free_result(res);
if (code != 0) {
monitorError("failed to exec sql:%s, reason:%s", tsMonitor.sql, tstrerror(code));
break;
} else {
monitorDebug("successfully to exec sql:%s", tsMonitor.sql);
}
}
if (tsMonitor.start) {
tsMonitor.state = MON_STATE_INITED;
}
}
if (tsMonitor.state == MON_STATE_INITED) {
if (accessTimes % tsMonitorInterval == 0) {
monitorSaveSystemInfo();
}
}
}
if (tsMonitorConn.conn == NULL) {
taos_connect_a(NULL, "monitor", tsInternalPass, "", 0, monitorInitConnCb, &tsMonitorConn, &(tsMonitorConn.conn));
} else {
monitorInitDatabase();
}
monitorInfo("monitor thread is stopped");
return NULL;
}
static void monitorInitConnCb(void *param, TAOS_RES *result, int32_t code) {
// free it firstly in any cases.
taos_free_result(result);
if (code != TSDB_CODE_SUCCESS) {
monitorError("monitor:%p, connect to database failed, reason:%s", tsMonitorConn.conn, tstrerror(code));
taos_close(tsMonitorConn.conn);
tsMonitorConn.conn = NULL;
tsMonitorConn.state = MONITOR_STATE_UN_INIT;
monitorStartSystemRetry();
return;
}
monitorDebug("monitor:%p, connect to database success, reason:%s", tsMonitorConn.conn, tstrerror(code));
monitorInitDatabase();
}
static void dnodeBuildMonitorSql(char *sql, int32_t cmd) {
static void monitorBuildMonitorSql(char *sql, int32_t cmd) {
memset(sql, 0, SQL_LENGTH);
if (cmd == MONITOR_CMD_CREATE_DB) {
if (cmd == MON_CMD_CREATE_DB) {
snprintf(sql, SQL_LENGTH,
"create database if not exists %s replica 1 days 10 keep 30 cache %d "
"blocks %d maxtables 16 precision 'us'",
"blocks %d precision 'us'",
tsMonitorDbName, TSDB_MIN_CACHE_BLOCK_SIZE, TSDB_MIN_TOTAL_BLOCKS);
} else if (cmd == MONITOR_CMD_CREATE_MT_DN) {
} else if (cmd == MON_CMD_CREATE_MT_DN) {
snprintf(sql, SQL_LENGTH,
"create table if not exists %s.dn(ts timestamp"
", cpu_taosd float, cpu_system float, cpu_cores int"
@ -166,10 +197,10 @@ static void dnodeBuildMonitorSql(char *sql, int32_t cmd) {
", req_http int, req_select int, req_insert int"
") tags (dnodeid int, fqdn binary(%d))",
tsMonitorDbName, TSDB_FQDN_LEN);
} else if (cmd == MONITOR_CMD_CREATE_TB_DN) {
} else if (cmd == MON_CMD_CREATE_TB_DN) {
snprintf(sql, SQL_LENGTH, "create table if not exists %s.dn%d using %s.dn tags(%d, '%s')", tsMonitorDbName,
dnodeGetDnodeId(), tsMonitorDbName, dnodeGetDnodeId(), tsLocalEp);
} else if (cmd == MONITOR_CMD_CREATE_MT_ACCT) {
} else if (cmd == MON_CMD_CREATE_MT_ACCT) {
snprintf(sql, SQL_LENGTH,
"create table if not exists %s.acct(ts timestamp "
", currentPointsPerSecond bigint, maxPointsPerSecond bigint"
@ -185,15 +216,15 @@ static void dnodeBuildMonitorSql(char *sql, int32_t cmd) {
", accessState smallint"
") tags (acctId binary(%d))",
tsMonitorDbName, TSDB_USER_LEN);
} else if (cmd == MONITOR_CMD_CREATE_TB_ACCT_ROOT) {
} else if (cmd == MON_CMD_CREATE_TB_ACCT_ROOT) {
snprintf(sql, SQL_LENGTH, "create table if not exists %s.acct_%s using %s.acct tags('%s')", tsMonitorDbName, TSDB_DEFAULT_USER,
tsMonitorDbName, TSDB_DEFAULT_USER);
} else if (cmd == MONITOR_CMD_CREATE_TB_SLOWQUERY) {
} else if (cmd == MON_CMD_CREATE_TB_SLOWQUERY) {
snprintf(sql, SQL_LENGTH,
"create table if not exists %s.slowquery(ts timestamp, username "
"binary(%d), created_time timestamp, time bigint, sql binary(%d))",
tsMonitorDbName, TSDB_TABLE_FNAME_LEN - 1, TSDB_SLOW_QUERY_SQL_LEN);
} else if (cmd == MONITOR_CMD_CREATE_TB_LOG) {
} else if (cmd == MON_CMD_CREATE_TB_LOG) {
snprintf(sql, SQL_LENGTH,
"create table if not exists %s.log(ts timestamp, level tinyint, "
"content binary(%d), ipaddr binary(%d))",
@ -203,75 +234,22 @@ static void dnodeBuildMonitorSql(char *sql, int32_t cmd) {
sql[SQL_LENGTH] = 0;
}
static void monitorInitDatabase() {
if (tsMonitorConn.cmdIndex < MONITOR_CMD_MAX) {
dnodeBuildMonitorSql(tsMonitorConn.sql, tsMonitorConn.cmdIndex);
taos_query_a(tsMonitorConn.conn, tsMonitorConn.sql, monitorInitDatabaseCb, NULL);
} else {
tsMonitorConn.state = MONITOR_STATE_INITIALIZED;
monitorExecuteSQLFp = monitorExecuteSQL;
monitorInfo("monitor service init success");
monitorStartTimer();
}
}
static void monitorInitDatabaseCb(void *param, TAOS_RES *result, int32_t code) {
if (code == TSDB_CODE_MND_TABLE_ALREADY_EXIST || code == TSDB_CODE_MND_DB_ALREADY_EXIST || code >= 0) {
monitorDebug("monitor:%p, sql success, reason:%s, %s", tsMonitorConn.conn, tstrerror(code), tsMonitorConn.sql);
if (tsMonitorConn.cmdIndex == MONITOR_CMD_CREATE_TB_LOG) {
monitorInfo("dnode:%s is started", tsLocalEp);
}
tsMonitorConn.cmdIndex++;
monitorInitDatabase();
} else {
monitorError("monitor:%p, sql failed, reason:%s, %s", tsMonitorConn.conn, tstrerror(code), tsMonitorConn.sql);
tsMonitorConn.state = MONITOR_STATE_UN_INIT;
monitorStartSystemRetry();
}
taos_free_result(result);
}
void monitorStopSystem() {
if (tsMonitorConn.state == MONITOR_STATE_STOPPED) return;
tsMonitorConn.state = MONITOR_STATE_STOPPED;
tsMonitor.start = 0;
tsMonitor.state = MON_STATE_NOT_INIT;
monitorExecuteSQLFp = NULL;
monitorInfo("monitor module is stopped");
if (tsMonitorConn.initTimer != NULL) {
taosTmrStopA(&(tsMonitorConn.initTimer));
}
if (tsMonitorConn.timer != NULL) {
taosTmrStopA(&(tsMonitorConn.timer));
}
if (tsMonitorConn.conn != NULL) {
taos_close(tsMonitorConn.conn);
tsMonitorConn.conn = NULL;
}
monitorInfo("monitor module stopped");
}
void monitorCleanUpSystem() {
tsMonitor.quiting = 1;
monitorStopSystem();
monitorInfo("monitor module cleanup");
}
static void monitorStartTimer() {
taosTmrReset(monitorSaveSystemInfo, tsMonitorInterval * 1000, NULL, tscTmr, &tsMonitorConn.timer);
}
static void dnodeMontiorLogCallback(void *param, TAOS_RES *result, int32_t code) {
int32_t c = taos_errno(result);
if (c != TSDB_CODE_SUCCESS) {
monitorError("monitor:%p, save %s failed, reason:%s", tsMonitorConn.conn, (char *)param, tstrerror(c));
} else {
int32_t rows = taos_affected_rows(result);
monitorDebug("monitor:%p, save %s succ, rows:%d", tsMonitorConn.conn, (char *)param, rows);
pthread_join(tsMonitor.thread, NULL);
if (tsMonitor.conn != NULL) {
taos_close(tsMonitor.conn);
tsMonitor.conn = NULL;
}
taos_free_result(result);
monitorInfo("monitor module is cleaned up");
}
// unit is MB
@ -279,13 +257,13 @@ static int32_t monitorBuildMemorySql(char *sql) {
float sysMemoryUsedMB = 0;
bool suc = taosGetSysMemory(&sysMemoryUsedMB);
if (!suc) {
monitorError("monitor:%p, get sys memory info failed.", tsMonitorConn.conn);
monitorDebug("failed to get sys memory info");
}
float procMemoryUsedMB = 0;
suc = taosGetProcMemory(&procMemoryUsedMB);
if (!suc) {
monitorError("monitor:%p, get proc memory info failed.", tsMonitorConn.conn);
monitorDebug("failed to get proc memory info");
}
return sprintf(sql, ", %f, %f, %d", procMemoryUsedMB, sysMemoryUsedMB, tsTotalMemoryMB);
@ -296,11 +274,11 @@ static int32_t monitorBuildCpuSql(char *sql) {
float sysCpuUsage = 0, procCpuUsage = 0;
bool suc = taosGetCpuUsage(&sysCpuUsage, &procCpuUsage);
if (!suc) {
monitorError("monitor:%p, get cpu usage failed.", tsMonitorConn.conn);
monitorDebug("failed to get cpu usage");
}
if (sysCpuUsage <= procCpuUsage) {
sysCpuUsage = procCpuUsage + (float)0.1;
sysCpuUsage = procCpuUsage + 0.1f;
}
return sprintf(sql, ", %f, %f, %d", procCpuUsage, sysCpuUsage, tsNumOfCores);
@ -316,14 +294,14 @@ static int32_t monitorBuildBandSql(char *sql) {
float bandSpeedKb = 0;
bool suc = taosGetBandSpeed(&bandSpeedKb);
if (!suc) {
monitorError("monitor:%p, get bandwidth speed failed.", tsMonitorConn.conn);
monitorDebug("failed to get bandwidth speed");
}
return sprintf(sql, ", %f", bandSpeedKb);
}
static int32_t monitorBuildReqSql(char *sql) {
SDnodeStatisInfo info = dnodeGetStatisInfo();
SDnodeStatisInfo info = dnodeGetStatisInfo();
return sprintf(sql, ", %d, %d, %d)", info.httpReqNum, info.queryReqNum, info.submitReqNum);
}
@ -331,20 +309,15 @@ static int32_t monitorBuildIoSql(char *sql) {
float readKB = 0, writeKB = 0;
bool suc = taosGetProcIO(&readKB, &writeKB);
if (!suc) {
monitorError("monitor:%p, get io info failed.", tsMonitorConn.conn);
monitorDebug("failed to get io info");
}
return sprintf(sql, ", %f, %f", readKB, writeKB);
}
static void monitorSaveSystemInfo() {
if (tsMonitorConn.state != MONITOR_STATE_INITIALIZED) {
monitorStartTimer();
return;
}
int64_t ts = taosGetTimestampUs();
char * sql = tsMonitorConn.sql;
char * sql = tsMonitor.sql;
int32_t pos = snprintf(sql, SQL_LENGTH, "insert into %s.dn%d values(%" PRId64, tsMonitorDbName, dnodeGetDnodeId(), ts);
pos += monitorBuildCpuSql(sql + pos);
@ -354,16 +327,31 @@ static void monitorSaveSystemInfo() {
pos += monitorBuildIoSql(sql + pos);
pos += monitorBuildReqSql(sql + pos);
monitorDebug("monitor:%p, save system info, sql:%s", tsMonitorConn.conn, sql);
taos_query_a(tsMonitorConn.conn, sql, dnodeMontiorLogCallback, "sys");
void *res = taos_query(tsMonitor.conn, tsMonitor.sql);
int code = taos_errno(res);
taos_free_result(res);
if (tsMonitorConn.timer != NULL && tsMonitorConn.state != MONITOR_STATE_STOPPED) {
monitorStartTimer();
if (code != 0) {
monitorError("failed to save system info, reason:%s, sql:%s", tstrerror(code), tsMonitor.sql);
} else {
monitorDebug("successfully to save system info, sql:%s", tsMonitor.sql);
}
}
static void montiorExecSqlCb(void *param, TAOS_RES *result, int32_t code) {
int32_t c = taos_errno(result);
if (c != TSDB_CODE_SUCCESS) {
monitorError("save %s failed, reason:%s", (char *)param, tstrerror(c));
} else {
int32_t rows = taos_affected_rows(result);
monitorDebug("save %s succ, rows:%d", (char *)param, rows);
}
taos_free_result(result);
}
void monitorSaveAcctLog(SAcctMonitorObj *pMon) {
if (tsMonitorConn.state != MONITOR_STATE_INITIALIZED) return;
if (tsMonitor.state != MON_STATE_INITED) return;
char sql[1024] = {0};
sprintf(sql,
@ -392,19 +380,16 @@ void monitorSaveAcctLog(SAcctMonitorObj *pMon) {
pMon->totalConns, pMon->maxConns,
pMon->accessState);
monitorDebug("monitor:%p, save account info, sql %s", tsMonitorConn.conn, sql);
taos_query_a(tsMonitorConn.conn, sql, dnodeMontiorLogCallback, "account");
monitorDebug("save account info, sql:%s", sql);
taos_query_a(tsMonitor.conn, sql, montiorExecSqlCb, "account info");
}
void monitorSaveLog(int32_t level, const char *const format, ...) {
if (tsMonitorConn.state != MONITOR_STATE_INITIALIZED) return;
if (tsMonitor.state != MON_STATE_INITED) return;
va_list argpointer;
char sql[SQL_LENGTH] = {0};
int32_t max_length = SQL_LENGTH - 30;
if (tsMonitorConn.state != MONITOR_STATE_INITIALIZED) return;
int32_t len = snprintf(sql, (size_t)max_length, "insert into %s.log values(%" PRId64 ", %d,'", tsMonitorDbName,
taosGetTimestampUs(), level);
@ -416,12 +401,13 @@ void monitorSaveLog(int32_t level, const char *const format, ...) {
len += sprintf(sql + len, "', '%s')", tsLocalEp);
sql[len++] = 0;
monitorDebug("monitor:%p, save log, sql: %s", tsMonitorConn.conn, sql);
taos_query_a(tsMonitorConn.conn, sql, dnodeMontiorLogCallback, "log");
monitorDebug("save log, sql: %s", sql);
taos_query_a(tsMonitor.conn, sql, montiorExecSqlCb, "log");
}
void monitorExecuteSQL(char *sql) {
if (tsMonitorConn.state != MONITOR_STATE_INITIALIZED) return;
monitorDebug("monitor:%p, execute sql: %s", tsMonitorConn.conn, sql);
taos_query_a(tsMonitorConn.conn, sql, dnodeMontiorLogCallback, "sql");
if (tsMonitor.state != MON_STATE_INITED) return;
monitorDebug("execute sql:%s", sql);
taos_query_a(tsMonitor.conn, sql, montiorExecSqlCb, "sql");
}

16
src/plugins/mqtt/CMakeLists.txt
View File

@ -2,21 +2,19 @@ CMAKE_MINIMUM_REQUIRED(VERSION 2.8)
PROJECT(TDengine)
INCLUDE_DIRECTORIES(inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/client/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/query/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/MQTT-C/include)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/MQTT-C/examples/templates)
AUX_SOURCE_DIRECTORY(src SRC)
IF (TD_LINUX)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/client/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/src/query/inc)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/MQTT-C/include)
INCLUDE_DIRECTORIES(${TD_COMMUNITY_DIR}/deps/MQTT-C/examples/templates)
ADD_LIBRARY(mqtt ${SRC})
TARGET_LINK_LIBRARIES(mqtt cJson mqttc)
IF (TD_SOMODE_STATIC)
TARGET_LINK_LIBRARIES(mqtt taos_static cJson mqttc)
TARGET_LINK_LIBRARIES(mqtt taos_static)
ELSE ()
TARGET_LINK_LIBRARIES(mqtt taos cJson mqttc)
ENDIF ()
IF (TD_ADMIN)
TARGET_LINK_LIBRARIES(mqtt admin cJson)
TARGET_LINK_LIBRARIES(mqtt taos)
ENDIF ()
ENDIF ()

28
src/plugins/mqtt/inc/mqttInit.h
View File

@ -23,11 +23,12 @@ extern "C" {
* @file
* A simple subscriber program that performs automatic reconnections.
*/
#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include "mqtt.h"
#include "taos.h"
#define QOS 1
#define TIMEOUT 10000L
#define MQTT_SEND_BUF_SIZE 102400
#define MQTT_RECV_BUF_SIZE 102400
/**
* @brief A structure that I will use to keep track of some data needed
@ -36,18 +37,12 @@ extern "C" {
* An instance of this struct will be created in my \c main(). Then, whenever
* \ref mqttReconnectClient is called, this instance will be passed.
*/
struct reconnect_state_t {
char* hostname;
char* port;
char* topic;
char* client_id;
char* user_name;
char* password;
typedef struct SMqttReconnectState {
uint8_t* sendbuf;
size_t sendbufsz;
uint8_t* recvbuf;
size_t recvbufsz;
};
} SMqttReconnectState;
/**
* @brief My reconnect callback. It will reestablish the connection whenever
@ -58,7 +53,7 @@ void mqttReconnectClient(struct mqtt_client* client, void** reconnect_state_vptr
/**
* @brief The function will be called whenever a PUBLISH message is received.
*/
void mqtt_PublishCallback(void** unused, struct mqtt_response_publish* published);
void mqttPublishCallback(void** unused, struct mqtt_response_publish* published);
/**
* @brief The client's refresher. This function triggers back-end routines to
@ -73,12 +68,7 @@ void* mqttClientRefresher(void* client);
/**
* @brief Safelty closes the \p sockfd and cancels the \p client_daemon before \c exit.
*/
void mqttCleanup(int status, int sockfd, pthread_t* client_daemon);
void mqttInitConnCb(void* param, TAOS_RES* result, int32_t code);
void mqttQueryInsertCallback(void* param, TAOS_RES* result, int32_t code);
#define QOS 1
#define TIMEOUT 10000L
void mqttCleanupRes(int status, int sockfd, pthread_t* client_daemon);
#ifdef __cplusplus
}

6
src/plugins/mqtt/inc/mqttPayload.h
View File

@ -15,11 +15,13 @@
#ifndef TDENGINE_MQTT_PLYLOAD_H
#define TDENGINE_MQTT_PLYLOAD_H
#ifdef __cplusplus
extern "C" {
#endif
char split(char str[], char delims[], char** p_p_cmd_part, int max);
char* converJsonToSql(char* json, char* _dbname, char* _tablename);
char* mqttConverJsonToSql(char* json, int maxSize);
#ifdef __cplusplus
}
#endif

182
src/plugins/mqtt/src/mqttPayload.c
View File

@ -14,52 +14,146 @@
*/
#define _DEFAULT_SOURCE
#include "mqttPayload.h"
#include "cJSON.h"
#include "string.h"
#include "taos.h"
#include "mqttLog.h"
#include "os.h"
char split(char str[], char delims[], char** p_p_cmd_part, int max) {
char* token = strtok(str, delims);
char part_index = 0;
char** tmp_part = p_p_cmd_part;
while (token) {
*tmp_part++ = token;
token = strtok(NULL, delims);
part_index++;
if (part_index >= max) break;
}
return part_index;
}
#include "cJSON.h"
#include "mqttLog.h"
#include "mqttPayload.h"
char* converJsonToSql(char* json, char* _dbname, char* _tablename) {
cJSON* jPlayload = cJSON_Parse(json);
char _names[102400] = {0};
char _values[102400] = {0};
int i = 0;
int count = cJSON_GetArraySize(jPlayload);
for (; i < count; i++)
{
cJSON* item = cJSON_GetArrayItem(jPlayload, i);
if (cJSON_Object == item->type) {
mqttInfo("The item '%s' is not supported", item->string);
} else {
strcat(_names, item->string);
if (i < count - 1) {
strcat(_names, ",");
}
char* __value_json = cJSON_Print(item);
strcat(_values, __value_json);
free(__value_json);
if (i < count - 1) {
strcat(_values, ",");
}
// subscribe message like this
/*
/test {
"timestamp": 1599121290,
"gateway": {
"name": "AcuLink 810 Gateway",
"model": "AcuLink810-868",
"serial": "S8P20200207"
},
"device": {
"name": "Acuvim L V3 .221",
"model": "Acuvim-L-V3",
"serial": "221",
"online": true,
"readings": [
{
"param": "Freq_Hz",
"value": "59.977539",
"unit": "Hz"
},
{
"param": "Va_V",
"value": "122.002907",
"unit": "V"
},
{
"param": "DI4",
"value": "5.000000",
"unit": ""
}
]
}
}
*/
// send msg cmd
// mosquitto_pub -h test.mosquitto.org -t "/test" -m '{"timestamp": 1599121290,"gateway": {"name": "AcuLink 810 Gateway","model": "AcuLink810-868","serial": "S8P20200207"},"device": {"name": "Acuvim L V3 .221","model": "Acuvim-L-V3","serial": "221","online": true,"readings": [{"param": "Freq_Hz","value": "59.977539","unit": "Hz"},{"param": "Va_V","value": "122.002907","unit": "V"},{"param": "DI4","value": "5.000000","unit": ""}]}}'
/*
* This is an example, this function needs to be implemented in order to parse the json file into a sql statement
* Note that you need to create a super table and database before writing data
* In this case:
* create database mqttdb;
* create table mqttdb.devices(ts timestamp, value bigint) tags(name binary(32), model binary(32), serial binary(16), param binary(16), unit binary(16));
*/
char* mqttConverJsonToSql(char* json, int maxSize) {
// const int32_t maxSize = 10240;
maxSize *= 5;
char* sql = malloc(maxSize);
cJSON* root = cJSON_Parse(json);
if (root == NULL) {
mqttError("failed to parse msg, invalid json format");
goto MQTT_PARSE_OVER;
}
cJSON_free(jPlayload);
int sqllen = strlen(_names) + strlen(_values) + strlen(_dbname) + strlen(_tablename) + 1024;
char* _sql = calloc(1, sqllen);
sprintf(_sql, "INSERT INTO %s.%s (%s) VALUES(%s);", _dbname, _tablename, _names, _values);
return _sql;
cJSON* timestamp = cJSON_GetObjectItem(root, "timestamp");
if (!timestamp || timestamp->type != cJSON_Number) {
mqttError("failed to parse msg, timestamp not found");
goto MQTT_PARSE_OVER;
}
cJSON* device = cJSON_GetObjectItem(root, "device");
if (!device) {
mqttError("failed to parse msg, device not found");
goto MQTT_PARSE_OVER;
}
cJSON* name = cJSON_GetObjectItem(device, "name");
if (!name || name->type != cJSON_String) {
mqttError("failed to parse msg, name not found");
goto MQTT_PARSE_OVER;
}
cJSON* model = cJSON_GetObjectItem(device, "model");
if (!model || model->type != cJSON_String) {
mqttError("failed to parse msg, model not found");
goto MQTT_PARSE_OVER;
}
cJSON* serial = cJSON_GetObjectItem(device, "serial");
if (!serial || serial->type != cJSON_String) {
mqttError("failed to parse msg, serial not found");
goto MQTT_PARSE_OVER;
}
cJSON* readings = cJSON_GetObjectItem(device, "readings");
if (!readings || readings->type != cJSON_Array) {
mqttError("failed to parse msg, readings not found");
goto MQTT_PARSE_OVER;
}
int count = cJSON_GetArraySize(readings);
if (count <= 0) {
mqttError("failed to parse msg, readings size smaller than 0");
goto MQTT_PARSE_OVER;
}
int len = snprintf(sql, maxSize, "insert into");
for (int i = 0; i < count; ++i) {
cJSON* reading = cJSON_GetArrayItem(readings, i);
if (reading == NULL) continue;
cJSON* param = cJSON_GetObjectItem(reading, "param");
if (!param || param->type != cJSON_String) {
mqttError("failed to parse msg, param not found");
goto MQTT_PARSE_OVER;
}
cJSON* value = cJSON_GetObjectItem(reading, "value");
if (!value || value->type != cJSON_String) {
mqttError("failed to parse msg, value not found");
goto MQTT_PARSE_OVER;
}
cJSON* unit = cJSON_GetObjectItem(reading, "unit");
if (!unit || unit->type != cJSON_String) {
mqttError("failed to parse msg, unit not found");
goto MQTT_PARSE_OVER;
}
len += snprintf(sql + len, maxSize - len,
" mqttdb.serial_%s_%s using mqttdb.devices tags('%s', '%s', '%s', '%s', '%s') values(%" PRId64 ", %s)",
serial->valuestring, param->valuestring, name->valuestring, model->valuestring, serial->valuestring,
param->valuestring, unit->valuestring, timestamp->valueint * 1000, value->valuestring);
}
cJSON_free(root);
return sql;
MQTT_PARSE_OVER:
cJSON_free(root);
free(sql);
return NULL;
}

267
src/plugins/mqtt/src/mqttSystem.c
View File

@ -14,244 +14,131 @@
*/
#define _DEFAULT_SOURCE
#include "cJSON.h"
#include "os.h"
#include "mqtt.h"
#include "mqttInit.h"
#include "mqttLog.h"
#include "mqttPayload.h"
#include "os.h"
#include "tmqtt.h"
#include "posix_sockets.h"
#include "string.h"
#include "taos.h"
#include "tglobal.h"
#include "tmqtt.h"
#include "tsclient.h"
#include "tsocket.h"
#include "ttimer.h"
#include "mqttSystem.h"
struct mqtt_client mqttClient = {0};
pthread_t clientDaemonThread = {0};
void* mqttConnect=NULL;
struct reconnect_state_t recntStatus = {0};
char* topicPath=NULL;
int mttIsRuning = 1;
#include "taoserror.h"
int32_t mqttInitSystem() {
int rc = 0;
#if 0
uint8_t sendbuf[2048];
uint8_t recvbuf[1024];
recntStatus.sendbuf = sendbuf;
recntStatus.sendbufsz = sizeof(sendbuf);
recntStatus.recvbuf = recvbuf;
recntStatus.recvbufsz = sizeof(recvbuf);
char* url = tsMqttBrokerAddress;
recntStatus.user_name = strstr(url, "@") != NULL ? strbetween(url, "//", ":") : NULL;
char * passStr = strstr(url, recntStatus.user_name);
if (passStr != NULL) {
recntStatus.password = strstr(url, "@") != NULL ? strbetween(passStr, ":", "@") : NULL;
}
struct SMqttReconnectState tsMqttStatus = {0};
struct mqtt_client tsMqttClient = {0};
static pthread_t tsMqttClientDaemonThread = {0};
static void* tsMqttConnect = NULL;
static bool tsMqttIsRuning = false;
if (strlen(url) == 0) {
mqttDebug("mqtt module not init, url is null");
return rc;
}
if (strstr(url, "@") != NULL) {
recntStatus.hostname = strbetween(url, "@", ":");
} else if (strstr(strstr(url, "://") + 3, ":") != NULL) {
recntStatus.hostname = strbetween(url, "//", ":");
} else {
recntStatus.hostname = strbetween(url, "//", "/");
}
char* _begin_hostname = strstr(url, recntStatus.hostname);
if (_begin_hostname != NULL && strstr(_begin_hostname, ":") != NULL) {
recntStatus.port = strbetween(_begin_hostname, ":", "/");
} else {
recntStatus.port = strbetween("'1883'", "'", "'");
}
char* portStr = recntStatus.hostname;
if (_begin_hostname != NULL) {
char* colonStr = strstr(_begin_hostname, ":");
if (colonStr != NULL) {
portStr = recntStatus.port;
}
}
char* topicStr = strstr(url, portStr);
if (topicStr != NULL) {
topicPath = strbetween(topicStr, "/", "/");
char* _topic = "+/+/+/";
int _tpsize = strlen(topicPath) + strlen(_topic) + 1;
recntStatus.topic = calloc(1, _tpsize);
sprintf(recntStatus.topic, "/%s/%s", topicPath, _topic);
recntStatus.client_id = strlen(tsMqttBrokerClientId) < 3 ? tsMqttBrokerClientId : "taos_mqtt";
mqttConnect = NULL;
} else {
topicPath = NULL;
}
#endif
return rc;
}
int32_t mqttInitSystem() { return 0; }
int32_t mqttStartSystem() {
int rc = 0;
#if 0
if (recntStatus.user_name != NULL && recntStatus.password != NULL) {
mqttInfo("connecting to mqtt://%s:%s@%s:%s/%s/", recntStatus.user_name, recntStatus.password,
recntStatus.hostname, recntStatus.port, topicPath);
} else if (recntStatus.user_name != NULL && recntStatus.password == NULL) {
mqttInfo("connecting to mqtt://%s@%s:%s/%s/", recntStatus.user_name, recntStatus.hostname, recntStatus.port,
topicPath);
tsMqttStatus.sendbufsz = MQTT_SEND_BUF_SIZE;
tsMqttStatus.recvbufsz = MQTT_RECV_BUF_SIZE;
tsMqttStatus.sendbuf = malloc(MQTT_SEND_BUF_SIZE);
tsMqttStatus.recvbuf = malloc(MQTT_RECV_BUF_SIZE);
tsMqttIsRuning = true;
mqtt_init_reconnect(&tsMqttClient, mqttReconnectClient, &tsMqttStatus, mqttPublishCallback);
if (pthread_create(&tsMqttClientDaemonThread, NULL, mqttClientRefresher, &tsMqttClient)) {
mqttError("mqtt failed to start daemon.");
mqttCleanupRes(EXIT_FAILURE, -1, NULL);
return -1;
}
mqtt_init_reconnect(&mqttClient, mqttReconnectClient, &recntStatus, mqtt_PublishCallback);
if (pthread_create(&clientDaemonThread, NULL, mqttClientRefresher, &mqttClient)) {
mqttError("Failed to start client daemon.");
mqttCleanup(EXIT_FAILURE, -1, NULL);
rc = -1;
} else {
mqttInfo("listening for '%s' messages.", recntStatus.topic);
}
#endif
return rc;
mqttInfo("mqtt listening for topic:%s messages", tsMqttTopic);
return 0;
}
void mqttStopSystem() {
#if 0
mqttClient.error = MQTT_ERROR_SOCKET_ERROR;
mttIsRuning = 0;
usleep(300000U);
mqttCleanup(EXIT_SUCCESS, mqttClient.socketfd, &clientDaemonThread);
mqttInfo("mqtt is stoped");
#endif
if (tsMqttIsRuning) {
tsMqttIsRuning = false;
tsMqttClient.error = MQTT_ERROR_SOCKET_ERROR;
taosMsleep(300);
mqttCleanupRes(EXIT_SUCCESS, tsMqttClient.socketfd, &tsMqttClientDaemonThread);
mqttInfo("mqtt is stopped");
}
}
void mqttCleanUpSystem() {
#if 0
mqttInfo("starting to cleanup mqtt");
free(recntStatus.user_name);
free(recntStatus.password);
free(recntStatus.hostname);
free(recntStatus.port);
free(recntStatus.topic);
free(topicPath);
mqttStopSystem();
mqttInfo("mqtt is cleaned up");
#endif
}
void mqtt_PublishCallback(void** unused, struct mqtt_response_publish* published) {
/* note that published->topic_name is NOT null-terminated (here we'll change it to a c-string) */
char* topic_name = (char*)malloc(published->topic_name_size + 1);
memcpy(topic_name, published->topic_name, published->topic_name_size);
topic_name[published->topic_name_size] = '\0';
mqttInfo("received publish('%s'): %s", topic_name, (const char*)published->application_message);
char _token[128] = {0};
char _dbname[128] = {0};
char _tablename[128] = {0};
if (mqttConnect == NULL) {
mqttInfo("connect database");
taos_connect_a(NULL, "_root", tsInternalPass, "", 0, mqttInitConnCb, &mqttClient, &mqttConnect);
}
if (topic_name[1]=='/' && strncmp((char*)&topic_name[1], topicPath, strlen(topicPath)) == 0) {
char* p_p_cmd_part[5] = {0};
char copystr[1024] = {0};
strncpy(copystr, topic_name, MIN(1024, published->topic_name_size));
char part_index = split(copystr, "/", p_p_cmd_part, 10);
if (part_index < 4) {
mqttError("The topic %s is't format '/path/token/dbname/table name/'. for expmle: '/taos/token/db/t'", topic_name);
} else {
strncpy(_token, p_p_cmd_part[1], 127);
strncpy(_dbname, p_p_cmd_part[2], 127);
strncpy(_tablename, p_p_cmd_part[3], 127);
mqttInfo("part count=%d,access token:%s,database name:%s, table name:%s", part_index, _token, _dbname,
_tablename);
void mqttPublishCallback(void** unused, struct mqtt_response_publish* published) {
const char* content = published->application_message;
mqttDebug("receive mqtt message, size:%d", (int)published->application_message_size);
if (mqttConnect != NULL) {
char* _sql = converJsonToSql((char*)published->application_message, _dbname, _tablename);
mqttInfo("query:%s", _sql);
taos_query_a(mqttConnect, _sql, mqttQueryInsertCallback, &mqttClient);
mqttInfo("free sql:%s", _sql);
free(_sql);
}
if (tsMqttConnect == NULL) {
tsMqttConnect = taos_connect(NULL, "_root", tsInternalPass, "", 0);
if (tsMqttConnect == NULL) {
mqttError("failed to connect to tdengine, reason:%s", tstrerror(terrno));
return;
} else {
mqttInfo("successfully connected to the tdengine");
}
}
free(topic_name);
mqttTrace("receive mqtt message, content:%s", content);
char* sql = mqttConverJsonToSql((char*)content, (int)published->application_message_size);
if (sql != NULL) {
void* res = taos_query(tsMqttConnect, sql);
int code = taos_errno(res);
if (code != 0) {
mqttError("failed to exec sql, reason:%s sql:%s", tstrerror(code), sql);
} else {
mqttTrace("successfully to exec sql:%s", sql);
}
taos_free_result(res);
} else {
mqttError("failed to parse mqtt message");
}
}
void* mqttClientRefresher(void* client) {
while (mttIsRuning) {
while (tsMqttIsRuning) {
mqtt_sync((struct mqtt_client*)client);
taosMsleep(100);
}
mqttDebug("quit refresher");
mqttDebug("mqtt quit refresher");
return NULL;
}
void mqttCleanup(int status, int sockfd, pthread_t* client_daemon) {
#if 0
void mqttCleanupRes(int status, int sockfd, pthread_t* client_daemon) {
mqttInfo("clean up mqtt module");
if (sockfd != -1) close(sockfd);
if (client_daemon != NULL) pthread_cancel(*client_daemon);
#endif
}
void mqttInitConnCb(void* param, TAOS_RES* result, int32_t code) {
if (code < 0) {
mqttError("mqtt:%d, connect to database failed, reason:%s", code, tstrerror(code));
taos_close(mqttConnect);
mqttConnect = NULL;
return;
if (sockfd != -1) {
close(sockfd);
}
mqttDebug("mqtt:%d, connect to database success, reason:%s", code, tstrerror(code));
}
void mqttQueryInsertCallback(void* param, TAOS_RES* result, int32_t code) {
if (code < 0) {
mqttError("mqtt:%d, save data failed, code:%s", code, tstrerror(code));
} else if (code == 0) {
mqttError("mqtt:%d, save data failed, affect rows:%d", code, code);
} else {
mqttInfo("mqtt:%d, save data success, code:%s", code, tstrerror(code));
if (client_daemon != NULL) {
pthread_cancel(*client_daemon);
}
}
void mqttReconnectClient(struct mqtt_client* client, void** reconnect_state_vptr) {
mqttInfo("reconnect client");
struct reconnect_state_t* reconnect_state = *((struct reconnect_state_t**)reconnect_state_vptr);
void mqttReconnectClient(struct mqtt_client* client, void** unused) {
mqttInfo("mqtt tries to connect to the mqtt server");
/* Close the clients socket if this isn't the initial reconnect call */
if (client->error != MQTT_ERROR_INITIAL_RECONNECT) {
close(client->socketfd);
}
/* Perform error handling here. */
if (client->error != MQTT_ERROR_INITIAL_RECONNECT) {
mqttError("mqttReconnectClient: called while client was in error state \"%s\"", mqtt_error_str(client->error));
mqttError("mqtt client was in error state %s", mqtt_error_str(client->error));
}
/* Open a new socket. */
int sockfd = open_nb_socket(reconnect_state->hostname, reconnect_state->port);
if (sockfd == -1) {
mqttError("failed to open socket: ");
mqttCleanup(EXIT_FAILURE, sockfd, NULL);
int sockfd = open_nb_socket(tsMqttHostName, tsMqttPort);
if (sockfd < 0) {
mqttError("mqtt client failed to open socket %s:%s", tsMqttHostName, tsMqttPort);
//mqttCleanupRes(EXIT_FAILURE, sockfd, NULL);
return;
}
/* Reinitialize the client. */
mqtt_reinit(client, sockfd, reconnect_state->sendbuf, reconnect_state->sendbufsz, reconnect_state->recvbuf,
reconnect_state->recvbufsz);
/* Ensure we have a clean session */
uint8_t connect_flags = MQTT_CONNECT_CLEAN_SESSION;
/* Send connection request to the broker. */
mqtt_connect(client, reconnect_state->client_id, NULL, NULL, 0, reconnect_state->user_name, reconnect_state->password,connect_flags, 400);
/* Subscribe to the topic. */
mqtt_subscribe(client, reconnect_state->topic, 0);
mqtt_reinit(client, sockfd, tsMqttStatus.sendbuf, tsMqttStatus.sendbufsz, tsMqttStatus.recvbuf, tsMqttStatus.recvbufsz);
mqtt_connect(client, tsMqttClientId, NULL, NULL, 0, tsMqttUser, tsMqttPass, MQTT_CONNECT_CLEAN_SESSION, 400);
mqtt_subscribe(client, tsMqttTopic, 0);
}

5
src/query/CMakeLists.txt
View File

@ -8,10 +8,9 @@ INCLUDE_DIRECTORIES(inc)
AUX_SOURCE_DIRECTORY(src SRC)
ADD_LIBRARY(query ${SRC})
SET_SOURCE_FILES_PROPERTIES(src/sql.c PROPERTIES COMPILE_FLAGS -w)
TARGET_LINK_LIBRARIES(query tsdb tutil)
IF (TD_LINUX)
TARGET_LINK_LIBRARIES(query tsdb tutil m rt)
TARGET_LINK_LIBRARIES(query m rt)
ADD_SUBDIRECTORY(tests)
ELSEIF (TD_WINDOWS)
TARGET_LINK_LIBRARIES(query tsdb tutil)
ENDIF ()

7
src/tsdb/CMakeLists.txt
View File

@ -3,13 +3,10 @@ PROJECT(TDengine)
INCLUDE_DIRECTORIES(inc)
AUX_SOURCE_DIRECTORY(src SRC)
ADD_LIBRARY(tsdb ${SRC})
TARGET_LINK_LIBRARIES(tsdb common tutil)
IF (TD_LINUX)
ADD_LIBRARY(tsdb ${SRC})
TARGET_LINK_LIBRARIES(tsdb common tutil)
# Someone has no gtest directory, so comment it
# ADD_SUBDIRECTORY(tests)
ELSEIF (TD_WINDOWS)
ADD_LIBRARY(tsdb ${SRC})
TARGET_LINK_LIBRARIES(tsdb common tutil)
ENDIF ()

28
src/tsdb/inc/tsdbMain.h
View File

@ -45,6 +45,8 @@ extern int tsdbDebugFlag;
#define TSDB_FILE_DELIMITER 0xF00AFA0F
#define TSDB_FILE_INIT_MAGIC 0xFFFFFFFF
#define TAOS_IN_RANGE(key, keyMin, keyLast) (((key) >= (keyMin)) && ((key) <= (keyMax)))
// NOTE: Any file format change must increase this version number by 1
// Also, implement the convert function
#define TSDB_FILE_VERSION ((uint32_t)0)
@ -318,6 +320,16 @@ typedef struct {
void* compBuffer; // Buffer for temperary compress/decompress purpose
} SRWHelper;
// ------------------ tsdbScan.c
typedef struct {
SFileGroup fGroup;
int numOfIdx;
SCompIdx* pCompIdx;
SCompInfo* pCompInfo;
void* pBuf;
FILE* tLogStream;
} STsdbScanHandle;
// Operations
// ------------------ tsdbMeta.c
#define TSDB_INIT_NTABLES 1024
@ -475,6 +487,7 @@ int tsdbUpdateFileHeader(SFile* pFile);
int tsdbEncodeSFileInfo(void** buf, const STsdbFileInfo* pInfo);
void* tsdbDecodeSFileInfo(void* buf, STsdbFileInfo* pInfo);
void tsdbRemoveFileGroup(STsdbRepo* pRepo, SFileGroup* pFGroup);
int tsdbLoadFileHeader(SFile* pFile, uint32_t* version);
void tsdbGetFileInfoImpl(char* fname, uint32_t* magic, int64_t* size);
void tsdbGetFidKeyRange(int daysPerFile, int8_t precision, int fileId, TSKEY *minKey, TSKEY *maxKey);
@ -513,7 +526,10 @@ int tsdbCommitTableData(SRWHelper* pHelper, SCommitIter* pCommitIter, SDataCols
int tsdbMoveLastBlockIfNeccessary(SRWHelper* pHelper);
int tsdbWriteCompInfo(SRWHelper* pHelper);
int tsdbWriteCompIdx(SRWHelper* pHelper);
int tsdbLoadCompIdxImpl(SFile* pFile, uint32_t offset, uint32_t len, void* buffer);
int tsdbDecodeSCompIdxImpl(void* buffer, uint32_t len, SCompIdx** ppCompIdx, int* numOfIdx);
int tsdbLoadCompIdx(SRWHelper* pHelper, void* target);
int tsdbLoadCompInfoImpl(SFile* pFile, SCompIdx* pIdx, SCompInfo** ppCompInfo);
int tsdbLoadCompInfo(SRWHelper* pHelper, void* target);
int tsdbLoadCompData(SRWHelper* phelper, SCompBlock* pcompblock, void* target);
void tsdbGetDataStatis(SRWHelper* pHelper, SDataStatis* pStatis, int numOfCols);
@ -537,7 +553,7 @@ static FORCE_INLINE int compTSKEY(const void* key1, const void* key2) {
#define TSDB_SUBMIT_MSG_HEAD_SIZE sizeof(SSubmitMsg)
char* tsdbGetMetaFileName(char* rootDir);
void tsdbGetDataFileName(STsdbRepo* pRepo, int fid, int type, char* fname);
void tsdbGetDataFileName(char* rootDir, int vid, int fid, int type, char* fname);
int tsdbLockRepo(STsdbRepo* pRepo);
int tsdbUnlockRepo(STsdbRepo* pRepo);
char* tsdbGetDataDirName(char* rootDir);
@ -546,6 +562,16 @@ STsdbMeta* tsdbGetMeta(TSDB_REPO_T* pRepo);
STsdbFileH* tsdbGetFile(TSDB_REPO_T* pRepo);
int tsdbCheckCommit(STsdbRepo* pRepo);
// ------------------ tsdbScan.c
int tsdbScanFGroup(STsdbScanHandle* pScanHandle, char* rootDir, int fid);
STsdbScanHandle* tsdbNewScanHandle();
void tsdbSetScanLogStream(STsdbScanHandle* pScanHandle, FILE* fLogStream);
int tsdbSetAndOpenScanFile(STsdbScanHandle* pScanHandle, char* rootDir, int fid);
int tsdbScanSCompIdx(STsdbScanHandle* pScanHandle);
int tsdbScanSCompBlock(STsdbScanHandle* pScanHandle, int idx);
int tsdbCloseScanFile(STsdbScanHandle* pScanHandle);
void tsdbFreeScanHandle(STsdbScanHandle* pScanHandle);
#ifdef __cplusplus
}
#endif

18
src/plugins/mqtt/inc/mqttSystem.h → src/tsdb/src/tsdbCompact.c
View File

@ -11,20 +11,4 @@
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#ifndef TDENGINE_MQTT_SYSTEM_H
#define TDENGINE_MQTT_SYSTEM_H
#ifdef __cplusplus
extern "C" {
#endif
#include <stdint.h>
int32_t mqttInitSystem();
int32_t mqttStartSystem();
void mqttStopSystem();
void mqttCleanUpSystem();
#ifdef __cplusplus
}
#endif
#endif
*/

74
src/tsdb/src/tsdbFile.c
View File

@ -302,7 +302,7 @@ int tsdbCreateFile(SFile *pFile, STsdbRepo *pRepo, int fid, int type) {
memset((void *)pFile, 0, sizeof(SFile));
pFile->fd = -1;
tsdbGetDataFileName(pRepo, fid, type, pFile->fname);
tsdbGetDataFileName(pRepo->rootDir, REPO_ID(pRepo), fid, type, pFile->fname);
if (access(pFile->fname, F_OK) == 0) {
tsdbError("vgId:%d file %s already exists", REPO_ID(pRepo), pFile->fname);
@ -424,33 +424,57 @@ void tsdbRemoveFileGroup(STsdbRepo *pRepo, SFileGroup *pFGroup) {
}
}
void tsdbGetFileInfoImpl(char *fname, uint32_t *magic, int64_t *size) {
char buf[TSDB_FILE_HEAD_SIZE] = "\0";
uint32_t version = 0;
STsdbFileInfo info = {0};
int tsdbLoadFileHeader(SFile *pFile, uint32_t *version) {
char buf[TSDB_FILE_HEAD_SIZE] = "\0";
int fd = open(fname, O_RDONLY);
if (fd < 0) goto _err;
if (lseek(pFile->fd, 0, SEEK_SET) < 0) {
tsdbError("failed to lseek file %s to start since %s", pFile->fname, strerror(errno));
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
if (taosTRead(fd, buf, TSDB_FILE_HEAD_SIZE) < TSDB_FILE_HEAD_SIZE) goto _err;
if (taosTRead(pFile->fd, buf, TSDB_FILE_HEAD_SIZE) < TSDB_FILE_HEAD_SIZE) {
tsdbError("failed to read file %s header part with %d bytes, reason:%s", pFile->fname, TSDB_FILE_HEAD_SIZE,
strerror(errno));
terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
return -1;
}
if (!taosCheckChecksumWhole((uint8_t *)buf, TSDB_FILE_HEAD_SIZE)) goto _err;
if (!taosCheckChecksumWhole((uint8_t *)buf, TSDB_FILE_HEAD_SIZE)) {
tsdbError("file %s header part is corrupted with failed checksum", pFile->fname);
terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
return -1;
}
void *pBuf = (void *)buf;
pBuf = taosDecodeFixedU32(pBuf, &version);
pBuf = tsdbDecodeSFileInfo(pBuf, &info);
pBuf = taosDecodeFixedU32(pBuf, version);
pBuf = tsdbDecodeSFileInfo(pBuf, &(pFile->info));
off_t offset = lseek(fd, 0, SEEK_END);
return 0;
}
void tsdbGetFileInfoImpl(char *fname, uint32_t *magic, int64_t *size) {
uint32_t version = 0;
SFile file;
SFile * pFile = &file;
strncpy(pFile->fname, fname, TSDB_FILENAME_LEN);
pFile->fd = -1;
if (tsdbOpenFile(pFile, O_RDONLY) < 0) goto _err;
if (tsdbLoadFileHeader(pFile, &version) < 0) goto _err;
off_t offset = lseek(pFile->fd, 0, SEEK_END);
if (offset < 0) goto _err;
close(fd);
tsdbCloseFile(pFile);
*magic = info.magic;
*magic = pFile->info.magic;
*size = offset;
return;
_err:
if (fd >= 0) close(fd);
tsdbCloseFile(pFile);
*magic = TSDB_FILE_INIT_MAGIC;
*size = 0;
}
@ -458,34 +482,23 @@ _err:
// ---------------- LOCAL FUNCTIONS ----------------
static int tsdbInitFile(SFile *pFile, STsdbRepo *pRepo, int fid, int type) {
uint32_t version;
char buf[512] = "\0";
tsdbGetDataFileName(pRepo, fid, type, pFile->fname);
tsdbGetDataFileName(pRepo->rootDir, REPO_ID(pRepo), fid, type, pFile->fname);
pFile->fd = -1;
if (tsdbOpenFile(pFile, O_RDONLY) < 0) goto _err;
if (taosTRead(pFile->fd, buf, TSDB_FILE_HEAD_SIZE) < TSDB_FILE_HEAD_SIZE) {
tsdbError("vgId:%d failed to read %d bytes from file %s since %s", REPO_ID(pRepo), TSDB_FILE_HEAD_SIZE,
pFile->fname, strerror(errno));
terrno = TAOS_SYSTEM_ERROR(errno);
if (tsdbLoadFileHeader(pFile, &version) < 0) {
tsdbError("vgId:%d failed to load file %s header part since %s", REPO_ID(pRepo), pFile->fname, tstrerror(terrno));
goto _err;
}
if (!taosCheckChecksumWhole((uint8_t *)buf, TSDB_FILE_HEAD_SIZE)) {
tsdbError("vgId:%d file %s head part is corrupted", REPO_ID(pRepo), pFile->fname);
terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
goto _err;
}
void *pBuf = buf;
pBuf = taosDecodeFixedU32(pBuf, &version);
pBuf = tsdbDecodeSFileInfo(pBuf, &(pFile->info));
if (pFile->info.size == TSDB_FILE_HEAD_SIZE) {
pFile->info.size = lseek(pFile->fd, 0, SEEK_END);
}
if (version != TSDB_FILE_VERSION) {
// TODO: deal with error
tsdbError("vgId:%d file %s version %u is not the same as program version %u which may cause problem",
REPO_ID(pRepo), pFile->fname, version, TSDB_FILE_VERSION);
}
@ -529,6 +542,7 @@ static void tsdbInitFileGroup(SFileGroup *pFGroup, STsdbRepo *pRepo) {
memset(&pFGroup->files[type].info, 0, sizeof(STsdbFileInfo));
pFGroup->files[type].info.magic = TSDB_FILE_INIT_MAGIC;
pFGroup->state = 1;
pRepo->state = TSDB_STATE_BAD_FILE;
terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
}
}

11
src/tsdb/src/tsdbMain.c
View File

@ -142,7 +142,6 @@ TSDB_REPO_T *tsdbOpenRepo(char *rootDir, STsdbAppH *pAppH) {
}
tsdbStartStream(pRepo);
// pRepo->state = TSDB_REPO_STATE_ACTIVE;
tsdbDebug("vgId:%d open tsdb repository succeed!", REPO_ID(pRepo));
@ -341,6 +340,10 @@ void tsdbReportStat(void *repo, int64_t *totalPoints, int64_t *totalStorage, int
*compStorage = pRepo->stat.compStorage;
}
int tsdbGetState(TSDB_REPO_T *repo) {
return ((STsdbRepo *)repo)->state;
}
// ----------------- INTERNAL FUNCTIONS -----------------
char *tsdbGetMetaFileName(char *rootDir) {
int tlen = (int)(strlen(rootDir) + strlen(TSDB_META_FILE_NAME) + 2);
@ -354,8 +357,8 @@ char *tsdbGetMetaFileName(char *rootDir) {
return fname;
}
void tsdbGetDataFileName(STsdbRepo *pRepo, int fid, int type, char *fname) {
snprintf(fname, TSDB_FILENAME_LEN, "%s/%s/v%df%d%s", pRepo->rootDir, TSDB_DATA_DIR_NAME, REPO_ID(pRepo), fid, tsdbFileSuffix[type]);
void tsdbGetDataFileName(char *rootDir, int vid, int fid, int type, char *fname) {
snprintf(fname, TSDB_FILENAME_LEN, "%s/%s/v%df%d%s", rootDir, TSDB_DATA_DIR_NAME, vid, fid, tsdbFileSuffix[type]);
}
int tsdbLockRepo(STsdbRepo *pRepo) {
@ -661,6 +664,8 @@ static STsdbRepo *tsdbNewRepo(char *rootDir, STsdbAppH *pAppH, STsdbCfg *pCfg) {
goto _err;
}
pRepo->state = TSDB_STATE_OK;
int code = pthread_mutex_init(&pRepo->mutex, NULL);
if (code != 0) {
terrno = TAOS_SYSTEM_ERROR(code);

168
src/tsdb/src/tsdbRWHelper.c
View File

@ -102,7 +102,8 @@ void tsdbResetHelper(SRWHelper *pHelper) {
int tsdbSetAndOpenHelperFile(SRWHelper *pHelper, SFileGroup *pGroup) {
ASSERT(pHelper != NULL && pGroup != NULL);
SFile *pFile = NULL;
SFile * pFile = NULL;
STsdbRepo *pRepo = pHelper->pRepo;
// Clear the helper object
tsdbResetHelper(pHelper);
@ -112,8 +113,10 @@ int tsdbSetAndOpenHelperFile(SRWHelper *pHelper, SFileGroup *pGroup) {
// Set the files
pHelper->files.fGroup = *pGroup;
if (helperType(pHelper) == TSDB_WRITE_HELPER) {
tsdbGetDataFileName(pHelper->pRepo, pGroup->fileId, TSDB_FILE_TYPE_NHEAD, helperNewHeadF(pHelper)->fname);
tsdbGetDataFileName(pHelper->pRepo, pGroup->fileId, TSDB_FILE_TYPE_NLAST, helperNewLastF(pHelper)->fname);
tsdbGetDataFileName(pRepo->rootDir, REPO_ID(pRepo), pGroup->fileId, TSDB_FILE_TYPE_NHEAD,
helperNewHeadF(pHelper)->fname);
tsdbGetDataFileName(pRepo->rootDir, REPO_ID(pRepo), pGroup->fileId, TSDB_FILE_TYPE_NLAST,
helperNewLastF(pHelper)->fname);
}
// Open the files
@ -443,10 +446,64 @@ int tsdbWriteCompIdx(SRWHelper *pHelper) {
return 0;
}
int tsdbLoadCompIdxImpl(SFile *pFile, uint32_t offset, uint32_t len, void *buffer) {
const char *prefixMsg = "failed to load SCompIdx part";
if (lseek(pFile->fd, offset, SEEK_SET) < 0) {
tsdbError("%s: seek to file %s offset %u failed since %s", prefixMsg, pFile->fname, offset, strerror(errno));
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
if (taosTRead(pFile->fd, buffer, len) < len) {
tsdbError("%s: read file %s offset %u len %u failed since %s", prefixMsg, pFile->fname, offset, len,
strerror(errno));
terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
return -1;
}
if (!taosCheckChecksumWhole((uint8_t *)buffer, len)) {
tsdbError("%s: file %s corrupted, offset %u len %u", prefixMsg, pFile->fname, offset, len);
terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
return -1;
}
return 0;
}
int tsdbDecodeSCompIdxImpl(void *buffer, uint32_t len, SCompIdx **ppCompIdx, int *numOfIdx) {
int nIdx = 0;
void *pPtr = buffer;
while (POINTER_DISTANCE(pPtr, buffer) < (int)(len - sizeof(TSCKSUM))) {
size_t tlen = taosTSizeof(*ppCompIdx);
if (tlen < sizeof(SCompIdx) * (nIdx + 1)) {
*ppCompIdx = (SCompIdx *)taosTRealloc(*ppCompIdx, (tlen == 0) ? 1024 : tlen * 2);
if (*ppCompIdx == NULL) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
return -1;
}
}
pPtr = tsdbDecodeSCompIdx(pPtr, &((*ppCompIdx)[nIdx]));
if (pPtr == NULL) {
tsdbError("failed to decode SCompIdx part, idx:%d", nIdx);
terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
return -1;
}
nIdx++;
ASSERT(nIdx == 1 || (*ppCompIdx)[nIdx - 1].tid > (*ppCompIdx)[nIdx - 2].tid);
ASSERT(POINTER_DISTANCE(pPtr, buffer) <= (int)(len - sizeof(TSCKSUM)));
}
*numOfIdx = nIdx;
return 0;
}
int tsdbLoadCompIdx(SRWHelper *pHelper, void *target) {
ASSERT(pHelper->state == TSDB_HELPER_FILE_SET_AND_OPEN);
SFile *pFile = helperHeadF(pHelper);
int fd = pFile->fd;
if (!helperHasState(pHelper, TSDB_HELPER_IDX_LOAD)) {
// If not load from file, just load it in object
@ -456,54 +513,18 @@ int tsdbLoadCompIdx(SRWHelper *pHelper, void *target) {
return -1;
}
if (lseek(fd, pFile->info.offset, SEEK_SET) < 0) {
tsdbError("vgId:%d failed to lseek file %s since %s", REPO_ID(pHelper->pRepo), pFile->fname, strerror(errno));
terrno = TAOS_SYSTEM_ERROR(errno);
// Load SCompIdx binary from file
if (tsdbLoadCompIdxImpl(pFile, pFile->info.offset, pFile->info.len, (void *)(pHelper->pBuffer)) < 0) {
return -1;
}
if (taosTRead(fd, (void *)(pHelper->pBuffer), pFile->info.len) < (int)pFile->info.len) {
tsdbError("vgId:%d failed to read %d bytes from file %s since %s", REPO_ID(pHelper->pRepo), pFile->info.len,
pFile->fname, strerror(errno));
terrno = TAOS_SYSTEM_ERROR(errno);
// Decode the SCompIdx part
if (tsdbDecodeSCompIdxImpl(pHelper->pBuffer, pFile->info.len, &(pHelper->idxH.pIdxArray),
&(pHelper->idxH.numOfIdx)) < 0) {
tsdbError("vgId:%d failed to decode SCompIdx part from file %s since %s", REPO_ID(pHelper->pRepo), pFile->fname,
tstrerror(errno));
return -1;
}
if (!taosCheckChecksumWhole((uint8_t *)(pHelper->pBuffer), pFile->info.len)) {
tsdbError("vgId:%d file %s SCompIdx part is corrupted. len %u", REPO_ID(pHelper->pRepo), pFile->fname,
pFile->info.len);
terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
return -1;
}
// Decode it
pHelper->idxH.numOfIdx = 0;
void *ptr = pHelper->pBuffer;
while (POINTER_DISTANCE(ptr, pHelper->pBuffer) < (int)(pFile->info.len - sizeof(TSCKSUM))) {
size_t tlen = taosTSizeof(pHelper->idxH.pIdxArray);
pHelper->idxH.numOfIdx++;
if (tlen < pHelper->idxH.numOfIdx * sizeof(SCompIdx)) {
pHelper->idxH.pIdxArray = (SCompIdx *)taosTRealloc(pHelper->idxH.pIdxArray, (tlen == 0) ? 1024 : tlen * 2);
if (pHelper->idxH.pIdxArray == NULL) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
return -1;
}
}
ptr = tsdbDecodeSCompIdx(ptr, &(pHelper->idxH.pIdxArray[pHelper->idxH.numOfIdx - 1]));
if (ptr == NULL) {
tsdbError("vgId:%d file %s SCompIdx part is corrupted. len %u", REPO_ID(pHelper->pRepo), pFile->fname,
pFile->info.len);
terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
return -1;
}
ASSERT(pHelper->idxH.numOfIdx == 1 || pHelper->idxH.pIdxArray[pHelper->idxH.numOfIdx - 1].tid >
pHelper->idxH.pIdxArray[pHelper->idxH.numOfIdx - 2].tid);
ASSERT(POINTER_DISTANCE(ptr, pHelper->pBuffer) <= (int)(pFile->info.len - sizeof(TSCKSUM)));
}
}
}
helperSetState(pHelper, TSDB_HELPER_IDX_LOAD);
@ -515,36 +536,49 @@ int tsdbLoadCompIdx(SRWHelper *pHelper, void *target) {
return 0;
}
int tsdbLoadCompInfoImpl(SFile *pFile, SCompIdx *pIdx, SCompInfo **ppCompInfo) {
const char *prefixMsg = "failed to load SCompInfo/SCompBlock part";
if (lseek(pFile->fd, pIdx->offset, SEEK_SET) < 0) {
tsdbError("%s: seek to file %s offset %u failed since %s", prefixMsg, pFile->fname, pIdx->offset, strerror(errno));
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
*ppCompInfo = taosTRealloc((void *)(*ppCompInfo), pIdx->len);
if (*ppCompInfo == NULL) {
terrno = TSDB_CODE_TDB_OUT_OF_MEMORY;
return -1;
}
if (taosTRead(pFile->fd, (void *)(*ppCompInfo), pIdx->len) < (int)pIdx->len) {
tsdbError("%s: read file %s offset %u len %u failed since %s", prefixMsg, pFile->fname, pIdx->offset, pIdx->len,
strerror(errno));
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
if (!taosCheckChecksumWhole((uint8_t *)(*ppCompInfo), pIdx->len)) {
tsdbError("%s: file %s corrupted, offset %u len %u", prefixMsg, pFile->fname, pIdx->offset, pIdx->len);
terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
return -1;
}
return 0;
}
int tsdbLoadCompInfo(SRWHelper *pHelper, void *target) {
ASSERT(helperHasState(pHelper, TSDB_HELPER_TABLE_SET));
SCompIdx *pIdx = &(pHelper->curCompIdx);
int fd = helperHeadF(pHelper)->fd;
SFile *pFile = helperHeadF(pHelper);
if (!helperHasState(pHelper, TSDB_HELPER_INFO_LOAD)) {
if (pIdx->offset > 0) {
ASSERT(pIdx->uid == pHelper->tableInfo.uid);
if (lseek(fd, pIdx->offset, SEEK_SET) < 0) {
tsdbError("vgId:%d failed to lseek file %s since %s", REPO_ID(pHelper->pRepo), helperHeadF(pHelper)->fname,
strerror(errno));
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
pHelper->pCompInfo = taosTRealloc((void *)pHelper->pCompInfo, pIdx->len);
if (taosTRead(fd, (void *)(pHelper->pCompInfo), pIdx->len) < (int)pIdx->len) {
tsdbError("vgId:%d failed to read %d bytes from file %s since %s", REPO_ID(pHelper->pRepo), pIdx->len,
helperHeadF(pHelper)->fname, strerror(errno));
terrno = TAOS_SYSTEM_ERROR(errno);
return -1;
}
if (!taosCheckChecksumWhole((uint8_t *)pHelper->pCompInfo, pIdx->len)) {
tsdbError("vgId:%d file %s SCompInfo part is corrupted, tid %d uid %" PRIu64, REPO_ID(pHelper->pRepo),
helperHeadF(pHelper)->fname, pHelper->tableInfo.tid, pHelper->tableInfo.uid);
terrno = TSDB_CODE_TDB_FILE_CORRUPTED;
return -1;
}
if (tsdbLoadCompInfoImpl(pFile, pIdx, &(pHelper->pCompInfo)) < 0) return -1;
ASSERT(pIdx->uid == pHelper->pCompInfo->uid && pIdx->tid == pHelper->pCompInfo->tid);
}

14
src/tsdb/src/tsdbRecover.c Normal file
View File

@ -0,0 +1,14 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/

36
src/tsdb/src/tsdbScan.c Normal file
View File

@ -0,0 +1,36 @@
/*
* Copyright (c) 2019 TAOS Data, Inc. <jhtao@taosdata.com>
*
* This program is free software: you can use, redistribute, and/or modify
* it under the terms of the GNU Affero General Public License, version 3
* or later ("AGPL"), as published by the Free Software Foundation.
*
* This program is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
#include "tsdbMain.h"
#ifndef _TSDB_PLUGINS
int tsdbScanFGroup(STsdbScanHandle* pScanHandle, char* rootDir, int fid) { return 0; }
STsdbScanHandle* tsdbNewScanHandle() { return NULL; }
void tsdbSetScanLogStream(STsdbScanHandle* pScanHandle, FILE* fLogStream) {}
int tsdbSetAndOpenScanFile(STsdbScanHandle* pScanHandle, char* rootDir, int fid) { return 0; }
int tsdbScanSCompIdx(STsdbScanHandle* pScanHandle) { return 0; }
int tsdbScanSCompBlock(STsdbScanHandle* pScanHandle, int idx) { return 0; }
int tsdbCloseScanFile(STsdbScanHandle* pScanHandle) { return 0; }
void tsdbFreeScanHandle(STsdbScanHandle* pScanHandle) {}
#endif

7
src/util/CMakeLists.txt
View File

@ -3,9 +3,10 @@ PROJECT(TDengine)
AUX_SOURCE_DIRECTORY(src SRC)
ADD_LIBRARY(tutil ${SRC})
TARGET_LINK_LIBRARIES(tutil pthread osdetail lz4)
IF (TD_LINUX)
TARGET_LINK_LIBRARIES(tutil pthread osdetail m rt lz4)
TARGET_LINK_LIBRARIES(tutil m rt)
ADD_SUBDIRECTORY(tests)
FIND_PATH(ICONV_INCLUDE_EXIST iconv.h /usr/include/ /usr/local/include/)
@ -24,7 +25,7 @@ IF (TD_LINUX)
ENDIF ()
ELSEIF (TD_WINDOWS)
TARGET_LINK_LIBRARIES(tutil iconv regex pthread osdetail winmm IPHLPAPI ws2_32 lz4 wepoll)
TARGET_LINK_LIBRARIES(tutil iconv regex winmm IPHLPAPI ws2_32 wepoll)
ELSEIF(TD_DARWIN)
TARGET_LINK_LIBRARIES(tutil iconv pthread osdetail lz4)
TARGET_LINK_LIBRARIES(tutil iconv)
ENDIF()

10
src/util/src/tnettest.c
View File

@ -73,7 +73,7 @@ static void *bindUdpPort(void *sarg) {
continue;
}
if (iDataNum > 0) {
printf("recv Client: %s pkg from UDP port: %d, pkg len: %d\n", inet_ntoa(clientAddr.sin_addr), port, iDataNum);
printf("recv Client: %s pkg from UDP port: %d, pkg len: %d\n", taosInetNtoa(clientAddr.sin_addr), port, iDataNum);
//printf("Read msg from udp:%s ... %s\n", buffer, buffer+iDataNum-16);
sendto(serverSocket, buffer, iDataNum, 0, (struct sockaddr *)&clientAddr, (int)sin_size);
@ -138,7 +138,7 @@ static void *bindTcpPort(void *sarg) {
if (errno == EINTR) {
continue;
} else {
printf("recv Client: %s pkg from TCP port: %d fail:%s.\n", inet_ntoa(clientAddr.sin_addr), port, strerror(errno));
printf("recv Client: %s pkg from TCP port: %d fail:%s.\n", taosInetNtoa(clientAddr.sin_addr), port, strerror(errno));
taosCloseSocket(serverSocket);
return NULL;
}
@ -149,7 +149,7 @@ static void *bindTcpPort(void *sarg) {
}
}
printf("recv Client: %s pkg from TCP port: %d, pkg len: %d\n", inet_ntoa(clientAddr.sin_addr), port, iDataNum);
printf("recv Client: %s pkg from TCP port: %d, pkg len: %d\n", taosInetNtoa(clientAddr.sin_addr), port, iDataNum);
if (iDataNum > 0) {
send(client, buffer, iDataNum, 0);
}
@ -197,7 +197,7 @@ static int checkTcpPort(info_s *info) {
struct in_addr ipStr;
memcpy(&ipStr, &info->hostIp, 4);
sprintf(sendbuf, "client send tcp pkg to %s:%d, content: 1122334455", inet_ntoa(ipStr), info->port);
sprintf(sendbuf, "client send tcp pkg to %s:%d, content: 1122334455", taosInetNtoa(ipStr), info->port);
sprintf(sendbuf + info->pktLen - 16, "1122334455667788");
send(clientSocket, sendbuf, info->pktLen, 0);
@ -267,7 +267,7 @@ static int checkUdpPort(info_s *info) {
struct in_addr ipStr;
memcpy(&ipStr, &info->hostIp, 4);
sprintf(sendbuf, "client send udp pkg to %s:%d, content: 1122334455", inet_ntoa(ipStr), info->port);
sprintf(sendbuf, "client send udp pkg to %s:%d, content: 1122334455", taosInetNtoa(ipStr), info->port);
sprintf(sendbuf + info->pktLen - 16, "1122334455667788");
socklen_t sin_size = sizeof(*(struct sockaddr *)&serverAddr);

2
src/util/src/tnote.c
View File

@ -253,7 +253,7 @@ void taosNotePrint(taosNoteInfo * pNote, const char * const format, ...)
ptm->tm_min, ptm->tm_sec, (int)timeSecs.tv_usec, taosGetPthreadId());
#else
len = sprintf(buffer, "%02d/%02d %02d:%02d:%02d.%06d %lx ", ptm->tm_mon + 1, ptm->tm_mday, ptm->tm_hour, ptm->tm_min,
ptm->tm_sec, (int)timeSecs.tv_usec, pthread_self());
ptm->tm_sec, (int)timeSecs.tv_usec, (unsigned long int)pthread_self());
#endif
va_start(argpointer, format);
len += vsnprintf(buffer + len, MAX_NOTE_LINE_SIZE - len, format, argpointer);

13
src/vnode/src/vnodeMain.c
View File

@ -47,6 +47,15 @@ static void vnodeNotifyRole(void *ahandle, int8_t role);
static void vnodeCtrlFlow(void *handle, int32_t mseconds);
static int vnodeNotifyFileSynced(void *ahandle, uint64_t fversion);
#ifndef _SYNC
tsync_h syncStart(const SSyncInfo *info) { return NULL; }
int32_t syncForwardToPeer(tsync_h shandle, void *pHead, void *mhandle, int qtype) { return 0; }
void syncStop(tsync_h shandle) {}
int32_t syncReconfig(tsync_h shandle, const SSyncCfg * cfg) { return 0; }
int syncGetNodesRole(tsync_h shandle, SNodesRole * cfg) { return 0; }
void syncConfirmForward(tsync_h shandle, uint64_t version, int32_t code) {}
#endif
int32_t vnodeInitResources() {
vnodeInitWriteFp();
vnodeInitReadFp();
@ -289,12 +298,16 @@ int32_t vnodeOpen(int32_t vnode, char *rootDir) {
syncInfo.notifyFileSynced = vnodeNotifyFileSynced;
pVnode->sync = syncStart(&syncInfo);
#ifndef _SYNC
pVnode->role = TAOS_SYNC_ROLE_MASTER;
#else
if (pVnode->sync == NULL) {
vError("vgId:%d, failed to open sync module, replica:%d reason:%s", pVnode->vgId, pVnode->syncCfg.replica,
tstrerror(terrno));
vnodeCleanUp(pVnode);
return terrno;
}
#endif
pVnode->qMgmt = qOpenQueryMgmt(pVnode->vgId);
if (pVnode->qMgmt == NULL) {

1
src/wal/CMakeLists.txt
View File

@ -7,6 +7,5 @@ AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR}/src SRC)
IF (TD_LINUX)
ADD_LIBRARY(twal ${SRC})
TARGET_LINK_LIBRARIES(twal tutil common)
ADD_SUBDIRECTORY(test)
ENDIF ()

2
tests/examples/JDBC/JDBCDemo/pom.xml
View File

@ -63,7 +63,7 @@
<dependency>
<groupId>com.taosdata.jdbc</groupId>
<artifactId>taos-jdbcdriver</artifactId>
<version>2.0.2</version>
<version>2.0.4</version>
</dependency>
</dependencies>
</project>

2
tests/examples/nodejs/node-example-raw.js
View File

@ -26,7 +26,7 @@ var c1 = conn.cursor();
// c1.execute(query) will execute the query
// Let's create a database named db
try {
c1.execute('create database db;');
c1.execute('create database if not exists db;');
}
catch(err) {
conn.close();

44
tests/examples/nodejs/node-example.js
View File

@ -22,8 +22,9 @@ var c1 = conn.cursor();
// c1.query(query) will return a TaosQuery object, of which then we can execute. The execute function then returns a promise
// Let's create a database named db
try {
var query = c1.query('create database db;');
query.execute();
c1.execute('create database if not exists db;');
//var query = c1.query('create database if not exists db;');
//query.execute();
}
catch(err) {
conn.close();
@ -71,6 +72,28 @@ catch (err) {
throw err;
}
Date.prototype.Format = function(fmt){
var o = {
'M+': this.getMonth() + 1,
'd+': this.getDate(),
'H+': this.getHours(),
'm+': this.getMinutes(),
's+': this.getSeconds(),
'S+': this.getMilliseconds()
};
if (/(y+)/.test(fmt)) {
fmt = fmt.replace(RegExp.$1, (this.getFullYear() + '').substr(4 - RegExp.$1.length));
}
for (var k in o) {
if (new RegExp('(' + k + ')').test(fmt)) {
fmt = fmt.replace(RegExp.$1, (RegExp.$1.length == 1) ? (o[k]) : (('00' + o[k]).substr(String(o[k]).length)));
}
}
return fmt;
}
// Let's try to insert some random generated data to test with
// We will use the bind function of the TaosQuery object to easily bind values to question marks in the query
// For Timestamps, a normal Datetime object or TaosTimestamp or milliseconds can be passed in through the bind function
@ -79,17 +102,21 @@ let interval = 1000;
try {
for (let i = 0; i < 1000; i++) {
stime.setMilliseconds(stime.getMilliseconds() + interval);
//console.log(stime.Format('yyyy-MM-dd HH:mm:ss.SSS'));
let insertData = [stime,
parseInt(Math.random()*100),
parseInt(Math.random()*300),
parseFloat(Math.random()*10 + 30),
"\"random note!\""];
"Note"];
//c1.execute('insert into db.weather values(' + insertData.join(',') + ' );');
var query = c1.query('insert into db.weather values(?, ?, ?, ?, ?);').bind(insertData);
query.execute();
//var query = c1.query('insert into db.weather values(?, ?, ?, ?, ?);').bind(insertData);
//query.execute();
c1.execute('insert into db.weather values(\"'+stime.Format('yyyy-MM-dd HH:mm:ss.SSS')+'\",'+parseInt(Math.random() * 100)+','+parseInt(Math.random() * 300)+','+parseFloat(Math.random()*10 + 30)+',"Note");');
}
}
catch (err) {
}catch (err) {
conn.close();
throw err;
}
@ -98,7 +125,8 @@ catch (err) {
var retrievedData;
try {
c1.query('select * from db.weather limit 5 offset 100;', true).then(function(result){
result.pretty();
//result.pretty();
console.log('=========>'+JSON.stringify(result));
// Neat!
});

23
tests/perftest-scripts/coverage_test.sh
View File

@ -74,21 +74,36 @@ function runTest {
runGeneralCaseOneByOne jenkins/basic.txt
totalSuccess=`grep 'success' $TDENGINE_COVERAGE_REPORT | wc -l`
sed -i "1i\SIM cases test result" $TDENGINE_COVERAGE_REPORT
totalSuccess=`grep 'success' $TDENGINE_COVERAGE_REPORT | wc -l`
if [ "$totalSuccess" -gt "0" ]; then
echo -e "\n${GREEN} ### Total $totalSuccess coverage test case(s) succeed! ### ${NC}" | tee -a $TDENGINE_COVERAGE_REPORT
sed -i -e "2i\ ### Total $totalSuccess SIM test case(s) succeed! ###" $TDENGINE_COVERAGE_REPORT
fi
totalFailed=`grep 'failed\|fault' $TDENGINE_COVERAGE_REPORT | wc -l`
if [ "$totalFailed" -ne "0" ]; then
echo -e "${RED} ### Total $totalFailed coverage test case(s) failed! ### ${NC}\n" | tee -a $TDENGINE_COVERAGE_REPORT
# exit $totalPyFailed
sed -i '3i\### Total $totalFailed SIM test case(s) failed! ###' $TDENGINE_COVERAGE_REPORT
else
sed -i '3i\\n' $TDENGINE_COVERAGE_REPORT
fi
cd $TDENGINE_DIR/tests
rm -rf ../sim
./test-all.sh full python | tee -a $TDENGINE_COVERAGE_REPORT
sed -i "4i\Python cases test result" $TDENGINE_COVERAGE_REPORT
totalPySuccess=`grep 'python case(s) succeed!' $TDENGINE_COVERAGE_REPORT | awk '{print $4}'`
if [ "$totalPySuccess" -gt "0" ]; then
sed -i -e "5i\ ### Total $totalPySuccess Python test case(s) succeed! ###" $TDENGINE_COVERAGE_REPORT
fi
totalPyFailed=`grep 'python case(s) failed!' $TDENGINE_COVERAGE_REPORT | awk '{print $4}'`
if [ -z $totalPyFailed ]; then
sed -i '6i\\n' $TDENGINE_COVERAGE_REPORT
else
sed -i '6i\### Total $totalPyFailed Python test case(s) failed! ###' $TDENGINE_COVERAGE_REPORT
fi
# Test Connector
stopTaosd

20
tests/perftest-scripts/perftest-taosdemo.sh
View File

@ -1,6 +1,7 @@
#!/bin/bash
WORK_DIR=/mnt/root
TDENGINE_DIR=/root/TDengine
walLevel=`grep "^walLevel" /etc/taos/taos.cfg | awk '{print $2}'`
if [[ "$walLevel" -eq "2" ]]; then
@ -71,9 +72,17 @@ function runCreateTableThenInsert {
restartTaosd
/usr/bin/time -f "Total: %e" -o totaltime.out bash -c "yes | taosdemo 2>&1 | tee -a taosdemo-$walPostfix-$today.log"
demoTableAndInsert=`grep "Total:" totaltime.out|awk '{print $2}'`
demoRPS=`grep "records\/second" taosdemo-$walPostfix-$today.log | tail -n1 | awk '{print $13}'`
}
demoTableAndInsert=`grep "Total:" totaltime.out|awk '{print $2}'`
demoRPS=`grep "records\/second" taosdemo-$walPostfix-$today.log | tail -n1 | awk '{print $13}'`
}
function queryPerformance {
echoInfo "Restart Taosd"
restartTaosd
cd $TDENGINE_DIR/tests/pytest
python3 query/queryPerformance.py
}
function generateTaosdemoPlot {
echo "${today} $walPostfix, demoCreateTableOnly: ${demoCreateTableOnly}, demoDeleteTableOnly: ${demoDeleteTableOnly}, demoTableAndInsert: ${demoTableAndInsert}" | tee -a taosdemo-$today.log
@ -101,13 +110,16 @@ today=`date +"%Y%m%d"`
cd $WORK_DIR
echoInfo "Test Create Table Only "
runCreateTableOnly
echoInfo "Test Create Table then Insert data"
echoInfo "Test Delete Table Only"
runDeleteTableOnly
echoInfo "Test Create Table then Insert data"
runCreateTableThenInsert
echoInfo "Query Performance for 10 Billion Records"
queryPerformance
echoInfo "Generate plot for taosdemo"
generateTaosdemoPlot
tar czf $WORK_DIR/taos-log-taosdemo-$today.tar.gz $logDir/*
echoInfo "End of TaosDemo Test" | tee -a $WORK_DIR/cron.log

82
tests/pytest/query/queryPerformance.py Normal file
View File

@ -0,0 +1,82 @@
###################################################################
# Copyright (c) 2016 by TAOS Technologies, Inc.
# All rights reserved.
#
# This file is proprietary and confidential to TAOS Technologies.
# No part of this file may be reproduced, stored, transmitted,
# disclosed or used in any form or by any means other than as
# expressly provided by the written permission from Jianhui Tao
#
###################################################################
# -*- coding: utf-8 -*-
import sys
import os
import taos
import time
class taosdemoQueryPerformace:
def initConnection(self):
self.host = "127.0.0.1"
self.user = "root"
self.password = "taosdata"
self.config = "/etc/taos"
self.conn = taos.connect(
self.host,
self.user,
self.password,
self.config)
def query(self):
cursor = self.conn.cursor()
cursor.execute("use test")
totalTime = 0
for i in range(100):
startTime = time.time()
cursor.execute("select count(*) from test.meters")
totalTime += time.time() - startTime
print("query time for: select count(*) from test.meters %f seconds" % (totalTime / 100))
totalTime = 0
for i in range(100):
startTime = time.time()
cursor.execute("select avg(f1), max(f2), min(f3) from test.meters")
totalTime += time.time() - startTime
print("query time for: select avg(f1), max(f2), min(f3) from test.meters %f seconds" % (totalTime / 100))
totalTime = 0
for i in range(100):
startTime = time.time()
cursor.execute("select count(*) from test.meters where loc='beijing'")
totalTime += time.time() - startTime
print("query time for: select count(*) from test.meters where loc='beijing' %f seconds" % (totalTime / 100))
totalTime = 0
for i in range(100):
startTime = time.time()
cursor.execute("select avg(f1), max(f2), min(f3) from test.meters where areaid=10")
totalTime += time.time() - startTime
print("query time for: select avg(f1), max(f2), min(f3) from test.meters where areaid=10 %f seconds" % (totalTime / 100))
totalTime = 0
for i in range(100):
startTime = time.time()
cursor.execute("select avg(f1), max(f2), min(f3) from test.t10 interval(10s)")
totalTime += time.time() - startTime
print("query time for: select avg(f1), max(f2), min(f3) from test.t10 interval(10s) %f seconds" % (totalTime / 100))
totalTime = 0
for i in range(100):
startTime = time.time()
cursor.execute("select last_row(*) from meters")
totalTime += time.time() - startTime
print("query time for: select last_row(*) from meters %f seconds" % (totalTime / 100))
if __name__ == '__main__':
perftest = taosdemoQueryPerformace()
perftest.initConnection()
perftest.query()

19
tests/script/general/connection/mqtt.sim Normal file
View File

@ -0,0 +1,19 @@
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/cfg.sh -n dnode1 -c walLevel -v 2
system sh/cfg.sh -n dnode1 -c numOfMnodes -v 1
system sh/cfg.sh -n dnode1 -c mnodeEqualVnodeNum -v 4
system sh/cfg.sh -n dnode1 -c maxTablesPerVnode -v 100000
system sh/cfg.sh -n dnode1 -c http -v 1
system sh/cfg.sh -n dnode1 -c mqtt -v 1
system sh/exec.sh -n dnode1 -s start
sleep 3000
sql connect
sql create database mqttdb;
sql create table mqttdb.devices(ts timestamp, value double) tags(name binary(32), model binary(32), serial binary(16), param binary(16), unit binary(16));
sleep 1000
system sh/exec.sh -n dnode1 -s stop -x SIGINT

2
tests/script/general/db/delete_reusevnode.sim
View File

@ -50,7 +50,7 @@ $tbPrefix = t
$i = 0
while $i < 10
$db = db . $i
sql create database $db maxTables 4
sql create database $db
sql use $db
sql create table st (ts timestamp, i int) tags(j int);

2
tests/script/general/db/delete_reusevnode2.sim
View File

@ -8,7 +8,7 @@ sql connect
print ======== step1
sql create database db maxTables 4;
sql create database db;
sql use db
$tbPrefix = t

2
tests/script/general/db/dropdnodes.sim
View File

@ -15,7 +15,7 @@ sql connect
sql create dnode $hostname2
sleep 2000
sql create database db maxTables 4
sql create database db
sql use db
print ========== step1

2
tests/script/general/parser/where.sim
View File

@ -21,7 +21,7 @@ $mt = $mtPrefix . $i
sql drop database if exits $db -x step1
step1:
sql create database if not exists $db maxTables 4
sql create database if not exists $db
sql use $db
sql create table $mt (ts timestamp, c1 int, c2 float, c3 bigint, c4 smallint, c5 tinyint, c6 double, c7 bool, c8 binary(10), c9 nchar(9)) TAGS(t1 int)

2
tests/script/general/stable/dnode3.sim
View File

@ -54,7 +54,7 @@ $i = 0
$db = $dbPrefix . $i
$mt = $mtPrefix . $i
sql create database $db maxTables 4
sql create database $db
sql use $db
sql create table $mt (ts timestamp, tbcol int) TAGS(tgcol int)

1
tests/script/jenkins/basic.txt
View File

@ -293,6 +293,7 @@ cd ../../../debug; make
./test.sh -f unique/stable/replica3_dnode6.sim
./test.sh -f unique/stable/replica3_vnode3.sim
./test.sh -f unique/mnode/mgmt20.sim
./test.sh -f unique/mnode/mgmt21.sim
./test.sh -f unique/mnode/mgmt22.sim
./test.sh -f unique/mnode/mgmt23.sim

2
tests/script/sh/deploy.sh
View File

@ -119,7 +119,7 @@ echo "tsdbDebugFlag 135" >> $TAOS_CFG
echo "cDebugFlag 135" >> $TAOS_CFG
echo "jnidebugFlag 135" >> $TAOS_CFG
echo "odbcdebugFlag 135" >> $TAOS_CFG
echo "httpDebugFlag 143" >> $TAOS_CFG
echo "httpDebugFlag 135" >> $TAOS_CFG
echo "monitorDebugFlag 135" >> $TAOS_CFG
echo "mqttDebugFlag 135" >> $TAOS_CFG
echo "qdebugFlag 135" >> $TAOS_CFG

2
tests/script/unique/arbitrator/dn3_mn1_vnode_change.sim
View File

@ -61,7 +61,7 @@ $totalTableNum = 10
$sleepTimer = 3000
$db = db
sql create database $db replica 2 maxTables $totalTableNum
sql create database $db replica 2
sql use $db
# create table , insert data

2
tests/script/unique/arbitrator/dn3_mn1_vnode_corruptFile_online.sim
View File

@ -61,7 +61,7 @@ $totalTableNum = 4
$sleepTimer = 3000
$db = db
sql create database $db cache 1 replica 2 maxTables $totalTableNum
sql create database $db cache 1 replica 2
sql use $db
# create table , insert data

4
tests/script/unique/arbitrator/sync_replica3_createTable.sim
View File

@ -61,8 +61,8 @@ $totalTableNum = 20
$sleepTimer = 3000
$db = db
print create database $db replica 3 maxTables $totalTableNum
sql create database $db replica 3 maxTables $totalTableNum
print create database $db replica 3
sql create database $db replica 3
sql use $db
# create table , insert data

4
tests/script/unique/arbitrator/sync_replica3_dnodeChang_DropAddAlterTableDropDb.sim
View File

@ -62,8 +62,8 @@ $sleepTimer = 3000
$maxTables = $totalTableNum * 2
$db = db
print create database $db replica 3 maxTables $maxTables
sql create database $db replica 3 maxTables $maxTables
print create database $db replica 3
sql create database $db replica 3
sql use $db
# create table , insert data

2
tests/script/unique/big/maxvnodes.sim
View File

@ -18,7 +18,7 @@ print ========== prepare data
system sh/exec.sh -n dnode1 -s start
sleep 3000
sql connect
sql create database db blocks 3 cache 1 maxTables $maxTables
sql create database db blocks 3 cache 1
sql use db
print ========== step1

2
tests/script/unique/big/restartSpeed.sim
View File

@ -16,7 +16,7 @@ print ========== prepare data
system sh/exec.sh -n dnode1 -s start
sleep 3000
sql connect
sql create database db blocks 3 cache 1 maxTables $maxTables
sql create database db blocks 3 cache 1
sql use db
print ========== step1

16
tests/script/unique/cluster/balance1.sim
View File

@ -40,7 +40,7 @@ print ========= start dnode1
system sh/exec.sh -n dnode1 -s start
sql connect
sql create database c_b1_d1 maxTables 4
sql create database c_b1_d1
sql use c_b1_d1
sql create table c_b1_t1 (t timestamp, i int)
@ -50,7 +50,7 @@ sql insert into c_b1_t1 values(1520000022013, 13)
sql insert into c_b1_t1 values(1520000023012, 12)
sql insert into c_b1_t1 values(1520000024011, 11)
sql create database c_b1_d2 maxTables 4
sql create database c_b1_d2
sql use c_b1_d2
sql create table c_b1_t2 (t timestamp, i int)
sql insert into c_b1_t2 values(1520000020025, 25)
@ -107,7 +107,7 @@ print dnode2 ==> $dnode2Role
print ============================== step3
print ========= add db3
sql create database c_b1_d3 maxTables 4
sql create database c_b1_d3
sql use c_b1_d3
sql create table c_b1_t3 (t timestamp, i int)
sql insert into c_b1_t3 values(1520000020035, 35)
@ -280,7 +280,7 @@ if $dnode4Role != slave then
endi
print ============================== step10
sql create database c_b1_d4 maxTables 4
sql create database c_b1_d4
sql use c_b1_d4
sql create table c_b1_t4 (t timestamp, i int)
sql insert into c_b1_t4 values(1520000020045, 45)
@ -318,7 +318,7 @@ sql use c_b1_d2
sql insert into c_b1_t2 values(1520000025026, 26)
print ============================== step12
sql create database c_b1_d5 maxTables 4
sql create database c_b1_d5
sql use c_b1_d5
sql_error create table c_b1_t5 (t timestamp, i int) -x error3
@ -343,7 +343,7 @@ sql insert into c_b1_t5 values(1520000022053, 53)
sql insert into c_b1_t5 values(1520000023052, 52)
sql insert into c_b1_t5 values(1520000024051, 51)
sql create database c_b1_d6 maxTables 4
sql create database c_b1_d6
sql use c_b1_d6
sql create table c_b1_t6 (t timestamp, i int)
sql insert into c_b1_t6 values(1520000020065, 65)
@ -375,7 +375,7 @@ sql create dnode $hostname6
system sh/exec.sh -n dnode6 -s start
sleep 15000
sql create database c_b1_d7 maxTables 4
sql create database c_b1_d7
sql use c_b1_d7
sql create table c_b1_t7 (t timestamp, i int)
sql insert into c_b1_t7 values(1520000020075, 75)
@ -384,7 +384,7 @@ sql insert into c_b1_t7 values(1520000022073, 73)
sql insert into c_b1_t7 values(1520000023072, 72)
sql insert into c_b1_t7 values(1520000024071, 71)
sql create database c_b1_d8 maxTables 4
sql create database c_b1_d8
sql use c_b1_d8
sql create table c_b1_t8 (t timestamp, i int)
sql insert into c_b1_t8 values(1520000020085, 85)

6
tests/script/unique/cluster/balance2.sim
View File

@ -47,7 +47,7 @@ system sh/exec.sh -n dnode2 -s start
system sh/exec.sh -n dnode3 -s start
sleep 4001
sql create database c_b2_d1 replica 2 maxTables 4
sql create database c_b2_d1 replica 2
sql use c_b2_d1
sql create table c_b2_t1 (t timestamp, i int)
sql insert into c_b2_t1 values(1520000020015, 15)
@ -56,7 +56,7 @@ sql insert into c_b2_t1 values(1520000022013, 13)
sql insert into c_b2_t1 values(1520000023012, 12)
sql insert into c_b2_t1 values(1520000024011, 11)
sql create database c_b2_d2 replica 2 maxTables 4
sql create database c_b2_d2 replica 2
sql use c_b2_d2
sql create table c_b2_t2 (t timestamp, i int)
sql insert into c_b2_t2 values(1520000020025, 25)
@ -65,7 +65,7 @@ sql insert into c_b2_t2 values(1520000022023, 23)
sql insert into c_b2_t2 values(1520000023022, 22)
sql insert into c_b2_t2 values(1520000024021, 21)
sql create database c_b2_d3 replica 2 maxTables 4
sql create database c_b2_d3 replica 2
sql use c_b2_d3
sql create table c_b2_t3 (t timestamp, i int)
sql insert into c_b2_t3 values(1520000020035, 35)

2
tests/script/unique/cluster/cluster_main.sim
View File

@ -57,7 +57,7 @@ sleep 3000
print ============== step2: create db1 with replica 3
$db = db1
print create database $db replica 3
#sql create database $db replica 3 maxTables $totalTableNum
#sql create database $db replica 3
sql create database $db replica 3
sql use $db

2
tests/script/unique/cluster/cluster_main0.sim
View File

@ -57,7 +57,7 @@ sleep 3000
print ============== step2: create db1 with replica 3
$db = db1
print create database $db replica 3
#sql create database $db replica 3 maxTables $totalTableNum
#sql create database $db replica 3
sql create database $db replica 3
sql use $db

2
tests/script/unique/cluster/cluster_main1.sim
View File

@ -58,7 +58,7 @@ print ============== step2: create db1 with replica 3
$replica = 3
$db = db1
print create database $db replica $replica
#sql create database $db replica 3 maxTables $totalTableNum
#sql create database $db replica 3
sql create database $db replica $replica
sql use $db

2
tests/script/unique/cluster/cluster_main2.sim
View File

@ -58,7 +58,7 @@ print ============== step2: create db1 with replica 3
$replica = 3
$db = db1
print create database $db replica $replica
#sql create database $db replica 3 maxTables $totalTableNum
#sql create database $db replica 3
sql create database $db replica $replica
sql use $db

2
tests/script/unique/dnode/alternativeRole.sim
View File

@ -56,7 +56,7 @@ if $data2_3 != slave then
endi
print ========== step2
sql create database d1 maxTables 4
sql create database d1
sql create table d1.t1 (ts timestamp, i int)
sql create table d1.t2 (ts timestamp, i int)
sql create table d1.t3 (ts timestamp, i int)

8
tests/script/unique/dnode/balance1.sim
View File

@ -30,7 +30,7 @@ system sh/exec.sh -n dnode1 -s start
sql connect
sleep 3000
sql create database d1 maxTables 4
sql create database d1
sql create table d1.t1 (t timestamp, i int)
sql insert into d1.t1 values(now+1s, 15)
sql insert into d1.t1 values(now+2s, 14)
@ -68,7 +68,7 @@ if $data2_2 != 1 then
endi
print ========== step3
sql create database d2 maxTables 4
sql create database d2
sql create table d2.t2 (t timestamp, i int)
sql insert into d2.t2 values(now+1s, 25)
sql insert into d2.t2 values(now+2s, 24)
@ -139,7 +139,7 @@ if $data2_3 != 2 then
endi
print ========== step6
sql create database d3 maxTables 4
sql create database d3
sql create table d3.t3 (t timestamp, i int)
sql insert into d3.t3 values(now+1s, 35)
sql insert into d3.t3 values(now+2s, 34)
@ -193,7 +193,7 @@ if $data2_4 != 1 then
endi
print ========== step8
sql create database d4 maxTables 4
sql create database d4
sql create table d4.t4 (t timestamp, i int)
sql insert into d4.t4 values(now+1s, 45)
sql insert into d4.t4 values(now+2s, 44)

6
tests/script/unique/dnode/balance2.sim
View File

@ -28,7 +28,7 @@ system sh/exec.sh -n dnode2 -s start
system sh/exec.sh -n dnode3 -s start
sleep 3000
sql create database d1 replica 2 maxTables 4
sql create database d1 replica 2
sql create table d1.t1 (t timestamp, i int)
sql insert into d1.t1 values(now+1s, 15)
sql insert into d1.t1 values(now+2s, 14)
@ -36,7 +36,7 @@ sql insert into d1.t1 values(now+3s, 13)
sql insert into d1.t1 values(now+4s, 12)
sql insert into d1.t1 values(now+5s, 11)
sql create database d2 replica 2 maxTables 4
sql create database d2 replica 2
sql create table d2.t2 (t timestamp, i int)
sql insert into d2.t2 values(now+1s, 25)
sql insert into d2.t2 values(now+2s, 24)
@ -117,7 +117,7 @@ if $data2_4 != 2 then
endi
print ========== step4
sql create database d3 replica 2 maxTables 4
sql create database d3 replica 2
sql create table d3.t3 (t timestamp, i int)
sql insert into d3.t3 values(now+1s, 35)
sql insert into d3.t3 values(now+2s, 34)

6
tests/script/unique/dnode/balance3.sim
View File

@ -33,7 +33,7 @@ system sh/exec.sh -n dnode3 -s start
system sh/exec.sh -n dnode4 -s start
sleep 3000
sql create database d1 replica 3 maxTables 4
sql create database d1 replica 3
sql create table d1.t1 (t timestamp, i int)
sql insert into d1.t1 values(now+1s, 15)
sql insert into d1.t1 values(now+2s, 14)
@ -41,7 +41,7 @@ sql insert into d1.t1 values(now+3s, 13)
sql insert into d1.t1 values(now+4s, 12)
sql insert into d1.t1 values(now+5s, 11)
sql create database d2 replica 3 maxTables 4
sql create database d2 replica 3
sql create table d2.t2 (t timestamp, i int)
sql insert into d2.t2 values(now+1s, 25)
sql insert into d2.t2 values(now+2s, 24)
@ -136,7 +136,7 @@ if $data2_5 != 2 then
endi
print ========== step4
sql create database d3 replica 3 maxTables 4
sql create database d3 replica 3
sql create table d3.t3 (t timestamp, i int)
sql insert into d3.t3 values(now+1s, 35)
sql insert into d3.t3 values(now+2s, 34)

6
tests/script/unique/dnode/balancex.sim
View File

@ -20,7 +20,7 @@ system sh/exec.sh -n dnode1 -s start
sql connect
sleep 3000
sql create database d1 maxTables 4
sql create database d1
sql create table d1.t1 (t timestamp, i int)
sql insert into d1.t1 values(now+1s, 15)
sql insert into d1.t1 values(now+2s, 14)
@ -28,7 +28,7 @@ sql insert into d1.t1 values(now+3s, 13)
sql insert into d1.t1 values(now+4s, 12)
sql insert into d1.t1 values(now+5s, 11)
sql create database d2 maxTables 4
sql create database d2
sql create table d2.t2 (t timestamp, i int)
sql insert into d2.t2 values(now+1s, 25)
sql insert into d2.t2 values(now+2s, 24)
@ -65,7 +65,7 @@ if $data2_2 != 2 then
endi
print ========== step3
sql create database d3 replica 2 maxTables 4
sql create database d3 replica 2
sql create table d3.t3 (t timestamp, i int)
sql insert into d3.t3 values(now+1s, 35)
sql insert into d3.t3 values(now+2s, 34)

2
tests/script/unique/dnode/offline2.sim
View File

@ -27,7 +27,7 @@ sql create dnode $hostname2
system sh/exec.sh -n dnode2 -s start
sleep 3000
sql create database d1 replica 2 maxTables 4
sql create database d1 replica 2
sql create table d1.t1(ts timestamp, i int)
sql insert into d1.t1 values(1588262400001, 1)

6
tests/script/unique/dnode/remove1.sim
View File

@ -20,7 +20,7 @@ system sh/exec.sh -n dnode1 -s start
sleep 3000
sql connect
sql create database d1 maxTables 4
sql create database d1
sql create table d1.t1 (t timestamp, i int)
sql insert into d1.t1 values(now+1s, 15)
sql insert into d1.t1 values(now+2s, 14)
@ -28,7 +28,7 @@ sql insert into d1.t1 values(now+3s, 13)
sql insert into d1.t1 values(now+4s, 12)
sql insert into d1.t1 values(now+5s, 11)
sql create database d2 maxTables 4
sql create database d2
sql create table d2.t2 (t timestamp, i int)
sql insert into d2.t2 values(now+1s, 25)
sql insert into d2.t2 values(now+2s, 24)
@ -47,7 +47,7 @@ sql create dnode $hostname2
system sh/exec.sh -n dnode2 -s start
sleep 9000
sql create database d3 replica 2 maxTables 4
sql create database d3 replica 2
sql create table d3.t3 (t timestamp, i int)
sql insert into d3.t3 values(now+1s, 35)
sql insert into d3.t3 values(now+2s, 34)

6
tests/script/unique/dnode/remove2.sim
View File

@ -20,7 +20,7 @@ system sh/exec.sh -n dnode1 -s start
sleep 3000
sql connect
sql create database d1 maxTables 4
sql create database d1
sql create table d1.t1 (t timestamp, i int)
sql insert into d1.t1 values(1588262400001, 15)
sql insert into d1.t1 values(1588262400002, 14)
@ -28,7 +28,7 @@ sql insert into d1.t1 values(1588262400003, 13)
sql insert into d1.t1 values(1588262400004, 12)
sql insert into d1.t1 values(1588262400005, 11)
sql create database d2 maxTables 4
sql create database d2
sql create table d2.t2 (t timestamp, i int)
sql insert into d2.t2 values(1588262400001, 25)
sql insert into d2.t2 values(1588262400002, 24)
@ -47,7 +47,7 @@ sql create dnode $hostname2
system sh/exec.sh -n dnode2 -s start
sleep 9000
sql create database d3 replica 2 maxTables 4
sql create database d3 replica 2
sql create table d3.t3 (t timestamp, i int)
sql insert into d3.t3 values(1588262400001, 35)
sql insert into d3.t3 values(1588262400002, 34)

6
tests/script/unique/dnode/vnode_clean.sim
View File

@ -19,7 +19,7 @@ print ========== step1
system sh/exec.sh -n dnode1 -s start
sql connect
sql create database d1 maxTables 4
sql create database d1
sql create table d1.t1 (t timestamp, i int)
sql insert into d1.t1 values(now+1s, 15)
sql insert into d1.t1 values(now+2s, 14)
@ -55,7 +55,7 @@ if $data2_2 != 1 then
endi
print ========== step3
sql create database d2 maxTables 4
sql create database d2
sql create table d2.t2 (t timestamp, i int)
sql insert into d2.t2 values(now+1s, 25)
@ -123,7 +123,7 @@ if $data2_3 != 2 then
endi
print ========== step6
sql create database d3 maxTables 4
sql create database d3
sql create table d3.t3 (t timestamp, i int)
sql insert into d3.t3 values(now+1s, 35)
sql insert into d3.t3 values(now+2s, 34)

71
tests/script/unique/mnode/mgmt20.sim Normal file
View File

@ -0,0 +1,71 @@
system sh/stop_dnodes.sh
system sh/deploy.sh -n dnode1 -i 1
system sh/deploy.sh -n dnode2 -i 2
system sh/cfg.sh -n dnode1 -c numOfMnodes -v 2
system sh/cfg.sh -n dnode2 -c numOfMnodes -v 2
system sh/cfg.sh -n dnode1 -c monitor -v 1
system sh/cfg.sh -n dnode2 -c monitor -v 1
print ============== step1
system sh/exec.sh -n dnode1 -s start
system sh/exec.sh -n dnode2 -s start
sleep 3000
sql connect
print ============== step2
sql create dnode $hostname2
$x = 0
show2:
$x = $x + 1
sleep 2000
if $x == 10 then
return -1
endi
sql show mnodes
print dnode1 ==> $data2_1
print dnode2 ==> $data2_2
if $data2_1 != master then
goto show2
endi
if $data2_2 != slave then
goto show2
endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT
system sh/exec.sh -n dnode2 -s stop -x SIGINT
print ============== step3
system sh/exec.sh -n dnode2 -s start
sleep 10000
system sh/exec.sh -n dnode1 -s start
sql connect
print =============== step4
sql select * from log.dn1
$d1_first = $rows
sql select * from log.dn2
$d2_first = $rows
sleep 3000
sql select * from log.dn1
$d1_second = $rows
sql select * from log.dn2
$d2_second = $rows
print dnode1 $d1_first $d1_second
print dnode2 $d2_first $d2_second
if $d1_first >= $d1_second then
return -1
endi
if $d2_first >= $d2_second then
return -1
endi
system sh/exec.sh -n dnode1 -s stop -x SIGINT
system sh/exec.sh -n dnode2 -s stop -x SIGINT

2
tests/script/unique/stable/balance_replica1.sim
View File

@ -29,7 +29,7 @@ $db = $dbPrefix
$mt = $mtPrefix
$st = $stPrefix . $i
sql create database $db maxTables 4
sql create database $db
sql use $db
sql create table $mt (ts timestamp, tbcol int, tbcol2 float) TAGS(tgcol int)

2
tests/script/unique/stable/dnode2.sim
View File

@ -38,7 +38,7 @@ $i = 0
$db = $dbPrefix . $i
$mt = $mtPrefix . $i
sql create database $db maxTables 4
sql create database $db
sql use $db
sql create table $mt (ts timestamp, tbcol int) TAGS(tgcol int)

2
tests/script/unique/stable/dnode2_stop.sim
View File

@ -43,7 +43,7 @@ $i = 0
$db = $dbPrefix . $i
$mt = $mtPrefix . $i
sql create database $db maxTables 4
sql create database $db
sql use $db
sql create table $mt (ts timestamp, tbcol int) TAGS(tgcol int)

2
tests/script/unique/stable/dnode3.sim
View File

@ -47,7 +47,7 @@ $i = 0
$db = $dbPrefix . $i
$mt = $mtPrefix . $i
sql create database $db maxTables 4
sql create database $db
sql use $db
sql create table $mt (ts timestamp, tbcol int) TAGS(tgcol int)

Some files were not shown because too many files have changed in this diff Show More