diff --git a/.gitignore b/.gitignore index 1883a5f83..54a687cd7 100644 --- a/.gitignore +++ b/.gitignore @@ -20,4 +20,5 @@ neuron-sdk log.html output.xml report.html -*.cache \ No newline at end of file +*.cache +.claude \ No newline at end of file diff --git a/docs/api/cn/http.md b/docs/api/cn/http.md index 3ede6496f..608b04c9a 100644 --- a/docs/api/cn/http.md +++ b/docs/api/cn/http.md @@ -34,7 +34,19 @@ Neuron 将为 IIoT 平台提供一系列 API 服务，用于查询基本信息 AUTO 必须用于可读点位；仅写 AUTO 点位返回 `2204`。识别后，单点及批量写入均按具体类型检查输入类型和数值范围，再交给驱动。例如 INT16 可接受整数 `123`，但拒绝 `40000` 和字符串 `"123"`。已排队写入若因点位或组 cache 重建失效，返回 `3011`，不会按另一类型重新解释。 -配置查询、导出和持久化存储始终保留 `type: 0`，不会回写识别结果，也不新增 `resolved_type` 字段。读取和上报值使用已识别的具体类型，并应用该类型适用的 precision、缩放及地址选项。非数值类型不进行数值缩放，缩放后的输出类型也不会改变设备侧已识别类型。 +配置查询、导出和持久化存储始终保留 `type: 0`，不会回写识别结果。读取和上报值使用已识别的具体类型，并应用该类型适用的 precision、缩放及地址选项。非数值类型不进行数值缩放，缩放后的输出类型也不会改变设备侧已识别类型。 + +### 点位运行时类型 + +`GET /api/v2/tags`、`POST /api/v2/read`、`POST /api/v2/read/paginate` 的每个点位对象顶层均返回整数 `resolved_type`： + +- 显式类型点位：等于配置 `type`，即使尚无样本、只写或读取报错。 +- AUTO 点位：未识别、cache 不存在或新配置尚待应用时为 `0`；识别后为设备侧实际类型。STRING/BYTES 的内部 PTR 样本返回 STRING/BYTES，不返回 PTR。 +- 已识别后的通信错误、类型不匹配、值过期或停用节点不清除识别结果；cache/配置重建或进程重启后重新识别。 +- `resolved_type` 不是缩放后的输出类型。例如 AUTO 识别为 INT16，`decimal: 0.1` 时可返回 `"resolved_type": 3, "value": 12.3`。前端写入校验还需结合 decimal、bias，并以后端结果为准。 +- 已识别类型不表示当前值有效、节点正在运行或具有写权限，应分别检查 error、节点状态和 attribute。 + +原有 `type` 仍表示配置类型；普通 `/read` 不因此新增配置 `type` 字段。`resolved_type` 不放入 `attributes`，普通读取存在同名 metas 时由该权威字段覆盖。配置导出 `/global/config`、`/global/drivers`、SQLite 和 MQTT 不新增此字段。 ### 波特率 @@ -1050,6 +1062,8 @@ HTTP 200 响应： *GET* /api/v2/tags +每个点位返回 `resolved_type`，含义见“点位运行时类型”；`type` 保持配置值。 + ### 请求参数 **node** 必需 @@ -1078,6 +1092,7 @@ HTTP 200 响应： "name": "tag1", //tag type "type": 4, + "resolved_type": 4, //tag address "address": "1!400001", //tag attribute @@ -1094,6 +1109,7 @@ HTTP 200 响应： { "name": "tag2", "type": 14, + "resolved_type": 14, "address": "1!00001", "attribute": 3, "description": "", @@ -1103,6 +1119,7 @@ HTTP 200 响应： { "name": "tag3", "type": 11, + "resolved_type": 11, "address": "1!400009", "attribute": 3, "description": "", @@ -1264,6 +1281,7 @@ HTTP 200 响应： "tags": [ { "name": "temperature", + "resolved_type": 10, "value": 25.5, "transferPrecision": 2, "timestamp": 1700000000000, @@ -1273,6 +1291,7 @@ HTTP 200 响应： }, { "name": "pressure", + "resolved_type": 3, "error": 3002, "quality": 0 } @@ -1328,6 +1347,7 @@ HTTP 200 响应： "items": [ { "name": "temperature", + "resolved_type": 10, "type": 10, "address": "1!400001", "attribute": 1, diff --git a/docs/api/english/http.md b/docs/api/english/http.md index 9894f029e..63becbc3f 100644 --- a/docs/api/english/http.md +++ b/docs/api/english/http.md @@ -34,7 +34,19 @@ The first valid, successful sample fixes the concrete runtime type for that tag' AUTO requires a readable tag; write-only AUTO is rejected with `2204`. After detection, single and batch writes are checked against the concrete type, including input type and numeric range, before the driver receives them. For example, an INT16 tag accepts integer `123`, but rejects `40000` and string `"123"`. Writes invalidated by a tag/group cache rebuild return `3011` rather than being reinterpreted with another type. -Configuration queries, exports and persistent storage continue to show `type: 0`; detection does not rewrite metadata or add a `resolved_type` field. Read/report values use the detected concrete type and its applicable precision, scaling and address options. Numeric scaling does not apply to nonnumeric types, and a scaled output value does not change the detected device type. +Configuration queries, exports and persistent storage continue to show `type: 0`; detection does not rewrite the configured type. Read/report values use the detected concrete type and its applicable precision, scaling and address options. Numeric scaling does not apply to nonnumeric types, and a scaled output value does not change the detected device type. + +### Runtime tag type + +Every tag object returned by `GET /api/v2/tags`, `POST /api/v2/read` and `POST /api/v2/read/paginate` includes a top-level integer `resolved_type`: + +- For explicitly typed tags it equals configured `type`, including tags without samples, write-only tags and read errors. +- For AUTO it is `0` while unresolved, without a cache entry, or while configuration changes await application. Once identified it is the device-side semantic type. Internally PTR-backed STRING/BYTES samples report STRING/BYTES, not PTR. +- Communication errors, type mismatches, value expiration and stopping a node do not themselves clear an identified type. Cache/configuration rebuilding or process restart requires identification again. +- It is not the scaled output type. For example, AUTO resolved as INT16 with `decimal: 0.1` can return `"resolved_type": 3, "value": 12.3`. Frontend write validation must also consider decimal and bias; backend validation remains authoritative. +- An identified type does not imply a valid current value, a running node or write permission. Check error, node state and attribute separately. + +Existing `type` remains the configured type. Ordinary `/read` does not gain a configured `type` field. `resolved_type` is separate from `attributes` and replaces any colliding ordinary-read metadata key. Configuration exports through `/global/config` and `/global/drivers`, SQLite and MQTT do not gain this field. ### Baud @@ -1031,6 +1043,8 @@ To update both group name and interval: *GET* /api/v2/tags +Each tag includes `resolved_type` as described in "Runtime tag type"; `type` remains the configured value. + ### Request Params **node** required @@ -1059,6 +1073,7 @@ To update both group name and interval: "name": "tag1", //tag type "type": 4, + "resolved_type": 4, //tag address "address": "1!400001", //tag attribute @@ -1075,6 +1090,7 @@ To update both group name and interval: { "name": "tag2", "type": 14, + "resolved_type": 14, "address": "1!00001", "attribute": 3, "description": "", @@ -1084,6 +1100,7 @@ To update both group name and interval: { "name": "tag3", "type": 11, + "resolved_type": 11, "address": "1!400009", "attribute": 3, "description": "", @@ -1245,6 +1262,7 @@ To update both group name and interval: "tags": [ { "name": "temperature", + "resolved_type": 10, "value": 25.5, "transferPrecision": 2, "timestamp": 1700000000000, @@ -1254,6 +1272,7 @@ To update both group name and interval: }, { "name": "pressure", + "resolved_type": 3, "error": 3002, "quality": 0 } @@ -1309,6 +1328,7 @@ Request headers are the same as for reading tag values. "items": [ { "name": "temperature", + "resolved_type": 10, "type": 10, "address": "1!400001", "attribute": 1, diff --git a/include/neuron/msg.h b/include/neuron/msg.h index 6f964ada1..316838a6a 100644 --- a/include/neuron/msg.h +++ b/include/neuron/msg.h @@ -848,8 +848,33 @@ typedef struct neu_req_get_tag { char name[NEU_TAG_NAME_LEN]; } neu_req_get_tag_t; +typedef struct { + neu_datatag_t tag; + neu_type_e resolved_type; +} neu_resp_tag_info_t; + +static inline void neu_resp_tag_info_copy(void *dst, const void *src) +{ + neu_resp_tag_info_t * to = (neu_resp_tag_info_t *) dst; + const neu_resp_tag_info_t *from = (const neu_resp_tag_info_t *) src; + neu_tag_get_icd()->copy(&to->tag, &from->tag); + to->resolved_type = from->resolved_type; +} + +static inline void neu_resp_tag_info_fini(void *item) +{ + neu_tag_fini(&((neu_resp_tag_info_t *) item)->tag); +} + +static inline UT_icd *neu_resp_tag_info_icd(void) +{ + static UT_icd icd = { sizeof(neu_resp_tag_info_t), NULL, + neu_resp_tag_info_copy, neu_resp_tag_info_fini }; + return &icd; +} + typedef struct neu_resp_get_tag { - UT_array *tags; // array neu_datatag_t + UT_array *tags; // array neu_resp_tag_info_t } neu_resp_get_tag_t; typedef struct { @@ -1126,6 +1151,7 @@ typedef struct neu_resp_tag_value_meta { neu_tag_meta_t *metas; int n_meta; neu_datatag_t datatag; + neu_type_e resolved_type; } neu_resp_tag_value_meta_t; static inline UT_icd *neu_resp_tag_value_meta_icd() @@ -1229,6 +1255,7 @@ typedef struct neu_resp_tag_value_meta_paginate { neu_tag_meta_t *metas; int n_meta; neu_datatag_t datatag; + neu_type_e resolved_type; } neu_resp_tag_value_meta_paginate_t; static inline UT_icd *neu_resp_tag_value_meta_paginate_icd() diff --git a/plugins/restful/datatag_handle.c b/plugins/restful/datatag_handle.c index b687d7426..8a235c240 100644 --- a/plugins/restful/datatag_handle.c +++ b/plugins/restful/datatag_handle.c @@ -29,6 +29,7 @@ #include "utils/http.h" #include "datatag_handle.h" +#include "read_response.h" void handle_add_tags(nng_aio *aio) { @@ -548,6 +549,7 @@ void handle_get_tags_resp(nng_aio *aio, neu_resp_get_tag_t *tags) neu_json_get_tags_resp_t tags_res = { 0 }; char * result = NULL; char sort_by[32] = { 0 }; + int error = NEU_ERR_EINTERNAL; ssize_t sort_len = neu_http_get_param_str(aio, "sort_by", sort_by, sizeof(sort_by)); @@ -557,21 +559,24 @@ void handle_get_tags_resp(nng_aio *aio, neu_resp_get_tag_t *tags) ? neu_tag_get_sort_cmp(sort_by) : NULL; if (!cmp) { - NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, { - neu_http_response(aio, NEU_ERR_PARAM_IS_WRONG, result_error); - }); - utarray_free(tags->tags); - return; + error = NEU_ERR_PARAM_IS_WRONG; + goto error; } utarray_sort(tags->tags, cmp); } tags_res.n_tag = utarray_len(tags->tags); - tags_res.tags = calloc(tags_res.n_tag, sizeof(neu_json_tag_t)); + if (tags_res.n_tag > 0) { + tags_res.tags = calloc(tags_res.n_tag, sizeof(neu_json_tag_t)); + if (tags_res.tags == NULL) { + goto error; + } + } - utarray_foreach(tags->tags, neu_datatag_t *, tag) + utarray_foreach(tags->tags, neu_resp_tag_info_t *, item) { - int index = utarray_eltidx(tags->tags, tag); + int index = utarray_eltidx(tags->tags, item); + neu_datatag_t *tag = &item->tag; tags_res.tags[index].name = tag->name; tags_res.tags[index].address = tag->address; @@ -585,10 +590,21 @@ void handle_get_tags_resp(nng_aio *aio, neu_resp_get_tag_t *tags) tags_res.tags[index].unit = tag->unit; } - neu_json_encode_by_fn(&tags_res, neu_json_encode_get_tags_resp, &result); + neu_http_get_tags_response_t context = { .response = &tags_res, + .tags = tags->tags }; + if (neu_json_encode_by_fn(&context, neu_http_encode_get_tags_response, + &result) != 0 || + result == NULL) { + goto error; + } neu_http_ok(aio, result); + goto end; +error: + NEU_JSON_RESPONSE_ERROR(error, + { neu_http_response(aio, error, result_error); }); +end: free(result); free(tags_res.tags); utarray_free(tags->tags); diff --git a/plugins/restful/global_config_handle.c b/plugins/restful/global_config_handle.c index 7467fcd53..da1204cf1 100644 --- a/plugins/restful/global_config_handle.c +++ b/plugins/restful/global_config_handle.c @@ -1136,9 +1136,10 @@ static int get_tags_resp(context_t *ctx, neu_resp_get_tag_t *tags) goto end; } - utarray_foreach(tags->tags, neu_datatag_t *, tag) + utarray_foreach(tags->tags, neu_resp_tag_info_t *, item) { - int index = utarray_eltidx(tags->tags, tag); + int index = utarray_eltidx(tags->tags, item); + neu_datatag_t *tag = &item->tag; tags_res.tags[index].name = tag->name; tags_res.tags[index].address = tag->address; @@ -1217,9 +1218,10 @@ static int get_driver_tags_resp(context_t *ctx, neu_resp_get_tag_t *tags) goto end; } - utarray_foreach(tags->tags, neu_datatag_t *, tag) + utarray_foreach(tags->tags, neu_resp_tag_info_t *, item) { - int index = utarray_eltidx(tags->tags, tag); + int index = utarray_eltidx(tags->tags, item); + neu_datatag_t *tag = &item->tag; gtag->tags[index].name = tag->name; gtag->tags[index].address = tag->address; diff --git a/plugins/restful/read_response.c b/plugins/restful/read_response.c index c4d45bf4d..f8bc68b96 100644 --- a/plugins/restful/read_response.c +++ b/plugins/restful/read_response.c @@ -1,4 +1,5 @@ #include "msg.h" +#include "parser/neu_json_tag.h" #include "read_response.h" @@ -38,10 +39,13 @@ int neu_http_encode_read_response(void *json_object, void *param) int count = ctx->paginate ? ((neu_json_read_paginate_resp_t *) ctx->response)->n_tag : ((neu_json_read_resp_t *) ctx->response)->n_tag; + void *output_tags = ctx->paginate + ? (void *) ((neu_json_read_paginate_resp_t *) ctx->response)->tags + : (void *) ((neu_json_read_resp_t *) ctx->response)->tags; unsigned int length = ctx->tags ? utarray_len(ctx->tags) : 0; if (!json_is_object(object) || count < 0 || (count > 0 && - (ctx->offset > length || + (output_tags == NULL || ctx->offset > length || (unsigned int) count > length - ctx->offset))) { if (ctx->paginate) { neu_json_read_paginate_resp_t *response = ctx->response; @@ -72,7 +76,47 @@ int neu_http_encode_read_response(void *json_object, void *param) const neu_tag_sample_attrs_t *attrs = ctx->paginate ? &((neu_resp_tag_value_meta_paginate_t *) tag)->value.attrs : &((neu_resp_tag_value_meta_t *) tag)->value.attrs; - if (encode_attrs(json_array_get(items, i), attrs) != 0) { + neu_type_e resolved_type = ctx->paginate + ? ((neu_resp_tag_value_meta_paginate_t *) tag)->resolved_type + : ((neu_resp_tag_value_meta_t *) tag)->resolved_type; + json_t *item = json_array_get(items, i); + if (encode_attrs(item, attrs) != 0 || + json_object_set_new(item, "resolved_type", + json_integer(resolved_type)) != 0) { + return -1; + } + } + return 0; +} + +int neu_http_encode_get_tags_response(void *json_object, void *param) +{ + neu_http_get_tags_response_t *ctx = param; + json_t * object = json_object; + if (ctx == NULL || ctx->response == NULL || !json_is_object(object)) { + return -1; + } + + neu_json_get_tags_resp_t *response = ctx->response; + unsigned int length = ctx->tags ? utarray_len(ctx->tags) : 0; + if (response->n_tag < 0 || (unsigned int) response->n_tag != length || + (length > 0 && response->tags == NULL)) { + return -1; + } + + int ret = neu_json_encode_get_tags_resp(json_object, response); + if (ret != 0) { + return ret; + } + + json_t *items = json_object_get(json_object, "tags"); + if (!json_is_array(items) || json_array_size(items) != length) { + return -1; + } + for (unsigned int i = 0; i < length; i++) { + const neu_resp_tag_info_t *item = utarray_eltptr(ctx->tags, i); + if (json_object_set_new(json_array_get(items, i), "resolved_type", + json_integer(item->resolved_type)) != 0) { return -1; } } diff --git a/plugins/restful/read_response.h b/plugins/restful/read_response.h index bb0b03293..0c60c1f95 100644 --- a/plugins/restful/read_response.h +++ b/plugins/restful/read_response.h @@ -12,6 +12,12 @@ typedef struct { bool paginate; } neu_http_read_response_t; +typedef struct { + void * response; + UT_array *tags; +} neu_http_get_tags_response_t; + int neu_http_encode_read_response(void *json_object, void *param); +int neu_http_encode_get_tags_response(void *json_object, void *param); #endif diff --git a/src/adapter/driver/driver.c b/src/adapter/driver/driver.c index 56f4c294e..3ddd61714 100644 --- a/src/adapter/driver/driver.c +++ b/src/adapter/driver/driver.c @@ -103,11 +103,13 @@ static int write_callback(void *usr_data); static void read_group(int64_t timestamp, int64_t timeout, neu_tag_cache_type_e cache_type, neu_driver_cache_t *cache, const char *group, - UT_array *tags, UT_array *tag_values); + UT_array *tags, UT_array *tag_values, + bool cache_current); static void read_group_paginate(int64_t timestamp, int64_t timeout, neu_tag_cache_type_e cache_type, neu_driver_cache_t *cache, const char *group, - UT_array *tags, UT_array *tag_values); + UT_array *tags, UT_array *tag_values, + bool cache_current); static void read_report_group(int64_t timestamp, int64_t timeout, neu_tag_cache_type_e cache_type, neu_driver_cache_t *cache, const char *group, @@ -145,6 +147,28 @@ static inline void start_group_timer(neu_adapter_driver_t *driver, group_t * grp); static inline void stop_group_timer(neu_adapter_driver_t *driver, group_t *grp); +static neu_type_e sample_resolved_type(const neu_datatag_t *tag, + neu_type_e cached_type, + bool cache_current) +{ + return tag->type != NEU_TYPE_AUTO + ? tag->type + : cache_current ? cached_type : NEU_TYPE_AUTO; +} + +static neu_type_e cached_resolved_type(neu_driver_cache_t * cache, + const char * group, + const neu_datatag_t *tag, + bool cache_current) +{ + neu_type_e type = NEU_TYPE_AUTO; + uint64_t generation = 0; + if (tag->type == NEU_TYPE_AUTO && cache_current) { + neu_driver_cache_type_get(cache, group, tag->name, &type, &generation); + } + return sample_resolved_type(tag, type, cache_current); +} + static neu_datatag_t effective_tag(const neu_datatag_t *tag, neu_type_e type) { neu_datatag_t effective = *tag; @@ -869,6 +893,78 @@ void neu_adapter_driver_stop_group_timer(neu_adapter_driver_t *driver) HASH_ITER(hh, driver->groups, el, tmp) { stop_group_timer(driver, el); } } +typedef struct { + group_t * group; + UT_array * values; + int64_t now; + neu_tag_cache_type_e cache_type; + bool paginate; + int error; + int total_count; +} read_snapshot_t; + +static void read_snapshot(void *arg, int64_t timestamp, uint32_t interval, + UT_array *tags, int total_count) +{ + read_snapshot_t * ctx = arg; + group_t * group = ctx->group; + neu_adapter_driver_t *driver = group->driver; + bool cache_current = timestamp == group->timestamp; + ctx->total_count = total_count; + if (ctx->error == NEU_ERR_SUCCESS) { + if (ctx->paginate) { + read_group_paginate( + ctx->now, (int64_t) interval * NEU_DRIVER_TAG_CACHE_EXPIRE_TIME, + ctx->cache_type, driver->cache, group->name, tags, ctx->values, + cache_current); + } else { + read_group(ctx->now, + (int64_t) interval * NEU_DRIVER_TAG_CACHE_EXPIRE_TIME, + ctx->cache_type, driver->cache, group->name, tags, + ctx->values, cache_current); + } + return; + } + + utarray_foreach(tags, neu_datatag_t *, tag) + { + neu_type_e type = cached_resolved_type(driver->cache, group->name, tag, + cache_current); + if (ctx->paginate) { + neu_resp_tag_value_meta_paginate_t value = { 0 }; + snprintf(value.tag, sizeof(value.tag), "%s", tag->name); + value.value.type = NEU_TYPE_ERROR; + value.value.value.i32 = ctx->error; + value.resolved_type = type; + neu_tag_get_icd()->copy(&value.datatag, tag); + utarray_push_back(ctx->values, &value); + } else { + neu_resp_tag_value_meta_t value = { 0 }; + snprintf(value.tag, sizeof(value.tag), "%s", tag->name); + value.value.type = NEU_TYPE_ERROR; + value.value.value.i32 = ctx->error; + value.resolved_type = type; + utarray_push_back(ctx->values, &value); + } + } +} + +static int prepare_read(neu_adapter_driver_t *driver, group_t *group, bool sync) +{ + if (driver->adapter.state != NEU_NODE_RUNNING_STATE_RUNNING) { + return NEU_ERR_PLUGIN_NOT_RUNNING; + } + if (sync) { + if (driver->adapter.module->intf_funs->driver.group_sync == NULL) { + return NEU_ERR_PLUGIN_NOT_SUPPORT_READ_SYNC; + } + stop_group_timer(driver, group); + driver->adapter.module->intf_funs->driver.group_sync( + driver->adapter.plugin, &group->grp); + } + return NEU_ERR_SUCCESS; +} + void neu_adapter_driver_read_group(neu_adapter_driver_t *driver, neu_reqresp_head_t * req) { @@ -882,55 +978,21 @@ void neu_adapter_driver_read_group(neu_adapter_driver_t *driver, return; } - neu_resp_read_group_t resp = { 0 }; - neu_group_t * group = g->group; - UT_array *tags = neu_group_query_read_tag(group, cmd->name, cmd->desc, - cmd->n_tag, cmd->tags); - + neu_resp_read_group_t resp = { 0 }; utarray_new(resp.tags, neu_resp_tag_value_meta_icd()); - - if (driver->adapter.state != NEU_NODE_RUNNING_STATE_RUNNING) { - utarray_foreach(tags, neu_datatag_t *, tag) - { - neu_resp_tag_value_meta_t tag_value = { 0 }; - snprintf(tag_value.tag, sizeof(tag_value.tag), "%s", tag->name); - tag_value.value.type = NEU_TYPE_ERROR; - tag_value.value.value.i32 = NEU_ERR_PLUGIN_NOT_RUNNING; - - utarray_push_back(resp.tags, &tag_value); - } - } else if (cmd->sync) { - if (NULL == driver->adapter.module->intf_funs->driver.group_sync) { - // plugin does not support sync read - utarray_foreach(tags, neu_datatag_t *, tag) - { - neu_resp_tag_value_meta_t tag_value = { 0 }; - snprintf(tag_value.tag, sizeof(tag_value.tag), "%s", tag->name); - tag_value.value.type = NEU_TYPE_ERROR; - tag_value.value.value.i32 = - NEU_ERR_PLUGIN_NOT_SUPPORT_READ_SYNC; - - utarray_push_back(resp.tags, &tag_value); - } - } else { - // sync read to update cache - stop_group_timer(driver, g); - driver->adapter.module->intf_funs->driver.group_sync( - driver->adapter.plugin, &g->grp); - // fetch updated data from cache - read_group(global_timestamp, - neu_group_get_interval(group) * - NEU_DRIVER_TAG_CACHE_EXPIRE_TIME, - neu_adapter_get_tag_cache_type(&driver->adapter), - driver->cache, cmd->group, tags, resp.tags); - start_group_timer(driver, g); - } - } else { - read_group(global_timestamp, - neu_group_get_interval(group) * - NEU_DRIVER_TAG_CACHE_EXPIRE_TIME, - neu_adapter_get_tag_cache_type(&driver->adapter), - driver->cache, cmd->group, tags, resp.tags); + neu_group_tag_query_t query = { .name = cmd->name, + .desc = cmd->desc, + .n_tagname = cmd->n_tag, + .tagnames = cmd->tags, + .readable = true }; + read_snapshot_t snapshot = { .group = g, + .values = resp.tags, + .error = prepare_read(driver, g, cmd->sync) }; + snapshot.now = global_timestamp; + snapshot.cache_type = neu_adapter_get_tag_cache_type(&driver->adapter); + neu_group_query_snapshot(g->group, &query, &snapshot, read_snapshot); + if (cmd->sync && snapshot.error == NEU_ERR_SUCCESS) { + start_group_timer(driver, g); } resp.driver = cmd->driver; @@ -938,7 +1000,6 @@ void neu_adapter_driver_read_group(neu_adapter_driver_t *driver, cmd->driver = NULL; // ownership moved cmd->group = NULL; // ownership moved - utarray_free(tags); neu_req_read_group_fini(cmd); req->type = NEU_RESP_READ_GROUP; @@ -959,92 +1020,27 @@ void neu_adapter_driver_read_group_paginate(neu_adapter_driver_t *driver, return; } - neu_resp_read_group_paginate_t resp = { 0 }; - neu_group_t * group = g->group; - UT_array * tags; - - neu_tag_cmp_fn cmp = neu_tag_get_sort_cmp(cmd->sort_by); - if (cmd->is_error != true && cmd->current_page > 0 && cmd->page_size > 0) { - tags = neu_group_query_read_tag_paginate( - group, cmd->name, cmd->desc, cmd->current_page, cmd->page_size, - &resp.total_count, cmp); - } else { - tags = neu_group_query_read_tag(group, cmd->name, cmd->desc, 0, NULL); - if (cmp) { - utarray_sort(tags, cmp); - } - resp.total_count = utarray_len(tags); - } - + neu_resp_read_group_paginate_t resp = { 0 }; + bool paginate = + !cmd->is_error && cmd->current_page > 0 && cmd->page_size > 0; + neu_group_tag_query_t query = { .name = cmd->name, + .desc = cmd->desc, + .current_page = + paginate ? cmd->current_page : 0, + .page_size = paginate ? cmd->page_size : 0, + .cmp = neu_tag_get_sort_cmp(cmd->sort_by), + .readable = !paginate }; utarray_new(resp.tags, neu_resp_tag_value_meta_paginate_icd()); - - if (driver->adapter.state != NEU_NODE_RUNNING_STATE_RUNNING) { - utarray_foreach(tags, neu_datatag_t *, tag) - { - neu_resp_tag_value_meta_paginate_t tag_value = { 0 }; - snprintf(tag_value.tag, sizeof(tag_value.tag), "%s", tag->name); - tag_value.value.type = NEU_TYPE_ERROR; - tag_value.value.value.i32 = NEU_ERR_PLUGIN_NOT_RUNNING; - - tag_value.datatag.name = strdup(tag->name); - tag_value.datatag.address = strdup(tag->address); - tag_value.datatag.attribute = tag->attribute; - tag_value.datatag.type = tag->type; - tag_value.datatag.precision = tag->precision; - tag_value.datatag.decimal = tag->decimal; - tag_value.datatag.bias = tag->bias; - tag_value.datatag.description = strdup(tag->description); - tag_value.datatag.option = tag->option; - tag_value.datatag.unit = strdup(tag->unit); - memcpy(tag_value.datatag.meta, tag->meta, NEU_TAG_META_LENGTH); - - utarray_push_back(resp.tags, &tag_value); - } - } else if (cmd->sync) { - if (NULL == driver->adapter.module->intf_funs->driver.group_sync) { - // plugin does not support sync read - utarray_foreach(tags, neu_datatag_t *, tag) - { - neu_resp_tag_value_meta_paginate_t tag_value = { 0 }; - snprintf(tag_value.tag, sizeof(tag_value.tag), "%s", tag->name); - tag_value.value.type = NEU_TYPE_ERROR; - tag_value.value.value.i32 = - NEU_ERR_PLUGIN_NOT_SUPPORT_READ_SYNC; - - tag_value.datatag.name = strdup(tag->name); - tag_value.datatag.address = strdup(tag->address); - tag_value.datatag.attribute = tag->attribute; - tag_value.datatag.type = tag->type; - tag_value.datatag.precision = tag->precision; - tag_value.datatag.decimal = tag->decimal; - tag_value.datatag.bias = tag->bias; - tag_value.datatag.description = strdup(tag->description); - tag_value.datatag.option = tag->option; - tag_value.datatag.unit = strdup(tag->unit); - memcpy(tag_value.datatag.meta, tag->meta, NEU_TAG_META_LENGTH); - - utarray_push_back(resp.tags, &tag_value); - } - } else { - // sync read to update cache - stop_group_timer(driver, g); - driver->adapter.module->intf_funs->driver.group_sync( - driver->adapter.plugin, &g->grp); - // fetch updated data from cache - read_group_paginate( - global_timestamp, - neu_group_get_interval(group) * - NEU_DRIVER_TAG_CACHE_EXPIRE_TIME, - neu_adapter_get_tag_cache_type(&driver->adapter), driver->cache, - cmd->group, tags, resp.tags); - start_group_timer(driver, g); - } - } else { - read_group_paginate(global_timestamp, - neu_group_get_interval(group) * - NEU_DRIVER_TAG_CACHE_EXPIRE_TIME, - neu_adapter_get_tag_cache_type(&driver->adapter), - driver->cache, cmd->group, tags, resp.tags); + read_snapshot_t snapshot = { .group = g, + .values = resp.tags, + .paginate = true, + .error = prepare_read(driver, g, cmd->sync) }; + snapshot.now = global_timestamp; + snapshot.cache_type = neu_adapter_get_tag_cache_type(&driver->adapter); + neu_group_query_snapshot(g->group, &query, &snapshot, read_snapshot); + resp.total_count = snapshot.total_count; + if (cmd->sync && snapshot.error == NEU_ERR_SUCCESS) { + start_group_timer(driver, g); } resp.driver = cmd->driver; @@ -1055,7 +1051,6 @@ void neu_adapter_driver_read_group_paginate(neu_adapter_driver_t *driver, cmd->driver = NULL; // ownership moved cmd->group = NULL; // ownership moved - utarray_free(tags); neu_req_read_group_paginate_fini(cmd); req->type = NEU_RESP_READ_GROUP_PAGINATE; @@ -2479,6 +2474,29 @@ int neu_adapter_driver_get_tag(neu_adapter_driver_t *driver, const char *group, return ret; } +typedef struct { + group_t * group; + UT_array *tags; +} tag_snapshot_t; + +static void query_tag_snapshot(void *arg, int64_t timestamp, uint32_t interval, + UT_array *tags, int total_count) +{ + tag_snapshot_t *ctx = arg; + (void) interval; + (void) total_count; + bool cache_current = timestamp == ctx->group->timestamp; + utarray_foreach(tags, neu_datatag_t *, tag) + { + utarray_extend_back(ctx->tags); + neu_resp_tag_info_t *info = utarray_back(ctx->tags); + info->resolved_type = cached_resolved_type( + ctx->group->driver->cache, ctx->group->name, tag, cache_current); + info->tag = *tag; + memset(tag, 0, sizeof(*tag)); + } +} + int neu_adapter_driver_query_tag(neu_adapter_driver_t *driver, const char *group, const char *name, UT_array **tags) @@ -2488,11 +2506,12 @@ int neu_adapter_driver_query_tag(neu_adapter_driver_t *driver, HASH_FIND_STR(driver->groups, group, find); if (find != NULL) { - if (strlen(name) > 0) { - *tags = neu_group_query_tag(find->group, name); - } else { - *tags = neu_group_get_tag(find->group); - } + neu_group_tag_query_t query = { .name = name }; + tag_snapshot_t snapshot = { .group = find }; + utarray_new(snapshot.tags, neu_resp_tag_info_icd()); + neu_group_query_snapshot(find->group, &query, &snapshot, + query_tag_snapshot); + *tags = snapshot.tags; } else { ret = NEU_ERR_GROUP_NOT_EXIST; } @@ -2594,7 +2613,7 @@ static void report_to_app(neu_adapter_driver_t *driver, group_t *group, neu_group_get_interval(group->group) * NEU_DRIVER_TAG_CACHE_EXPIRE_TIME, neu_adapter_get_tag_cache_type(&driver->adapter), driver->cache, - group->name, tags, data->tags); + group->name, tags, data->tags, true); nlog_info("report group: %s, all tags: %d, report tags: %d", group->name, utarray_len(tags), utarray_len(data->tags)); @@ -2963,6 +2982,7 @@ static void read_report_group(int64_t timestamp, int64_t timeout, { neu_driver_cache_value_t value = { 0 }; neu_resp_tag_value_meta_t tag_value = { 0 }; + tag_value.resolved_type = tag->type; if (neu_tag_attribute_test(tag, NEU_ATTRIBUTE_SUBSCRIBE)) { if (neu_driver_cache_meta_get_changed(cache, group, tag->name, @@ -2989,6 +3009,8 @@ static void read_report_group(int64_t timestamp, int64_t timeout, tag_value.datatag.bias = tag->bias; tag_value.value.attrs = value.value.attrs; + tag_value.resolved_type = + sample_resolved_type(tag, value.resolved_type, true); if (value.value.type == NEU_TYPE_ERROR) { tag_value.value = value.value; @@ -3162,7 +3184,7 @@ static void read_report_group(int64_t timestamp, int64_t timeout, static void read_group(int64_t timestamp, int64_t timeout, neu_tag_cache_type_e cache_type, neu_driver_cache_t *cache, const char *group, - UT_array *tags, UT_array *tag_values) + UT_array *tags, UT_array *tag_values, bool cache_current) { utarray_foreach(tags, neu_datatag_t *, tag) { @@ -3170,7 +3192,8 @@ static void read_group(int64_t timestamp, int64_t timeout, neu_driver_cache_value_t value = { 0 }; snprintf(tag_value.tag, sizeof(tag_value.tag), "%s", tag->name); - tag_value.datatag = *tag; + tag_value.datatag = *tag; + tag_value.resolved_type = tag->type; tag_value.datatag.bias = tag->bias; @@ -3185,6 +3208,8 @@ static void read_group(int64_t timestamp, int64_t timeout, } tag_value.value.attrs = value.value.attrs; + tag_value.resolved_type = + sample_resolved_type(tag, value.resolved_type, cache_current); if (value.value.type == NEU_TYPE_ERROR) { tag_value.value = value.value; @@ -3355,7 +3380,8 @@ static void read_group(int64_t timestamp, int64_t timeout, static void read_group_paginate(int64_t timestamp, int64_t timeout, neu_tag_cache_type_e cache_type, neu_driver_cache_t *cache, const char *group, - UT_array *tags, UT_array *tag_values) + UT_array *tags, UT_array *tag_values, + bool cache_current) { utarray_foreach(tags, neu_datatag_t *, tag) { @@ -3375,6 +3401,7 @@ static void read_group_paginate(int64_t timestamp, int64_t timeout, tag_value.datatag.option = tag->option; tag_value.datatag.unit = strdup(tag->unit); memcpy(tag_value.datatag.meta, tag->meta, NEU_TAG_META_LENGTH); + tag_value.resolved_type = tag->type; if (tag->attribute == NEU_ATTRIBUTE_WRITE) { tag_value.value.type = NEU_TYPE_STRING; @@ -3394,6 +3421,8 @@ static void read_group_paginate(int64_t timestamp, int64_t timeout, } tag_value.value.attrs = value.value.attrs; + tag_value.resolved_type = + sample_resolved_type(tag, value.resolved_type, cache_current); if (value.value.type == NEU_TYPE_ERROR) { tag_value.value = value.value; diff --git a/src/base/group.c b/src/base/group.c index 92bb5158f..102cd1a7e 100644 --- a/src/base/group.c +++ b/src/base/group.c @@ -133,15 +133,19 @@ uint32_t neu_group_get_interval(const neu_group_t *group) void neu_group_set_interval(neu_group_t *group, uint32_t interval) { + pthread_mutex_lock(&group->mtx); group->interval = interval; + pthread_mutex_unlock(&group->mtx); } int neu_group_update(neu_group_t *group, uint32_t interval) { + pthread_mutex_lock(&group->mtx); if (group->interval != interval) { group->interval = interval; update_timestamp(group); } + pthread_mutex_unlock(&group->mtx); return 0; } @@ -308,14 +312,14 @@ static inline UT_array *filter_tags_by_page( tag_elem_t *el = NULL, *tmp = NULL; UT_array * array = NULL; int count = 0; - int start_index = (current_page - 1) * page_size; - int end_index = start_index + page_size; + size_t start_index = (size_t)(current_page - 1) * (size_t) page_size; + size_t end_index = start_index + (size_t) page_size; utarray_new(array, neu_tag_get_icd()); HASH_ITER(hh, tags, el, tmp) { if (predicate(el->tag, data)) { - if (count >= start_index && count < end_index) { + if ((size_t) count >= start_index && (size_t) count < end_index) { utarray_push_back(array, el->tag); } count++; @@ -378,6 +382,55 @@ static inline bool is_readable_and_match_query(const neu_datatag_t *tag, return is_readable(tag, NULL) && match_query(tag, data); } +static UT_array *query_tags_locked(neu_group_t * group, + const neu_group_tag_query_t *query, + int * total_count) +{ + struct query q = { .name = (char *) query->name, + .desc = (char *) query->desc, + .n_tagname = query->n_tagname, + .tagnames = query->tagnames }; + bool (*predicate)(const neu_datatag_t *, void *) = + query->readable ? is_readable_and_match_query : match_query; + bool paginate = query->current_page > 0 && query->page_size > 0; + if (paginate && query->cmp == NULL) { + return filter_tags_by_page(group->tags, predicate, &q, + query->current_page, query->page_size, + total_count); + } + + UT_array *tags = filter_tags(group->tags, predicate, &q); + if (query->cmp) { + utarray_sort(tags, query->cmp); + } + *total_count = utarray_len(tags); + if (paginate) { + size_t start = + (size_t)(query->current_page - 1) * (size_t) query->page_size; + if (start >= utarray_len(tags)) { + utarray_clear(tags); + } else { + utarray_erase(tags, 0, start); + if (utarray_len(tags) > (unsigned int) query->page_size) { + utarray_resize(tags, query->page_size); + } + } + } + return tags; +} + +void neu_group_query_snapshot(neu_group_t * group, + const neu_group_tag_query_t *query, void *arg, + neu_group_snapshot_fn fn) +{ + int total_count = 0; + pthread_mutex_lock(&group->mtx); + UT_array *tags = query_tags_locked(group, query, &total_count); + fn(arg, group->timestamp, group->interval, tags, total_count); + utarray_free(tags); + pthread_mutex_unlock(&group->mtx); +} + UT_array *neu_group_query_tag(neu_group_t *group, const char *name) { UT_array *array = NULL; @@ -414,37 +467,18 @@ UT_array *neu_group_query_read_tag_paginate(neu_group_t *group, int * total_count, neu_tag_cmp_fn cmp) { - UT_array * array = NULL; - struct query q = { - .name = (char *) name, - .desc = (char *) desc, - }; - + neu_group_tag_query_t query = { .name = name, + .desc = desc, + .current_page = current_page, + .page_size = page_size, + .cmp = cmp }; + int total = 0; pthread_mutex_lock(&group->mtx); - if (cmp) { - array = filter_tags(group->tags, match_query, &q); - } else { - array = filter_tags_by_page(group->tags, match_query, &q, current_page, - page_size, total_count); - } + UT_array *array = query_tags_locked(group, &query, &total); pthread_mutex_unlock(&group->mtx); - - if (cmp) { - utarray_sort(array, cmp); - if (total_count) { - *total_count = utarray_len(array); - } - size_t start = (size_t)(current_page - 1) * page_size; - if (start >= utarray_len(array)) { - utarray_clear(array); - } else { - utarray_erase(array, 0, start); - if (utarray_len(array) > (unsigned int) page_size) { - utarray_resize(array, page_size); - } - } + if (total_count) { + *total_count = total; } - return array; } diff --git a/src/base/group.h b/src/base/group.h index 6d0f7e41a..c359b0acf 100644 --- a/src/base/group.h +++ b/src/base/group.h @@ -28,6 +28,24 @@ typedef struct neu_group neu_group_t; +typedef struct { + const char * name; + const char * desc; + uint16_t n_tagname; + char ** tagnames; + int current_page; + int page_size; + neu_tag_cmp_fn cmp; + bool readable; +} neu_group_tag_query_t; + +typedef void (*neu_group_snapshot_fn)(void *arg, int64_t timestamp, + uint32_t interval, UT_array *tags, + int total_count); +void neu_group_query_snapshot(neu_group_t * group, + const neu_group_tag_query_t *query, void *arg, + neu_group_snapshot_fn fn); + neu_group_t *neu_group_new(const char *name, uint32_t interval); const char * neu_group_get_name(const neu_group_t *group); int neu_group_set_name(neu_group_t *group, const char *name); diff --git a/tests/ft/auto_type_http.py b/tests/ft/auto_type_http.py index bcc9e2f3b..a77016d23 100644 --- a/tests/ft/auto_type_http.py +++ b/tests/ft/auto_type_http.py @@ -29,6 +29,7 @@ import unittest import urllib.error import urllib.request +from urllib.parse import urlencode REPO = Path(__file__).resolve().parents[2] @@ -184,10 +185,29 @@ def read(self, name="a", *, group="g", paginate=False, sync=False, query=None): if query is not None: payload["query"] = query response = self.http("POST", "/read/paginate" if paginate else "/read", payload) - return next(tag for tag in response["items" if paginate else "tags"] - if tag["name"] == name) + tag = next(tag for tag in response["items" if paginate else "tags"] + if tag["name"] == name) + self.assertIs(int, type(tag["resolved_type"]), tag) + return tag - def await_sample(self, value=None, error=None, attrs=None, **kwargs): + def list_tags(self, **query): + params = {"node": self.node, "group": "g"} + params.update(query) + return self.http("GET", "/tags?" + urlencode(params))["tags"] + + def await_types(self, expected): + deadline = time.monotonic() + 8 + actual = None + while time.monotonic() < deadline: + tags = self.list_tags() + self.assertTrue(all(type(tag["resolved_type"]) is int for tag in tags)) + actual = {tag["name"]: tag["resolved_type"] for tag in tags} + if actual == expected: + return tags + time.sleep(0.05) + self.fail(f"resolved types did not become {expected!r}: {actual!r}") + + def await_sample(self, value=None, error=None, attrs=None, resolved_type=None, **kwargs): deadline = time.monotonic() + 8 actual = None while time.monotonic() < deadline: @@ -195,11 +215,12 @@ def await_sample(self, value=None, error=None, attrs=None, **kwargs): sample_attrs = {name: actual[name] for name in SAMPLE_ATTRS if name in actual} matches = (actual.get("error") == error if error is not None else "error" not in actual and actual.get("value") == value) - if matches and (attrs is None or sample_attrs == attrs): + if (matches and (attrs is None or sample_attrs == attrs) and + (resolved_type is None or actual["resolved_type"] == resolved_type)): return actual time.sleep(0.05) self.fail(f"sample did not become value={value!r}, error={error!r}, " - f"attrs={attrs!r}: {actual}") + f"attrs={attrs!r}, resolved_type={resolved_type!r}: {actual}") def write(self, value, *, name="a", error=0): status, response = self.request("POST", "/write", { @@ -220,6 +241,11 @@ def metadata_is_auto(self): exported = self.http("GET", "/global/drivers?name=" + self.node) self.assertTrue(all(tag["type"] == 0 for node in exported["nodes"] for group in node["groups"] for tag in group["tags"])) + self.assertTrue(all("resolved_type" not in tag for node in exported["nodes"] + for group in node["groups"] for tag in group["tags"])) + global_config = self.http("GET", "/global/config") + self.assertTrue(all("resolved_type" not in tag for group in global_config["tags"] + for tag in group["tags"])) with sqlite3.connect(f"file:{self.root}/persistence/sqlite.db?mode=ro", uri=True) as db: rows = db.execute("SELECT type FROM tags WHERE driver_name = ?", (self.node,)).fetchall() self.assertTrue(rows) @@ -264,6 +290,126 @@ def test_resolution_writes_mismatch_recovery_and_restart(self): self.await_sample("new lifecycle") self.metadata_is_auto() + def test_resolved_type_exports_roundtrip_units(self): + for unit in ("°C", "", None): + tags = [self.tag(), self.tag("explicit", type=3)] + if unit is not None: + tags[0]["unit"] = unit + data = { + "nodes": [{"name": self.node, "plugin": "auto_type"}], + "groups": [{"driver": self.node, "group": "g", "interval": 100, + "tag_count": len(tags)}], + "tags": [{"driver": self.node, "group": "g", "tags": tags}], + "subscriptions": [], "settings": []} + expected_units = {"a": unit or "", "explicit": ""} + for cycle in range(2): + self.http("PUT", "/global/config", data, error=0) + listed = self.await_types({"a": 0, "explicit": 3}) + self.assertEqual(expected_units, {tag["name"]: tag["unit"] for tag in listed}) + data = self.http("GET", "/global/config") + exported = next(group for group in data["tags"] + if group["driver"] == self.node and group["group"] == "g")["tags"] + self.assertEqual(expected_units, {tag["name"]: tag["unit"] for tag in exported}) + self.assertEqual({"a": 0, "explicit": 3}, + {tag["name"]: tag["type"] for tag in exported}) + self.assertTrue(all("resolved_type" not in tag for tag in exported)) + drivers = self.http("GET", "/global/drivers?name=" + self.node) + exported = drivers["nodes"][0]["groups"][0]["tags"] + self.assertEqual(expected_units, {tag["name"]: tag["unit"] for tag in exported}) + self.assertTrue(all("resolved_type" not in tag for tag in exported)) + + def test_resolved_type_lifecycle(self): + self.add_tags(self.tag(), self.tag("explicit", type=3), + self.tag("write_only", type=5, attribute=2)) + self.control(hold=True) + self.await_types({"a": 0, "explicit": 3, "write_only": 5}) + for paginate in (False, True): + self.await_sample(error=3011, resolved_type=0, paginate=paginate) + self.await_sample(error=3011, resolved_type=3, name="explicit", paginate=paginate) + self.control() + for paginate in (False, True): + self.await_sample(123, resolved_type=3, paginate=paginate) + self.await_sample(123, resolved_type=3, sync=True) + self.await_types({"a": 3, "explicit": 3, "write_only": 5}) + self.control(hold=True) + self.http("POST", "/node/ctl", {"node": self.node, "cmd": 1}, error=0) + for paginate in (False, True): + self.await_sample(error=3010, resolved_type=3, paginate=paginate) + self.await_types({"a": 3, "explicit": 3, "write_only": 5}) + self.add_tags(self.tag(address="changed"), method="PUT") + self.await_types({"a": 0, "explicit": 3, "write_only": 5}) + for paginate in (False, True): + self.await_sample(error=3010, resolved_type=0, paginate=paginate) + self.control(type=10, value=12.5) + self.http("POST", "/node/ctl", {"node": self.node, "cmd": 0}, error=0) + self.await_sample(12.5, resolved_type=10) + self.await_types({"a": 10, "explicit": 3, "write_only": 5}) + for params, expected_error in (({"type": 10, "error": 3002}, 3002), + ({"type": 3}, 3007)): + self.control(**params) + for paginate in (False, True): + self.await_sample(error=expected_error, resolved_type=10, paginate=paginate) + self.await_types({"a": 10, "explicit": 3, "write_only": 5}) + self.control(type=10, value=12.5) + self.await_sample(12.5, resolved_type=10) + self.control(hold=True) + self.stop() + self.start() + self.await_sample(error=3011, resolved_type=0) + self.await_types({"a": 0, "explicit": 3, "write_only": 5}) + self.control(type=13, value="new type", pointer=True) + for paginate in (False, True): + self.await_sample("new type", resolved_type=13, paginate=paginate) + self.await_types({"a": 13, "explicit": 3, "write_only": 5}) + exported = self.http("GET", "/global/drivers?name=" + self.node) + tags = exported["nodes"][0]["groups"][0]["tags"] + self.assertEqual(0, next(tag for tag in tags if tag["name"] == "a")["type"]) + self.assertTrue(all("resolved_type" not in tag for tag in tags)) + + def test_resolved_type_sort_filter_and_pages(self): + self.add_tags(self.tag("a", address="z"), + self.tag("b", type=3, address="a", description="needle"), + self.tag("c", type=10, address="m"), + self.tag("w", type=5, attribute=2)) + self.control() + self.await_sample(123, resolved_type=3) + expected = {"a": 3, "b": 3, "c": 10, "w": 5} + self.await_types(expected) + for sort in ("+name", "-name", "+type", "-type", "+address"): + tags = self.list_tags(sort_by=sort) + self.assertEqual(expected, {tag["name"]: tag["resolved_type"] for tag in tags}) + self.assertEqual(["b"], [tag["name"] for tag in self.list_tags(name="needle")]) + self.assertEqual([], self.list_tags(name="not-present")) + self.http("GET", "/tags?" + urlencode({"node": self.node, "group": "g", + "sort_by": "invalid"}), status=400, error=1003) + request = {"node": self.node, "group": "g"} + for sort in ("+name", "-type", "+address"): + full = self.http("POST", "/read/paginate", dict(request, query={ + "sort_by": sort, "currentPage": 1, "pageSize": 10}))["items"] + self.assertEqual(expected, {tag["name"]: tag["resolved_type"] for tag in full}) + for index, tag in enumerate(full): + page = self.http("POST", "/read/paginate", dict(request, query={ + "sort_by": sort, "currentPage": index + 1, "pageSize": 1}))["items"] + self.assertEqual([tag["name"]], [item["name"] for item in page]) + self.assertEqual(expected[tag["name"]], page[0]["resolved_type"]) + self.control(error=3002) + for name in ("a", "b", "c"): + self.await_sample(error=3002, name=name, resolved_type=expected[name]) + self.control(hold=True) + full = self.http("POST", "/read/paginate", dict(request, query={ + "isError": True, "sort_by": "-type"}))["items"] + self.assertEqual({name: expected[name] for name in ("a", "b", "c")}, + {tag["name"]: tag["resolved_type"] for tag in full}) + for index, tag in enumerate(full): + page = self.http("POST", "/read/paginate", dict(request, query={ + "isError": True, "sort_by": "-type", "currentPage": index + 1, + "pageSize": 1}))["items"] + self.assertEqual([tag], page) + for is_error in (False, True): + response = self.http("POST", "/read/paginate", dict(request, query={ + "isError": is_error, "currentPage": 99, "pageSize": 1})) + self.assertEqual([], response["items"]) + def test_all_update_callbacks(self): for index, mode in enumerate(("update", "meta", "trace", "im", "im_f_m")): with self.subTest(mode=mode): @@ -347,8 +493,10 @@ def test_sample_attrs_pagination(self): def test_scaled_value_and_byte_order(self): self.add_tags(self.tag(decimal=0.5, bias=1), self.tag("big", address="sample#B")) self.control(value=256) - self.await_sample(129) - self.await_sample(1, name="big") + self.await_sample(129, resolved_type=3) + self.await_sample(129, resolved_type=3, paginate=True) + self.await_types({"a": 3, "big": 3}) + self.await_sample(1, name="big", resolved_type=3) self.write(51) self.assertEqual(100, self.records()[-1]["value"]) self.write(1, name="big") diff --git a/tests/ft/http_api/test_http_api.py b/tests/ft/http_api/test_http_api.py index 91af4f6b8..e6709d688 100644 --- a/tests/ft/http_api/test_http_api.py +++ b/tests/ft/http_api/test_http_api.py @@ -322,7 +322,9 @@ def test_global_config_tag_unit_roundtrip(self, unit): response = api.get_tags(driver["name"], group["group"]) assert 200 == response.status_code - assert expected_tags == response.json()["tags"] + expected_live = [dict(tag, resolved_type=tag["type"]) + for tag in expected_tags] + assert expected_live == response.json()["tags"] response = api.get_global_config() assert 200 == response.status_code @@ -354,7 +356,9 @@ def test_driver_tag_unit_roundtrip(self, unit): response = api.get_tags(driver["name"], driver["groups"][0]["group"]) assert 200 == response.status_code - assert expected["groups"][0]["tags"] == response.json()["tags"] + expected_live = [dict(tag, resolved_type=tag["type"]) + for tag in expected["groups"][0]["tags"]] + assert expected_live == response.json()["tags"] response = api.get_driver(driver["name"]) assert 200 == response.status_code diff --git a/tests/ut/driver_auto_helpers.c b/tests/ut/driver_auto_helpers.c index 0deb77291..8994df869 100644 --- a/tests/ut/driver_auto_helpers.c +++ b/tests/ut/driver_auto_helpers.c @@ -1,3 +1,4 @@ +#include "driver_auto_helpers.h" #include "../../src/adapter/driver/driver.c" bool sub_filter_err = false; @@ -103,7 +104,8 @@ int auto_test_metadata(neu_type_e resolved, bool pending, int *count, } void auto_test_read(neu_dvalue_t sample, double decimal, const char *address, - int path, neu_dvalue_t *output, neu_type_e *config_type) + int path, neu_dvalue_t *output, neu_type_e *config_type, + neu_type_e *resolved_type) { neu_driver_cache_t *cache = neu_driver_cache_new(); neu_datatag_t tag = { .name = "a", @@ -123,16 +125,17 @@ void auto_test_read(neu_dvalue_t sample, double decimal, const char *address, if (path == 2) { utarray_new(values, neu_resp_tag_value_meta_paginate_icd()); read_group_paginate(100, 1000, NEU_TAG_CACHE_TYPE_NEVER, cache, "g", - tags, values); + tags, values, true); neu_resp_tag_value_meta_paginate_t *v = utarray_front(values); *output = v->value; *config_type = v->datatag.type; + *resolved_type = v->resolved_type; neu_tag_fini(&v->datatag); } else { utarray_new(values, neu_resp_tag_value_meta_icd()); if (path == 0) { read_group(100, 1000, NEU_TAG_CACHE_TYPE_NEVER, cache, "g", tags, - values); + values, true); } else { read_report_group(100, 1000, NEU_TAG_CACHE_TYPE_NEVER, cache, "g", tags, values); @@ -140,8 +143,85 @@ void auto_test_read(neu_dvalue_t sample, double decimal, const char *address, neu_resp_tag_value_meta_t *v = utarray_front(values); *output = v->value; *config_type = v->datatag.type; + *resolved_type = v->resolved_type; } utarray_free(values); utarray_free(tags); neu_driver_cache_destroy(cache); } + +void auto_test_snapshot(const auto_test_snapshot_case_t *input, bool paginate, + auto_test_snapshot_result_t *output) +{ + neu_adapter_driver_t driver = { 0 }; + group_t group = { .name = "g", .driver = &driver }; + neu_datatag_t tag = { .name = "a", + .address = "sample", + .description = "", + .unit = "", + .type = input->configured_type, + .attribute = input->write_only + ? NEU_ATTRIBUTE_WRITE + : NEU_ATTRIBUTE_READ | NEU_ATTRIBUTE_WRITE }; + driver.cache = neu_driver_cache_new(); + group.group = neu_group_new("g", 100); + HASH_ADD_STR(driver.groups, name, &group); + neu_group_add_tag(group.group, &tag); + neu_group_change_test(group.group, 0, &group, capture_timestamp); + if (!input->missing_cache) { + neu_dvalue_t initial = { .type = NEU_TYPE_ERROR, + .value.i32 = NEU_ERR_PLUGIN_TAG_NOT_READY }; + neu_driver_cache_add_tag(driver.cache, "g", &tag, initial); + if (input->learned_type != NEU_TYPE_AUTO) { + neu_dvalue_t learned = { .type = input->learned_type }; + neu_driver_cache_update(driver.cache, "g", "a", 100, learned, NULL, + 0); + } + neu_driver_cache_update(driver.cache, "g", "a", 100, input->sample, + NULL, 0); + } + if (input->pending) { + group.timestamp = 0; + } + + UT_array *listed = NULL; + neu_adapter_driver_query_tag(&driver, "g", "", &listed); + neu_resp_tag_info_t *info = utarray_front(listed); + output->configured_type = info->tag.type; + output->listed_type = info->resolved_type; + utarray_free(listed); + + UT_array *values; + utarray_new(values, + paginate ? neu_resp_tag_value_meta_paginate_icd() + : neu_resp_tag_value_meta_icd()); + neu_group_tag_query_t query = { .current_page = paginate ? 1 : 0, + .page_size = paginate ? 10 : 0, + .readable = !paginate }; + read_snapshot_t snapshot = { .group = &group, + .values = values, + .now = input->expired ? 1000000 : 100, + .cache_type = input->expired + ? NEU_TAG_CACHE_TYPE_INTERVAL + : NEU_TAG_CACHE_TYPE_NEVER, + .paginate = paginate, + .error = input->error }; + neu_group_query_snapshot(group.group, &query, &snapshot, read_snapshot); + output->read_count = utarray_len(values); + if (output->read_count) { + if (paginate) { + neu_resp_tag_value_meta_paginate_t *value = utarray_front(values); + output->value = value->value; + output->read_type = value->resolved_type; + neu_tag_fini(&value->datatag); + } else { + neu_resp_tag_value_meta_t *value = utarray_front(values); + output->value = value->value; + output->read_type = value->resolved_type; + } + } + utarray_free(values); + HASH_DEL(driver.groups, &group); + neu_group_destroy(group.group); + neu_driver_cache_destroy(driver.cache); +} diff --git a/tests/ut/driver_auto_helpers.h b/tests/ut/driver_auto_helpers.h new file mode 100644 index 000000000..a3f41b208 --- /dev/null +++ b/tests/ut/driver_auto_helpers.h @@ -0,0 +1,35 @@ +#ifndef NEU_DRIVER_AUTO_TEST_HELPERS_H +#define NEU_DRIVER_AUTO_TEST_HELPERS_H + +#include "type.h" + +int is_value_in_range(neu_type_e, int64_t, double, neu_type_e, double); +int auto_test_prepare(neu_type_e, neu_dvalue_t *, double, double, const char *, + int, bool *, neu_type_e *); +void auto_test_read(neu_dvalue_t, double, const char *, int, neu_dvalue_t *, + neu_type_e *, neu_type_e *); +int auto_test_metadata(neu_type_e, bool, int *, neu_type_e *); + +typedef struct { + neu_type_e configured_type; + neu_type_e learned_type; + neu_dvalue_t sample; + bool missing_cache; + bool pending; + bool expired; + bool write_only; + int error; +} auto_test_snapshot_case_t; + +typedef struct { + neu_dvalue_t value; + neu_type_e configured_type; + neu_type_e listed_type; + neu_type_e read_type; + unsigned int read_count; +} auto_test_snapshot_result_t; + +void auto_test_snapshot(const auto_test_snapshot_case_t *, bool, + auto_test_snapshot_result_t *); + +#endif diff --git a/tests/ut/driver_auto_test.cc b/tests/ut/driver_auto_test.cc index 73909140b..57568de49 100644 --- a/tests/ut/driver_auto_test.cc +++ b/tests/ut/driver_auto_test.cc @@ -1,15 +1,15 @@ #include #include +#include +#include +#include + extern "C" { +#include "adapter/driver/cache.h" +#include "base/group.h" +#include "driver_auto_helpers.h" #include "errcodes.h" -#include "type.h" -int is_value_in_range(neu_type_e, int64_t, double, neu_type_e, double); -int auto_test_prepare(neu_type_e, neu_dvalue_t *, double, double, const char *, - int, bool *, neu_type_e *); -void auto_test_read(neu_dvalue_t, double, const char *, int, neu_dvalue_t *, - neu_type_e *); -int auto_test_metadata(neu_type_e, bool, int *, neu_type_e *); } TEST(DriverAuto, UnresolvedWrite) @@ -222,6 +222,262 @@ static void expect_attrs(const neu_tag_sample_attrs_t &expected, EXPECT_EQ(expected.has_quality ? expected.quality : 0, actual.quality); } +TEST(DriverAuto, QueryAndReadUnresolvedOrMissingCache) +{ + for (bool paginate : { false, true }) { + for (bool missing : { false, true }) { + for (neu_type_e config : { NEU_TYPE_AUTO, NEU_TYPE_INT16 }) { + auto_test_snapshot_case_t input = {}; + input.configured_type = config; + input.sample.type = NEU_TYPE_ERROR; + input.sample.value.i32 = NEU_ERR_PLUGIN_TAG_NOT_READY; + input.missing_cache = missing; + auto_test_snapshot_result_t output = {}; + auto_test_snapshot(&input, paginate, &output); + EXPECT_EQ(config, output.configured_type); + EXPECT_EQ(config, output.listed_type); + EXPECT_EQ(config, output.read_type); + EXPECT_EQ(1u, output.read_count); + EXPECT_EQ(NEU_TYPE_ERROR, output.value.type); + } + } + } +} + +TEST(DriverAuto, QueryAndReadRetainResolvedTypeOnError) +{ + for (bool paginate : { false, true }) { + for (bool mismatch : { false, true }) { + auto_test_snapshot_case_t input = {}; + input.learned_type = NEU_TYPE_INT16; + input.sample.type = mismatch ? NEU_TYPE_DOUBLE : NEU_TYPE_ERROR; + if (mismatch) { + input.sample.value.d64 = 1.5; + } else { + input.sample.value.i32 = NEU_ERR_PLUGIN_READ_FAILURE; + } + auto_test_snapshot_result_t output = {}; + auto_test_snapshot(&input, paginate, &output); + EXPECT_EQ(NEU_TYPE_AUTO, output.configured_type); + EXPECT_EQ(NEU_TYPE_INT16, output.listed_type); + EXPECT_EQ(NEU_TYPE_INT16, output.read_type); + EXPECT_EQ(NEU_TYPE_ERROR, output.value.type); + EXPECT_EQ(mismatch ? NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH + : NEU_ERR_PLUGIN_READ_FAILURE, + output.value.value.i32); + } + } +} + +TEST(DriverAuto, QueryAndReadPendingAndExpiredTypes) +{ + for (bool paginate : { false, true }) { + for (bool pending : { false, true }) { + auto_test_snapshot_case_t input = {}; + input.learned_type = NEU_TYPE_INT16; + input.sample.type = NEU_TYPE_INT16; + input.sample.value.i16 = 123; + input.pending = pending; + input.expired = true; + auto_test_snapshot_result_t output = {}; + auto_test_snapshot(&input, paginate, &output); + EXPECT_EQ(pending ? NEU_TYPE_AUTO : NEU_TYPE_INT16, + output.listed_type); + EXPECT_EQ(pending ? NEU_TYPE_AUTO : NEU_TYPE_INT16, + output.read_type); + EXPECT_EQ(NEU_TYPE_ERROR, output.value.type); + EXPECT_EQ(NEU_ERR_PLUGIN_TAG_VALUE_EXPIRED, output.value.value.i32); + } + } +} + +TEST(DriverAuto, QueryAndReadSemanticPointerTypes) +{ + for (bool paginate : { false, true }) { + for (neu_type_e type : { NEU_TYPE_STRING, NEU_TYPE_BYTES }) { + char text[] = "hello"; + auto_test_snapshot_case_t input = {}; + input.sample.type = NEU_TYPE_PTR; + input.sample.value.ptr.type = type; + input.sample.value.ptr.length = sizeof(text); + input.sample.value.ptr.ptr = reinterpret_cast(text); + auto_test_snapshot_result_t output = {}; + auto_test_snapshot(&input, paginate, &output); + EXPECT_EQ(type, output.listed_type); + EXPECT_EQ(type, output.read_type); + ASSERT_EQ(NEU_TYPE_PTR, output.value.type); + EXPECT_NE(input.sample.value.ptr.ptr, output.value.value.ptr.ptr); + EXPECT_EQ(0, + memcmp(text, output.value.value.ptr.ptr, sizeof(text))); + free(output.value.value.ptr.ptr); + } + } +} + +TEST(DriverAuto, SyntheticReadErrorsKeepKnownTypes) +{ + for (bool paginate : { false, true }) { + for (neu_type_e config : { NEU_TYPE_AUTO, NEU_TYPE_INT16 }) { + for (int error : { NEU_ERR_PLUGIN_NOT_RUNNING, + NEU_ERR_PLUGIN_NOT_SUPPORT_READ_SYNC }) { + for (bool pending : { false, true }) { + auto_test_snapshot_case_t input = {}; + input.configured_type = config; + input.learned_type = NEU_TYPE_INT16; + input.sample.type = NEU_TYPE_ERROR; + input.sample.value.i32 = NEU_ERR_PLUGIN_READ_FAILURE; + input.pending = pending; + input.error = error; + auto_test_snapshot_result_t output = {}; + auto_test_snapshot(&input, paginate, &output); + neu_type_e expected = config == NEU_TYPE_AUTO && pending + ? NEU_TYPE_AUTO + : NEU_TYPE_INT16; + EXPECT_EQ(expected, output.listed_type); + EXPECT_EQ(expected, output.read_type); + EXPECT_EQ(NEU_TYPE_ERROR, output.value.type); + EXPECT_EQ(error, output.value.value.i32); + } + } + } + } +} + +TEST(DriverAuto, WriteOnlyPaginationUsesConfiguredType) +{ + for (bool paginate : { false, true }) { + auto_test_snapshot_case_t input = {}; + input.configured_type = NEU_TYPE_INT32; + input.write_only = true; + input.missing_cache = true; + auto_test_snapshot_result_t output = {}; + auto_test_snapshot(&input, paginate, &output); + EXPECT_EQ(NEU_TYPE_INT32, output.listed_type); + EXPECT_EQ(paginate ? 1u : 0u, output.read_count); + if (paginate) { + EXPECT_EQ(NEU_TYPE_INT32, output.read_type); + EXPECT_EQ(NEU_TYPE_STRING, output.value.type); + } + } +} + +TEST(DriverAuto, GroupSnapshotDoesNotObservePartialRebuild) +{ + struct Context { + neu_driver_cache_t * cache; + int64_t applied = 0; + std::mutex mutex; + std::condition_variable changed; + bool rebuilding = false; + bool query_started = false; + bool release = false; + bool complete = false; + bool observed_complete = false; + neu_type_e first = NEU_TYPE_AUTO; + neu_type_e second = NEU_TYPE_AUTO; + } ctx; + ctx.cache = neu_driver_cache_new(); + neu_group_t * group = neu_group_new("g", 100); + neu_datatag_t tag = {}; + tag.name = const_cast("a"); + tag.address = const_cast("sample"); + tag.type = NEU_TYPE_AUTO; + tag.attribute = NEU_ATTRIBUTE_READ; + ASSERT_EQ(0, neu_group_add_tag(group, &tag)); + neu_dvalue_t initial = {}; + initial.type = NEU_TYPE_ERROR; + initial.value.i32 = NEU_ERR_PLUGIN_TAG_NOT_READY; + neu_driver_cache_add_tag(ctx.cache, "g", &tag, initial); + neu_dvalue_t sample = {}; + sample.type = NEU_TYPE_INT16; + sample.value.i16 = 123; + neu_driver_cache_update(ctx.cache, "g", "a", 100, sample, nullptr, 0); + tag.name = const_cast("b"); + ASSERT_EQ(0, neu_group_add_tag(group, &tag)); + + auto inspect = [](void *arg, int64_t timestamp, uint32_t, UT_array *tags, + int) { + auto *state = static_cast(arg); + { + std::lock_guard lock(state->mutex); + state->observed_complete = state->complete; + } + for (unsigned i = 0; i < utarray_len(tags); ++i) { + auto *item = static_cast(utarray_eltptr(tags, i)); + neu_type_e type = NEU_TYPE_AUTO; + uint64_t generation = 0; + if (timestamp == state->applied) { + neu_driver_cache_type_get(state->cache, "g", item->name, &type, + &generation); + } + (strcmp(item->name, "a") == 0 ? state->first : state->second) = + type; + } + }; + neu_group_tag_query_t query = {}; + neu_group_query_snapshot(group, &query, &ctx, inspect); + EXPECT_EQ(NEU_TYPE_AUTO, ctx.first); + EXPECT_EQ(NEU_TYPE_AUTO, ctx.second); + + std::thread rebuild([&] { + neu_group_change_test( + group, 0, &ctx, + [](void *arg, int64_t timestamp, UT_array *tags, uint32_t) { + auto *state = static_cast(arg); + state->applied = timestamp; + { + std::unique_lock lock(state->mutex); + state->rebuilding = true; + state->changed.notify_all(); + state->changed.wait(lock, [&] { return state->release; }); + } + for (unsigned i = 0; i < utarray_len(tags); ++i) { + auto *item = + static_cast(utarray_eltptr(tags, i)); + neu_dvalue_t value = {}; + value.type = NEU_TYPE_ERROR; + value.value.i32 = NEU_ERR_PLUGIN_TAG_NOT_READY; + neu_driver_cache_add_tag(state->cache, "g", item, value); + } + neu_dvalue_t value = {}; + value.type = NEU_TYPE_DOUBLE; + value.value.d64 = 1.5; + neu_driver_cache_update(state->cache, "g", "a", 100, value, + nullptr, 0); + { + std::lock_guard lock(state->mutex); + state->complete = true; + } + utarray_free(tags); + }); + }); + { + std::unique_lock lock(ctx.mutex); + ctx.changed.wait(lock, [&] { return ctx.rebuilding; }); + } + std::thread reader([&] { + { + std::lock_guard lock(ctx.mutex); + ctx.query_started = true; + ctx.changed.notify_all(); + } + neu_group_query_snapshot(group, &query, &ctx, inspect); + }); + { + std::unique_lock lock(ctx.mutex); + ctx.changed.wait(lock, [&] { return ctx.query_started; }); + ctx.release = true; + ctx.changed.notify_all(); + } + rebuild.join(); + reader.join(); + EXPECT_TRUE(ctx.observed_complete); + EXPECT_EQ(NEU_TYPE_DOUBLE, ctx.first); + EXPECT_EQ(NEU_TYPE_AUTO, ctx.second); + neu_group_destroy(group); + neu_driver_cache_destroy(ctx.cache); +} + TEST(DriverAuto, ReadPathsScaleWithoutChangingConfig) { for (int path : { 0, 1, 2 }) { @@ -230,8 +486,10 @@ TEST(DriverAuto, ReadPathsScaleWithoutChangingConfig) sample.value.i16 = -246; sample.attrs = sample_attrs(); neu_dvalue_t output = {}; - neu_type_e config; - auto_test_read(sample, 0.5, "sample", path, &output, &config); + neu_type_e config, resolved; + auto_test_read(sample, 0.5, "sample", path, &output, &config, + &resolved); + EXPECT_EQ(NEU_TYPE_INT16, resolved); EXPECT_EQ(NEU_TYPE_AUTO, config); EXPECT_EQ(NEU_TYPE_DOUBLE, output.type); EXPECT_DOUBLE_EQ(-123, output.value.d64); @@ -247,8 +505,10 @@ TEST(DriverAuto, ReadPathsResolveEndian) sample.value.u16 = htons(123); sample.attrs = sample_attrs(); neu_dvalue_t output = {}; - neu_type_e config; - auto_test_read(sample, 0, "sample#B", path, &output, &config); + neu_type_e config, resolved; + auto_test_read(sample, 0, "sample#B", path, &output, &config, + &resolved); + EXPECT_EQ(NEU_TYPE_INT16, resolved); EXPECT_EQ(NEU_TYPE_AUTO, config); EXPECT_EQ(NEU_TYPE_INT16, output.type); EXPECT_EQ(123, output.value.i16); @@ -283,8 +543,13 @@ TEST(DriverAuto, ReadPathsPreserveOptionalTimestamps) sample.value.i16 = 123; } neu_dvalue_t output = {}; - neu_type_e config; - auto_test_read(sample, 0, "sample", path, &output, &config); + neu_type_e config, resolved; + auto_test_read(sample, 0, "sample", path, &output, &config, + &resolved); + EXPECT_EQ(type == NEU_TYPE_PTR + ? NEU_TYPE_STRING + : type == NEU_TYPE_ERROR ? NEU_TYPE_AUTO : type, + resolved); EXPECT_EQ(NEU_TYPE_AUTO, config); EXPECT_EQ(type, output.type); expect_attrs(sample.attrs, output.attrs); diff --git a/tests/ut/http_read_attrs_test.cc b/tests/ut/http_read_attrs_test.cc index b5be5e690..06fa314ff 100644 --- a/tests/ut/http_read_attrs_test.cc +++ b/tests/ut/http_read_attrs_test.cc @@ -1,21 +1,37 @@ #include +#include +#include #include #include #include +#include extern "C" { #include "msg.h" +#include "parser/neu_json_tag.h" #include "restful/read_response.h" #include "utils/log.h" +#include "json/neu_json_fn.h" zlog_category_t *neuron = nullptr; } +static_assert(offsetof(neu_resp_tag_info_t, tag) == 0, + "GET-tag items must support configuration tag comparators"); +static_assert(sizeof(neu_resp_tag_value_meta_t) == + sizeof(neu_resp_tag_value_meta_paginate_t), + "Read items must retain their common layout"); +static_assert(offsetof(neu_resp_tag_value_meta_t, resolved_type) == + offsetof(neu_resp_tag_value_meta_paginate_t, resolved_type), + "Read resolved types must have a common offset"); + using Json = std::unique_ptr; static Json encode_sample(bool paginate, neu_tag_sample_attrs_t attrs, - bool error = false, bool legacy = false) + bool error = false, bool legacy = false, + neu_type_e resolved_type = NEU_TYPE_AUTO, + neu_type_e configured_type = NEU_TYPE_AUTO) { UT_array *source; UT_icd icd = { paginate ? sizeof(neu_resp_tag_value_meta_paginate_t) @@ -25,19 +41,26 @@ static Json encode_sample(bool paginate, neu_tag_sample_attrs_t attrs, neu_resp_tag_value_meta_t tag = {}; neu_resp_tag_value_meta_paginate_t paginated_tag = {}; tag.value.attrs = paginated_tag.value.attrs = attrs; + tag.value.type = paginated_tag.value.type = + error ? NEU_TYPE_ERROR : NEU_TYPE_DOUBLE; + tag.resolved_type = paginated_tag.resolved_type = resolved_type; + tag.datatag.type = paginated_tag.datatag.type = configured_type; if (paginate) { utarray_push_back(source, &paginated_tag); } else { utarray_push_back(source, &tag); } - neu_json_tag_meta_t metas[2] = {}; + neu_json_tag_meta_t metas[3] = {}; metas[0].name = const_cast("timestamp"); metas[0].t = NEU_JSON_INT; metas[0].value.val_int = 11; metas[1].name = const_cast("quality"); metas[1].t = NEU_JSON_INT; metas[1].value.val_int = 22; + metas[2].name = const_cast("resolved_type"); + metas[2].t = NEU_JSON_INT; + metas[2].value.val_int = NEU_TYPE_ERROR; neu_json_read_resp_tag_t item = {}; neu_json_read_paginate_resp_tag_t page_item = {}; @@ -46,7 +69,8 @@ static Json encode_sample(bool paginate, neu_tag_sample_attrs_t attrs, item.t = page_item.t = NEU_JSON_DOUBLE; item.value.val_double = page_item.value.val_double = 12.5; item.precision = page_item.precision = 2; - item.n_meta = page_item.n_meta = legacy ? 2 : 0; + item.n_meta = page_item.n_meta = legacy ? 3 : 0; + item.datatag.type = page_item.datatag.type = configured_type; item.metas = page_item.metas = metas; neu_json_read_resp_t response = { 1, &item }; neu_json_read_paginate_resp_t page = {}; @@ -153,6 +177,16 @@ TEST(HttpReadAttrs, LegacyMetadataCollisionAndNesting) json_integer_value(json_object_get(legacy, "timestamp"))); EXPECT_EQ(paginate || !present ? 22 : 192, json_integer_value(json_object_get(legacy, "quality"))); + ASSERT_TRUE( + json_is_integer(json_object_get(item, "resolved_type"))); + EXPECT_EQ( + NEU_TYPE_AUTO, + json_integer_value(json_object_get(item, "resolved_type"))); + if (paginate) { + EXPECT_EQ(NEU_TYPE_ERROR, + json_integer_value( + json_object_get(legacy, "resolved_type"))); + } if (paginate) { EXPECT_EQ(present, json_object_get(item, "timestamp") != nullptr); @@ -179,6 +213,9 @@ TEST(HttpReadAttrs, FilteredPageUsesSourceOffset) neu_resp_tag_value_meta_paginate_t tag = {}; tag.value.attrs.has_quality = true; tag.value.attrs.quality = 100 + i; + tag.value.type = NEU_TYPE_ERROR; + tag.resolved_type = + i == 0 ? NEU_TYPE_BOOL : i == 1 ? NEU_TYPE_AUTO : NEU_TYPE_INT16; utarray_push_back(source, &tag); } neu_json_read_paginate_resp_tag_t items[2] = {}; @@ -200,6 +237,13 @@ TEST(HttpReadAttrs, FilteredPageUsesSourceOffset) EXPECT_EQ(102, json_integer_value( json_object_get(json_array_get(output, 1), "quality"))); + for (int i = 0; i < 2; i++) { + auto item = json_array_get(output, i); + ASSERT_TRUE(json_is_integer(json_object_get(item, "resolved_type"))); + EXPECT_EQ(i == 0 ? NEU_TYPE_AUTO : NEU_TYPE_INT16, + json_integer_value(json_object_get(item, "resolved_type"))); + EXPECT_EQ(3002, json_integer_value(json_object_get(item, "error"))); + } utarray_free(source); } @@ -253,3 +297,395 @@ TEST(HttpReadAttrs, EmptyPagesAndInvalidSourceBounds) EXPECT_EQ(-1, neu_http_encode_read_response(root.get(), &context)); } } + +TEST(HttpReadResolvedType, + RejectsNullOutputAndOffsetBoundsWithConfigurationCleanup) +{ + for (bool paginate : { false, true }) { + UT_array *source; + utarray_new(source, + paginate ? neu_resp_tag_value_meta_paginate_icd() + : neu_resp_tag_value_meta_icd()); + neu_resp_tag_value_meta_t read_tag = {}; + neu_resp_tag_value_meta_paginate_t page_tag = {}; + for (int i = 0; i < 2; i++) { + if (paginate) { + utarray_push_back(source, &page_tag); + } else { + utarray_push_back(source, &read_tag); + } + } + neu_json_read_resp_t read_response = {}; + neu_json_read_paginate_resp_t page_response = {}; + read_response.n_tag = page_response.n_tag = 1; + neu_http_read_response_t context = { + paginate ? static_cast(&page_response) + : static_cast(&read_response), + source, 0, paginate + }; + Json root(json_object(), json_decref); + EXPECT_EQ(-1, neu_http_encode_read_response(root.get(), &context)); + for (unsigned int offset : + { 1u, 2u, 3u, std::numeric_limits::max() }) { + neu_json_read_resp_tag_t read_items[2] = {}; + neu_json_read_paginate_resp_tag_t page_items[2] = {}; + int count = offset == 1 ? 2 : 1; + for (int i = 0; i < count && paginate; i++) { + page_items[i].datatag.name = strdup("a"); + page_items[i].datatag.address = strdup("source"); + page_items[i].datatag.description = strdup("description"); + page_items[i].datatag.unit = strdup("C"); + } + read_response.n_tag = page_response.n_tag = count; + read_response.tags = read_items; + page_response.tags = page_items; + context.offset = offset; + EXPECT_EQ(-1, neu_http_encode_read_response(root.get(), &context)); + } + utarray_free(source); + } +} + +TEST(HttpReadResolvedType, AuthoritativeTypeIsIndependentOfOutputAndErrors) +{ + const struct { + neu_type_e configured; + neu_type_e resolved; + } cases[] = { + { NEU_TYPE_AUTO, NEU_TYPE_AUTO }, { NEU_TYPE_AUTO, NEU_TYPE_INT16 }, + { NEU_TYPE_AUTO, NEU_TYPE_STRING }, { NEU_TYPE_AUTO, NEU_TYPE_BYTES }, + { NEU_TYPE_INT32, NEU_TYPE_INT32 }, { NEU_TYPE_BOOL, NEU_TYPE_BOOL }, + }; + for (bool paginate : { false, true }) { + for (bool error : { false, true }) { + for (const auto &test : cases) { + SCOPED_TRACE(::testing::Message() << paginate << ":" << error + << ":" << test.resolved); + auto root = encode_sample(paginate, {}, error, true, + test.resolved, test.configured); + auto item = first_item(root, paginate); + auto type = json_object_get(item, "resolved_type"); + ASSERT_TRUE(json_is_integer(type)); + EXPECT_EQ(test.resolved, json_integer_value(type)); + if (paginate) { + EXPECT_EQ( + test.configured, + json_integer_value(json_object_get(item, "type"))); + EXPECT_EQ(NEU_TYPE_ERROR, + json_integer_value(json_object_get( + json_object_get(item, "attributes"), + "resolved_type"))); + } else { + EXPECT_EQ(nullptr, json_object_get(item, "type")); + } + if (error) { + EXPECT_EQ( + 3002, + json_integer_value(json_object_get(item, "error"))); + EXPECT_EQ(nullptr, json_object_get(item, "value")); + } else { + EXPECT_DOUBLE_EQ( + 12.5, + json_number_value(json_object_get(item, "value"))); + } + } + } + } +} + +static void free_tags(UT_array *tags) +{ + utarray_free(tags); +} + +using Tags = std::unique_ptr; + +static Tags new_tags(UT_icd *icd) +{ + UT_array *tags; + utarray_new(tags, icd); + return Tags(tags, free_tags); +} + +static neu_json_tag_t configuration_tag(const neu_datatag_t &tag) +{ + neu_json_tag_t result = {}; + result.name = tag.name; + result.address = tag.address; + result.description = tag.description; + result.unit = tag.unit; + result.type = tag.type; + result.attribute = tag.attribute; + result.precision = tag.precision; + result.decimal = tag.decimal; + result.bias = tag.bias; + result.t = NEU_JSON_UNDEFINE; + return result; +} + +TEST(HttpGetTagsResolvedType, + SortedMixedItemsRetainConfigurationAndRuntimeTypes) +{ + auto source = new_tags(neu_resp_tag_info_icd()); + const char * names[] = { "pending", "scaled", "fixed", "bytes" }; + const char * units[] = { "C", "m", "V", "B" }; + const neu_type_e configured[] = { NEU_TYPE_AUTO, NEU_TYPE_AUTO, + NEU_TYPE_BOOL, NEU_TYPE_AUTO }; + const neu_type_e resolved[] = { NEU_TYPE_AUTO, NEU_TYPE_INT16, + NEU_TYPE_BOOL, NEU_TYPE_BYTES }; + for (int i = 0; i < 4; i++) { + neu_resp_tag_info_t item = {}; + item.tag.name = const_cast(names[i]); + item.tag.address = const_cast(names[3 - i]); + item.tag.description = const_cast(names[i]); + item.tag.unit = const_cast(units[i]); + item.tag.type = configured[i]; + item.tag.attribute = NEU_ATTRIBUTE_READ; + item.tag.precision = i; + item.tag.decimal = i * 0.5; + item.tag.bias = i; + item.resolved_type = resolved[i]; + utarray_push_back(source.get(), &item); + } + for (const char *sort : { "+name", "-name", "+type", "-type", "+address", + "-address", "+unit", "-unit" }) { + SCOPED_TRACE(sort); + auto cmp = neu_tag_get_sort_cmp(sort); + ASSERT_NE(nullptr, cmp); + utarray_sort(source.get(), cmp); + std::vector items; + utarray_foreach(source.get(), neu_resp_tag_info_t *, item) + { + items.push_back(configuration_tag(item->tag)); + } + neu_json_get_tags_resp_t response = { static_cast(items.size()), + items.data() }; + neu_http_get_tags_response_t context = { &response, source.get() }; + char * encoded = nullptr; + ASSERT_EQ(0, + neu_json_encode_by_fn( + &context, neu_http_encode_get_tags_response, &encoded)); + ASSERT_NE(nullptr, encoded); + Json root(json_loads(encoded, JSON_REJECT_DUPLICATES, nullptr), + json_decref); + free(encoded); + ASSERT_NE(nullptr, root.get()); + Json exported(json_object(), json_decref); + ASSERT_EQ(0, neu_json_encode_get_tags_resp(exported.get(), &response)); + auto output = json_object_get(root.get(), "tags"); + ASSERT_EQ(4u, json_array_size(output)); + for (unsigned int i = 0; i < 4; i++) { + auto item = json_array_get(output, i); + auto original = static_cast( + utarray_eltptr(source.get(), i)); + const char *name = json_string_value(json_object_get(item, "name")); + ASSERT_NE(nullptr, name); + int index = 0; + while (index < 4 && strcmp(names[index], name) != 0) { + index++; + } + ASSERT_LT(index, 4); + EXPECT_STREQ(original->tag.name, name); + ASSERT_TRUE( + json_is_integer(json_object_get(item, "resolved_type"))); + EXPECT_EQ( + resolved[index], + json_integer_value(json_object_get(item, "resolved_type"))); + EXPECT_EQ(configured[index], + json_integer_value(json_object_get(item, "type"))); + EXPECT_STREQ(units[index], + json_string_value(json_object_get(item, "unit"))); + auto config = + json_array_get(json_object_get(exported.get(), "tags"), i); + EXPECT_EQ(nullptr, json_object_get(config, "resolved_type")); + EXPECT_EQ(configured[index], + json_integer_value(json_object_get(config, "type"))); + EXPECT_STREQ(units[index], + json_string_value(json_object_get(config, "unit"))); + if (i > 0) { + EXPECT_LE(cmp(utarray_eltptr(source.get(), i - 1), original), + 0); + } + } + } +} + +TEST(HttpGetTagsResolvedType, CombinedItemsDeepCopyAndMoveExportStrings) +{ + neu_resp_tag_info_t original = {}; + original.tag.name = strdup("auto"); + original.tag.address = strdup("source"); + original.tag.description = strdup("description"); + original.tag.unit = strdup("C"); + original.tag.type = NEU_TYPE_AUTO; + original.tag.decimal = 0.5; + original.resolved_type = NEU_TYPE_INT16; + auto source = new_tags(neu_resp_tag_info_icd()); + utarray_push_back(source.get(), &original); + auto copy = static_cast(utarray_front(source.get())); + EXPECT_NE(original.tag.name, copy->tag.name); + EXPECT_NE(original.tag.address, copy->tag.address); + EXPECT_NE(original.tag.description, copy->tag.description); + EXPECT_NE(original.tag.unit, copy->tag.unit); + neu_tag_fini(&original.tag); + + auto second = new_tags(neu_resp_tag_info_icd()); + utarray_push_back(second.get(), copy); + auto moved = + static_cast(utarray_front(second.get())); + EXPECT_NE(copy->tag.name, moved->tag.name); + EXPECT_NE(copy->tag.address, moved->tag.address); + EXPECT_NE(copy->tag.description, moved->tag.description); + EXPECT_NE(copy->tag.unit, moved->tag.unit); + source.reset(); + EXPECT_EQ(NEU_TYPE_INT16, moved->resolved_type); + EXPECT_EQ(NEU_TYPE_AUTO, moved->tag.type); + EXPECT_DOUBLE_EQ(0.5, moved->tag.decimal); + neu_json_tag_t config = configuration_tag(moved->tag); + moved->tag.name = moved->tag.address = moved->tag.description = + moved->tag.unit = nullptr; + second.reset(); + EXPECT_STREQ("auto", config.name); + EXPECT_STREQ("source", config.address); + EXPECT_STREQ("description", config.description); + EXPECT_STREQ("C", config.unit); + Json root(json_object(), json_decref); + EXPECT_EQ(0, neu_json_encode_tag(root.get(), &config)); + EXPECT_EQ(nullptr, json_object_get(root.get(), "resolved_type")); + EXPECT_EQ(NEU_TYPE_AUTO, + json_integer_value(json_object_get(root.get(), "type"))); + EXPECT_STREQ("C", json_string_value(json_object_get(root.get(), "unit"))); + neu_json_decode_tag_fini(&config); +} + +TEST(HttpGetTagsResolvedType, EmptyItemsAndInvalidContextsAndBounds) +{ + Json root(json_object(), json_decref); + EXPECT_EQ(-1, neu_http_encode_get_tags_response(root.get(), nullptr)); + neu_http_get_tags_response_t context = {}; + EXPECT_EQ(-1, neu_http_encode_get_tags_response(root.get(), &context)); + neu_json_get_tags_resp_t response = {}; + context.response = &response; + EXPECT_EQ(0, neu_http_encode_get_tags_response(root.get(), &context)); + EXPECT_EQ(0u, json_array_size(json_object_get(root.get(), "tags"))); + response.n_tag = 1; + EXPECT_EQ(-1, neu_http_encode_get_tags_response(root.get(), &context)); + response.n_tag = -1; + EXPECT_EQ(-1, neu_http_encode_get_tags_response(root.get(), &context)); + response.n_tag = 0; + auto source = new_tags(neu_resp_tag_info_icd()); + neu_resp_tag_info_t tag = {}; + utarray_push_back(source.get(), &tag); + context.tags = source.get(); + EXPECT_EQ(-1, neu_http_encode_get_tags_response(root.get(), &context)); + response.n_tag = 2; + EXPECT_EQ(-1, neu_http_encode_get_tags_response(root.get(), &context)); + response.n_tag = 1; + EXPECT_EQ(-1, neu_http_encode_get_tags_response(root.get(), &context)); + neu_json_tag_t item = {}; + response.tags = &item; + EXPECT_EQ(-1, neu_http_encode_get_tags_response(nullptr, &context)); + Json array(json_array(), json_decref); + EXPECT_EQ(-1, neu_http_encode_get_tags_response(array.get(), &context)); + EXPECT_EQ(0, neu_http_encode_get_tags_response(root.get(), &context)); + EXPECT_TRUE(json_is_integer( + json_object_get(json_array_get(json_object_get(root.get(), "tags"), 0), + "resolved_type"))); + EXPECT_EQ(1u, utarray_len(source.get())); + EXPECT_NE(nullptr, neu_tag_get_sort_cmp("+name")); + EXPECT_EQ(nullptr, neu_tag_get_sort_cmp("+resolved_type")); +} + +static json_malloc_t saved_json_malloc; +static json_free_t saved_json_free; +static size_t json_allocation_count; +static size_t json_fail_at; + +static void *probe_json_malloc(size_t size) +{ + json_allocation_count++; + return json_fail_at == json_allocation_count ? nullptr + : saved_json_malloc(size); +} + +class JsonAllocationScope { +public: + explicit JsonAllocationScope(size_t fail_at = 0) + { + json_get_alloc_funcs(&saved_json_malloc, &saved_json_free); + json_allocation_count = 0; + json_fail_at = fail_at; + json_set_alloc_funcs(probe_json_malloc, saved_json_free); + } + ~JsonAllocationScope() + { + json_set_alloc_funcs(saved_json_malloc, saved_json_free); + } +}; + +TEST(HttpResolvedType, InsertionFailureReleasesDomAndPreservesSourceOwnership) +{ + for (int mode = 0; mode < 3; mode++) { + SCOPED_TRACE(mode); + auto source = new_tags(mode == 2 ? neu_resp_tag_info_icd() + : mode == 1 + ? neu_resp_tag_value_meta_paginate_icd() + : neu_resp_tag_value_meta_icd()); + neu_resp_tag_info_t get_tag = {}; + get_tag.tag.name = const_cast("a"); + get_tag.resolved_type = NEU_TYPE_INT16; + neu_resp_tag_value_meta_t read_tag = {}; + read_tag.resolved_type = NEU_TYPE_INT16; + neu_resp_tag_value_meta_paginate_t page_tag = {}; + page_tag.resolved_type = NEU_TYPE_INT16; + const void *tag = mode == 2 + ? static_cast(&get_tag) + : mode == 1 ? static_cast(&page_tag) + : static_cast(&read_tag); + utarray_push_back(source.get(), tag); + neu_json_tag_t get_item = configuration_tag(get_tag.tag); + neu_json_get_tags_resp_t get_response = { 1, &get_item }; + neu_http_get_tags_response_t get_context = { &get_response, + source.get() }; + neu_json_read_resp_tag_t read_item = {}; + neu_json_read_resp_t read_response = { 1, &read_item }; + neu_json_read_paginate_resp_tag_t page_item = {}; + neu_json_read_paginate_resp_t page_response = {}; + page_response.n_tag = 1; + page_response.tags = &page_item; + neu_http_read_response_t read_context = { + mode == 1 ? static_cast(&page_response) + : static_cast(&read_response), + source.get(), 0, mode == 1 + }; + auto encode = mode == 2 ? neu_http_encode_get_tags_response + : neu_http_encode_read_response; + void *context = mode == 2 ? static_cast(&get_context) + : static_cast(&read_context); + Json baseline(json_object(), json_decref); + size_t allocations = 0; + { + JsonAllocationScope probe; + EXPECT_EQ(0, encode(baseline.get(), context)); + allocations = json_allocation_count; + } + ASSERT_GT(allocations, 0u); + Json failed(json_object(), json_decref); + { + JsonAllocationScope probe(allocations); + EXPECT_EQ(-1, encode(failed.get(), context)); + } + EXPECT_EQ(1u, utarray_len(source.get())); + if (mode == 2) { + Json failed_array(json_object(), json_decref); + { + JsonAllocationScope probe(1); + EXPECT_EQ(-1, encode(failed_array.get(), context)); + } + auto item = + static_cast(utarray_front(source.get())); + EXPECT_STREQ("a", item->tag.name); + EXPECT_EQ(NEU_TYPE_INT16, item->resolved_type); + } + } +}