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

[TD-6443]<feature>: Support OpenTSDB HTTP JSON data import format

This commit is contained in:
Ganlin Zhao 2021-09-06 12:32:49 +08:00
parent 691bab64ba
commit 4bd405f64b
1 changed files with 51 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

54
src/client/src/tscParseOpenTSDB.c
View File

@ -464,7 +464,7 @@ int32_t parseTimestampFromJSON(cJSON *root, TAOS_SML_KV **pTS, int *num_kvs, SSm
cJSON *timestamp = cJSON_GetObjectItem(root, "timestamp");
if (timestamp == NULL || timestamp->type != cJSON_Number) {
tscError("OTD:0x%"PRIx64" failed to parse timestamp from JSON Payload", info->id);
return TSDB_CODE_TSC_INVALID_JSON;
return TSDB_CODE_TSC_INVALID_JSON;
}
//allocate fields for timestamp and value
@ -473,7 +473,7 @@ int32_t parseTimestampFromJSON(cJSON *root, TAOS_SML_KV **pTS, int *num_kvs, SSm
tsVal = convertTimePrecision(timestamp->valueint, TSDB_TIME_PRECISION_MICRO, TSDB_TIME_PRECISION_NANO);
(*pTS)->key = tcalloc(sizeof(key), 1);
if ((*pTS)->key == NULL){
if ((*pTS)->key == NULL) {
tfree(*pTS);
return TSDB_CODE_TSC_OUT_OF_MEMORY;
}
@ -483,7 +483,7 @@ int32_t parseTimestampFromJSON(cJSON *root, TAOS_SML_KV **pTS, int *num_kvs, SSm
(*pTS)->length = (int16_t)tDataTypes[(*pTS)->type].bytes;
(*pTS)->value = tcalloc((*pTS)->length, 1);
if ((*pTS)->value == NULL){
if ((*pTS)->value == NULL) {
tfree((*pTS)->key);
tfree(*pTS);
return TSDB_CODE_TSC_OUT_OF_MEMORY;
@ -495,6 +495,54 @@ int32_t parseTimestampFromJSON(cJSON *root, TAOS_SML_KV **pTS, int *num_kvs, SSm
}
int32_t parseMetricValueFromJSON(cJSON *root, TAOS_SML_KV **pKVs, int *num_kvs, SSmlLinesInfo* info) {
//skip timestamp
TAOS_SML_KV *pVal = *pKVs + 1;
char key[] = OTS_METRIC_VALUE_COLUMN_NAME;
cJSON *metricVal = cJSON_GetObjectItem(root, "value");
if (metricVal == NULL) {
tscError("OTD:0x%"PRIx64" failed to parse metric value from JSON Payload", info->id);
return TSDB_CODE_TSC_INVALID_JSON;
}
switch (metricVal->type) {
case cJSON_True:
case cJSON_False: {
pVal->type = TSDB_DATA_TYPE_BOOL;
pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
pVal->value = tcalloc(pVal->length, 1);
*(bool *)(pVal->value) = metricVal->type ? true : false;
break;
}
case cJSON_Number: {
//convert default JSON Number type to float
pVal->type = TSDB_DATA_TYPE_FLOAT;
pVal->length = (int16_t)tDataTypes[pVal->type].bytes;
pVal->value = tcalloc(pVal->length, 1);
*(float *)(pVal->value) = (float)(metricVal->valuedouble);
break;
}
case cJSON_String: {
//convert default JSON String type to nchar
pVal->type = TSDB_DATA_TYPE_NCHAR;
pVal->length = wcslen((wchar_t *)metricVal->string) * TSDB_NCHAR_SIZE;
pVal->value = tcalloc(pVal->length + 1, 1);
memcpy(pVal->value, metricVal->string, pVal->length);
break;
}
default:
return TSDB_CODE_TSC_INVALID_JSON;
}
pVal->key = tcalloc(sizeof(key), 1);
memcpy(pVal->key, key, sizeof(key));
*num_kvs += 1;
return TSDB_CODE_SUCCESS;
}
int32_t tscParseJSONPayload(const char* payload, TAOS_SML_DATA_POINT* smlData, SSmlLinesInfo* info) {
int32_t ret = TSDB_CODE_SUCCESS;