Compare commits

...
99 Commits
Author SHA1 Message Date
fengzeroz bca973293d Merge pull request #1866 from fengzeroz/v2.7
version: 2.7.4
2024-02-29 15:57:15 +08:00
fengzero 1ef579ae45 version: 2.7.4 2024-02-29 07:56:28 +00:00
fengzeroz cd0d26c110 Merge pull request #1864 from ZhouBox/v2.7
[v2.7]remove the SIGSEGV signal handler
2024-02-27 14:49:25 +08:00
xiang.zhou b77ba27b24 remove the SIGSEGV signal handler 2024-02-27 14:35:16 +08:00
fengzeroz 7055ac2fb8 Merge pull request #1862 from eeff/v2.7
Pick to v2.7 chore: remove useless code
2024-02-27 10:28:26 +08:00
eeff d3358cfb6f chore: remove useless code 2024-02-26 23:30:48 +08:00
fengzeroz e90c1352d7 Merge pull request #1854 from ZhouBox/v2.7
[v2.7]riscv64:add build rule
2024-02-19 11:07:06 +08:00
xiang.zhou 0499f6050c riscv64:add build rule 2024-02-19 10:57:00 +08:00
fengzeroz 6424cfc33b Merge pull request #1850 from eeff/v2.7
Pick to v2.7: feat(ekuiper): add rolling metrics
2024-02-19 10:40:32 +08:00
eeff 88fcb9dcff feat(ekuiper): add rolling metrics 2024-02-05 17:17:24 +08:00
eeff ed3fb2ef60 fix(restful): force clean stale rolling metrics 2024-02-05 17:17:24 +08:00
eeff 3d705aa772 refactor(metrics): add lock to avoid data race 2024-02-05 17:17:24 +08:00
eeff 31bd132398 fix: rolling counter integer overflow 2024-02-05 16:53:08 +08:00
fengzeroz bf8a1a7891 Merge pull request #1846 from eeff/v2.7
Pick to v2.7 feat(mqtt): add rolling metrics
2024-02-05 15:48:59 +08:00
eeff be831d7333 fix(metrics): correct registered rolling counter entry ref count 2024-02-04 18:40:08 +08:00
eeff 59f1f5d97d feat(mqtt): add rolling metrics 2024-02-04 15:56:21 +08:00
eeff 31ffde8406 feat(metrics): support distribution counter 2024-02-04 15:56:21 +08:00
eeff a03385af37 test: rolling counter unit tests 2024-02-04 15:56:19 +08:00
eeff 1f7e0ed700 feat: add rolling counter utility 2024-02-04 15:55:37 +08:00
fengzeroz c9531705bd Merge pull request #1845 from hxy7yx/v2.7-2
fix: tag name sql schema
2024-02-04 10:19:04 +08:00
hxy7yx 60f4d77501 fix: tag name sql schema 2024-02-02 17:54:00 +08:00
fengzeroz a71e2e3cbe Merge pull request #1843 from hxy7yx/v2.7-2
feat: add tag name sql schema
2024-02-02 17:14:41 +08:00
hxy7yx 776563a2ff feat: add tag name sql schema 2024-02-02 16:57:21 +08:00
fengzeroz 5360d20da1 Merge pull request #1840 from fengzeroz/v2.7
version: 2.7.3
2024-02-02 11:08:31 +08:00
fengzero 8be6cda1b1 version: 2.7.3 2024-02-02 03:07:40 +00:00
fengzeroz f28842224d Merge pull request #1836 from eeff/v2.7
Pick to v2.7 build: cmake link libatomic
2024-01-31 17:37:26 +08:00
eeff f8e879c989 build: cmake link libatomic 2024-01-31 17:32:22 +08:00
fengzeroz a8ce8c9da0 Merge pull request #1834 from hxy7yx/v2.7-1
update tag name length max to 128
2024-01-31 10:34:43 +08:00
hxy7yx ebc6a076fc update tag name length max to 128 2024-01-30 17:45:07 +08:00
fengzeroz bbad12cdd6 Merge pull request #1828 from ZhouBox/v2.7
[v2.7]judge cache type when report group
2024-01-29 14:37:10 +08:00
xiang.zhou 9614b509b1 feat:judge cache type when report group 2024-01-29 14:15:04 +08:00
fengzeroz 11db02ccf5 Merge pull request #1812 from hxy7yx/v2.7-1
fix(add gtag): memory overflow
2024-01-12 17:03:19 +08:00
hxy7yx c0dbdf39ff fix(add gtag): memory overflow 2024-01-12 16:55:32 +08:00
fengzeroz 10c89daf5e Merge pull request #1811 from fengzeroz/v2.7
plugin version: 2.7.2
2024-01-11 17:00:10 +08:00
fengzero 94e492652a plugin version: 2.7.2 2024-01-11 08:59:33 +00:00
fengzeroz 87aac219d7 Merge pull request #1810 from fengzeroz/v2.7
version: 2.7.2
2024-01-11 16:26:54 +08:00
fengzero 7fdd4c4f79 version: 2.7.2 2024-01-11 08:26:19 +00:00
fengzeroz 7c46a3ec06 Merge pull request #1809 from eeff/v2.7
Pick to v2.7 fix: enlarge plugin lib length
2024-01-11 16:06:15 +08:00
eeff 065d77dd5d fix: enlarge plugin lib length 2024-01-11 15:59:16 +08:00
fengzeroz 68d8632fcd Merge pull request #1806 from hxy7yx/v2.7-1
fix(add_gtag): parser memory free
2024-01-10 17:31:51 +08:00
hxy7yx 84ca0952e8 fix(add_gtag): payload parser memory free 2024-01-10 17:25:01 +08:00
fengzeroz f74d55a98c Merge pull request #1803 from fengzeroz/v2.7
version: 2.7.1
2024-01-05 10:02:33 +08:00
fengzero f0d20ec77b version: 2.7.1 2024-01-05 02:01:49 +00:00
fengzeroz 9693d50dd0 Merge pull request #1800 from fengzeroz/v2.7
version: 2.7.1
2024-01-05 09:48:49 +08:00
fengzero e34d7bf3e5 version: 2.7.1 2024-01-05 01:48:11 +00:00
fengzeroz 8fe1202d3f Merge pull request #1796 from hxy7yx/v2.7-2
limit the number of groups to a maximum of 512
2024-01-04 16:20:41 +08:00
hxy7yx 11672c62ee limit the number of groups to a maximum of 512 2024-01-04 16:04:42 +08:00
fengzeroz af8d29f9e9 Merge pull request #1798 from eeff/v2.7
Pick to v2.7 feat(core): get subscribe group request support filter
2024-01-04 14:53:56 +08:00
eeff 96fc6ebf6e test: get subscribe info ft 2024-01-04 14:23:15 +08:00
eeff b922905e39 feat(restful): get subscribe group api support filtering 2024-01-04 14:23:15 +08:00
eeff b49f39729a feat(core): get subscribe group request support filtering 2024-01-04 14:23:13 +08:00
fengzeroz 128bc2820b Merge pull request #1794 from fengzeroz/v2.7
get tags: query description
2024-01-04 10:21:51 +08:00
fengzero 5e33fc9d17 get tags: query description 2024-01-04 02:12:14 +00:00
fengzeroz b1d410e497 Merge pull request #1793 from eeff/v2.7
Pick to v2.7 feat(core): read group support filter
2024-01-03 17:46:56 +08:00
eeff 2c09db89c0 test: read group query ft 2024-01-03 17:32:15 +08:00
eeff 40025265f2 feat(restful): read group handler support query 2024-01-03 17:32:15 +08:00
eeff 725618d345 feat(core): read group support query matching 2024-01-03 17:32:14 +08:00
fengzeroz e61826de1c Merge pull request #1788 from hxy7yx/v2.7-1
increase the size of EVENT_SIZE
2024-01-03 10:03:55 +08:00
fengzeroz 4beaa8c113 Merge pull request #1786 from fengzeroz/v2.7
msg q
2024-01-03 10:03:23 +08:00
hxy7yx fbe5c2f42b increase the size of EVENT_SIZE 2024-01-02 18:06:01 +08:00
fengzero 6d90208a6e msg q 2023-12-29 06:53:26 +00:00
fengzeroz 9c7d95d3a1 Merge pull request #1783 from fengzeroz/v2.7
lock tag table
2023-12-29 11:06:37 +08:00
fengzero c2a59e9a23 lock tag table 2023-12-29 02:03:03 +00:00
fengzeroz 71227496ac Merge pull request #1779 from fengzeroz/v2.7
fix event leak
2023-12-25 10:55:21 +08:00
fengzero 52d15c31de event log 2023-12-25 02:46:30 +00:00
fengzero 3fb501630a fix(event): timer/io index set 2023-12-25 02:42:32 +00:00
fengzeroz 2b814b8fe2 Merge pull request #1776 from fengzeroz/v2.7
V2.7
2023-12-22 15:59:12 +08:00
fengzero 336563834b version: 2.7.0 2023-12-22 07:52:13 +00:00
hxy7yx 59e7ff3e60 fix(add_gtags):license tag count 2023-12-22 07:52:13 +00:00
hxy7yx b145220585 fix http status 2023-12-22 07:52:13 +00:00
hxy7yx 3e4489d2c2 fix:add_tags and add_gtags http resp code 2023-12-22 07:52:13 +00:00
fengzero 74dc6b9489 fix(conn): lock conn buf 2023-12-22 07:52:13 +00:00
xiang.zhou d6e94f7009 fix:add library update fail error code 2023-12-22 07:52:13 +00:00
fengzeroz 8930b16948 Merge pull request #1774 from eeff/v2.7
Pick to v2.7 fix(core): replace path separator char in adapter log category
2023-12-22 14:50:51 +08:00
eeff d95137d1a3 fix(core): replace path separator char in adapter log category 2023-12-22 14:47:21 +08:00
fengzeroz 21ff7aea65 Merge pull request #1754 from ZhouBox/v2.7
[v2.7]fixbug:no update plugin when update plugin library
2023-12-15 09:29:40 +08:00
xiang.zhou 58cc696e5e fixbug:no update plugin when update plugin library 2023-12-14 19:11:41 +08:00
fengzeroz 76ea51be2b Merge pull request #1747 from hxy7yx/v2.7-1
fix(NEU_REQ_ADD_GTAG): del added tags when add failed
2023-12-13 10:00:21 +08:00
hxy7yx b91b2ef55d fix(NEU_REQ_ADD_GTAG): del all tags when add failed 2023-12-12 15:53:03 +08:00
fengzeroz 8fc783e679 Merge pull request #1750 from fengzeroz/v2.7
event: log
2023-12-12 15:29:55 +08:00
fengzero 36e66e1521 event: log 2023-12-12 07:21:33 +00:00
fengzeroz 2ad3c5e87c Merge pull request #1744 from fengzeroz/v2.7
fix write tag in arm64
2023-12-12 10:30:56 +08:00
fengzero d51ef29568 fix write tag 2023-12-12 02:24:46 +00:00
fengzero b3d88977bf fix write tag in arm64 2023-12-11 10:34:05 +00:00
fengzeroz 330a58f825 Merge pull request #1740 from fengzeroz/v2.7
reg sigkill
2023-12-11 17:07:01 +08:00
fengzero c779d86da2 reg sigkill 2023-12-11 09:03:48 +00:00
fengzeroz 0b3075e94d Merge pull request #1735 from hxy7yx/v2.7-1
add ft case for negative decimal
2023-12-11 16:17:05 +08:00
fengzeroz 33d317551c Merge pull request #1737 from fengzeroz/v2.7
Revert "fix(conn): lock conn buf"
2023-12-11 15:50:15 +08:00
fengzero 1f019eb02b Revert "fix(conn): lock conn buf"
This reverts commit 1151bb43b8.
2023-12-11 07:48:16 +00:00
hxy7yx 2e407c3e6e add ft case for negative decimal 2023-12-11 14:42:58 +08:00
fengzeroz a4fb604db7 Merge pull request #1733 from fengzeroz/v2.7
ci: v2.7
2023-12-08 17:13:00 +08:00
fengzero 090650c5c4 ci: v2.7 2023-12-08 09:10:27 +00:00
fengzeroz d6c0c04feb Merge pull request #1732 from fengzeroz/v2.7
cmake v2.7
2023-12-08 17:06:49 +08:00
fengzero bf548c974f cmake v2.7 2023-12-08 09:06:06 +00:00
fengzeroz 54aa2a15b3 Merge pull request #1731 from fengzeroz/v2.7
V2.7
2023-12-08 16:55:37 +08:00
fengzero e3040ea184 rm codecov 2023-12-08 08:55:09 +00:00
fengzero 9d02f1faec branch v2.7 2023-12-08 08:55:09 +00:00
fengzeroz 34f64c832f Merge pull request #1730 from fengzeroz/v2.7
version: 2.7.0
2023-12-08 16:50:19 +08:00
fengzero 93b488a02c version: 2.7.0 2023-12-08 08:48:57 +00:00
60 changed files with 1758 additions and 431 deletions
+3 -32
View File
@@ -41,7 +41,7 @@ jobs:
ut:
runs-on: ubuntu-20.04
container: ghcr.io/neugates/build:x86_64-main
container: ghcr.io/neugates/build:x86_64-v2.7
steps:
@@ -74,17 +74,9 @@ jobs:
build/result/index.html
build/neuron.log
- name: Upload coverage to Codecov
uses: codecov/codecov-action@v3
with:
gcov: true
fail_ci_if_error: true
gcov_include: include plugins src
gcov_ignore: tests simulator
ft:
runs-on: ubuntu-20.04
container: ghcr.io/neugates/build:x86_64-main
container: ghcr.io/neugates/build:x86_64-v2.7
steps:
- uses: actions/checkout@v4
@@ -115,16 +107,9 @@ jobs:
ft/reports/report.html
build/neuron.log
- name: Upload coverage to Codecov
uses: codecov/codecov-action@v3
with:
gcov: true
gcov_include: include plugins src
gcov_ignore: tests simulator
pft:
runs-on: ubuntu-20.04
container: ghcr.io/neugates/build:x86_64-main
container: ghcr.io/neugates/build:x86_64-v2.7
strategy:
matrix:
plugin: [core, modbus, mqtt, metrics]
@@ -163,17 +148,3 @@ jobs:
else
pytest -s -v tests/ft/metrics/"test_metrics.py"
fi
- name: create cov report
run: |
sudo apt-get update
sudo apt-get -y install lcov
mkdir -p cov_report
./create_cov_report.sh ${{ matrix.plugin }}
- name: Upload coverage to Codecov
uses: codecov/codecov-action@v3
with:
verbose: true
directory: ./cov_report/
files: cov-${{ matrix.plugin }}.info
+5 -5
View File
@@ -8,9 +8,9 @@ on:
- published
env:
VERSION: 2.7.0
UI_VERSION: 2.6.1
BRANCH: main
VERSION: 2.7.4
UI_VERSION: 2.6.3
BRANCH: v2.7
jobs:
cross-build:
@@ -21,7 +21,7 @@ jobs:
- [aarch64-linux-gnu, aarch64, arm64]
- [arm-linux-gnueabihf, armv4, armhf]
- [x86_64-linux-gnu, x86_64, amd64]
container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-main
container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-v2.7
steps:
- uses: actions/checkout@v4
@@ -64,7 +64,7 @@ jobs:
matrix:
arch:
- [x86_64-linux-gnu, x86_64, amd64]
container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-main
container: ghcr.io/neugates/build:${{ matrix.arch[1] }}-v2.7
steps:
- uses: actions/checkout@v4
+1
View File
@@ -128,6 +128,7 @@ set(NEURON_SOURCES
src/core/node_manager.c
src/core/template_manager.c
src/core/storage.c
src/adapter/msg_q.c
src/adapter/storage.c
src/adapter/adapter.c
src/adapter/driver/cache.c
+1 -1
View File
@@ -1,7 +1,7 @@
set(CMAKE_SYSTEM_NAME Linux)
set(COMPILER_PREFIX aarch64-linux-gnu)
set(CMAKE_SYSTEM_PROCESSOR aarch64)
set(LIBRARY_DIR /home/neuron/main/libs)
set(LIBRARY_DIR /home/neuron/v2.7/libs)
set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)
+1 -1
View File
@@ -1,7 +1,7 @@
set(CMAKE_SYSTEM_NAME Linux)
set(COMPILER_PREFIX arm-linux-gnueabi)
set(CMAKE_SYSTEM_PROCESSOR armv5tejl)
set(LIBRARY_DIR /home/neuron/main/libs)
set(LIBRARY_DIR /home/neuron/v2.7/libs)
set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)
+1 -1
View File
@@ -1,7 +1,7 @@
set(CMAKE_SYSTEM_NAME Linux)
set(COMPILER_PREFIX arm-linux-gnueabihf)
set(CMAKE_SYSTEM_PROCESSOR armv7l)
set(LIBRARY_DIR /home/neuron/main/libs)
set(LIBRARY_DIR /home/neuron/v2.7/libs)
set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)
+4
View File
@@ -21,3 +21,7 @@ link_directories(${CMAKE_STAGING_PREFIX})
if(CMAKE_SYSTEM_PROCESSOR MATCHES "mips32")
link_libraries(atomic)
endif()
if(CMAKE_SYSTEM_PROCESSOR MATCHES "riscv64")
link_libraries(atomic)
endif()
+1 -1
View File
@@ -1,7 +1,7 @@
set(CMAKE_SYSTEM_NAME Linux)
set(COMPILER_PREFIX x86_64-linux-gnu)
set(CMAKE_SYSTEM_PROCESSOR x86_64)
set(LIBRARY_DIR /home/neuron/main/libs)
set(LIBRARY_DIR /home/neuron/v2.7/libs)
set(CMAKE_C_COMPILER ${COMPILER_PREFIX}-gcc)
set(CMAKE_CXX_COMPILER ${COMPILER_PREFIX}-g++)
+2 -2
View File
@@ -59,7 +59,7 @@ Add different types of tags to the group, it should return success
Add a list of tags with errors, it should return the number of successful additions.
${res}= Add Tags modbus-node group ${tag1},${tag2},${tag3},${tag4}
Check Response Status ${res} 200
Check Response Status ${res} 409
Check Error Code ${res} ${NEU_ERR_TAG_NAME_CONFLICT}
Should Be Equal As Integers ${res}[index] 0
@@ -139,7 +139,7 @@ Delete tag from non-existent group, it should return success
Update non-existent tag, it should return failure.
${res}= Update Tags modbus-node group ${tag4}
Check Response Status ${res} 200
Check Response Status ${res} 404
Check Error Code ${res} ${NEU_ERR_TAG_NOT_EXIST}
${res}= Get Tags modbus-node group
+4 -4
View File
@@ -96,7 +96,7 @@ Create a template with invalid tag type, it should fail.
${groups} = Create List ${group}
${res} = Add Template ${g_template} ${g_plugin} ${groups}
Check Response Status ${res} 206
Check Response Status ${res} 400
Check Error Code ${res} ${NEU_ERR_TAG_TYPE_NOT_SUPPORT}
[Teardown] Del Template ${g_template}
@@ -108,7 +108,7 @@ Create a template with invalid tag address, it should fail.
${groups} = Create List ${group}
${res} = Add Template ${g_template} ${g_plugin} ${groups}
Check Response Status ${res} 206
Check Response Status ${res} 400
Check Error Code ${res} ${NEU_ERR_TAG_ADDRESS_FORMAT_INVALID}
[Teardown] Del Template ${g_template}
@@ -802,7 +802,7 @@ Add tag with too long name to existing template group, it should fail.
&{tag} = New Tag From ${g_tag1} name=${g_long_str}
${res} = Add Template Tags ${g_template} ${g_group1}[name] ${tag}
Check Response Status ${res} 206
Check Response Status ${res} 400
Check Error Code ${res} ${NEU_ERR_TAG_NAME_TOO_LONG}
[Teardown] Del Template ${g_template}
@@ -1026,7 +1026,7 @@ Update tag name to invalid value, it should fail.
&{tag} = New Tag From ${g_tag1} name=${g_long_str}
${res} = Put Template Tags ${g_template} ${g_group1}[name] ${tag}
Check Response Status ${res} 206
Check Response Status ${res} 400
Check Error Code ${res} ${NEU_ERR_TAG_NAME_TOO_LONG}
[Teardown] Del Template ${g_template}
+4 -3
View File
@@ -24,22 +24,23 @@
#define NEU_VERSION_MAJOR 2
#define NEU_VERSION_MINOR 7
#define NEU_VERSION_FIX 0
#define NEU_VERSION_FIX 4
#define NEU_GET_VERSION_MAJOR(V) ((0xffff0000 & V) >> 16)
#define NEU_GET_VERSION_MINOR(V) ((0x0000ff00 & V) >> 8)
#define NEU_GET_VERSION_FIX(V) ((0x000000ff & V))
#define NEU_VERSION(MAJOR, MINOR, FIX) ((MAJOR << 16) | (MINOR << 8) | FIX)
#define NEU_TAG_NAME_LEN 64
#define NEU_TAG_NAME_LEN 128
#define NEU_TAG_ADDRESS_LEN 128
#define NEU_TAG_DESCRIPTION_LEN 128
#define NEU_GROUP_MAX_PER_NODE 512
#define NEU_GROUP_NAME_LEN 128
#define NEU_GROUP_INTERVAL_LIMIT 100
#define NEU_DEFAULT_GROUP_INTERVAL 100
#define NEU_NODE_NAME_LEN 128
#define NEU_PLUGIN_NAME_LEN 32
#define NEU_PLUGIN_LIBRARY_LEN 32
#define NEU_PLUGIN_LIBRARY_LEN 64
#define NEU_PLUGIN_DESCRIPTION_LEN 512
#define NEU_TEMPLATE_NAME_LEN 128
#define NEU_DRIVER_TAG_CACHE_EXPIRE_TIME 60
+2
View File
@@ -70,6 +70,7 @@ typedef enum {
NEU_ERR_GROUP_PARAMETER_INVALID = 2105,
NEU_ERR_GROUP_NOT_EXIST = 2106,
NEU_ERR_GROUP_NAME_TOO_LONG = 2107,
NEU_ERR_GROUP_MAX_GROUPS = 2108,
NEU_ERR_TAG_NOT_EXIST = 2201,
NEU_ERR_TAG_NAME_CONFLICT = 2202,
@@ -98,6 +99,7 @@ typedef enum {
NEU_ERR_LIBRARY_MODULE_VERSION_NOT_MATCH = 2314,
NEU_ERR_LIBRARY_NAME_NOT_CONFORM = 2315,
NEU_ERR_LIBRARY_CLIB_NOT_MATCH = 2316,
NEU_ERR_LIBRARY_UPDATE_FAIL = 2317,
NEU_ERR_LICENSE_NOT_FOUND = 2400,
NEU_ERR_LICENSE_INVALID = 2401,
+2
View File
@@ -126,6 +126,8 @@ void neu_json_decode_write_free(neu_json_write_t *req);
typedef struct {
char *group;
char *node;
char *name;
char *desc;
bool sync;
} neu_json_read_req_t;
+277 -13
View File
@@ -26,14 +26,21 @@ extern "C" {
#include <stdint.h>
#include <pthread.h>
#include "define.h"
#include "type.h"
#include "utils/rolling_counter.h"
#include "utils/utextend.h"
#include "utils/uthash.h"
extern int64_t global_timestamp;
typedef enum {
NEU_METRIC_TYPE_COUNTER,
NEU_METRIC_TYPE_GAUAGE,
NEU_METRIC_TYPE_COUNTER_SET,
NEU_METRIC_TYPE_ROLLING_COUNTER,
} neu_metric_type_e;
// node running state
@@ -131,6 +138,96 @@ typedef enum {
#define NEU_METRIC_RECV_MSGS_TOTAL_TYPE NEU_METRIC_TYPE_COUNTER
#define NEU_METRIC_RECV_MSGS_TOTAL_HELP "Total number of messages received"
// number of trans data message within the last 5 seconds
#define NEU_METRIC_TRANS_DATA_5S "last_5s_trans_data_msgs"
#define NEU_METRIC_TRANS_DATA_5S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
#define NEU_METRIC_TRANS_DATA_5S_HELP \
"Number of internal trans data message within the last 5 seconds"
// number of trans data message within the last 30 seconds
#define NEU_METRIC_TRANS_DATA_30S "last_30s_trans_data_msgs"
#define NEU_METRIC_TRANS_DATA_30S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
#define NEU_METRIC_TRANS_DATA_30S_HELP \
"Number of internal trans data message within the last 30 seconds"
// number of trans data message within the last 60 seconds
#define NEU_METRIC_TRANS_DATA_60S "last_60s_trans_data_msgs"
#define NEU_METRIC_TRANS_DATA_60S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
#define NEU_METRIC_TRANS_DATA_60S_HELP \
"Number of internal trans data message within the last 60 seconds"
// number of bytes sent within the last 5 seconds
#define NEU_METRIC_SEND_BYTES_5S "last_5s_send_bytes"
#define NEU_METRIC_SEND_BYTES_5S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
#define NEU_METRIC_SEND_BYTES_5S_HELP \
"Number of bytes sent within the last 5 seconds"
// number of bytes sent within the last 30 seconds
#define NEU_METRIC_SEND_BYTES_30S "last_30s_send_bytes"
#define NEU_METRIC_SEND_BYTES_30S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
#define NEU_METRIC_SEND_BYTES_30S_HELP \
"Number of bytes sent within the last 30 seconds"
// number of bytes sent within the last 60 seconds
#define NEU_METRIC_SEND_BYTES_60S "last_60s_send_bytes"
#define NEU_METRIC_SEND_BYTES_60S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
#define NEU_METRIC_SEND_BYTES_60S_HELP \
"Number of bytes sent within the last 60 seconds"
// number of bytes received within the last 5 seconds
#define NEU_METRIC_RECV_BYTES_5S "last_5s_recv_bytes"
#define NEU_METRIC_RECV_BYTES_5S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
#define NEU_METRIC_RECV_BYTES_5S_HELP \
"Number of bytes received within the last 5 seconds"
// number of bytes received within the last 30 seconds
#define NEU_METRIC_RECV_BYTES_30S "last_30s_recv_bytes"
#define NEU_METRIC_RECV_BYTES_30S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
#define NEU_METRIC_RECV_BYTES_30S_HELP \
"Number of bytes received within the last 30 seconds"
// number of bytes received within the last 60 seconds
#define NEU_METRIC_RECV_BYTES_60S "last_60s_recv_bytes"
#define NEU_METRIC_RECV_BYTES_60S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
#define NEU_METRIC_RECV_BYTES_60S_HELP \
"Number of bytes received within the last 60 seconds"
// number of messages received within the last 5 seconds
#define NEU_METRIC_RECV_MSGS_5S "last_5s_recv_msgs"
#define NEU_METRIC_RECV_MSGS_5S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
#define NEU_METRIC_RECV_MSGS_5S_HELP \
"Number of messages received within the last 5 seconds"
// number of messages received within the last 30 seconds
#define NEU_METRIC_RECV_MSGS_30S "last_30s_recv_msgs"
#define NEU_METRIC_RECV_MSGS_30S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
#define NEU_METRIC_RECV_MSGS_30S_HELP \
"Number of messages received within the last 30 seconds"
// number of messages received within the last 60 seconds
#define NEU_METRIC_RECV_MSGS_60S "last_60s_recv_msgs"
#define NEU_METRIC_RECV_MSGS_60S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
#define NEU_METRIC_RECV_MSGS_60S_HELP \
"Number of messages received within the last 60 seconds"
// number of disconnection within the last 60 seconds
#define NEU_METRIC_DISCONNECTION_60S "last_60s_disconnections"
#define NEU_METRIC_DISCONNECTION_60S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
#define NEU_METRIC_DISCONNECTION_60S_HELP \
"Number of disconnection within the last 60 seconds"
// number of disconnection within the last 600 seconds
#define NEU_METRIC_DISCONNECTION_600S "last_600s_disconnections"
#define NEU_METRIC_DISCONNECTION_600S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
#define NEU_METRIC_DISCONNECTION_600S_HELP \
"Number of disconnection within the last 600 seconds"
// number of disconnection within the last 1800 seconds
#define NEU_METRIC_DISCONNECTION_1800S "last_1800s_disconnections"
#define NEU_METRIC_DISCONNECTION_1800S_TYPE NEU_METRIC_TYPE_ROLLING_COUNTER
#define NEU_METRIC_DISCONNECTION_1800S_HELP \
"Number of disconnection within the last 1800 seconds"
typedef enum {
NEU_METRICS_CATEGORY_GLOBAL,
NEU_METRICS_CATEGORY_DRIVER,
@@ -140,11 +237,12 @@ typedef enum {
// metric entry
typedef struct {
const char * name; // NOTE: should points to string literal
const char * help; // NOTE: should points to string literal
neu_metric_type_e type; //
uint64_t value; //
UT_hash_handle hh; // ordered by name
const char * name; // NOTE: should points to string literal
const char * help; // NOTE: should points to string literal
neu_metric_type_e type; //
uint64_t value; //
neu_rolling_counter_t *rcnt; //
UT_hash_handle hh; // ordered by name
} neu_metric_entry_t;
// group metrics
@@ -156,6 +254,7 @@ typedef struct {
// node metrics
typedef struct {
pthread_mutex_t lock; // spin lock
neu_node_type_e type; // node type
char * name; // node name
neu_metric_entry_t * entries; // node metric entries
@@ -192,6 +291,16 @@ typedef struct {
neu_metric_entry_t *registered_metrics;
} neu_metrics_t;
void neu_metrics_init();
void neu_metrics_add_node(const neu_adapter_t *adapter);
void neu_metrics_del_node(const neu_adapter_t *adapter);
int neu_metrics_register_entry(const char *name, const char *help,
neu_metric_type_e type);
void neu_metrics_unregister_entry(const char *name);
typedef void (*neu_metrics_cb_t)(const neu_metrics_t *metrics, void *data);
void neu_metrics_visist(neu_metrics_cb_t cb, void *data);
static inline const char *neu_metric_type_str(neu_metric_type_e type)
{
if (NEU_METRIC_TYPE_COUNTER == type) {
@@ -207,6 +316,9 @@ int neu_metric_entries_add(neu_metric_entry_t **entries, const char *name,
static inline void neu_metric_entry_free(neu_metric_entry_t *entry)
{
if (NEU_METRIC_TYPE_ROLLING_COUNTER == entry->type) {
neu_rolling_counter_free(entry->rcnt);
}
free(entry);
}
@@ -220,22 +332,41 @@ static inline void neu_group_metrics_free(neu_group_metrics_t *group_metrics)
HASH_ITER(hh, group_metrics->entries, e, tmp)
{
HASH_DEL(group_metrics->entries, e);
neu_metrics_unregister_entry(e->name);
neu_metric_entry_free(e);
}
free(group_metrics->name);
free(group_metrics);
}
static inline neu_node_metrics_t *
neu_node_metrics_new(neu_adapter_t *adapter, neu_node_type_e type, char *name)
{
neu_node_metrics_t *node_metrics =
(neu_node_metrics_t *) calloc(1, sizeof(*node_metrics));
if (NULL != node_metrics) {
pthread_mutex_init(&node_metrics->lock, NULL);
node_metrics->type = type;
node_metrics->name = name;
node_metrics->adapter = adapter;
// neu_metrics_add_node(adapter);
}
return node_metrics;
}
static inline void neu_node_metrics_free(neu_node_metrics_t *node_metrics)
{
if (NULL == node_metrics) {
return;
}
pthread_mutex_destroy(&node_metrics->lock);
neu_metric_entry_t *e = NULL, *etmp = NULL;
HASH_ITER(hh, node_metrics->entries, e, etmp)
{
HASH_DEL(node_metrics->entries, e);
neu_metrics_unregister_entry(e->name);
neu_metric_entry_free(e);
}
@@ -249,15 +380,148 @@ static inline void neu_node_metrics_free(neu_node_metrics_t *node_metrics)
free(node_metrics);
}
void neu_metrics_init();
void neu_metrics_add_node(const neu_adapter_t *adapter);
void neu_metrics_del_node(const neu_adapter_t *adapter);
int neu_metrics_register_entry(const char *name, const char *help,
neu_metric_type_e type);
void neu_metrics_unregister_entry(const char *name);
static inline int neu_node_metrics_add(neu_node_metrics_t *node_metrics,
const char *group_name, const char *name,
const char *help, neu_metric_type_e type,
uint64_t init)
{
int rv = 0;
typedef void (*neu_metrics_cb_t)(const neu_metrics_t *metrics, void *data);
void neu_metrics_visist(neu_metrics_cb_t cb, void *data);
if (0 > neu_metrics_register_entry(name, help, type)) {
return -1;
}
pthread_mutex_lock(&node_metrics->lock);
if (NULL == group_name) {
rv = neu_metric_entries_add(&node_metrics->entries, name, help, type,
init);
} else {
neu_group_metrics_t *group_metrics = NULL;
HASH_FIND_STR(node_metrics->group_metrics, group_name, group_metrics);
if (NULL != group_metrics) {
rv = neu_metric_entries_add(&group_metrics->entries, name, help,
type, init);
} else {
group_metrics =
(neu_group_metrics_t *) calloc(1, sizeof(*group_metrics));
if (NULL != group_metrics &&
NULL != (group_metrics->name = strdup(group_name))) {
HASH_ADD_STR(node_metrics->group_metrics, name, group_metrics);
rv = neu_metric_entries_add(&group_metrics->entries, name, help,
type, init);
} else {
free(group_metrics);
rv = -1;
}
}
}
pthread_mutex_unlock(&node_metrics->lock);
if (0 != rv) {
neu_metrics_unregister_entry(name);
}
return rv;
}
static inline int neu_node_metrics_update(neu_node_metrics_t *node_metrics,
const char * group,
const char *metric_name, uint64_t n)
{
neu_metric_entry_t *entry = NULL;
pthread_mutex_lock(&node_metrics->lock);
if (NULL == group) {
HASH_FIND_STR(node_metrics->entries, metric_name, entry);
} else if (NULL != node_metrics->group_metrics) {
neu_group_metrics_t *g = NULL;
HASH_FIND_STR(node_metrics->group_metrics, group, g);
if (NULL != g) {
HASH_FIND_STR(g->entries, metric_name, entry);
}
}
if (NULL == entry) {
pthread_mutex_unlock(&node_metrics->lock);
return -1;
}
if (NEU_METRIC_TYPE_COUNTER == entry->type) {
entry->value += n;
} else if (NEU_METRIC_TYPE_ROLLING_COUNTER == entry->type) {
entry->value =
neu_rolling_counter_inc(entry->rcnt, global_timestamp, n);
} else {
entry->value = n;
}
pthread_mutex_unlock(&node_metrics->lock);
return 0;
}
static inline void neu_node_metrics_reset(neu_node_metrics_t *node_metrics)
{
neu_metric_entry_t *entry = NULL;
pthread_mutex_lock(&node_metrics->lock);
HASH_LOOP(hh, node_metrics->entries, entry)
{
entry->value = 0;
if (NEU_METRIC_TYPE_ROLLING_COUNTER == entry->type) {
neu_rolling_counter_reset(entry->rcnt);
}
}
neu_group_metrics_t *g = NULL;
HASH_LOOP(hh, node_metrics->group_metrics, g)
{
HASH_LOOP(hh, g->entries, entry)
{
entry->value = 0;
if (NEU_METRIC_TYPE_ROLLING_COUNTER == entry->type) {
neu_rolling_counter_reset(entry->rcnt);
}
}
}
pthread_mutex_unlock(&node_metrics->lock);
}
static inline int
neu_node_metrics_update_group(neu_node_metrics_t *node_metrics,
const char * group_name,
const char * new_group_name)
{
int rv = -1;
neu_group_metrics_t *group_metrics = NULL;
pthread_mutex_lock(&node_metrics->lock);
HASH_FIND_STR(node_metrics->group_metrics, group_name, group_metrics);
if (NULL != group_metrics) {
char *name = strdup(new_group_name);
if (NULL != name) {
HASH_DEL(node_metrics->group_metrics, group_metrics);
free(group_metrics->name);
group_metrics->name = name;
HASH_ADD_STR(node_metrics->group_metrics, name, group_metrics);
rv = 0;
}
}
pthread_mutex_unlock(&node_metrics->lock);
return rv;
}
static inline void neu_node_metrics_del_group(neu_node_metrics_t *node_metrics,
const char * group_name)
{
neu_group_metrics_t *gm = NULL;
pthread_mutex_lock(&node_metrics->lock);
HASH_FIND_STR(node_metrics->group_metrics, group_name, gm);
if (NULL != gm) {
HASH_DEL(node_metrics->group_metrics, gm);
neu_group_metrics_free(gm);
}
pthread_mutex_unlock(&node_metrics->lock);
}
#ifdef __cplusplus
}
+11 -1
View File
@@ -515,7 +515,13 @@ neu_req_subscribe_groups_fini(neu_req_subscribe_groups_t *req)
typedef struct neu_req_get_subscribe_group {
char app[NEU_NODE_NAME_LEN];
} neu_req_get_subscribe_group_t, neu_req_get_sub_driver_tags_t;
char driver[NEU_NODE_NAME_LEN];
char group[NEU_GROUP_NAME_LEN];
} neu_req_get_subscribe_group_t;
typedef struct {
char app[NEU_NODE_NAME_LEN];
} neu_req_get_sub_driver_tags_t;
typedef struct neu_resp_subscribe_info {
char app[NEU_NODE_NAME_LEN];
@@ -779,6 +785,8 @@ static inline void neu_req_inst_templates_fini(neu_req_inst_templates_t *req)
typedef struct neu_req_read_group {
char *driver;
char *group;
char *name;
char *desc;
bool sync;
} neu_req_read_group_t;
@@ -786,6 +794,8 @@ static inline void neu_req_read_group_fini(neu_req_read_group_t *req)
{
free(req->driver);
free(req->group);
free(req->name);
free(req->desc);
}
typedef struct neu_req_write_tag {
+8
View File
@@ -36,6 +36,14 @@ extern "C" {
#define NEURON_PLUGIN_VER_1_0 \
NEU_VERSION(NEU_VERSION_MAJOR, NEU_VERSION_MINOR, NEU_VERSION_FIX)
#define NEU_PLUGIN_REGISTER_METRIC(plugin, name, init) \
plugin->common.adapter_callbacks->register_metric( \
plugin->common.adapter, name, name##_HELP, name##_TYPE, init)
#define NEU_PLUGIN_UPDATE_METRIC(plugin, name, val, grp) \
plugin->common.adapter_callbacks->update_metric(plugin->common.adapter, \
name, val, grp)
extern int64_t global_timestamp;
typedef struct neu_plugin_common {
+115
View File
@@ -0,0 +1,115 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2024 EMQ Technologies Co., Ltd All rights reserved.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 3 of the License, or (at your option) any later version.
*
* 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. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with this program; if not, write to the Free Software Foundation,
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#ifndef NEURON_UTILS_ROLLING_COUNTER_H
#define NEURON_UTILS_ROLLING_COUNTER_H
#ifdef __cplusplus
extern "C" {
#endif
#include <assert.h>
#include <stdbool.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
/** Rolling counter.
*
* This counter is for counting values within some latest time span, like
* network bytes sent within the last 5 seconds etc.
*/
typedef struct {
uint64_t val; // accumulator
uint64_t ts; // head time stamp in milliseconds
uint32_t res : 21; // time resolution in milliseconds
uint32_t hd : 5; // head position
uint32_t n : 6; // number of counters
uint32_t counts[]; // bins of counters
} neu_rolling_counter_t;
/** Create rolling counter.
*
* @param span time span in milliseconds
*/
static inline neu_rolling_counter_t *neu_rolling_counter_new(unsigned span)
{
unsigned n = span <= 6000 ? 4 : span <= 32000 ? 8 : span <= 64000 ? 16 : 32;
assert(span / n < (1 << 22)); // should not overflow ti
neu_rolling_counter_t *counter = (neu_rolling_counter_t *) calloc(
1, sizeof(*counter) + sizeof(counter->counts[0]) * n);
if (counter) {
counter->res = span / n;
counter->n = n;
}
return counter;
}
/** Destructs the rolling counter.
*/
static inline void neu_rolling_counter_free(neu_rolling_counter_t *counter)
{
if (counter) {
free(counter);
}
}
/** Increment the rolling counter and return the value;
*
* @param ts time stamp in milliseconds, should be monotonic
* @param dt delta value to increment by
*/
static inline uint64_t neu_rolling_counter_inc(neu_rolling_counter_t *counter,
uint64_t ts, unsigned dt)
{
uint64_t step = (ts - counter->ts) / counter->res;
for (unsigned i = 0; i < step && i < counter->n; ++i) {
counter->hd = (counter->hd + 1) & (counter->n - 1);
counter->val -= counter->counts[counter->hd];
counter->counts[counter->hd] = 0;
}
counter->val += dt;
counter->counts[counter->hd] += dt;
counter->ts += step * counter->res;
return counter->val;
}
/** Reset the counter.
*/
static inline void neu_rolling_counter_reset(neu_rolling_counter_t *counter)
{
counter->val = 0;
counter->hd = 0;
memset(counter->counts, 0, counter->n * sizeof(counter->counts[0]));
}
/** Return the counter value.
*
* NOTE: may return stale value if the counter is not updated frequent enough.
*/
static inline uint64_t neu_rolling_counter_value(neu_rolling_counter_t *counter)
{
return counter->val;
}
#ifdef __cplusplus
}
#endif
#endif
@@ -0,0 +1,40 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2023 EMQ Technologies Co., Ltd All rights reserved.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 3 of the License, or (at your option) any later version.
*
* 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. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with this program; if not, write to the Free Software Foundation,
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
BEGIN TRANSACTION;
ALTER TABLE tags RENAME TO temp_tags;
CREATE TABLE IF NOT EXISTS
tags (
driver_name TEXT NOT NULL,
group_name TEXT NOT NULL,
name TEXT NULL check(length(name) <= 128),
address TEXT NULL check(length(address) <= 128),
attribute INTEGER NOT NULL check(attribute BETWEEN 0 AND 15),
precision INTEGER NOT NULL check(precision BETWEEN 0 AND 17),
decimal REAL NOT NULL,
type INTEGER NOT NULL check(type BETWEEN 0 AND 19),
description TEXT NULL check(length(description) <= 512),
value TEXT,
UNIQUE (driver_name, group_name, name),
FOREIGN KEY (driver_name, group_name) REFERENCES groups (driver_name, name) ON UPDATE CASCADE ON DELETE CASCADE
);
INSERT INTO tags SELECT * FROM temp_tags;
DROP TABLE temp_tags;
COMMIT;
+24
View File
@@ -71,6 +71,10 @@ static void pipe_rm_cb(nng_pipe p, nng_pipe_ev ev, void *arg)
nng_mtx_lock(plugin->mtx);
plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
nng_mtx_unlock(plugin->mtx);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_DISCONNECTION_60S, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_DISCONNECTION_600S, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_DISCONNECTION_1800S, 1, NULL);
}
static int ekuiper_plugin_init(neu_plugin_t *plugin, bool load)
@@ -96,6 +100,22 @@ static int ekuiper_plugin_init(neu_plugin_t *plugin, bool load)
plugin->recv_aio = recv_aio;
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_TRANS_DATA_5S, 5000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_TRANS_DATA_30S, 30000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_TRANS_DATA_60S, 60000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_BYTES_5S, 5000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_BYTES_30S, 30000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_BYTES_60S, 60000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_BYTES_5S, 5000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_BYTES_30S, 30000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_BYTES_60S, 60000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_MSGS_5S, 5000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_MSGS_30S, 30000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_MSGS_60S, 60000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_DISCONNECTION_60S, 60000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_DISCONNECTION_600S, 600000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_DISCONNECTION_1800S, 1800000);
plog_notice(plugin, "plugin initialized");
return rv;
}
@@ -303,6 +323,10 @@ static int ekuiper_plugin_request(neu_plugin_t * plugin,
case NEU_REQRESP_TRANS_DATA: {
neu_reqresp_trans_data_t *trans_data = data;
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_5S, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_30S, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_60S, 1, NULL);
if (disconnected) {
plog_debug(plugin, "not connected");
} else {
+16 -15
View File
@@ -30,9 +30,6 @@ void send_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *trans_data)
{
int rv = 0;
neu_adapter_update_metric_cb_t update_metric =
plugin->common.adapter_callbacks->update_metric;
char * json_str = NULL;
json_read_resp_t resp = {
.plugin = plugin,
@@ -59,15 +56,18 @@ void send_data(neu_plugin_t *plugin, neu_reqresp_trans_data_t *trans_data)
rv = nng_sendmsg(plugin->sock, msg,
NNG_FLAG_NONBLOCK); // TODO: use aio to send message
if (0 == rv) {
update_metric(plugin->common.adapter, NEU_METRIC_SEND_BYTES, json_len,
NULL);
update_metric(plugin->common.adapter, NEU_METRIC_SEND_MSGS_TOTAL, 1,
NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSGS_TOTAL, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_5S, json_len,
NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_30S, json_len,
NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_60S, json_len,
NULL);
} else {
plog_error(plugin, "nng cannot send msg: %s", nng_strerror(rv));
nng_msg_free(msg);
update_metric(plugin->common.adapter, NEU_METRIC_SEND_MSG_ERRORS_TOTAL,
1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, 1,
NULL);
}
}
@@ -80,9 +80,6 @@ void recv_data_callback(void *arg)
char * json_str = NULL;
json_write_req_t *req = NULL;
neu_adapter_update_metric_cb_t update_metric =
plugin->common.adapter_callbacks->update_metric;
rv = nng_aio_result(plugin->recv_aio);
if (0 != rv) {
plog_error(plugin, "nng_recv error: %s", nng_strerror(rv));
@@ -93,9 +90,13 @@ void recv_data_callback(void *arg)
json_str = nng_msg_body(msg);
json_len = nng_msg_len(msg);
plog_debug(plugin, "<< %.*s", (int) json_len, json_str);
update_metric(plugin->common.adapter, NEU_METRIC_RECV_BYTES, json_len,
NULL);
update_metric(plugin->common.adapter, NEU_METRIC_RECV_MSGS_TOTAL, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_TOTAL, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_5S, json_len, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_30S, json_len, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_60S, json_len, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_5S, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_30S, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_60S, 1, NULL);
if (json_decode_write_req(json_str, json_len, &req) < 0) {
plog_error(plugin, "fail decode write request json: %.*s",
(int) nng_msg_len(msg), json_str);
+22 -17
View File
@@ -276,15 +276,14 @@ static void publish_cb(int errcode, neu_mqtt_qos_e qos, char *topic,
neu_plugin_t *plugin = data;
neu_adapter_update_metric_cb_t update_metric =
plugin->common.adapter_callbacks->update_metric;
if (0 == errcode) {
update_metric(plugin->common.adapter, NEU_METRIC_SEND_MSGS_TOTAL, 1,
NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSGS_TOTAL, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_5S, len, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_30S, len, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_BYTES_60S, len, NULL);
} else {
update_metric(plugin->common.adapter, NEU_METRIC_SEND_MSG_ERRORS_TOTAL,
1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, 1,
NULL);
}
free(payload);
@@ -293,16 +292,14 @@ static void publish_cb(int errcode, neu_mqtt_qos_e qos, char *topic,
static inline int publish(neu_plugin_t *plugin, neu_mqtt_qos_e qos, char *topic,
char *payload, size_t payload_len)
{
neu_adapter_update_metric_cb_t update_metric =
plugin->common.adapter_callbacks->update_metric;
int rv =
neu_mqtt_client_publish(plugin->client, qos, topic, (uint8_t *) payload,
(uint32_t) payload_len, plugin, publish_cb);
if (0 != rv) {
plog_error(plugin, "pub [%s, QoS%d] fail", topic, qos);
update_metric(plugin->common.adapter, NEU_METRIC_SEND_MSG_ERRORS_TOTAL,
1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_SEND_MSG_ERRORS_TOTAL, 1,
NULL);
free(payload);
rv = NEU_ERR_MQTT_PUBLISH_FAILURE;
}
@@ -320,9 +317,13 @@ void handle_write_req(neu_mqtt_qos_e qos, const char *topic,
(void) qos;
(void) topic;
neu_adapter_update_metric_cb_t update_metric =
plugin->common.adapter_callbacks->update_metric;
update_metric(plugin->common.adapter, NEU_METRIC_RECV_MSGS_TOTAL, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_TOTAL, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_5S, len, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_30S, len, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_60S, len, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_5S, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_30S, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_60S, 1, NULL);
char *json_str = malloc(len + 1);
if (NULL == json_str) {
@@ -407,9 +408,13 @@ void handle_read_req(neu_mqtt_qos_e qos, const char *topic,
(void) qos;
(void) topic;
neu_adapter_update_metric_cb_t update_metric =
plugin->common.adapter_callbacks->update_metric;
update_metric(plugin->common.adapter, NEU_METRIC_RECV_MSGS_TOTAL, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_TOTAL, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_5S, len, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_30S, len, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_BYTES_60S, len, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_5S, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_30S, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_RECV_MSGS_60S, 1, NULL);
char *json_str = malloc(len + 1);
if (NULL == json_str) {
+27 -11
View File
@@ -37,6 +37,9 @@ static void disconnect_cb(void *data)
{
neu_plugin_t *plugin = data;
plugin->common.link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_DISCONNECTION_60S, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_DISCONNECTION_600S, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_DISCONNECTION_1800S, 1, NULL);
plog_notice(plugin, "plugin `%s` disconnected",
neu_plugin_module.module_name);
}
@@ -60,13 +63,26 @@ static int mqtt_plugin_close(neu_plugin_t *plugin)
static int mqtt_plugin_init(neu_plugin_t *plugin, bool load)
{
(void) load;
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_CACHED_MSGS_NUM, 0);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_TRANS_DATA_5S, 5000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_TRANS_DATA_30S, 30000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_TRANS_DATA_60S, 60000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_BYTES_5S, 5000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_BYTES_30S, 30000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_SEND_BYTES_60S, 60000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_BYTES_5S, 5000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_BYTES_30S, 30000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_BYTES_60S, 60000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_MSGS_5S, 5000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_MSGS_30S, 30000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_RECV_MSGS_60S, 60000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_DISCONNECTION_60S, 60000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_DISCONNECTION_600S, 600000);
NEU_PLUGIN_REGISTER_METRIC(plugin, NEU_METRIC_DISCONNECTION_1800S, 1800000);
plog_notice(plugin, "initialize plugin `%s` success",
neu_plugin_module.module_name);
neu_adapter_register_metric_cb_t register_metric =
plugin->common.adapter_callbacks->register_metric;
register_metric(plugin->common.adapter, NEU_METRIC_CACHED_MSGS_NUM,
NEU_METRIC_CACHED_MSGS_NUM_HELP,
NEU_METRIC_CACHED_MSGS_NUM_TYPE, 0);
return NEU_ERR_SUCCESS;
}
@@ -333,15 +349,12 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
{
neu_err_code_e error = NEU_ERR_SUCCESS;
neu_adapter_update_metric_cb_t update_metric =
plugin->common.adapter_callbacks->update_metric;
// update cached messages number per seconds
if (NULL != plugin->client &&
(global_timestamp - plugin->cache_metric_update_ts) >= 1000) {
update_metric(plugin->common.adapter, NEU_METRIC_CACHED_MSGS_NUM,
neu_mqtt_client_get_cached_msgs_num(plugin->client),
NULL);
NEU_PLUGIN_UPDATE_METRIC(
plugin, NEU_METRIC_CACHED_MSGS_NUM,
neu_mqtt_client_get_cached_msgs_num(plugin->client), NULL);
plugin->cache_metric_update_ts = global_timestamp;
}
@@ -353,6 +366,9 @@ static int mqtt_plugin_request(neu_plugin_t *plugin, neu_reqresp_head_t *head,
error = handle_read_response(plugin, head->ctx, data);
break;
case NEU_REQRESP_TRANS_DATA: {
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_5S, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_30S, 1, NULL);
NEU_PLUGIN_UPDATE_METRIC(plugin, NEU_METRIC_TRANS_DATA_60S, 1, NULL);
error = handle_trans_data(plugin, data);
break;
}
+4 -3
View File
@@ -92,7 +92,8 @@ void handle_add_tags_resp(nng_aio *aio, neu_resp_add_tag_t *resp)
neu_json_encode_by_fn(&res, neu_json_encode_au_tags_resp, &result);
neu_http_ok(aio, result);
NEU_JSON_RESPONSE_ERROR(resp->error,
{ neu_http_response(aio, resp->error, result); });
free(result);
}
@@ -183,8 +184,8 @@ void handle_add_gtags_resp(nng_aio *aio, neu_resp_add_tag_t *resp)
res.index = resp->index;
neu_json_encode_by_fn(&res, neu_json_encode_au_gtags_resp, &result);
neu_http_ok(aio, result);
NEU_JSON_RESPONSE_ERROR(resp->error,
{ neu_http_response(aio, resp->error, result); });
free(result);
}
+23 -5
View File
@@ -356,23 +356,41 @@ void handle_grp_get_subscribe(nng_aio *aio)
{
int ret = 0;
neu_plugin_t * plugin = neu_rest_get_plugin();
char node_name[NEU_NODE_NAME_LEN] = { 0 };
neu_req_get_subscribe_group_t cmd = { 0 };
neu_reqresp_head_t header = {
neu_req_get_subscribe_group_t cmd = { 0 };
neu_reqresp_head_t header = {
.ctx = aio,
.type = NEU_REQ_GET_SUBSCRIBE_GROUP,
};
NEU_VALIDATE_JWT(aio);
if (neu_http_get_param_str(aio, "app", node_name, sizeof(node_name)) <= 0) {
// required parameter
ret = neu_http_get_param_str(aio, "app", cmd.app, sizeof(cmd.app));
if (ret <= 0 || (size_t) ret == sizeof(cmd.driver)) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
neu_http_response(aio, error_code.error, result_error);
})
return;
}
// optional parameter
ret = neu_http_get_param_str(aio, "driver", cmd.driver, sizeof(cmd.driver));
if (-1 == ret || (size_t) ret == sizeof(cmd.driver)) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
neu_http_response(aio, error_code.error, result_error);
})
return;
}
// optional parameter
ret = neu_http_get_param_str(aio, "group", cmd.group, sizeof(cmd.group));
if (-1 == ret || (size_t) ret == sizeof(cmd.driver)) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_PARAM_IS_WRONG, {
neu_http_response(aio, error_code.error, result_error);
})
return;
}
strcpy(cmd.app, node_name);
ret = neu_plugin_op(plugin, header, &cmd);
if (ret != 0) {
NEU_JSON_RESPONSE_ERROR(NEU_ERR_IS_BUSY, {
+31 -4
View File
@@ -19,8 +19,11 @@
#include <stdio.h>
#include <pthread.h>
#include "define.h"
#include "metrics.h"
#include "plugin.h"
#include "utils/asprintf.h"
#include "utils/http.h"
#include "utils/http_handler.h"
@@ -164,8 +167,8 @@ static inline void gen_global_metrics(const neu_metrics_t *metrics,
metrics->south_running_nodes, metrics->south_disconnected_nodes);
}
static inline void
gen_single_node_metrics(const neu_node_metrics_t *node_metrics, FILE *stream)
static inline void gen_single_node_metrics(neu_node_metrics_t *node_metrics,
FILE * stream)
{
fprintf(stream,
"# HELP node_type Driver(1) or APP(2)\n"
@@ -175,8 +178,13 @@ gen_single_node_metrics(const neu_node_metrics_t *node_metrics, FILE *stream)
neu_metric_entry_t *e = NULL;
pthread_mutex_lock(&node_metrics->lock);
HASH_LOOP(hh, node_metrics->entries, e)
{
if (NEU_METRIC_TYPE_ROLLING_COUNTER == e->type) {
// force clean stale value
e->value = neu_rolling_counter_inc(e->rcnt, global_timestamp, 0);
}
fprintf(stream,
"# HELP %s %s\n# TYPE %s %s\n%s{node=\"%s\"} %" PRIu64 "\n",
e->name, e->help, e->name, neu_metric_type_str(e->type),
@@ -188,6 +196,11 @@ gen_single_node_metrics(const neu_node_metrics_t *node_metrics, FILE *stream)
{
HASH_LOOP(hh, g->entries, e)
{
if (NEU_METRIC_TYPE_ROLLING_COUNTER == e->type) {
// force clean stale value
e->value =
neu_rolling_counter_inc(e->rcnt, global_timestamp, 0);
}
fprintf(stream,
"# HELP %s %s\n# TYPE %s %s\n%s{node=\"%s\",group=\"%s\"} "
"%" PRIu64 "\n",
@@ -195,6 +208,7 @@ gen_single_node_metrics(const neu_node_metrics_t *node_metrics, FILE *stream)
e->name, node_metrics->name, g->name, e->value);
}
}
pthread_mutex_unlock(&node_metrics->lock);
}
static inline bool has_entry(neu_node_metrics_t *node_metrics, const char *name)
@@ -216,7 +230,7 @@ static inline bool has_entry(neu_node_metrics_t *node_metrics, const char *name)
return false;
}
static void gen_all_node_metrics(const neu_metrics_t *metrics, int type_filter,
static void gen_all_node_metrics(neu_metrics_t *metrics, int type_filter,
FILE *stream)
{
neu_metric_entry_t * e = NULL, *r = NULL;
@@ -261,11 +275,18 @@ static void gen_all_node_metrics(const neu_metrics_t *metrics, int type_filter,
r->help, r->name, neu_metric_type_str(r->type));
}
pthread_mutex_lock(&n->lock);
HASH_FIND_STR(n->entries, r->name, e);
if (e) {
if (NEU_METRIC_TYPE_ROLLING_COUNTER == e->type) {
// force clean stale value
e->value =
neu_rolling_counter_inc(e->rcnt, global_timestamp, 0);
}
fprintf(stream, "%s{node=\"%s\"} %" PRIu64 "\n", e->name,
n->name, e->value);
pthread_mutex_unlock(&n->lock);
continue;
}
@@ -273,11 +294,17 @@ static void gen_all_node_metrics(const neu_metrics_t *metrics, int type_filter,
{
HASH_FIND_STR(g->entries, r->name, e);
if (e) {
if (NEU_METRIC_TYPE_ROLLING_COUNTER == e->type) {
// force clean stale value
e->value = neu_rolling_counter_inc(e->rcnt,
global_timestamp, 0);
}
fprintf(stream,
"%s{node=\"%s\",group=\"%s\"} %" PRIu64 "\n",
e->name, n->name, g->name, e->value);
}
}
pthread_mutex_unlock(&n->lock);
}
}
}
@@ -289,7 +316,7 @@ struct context {
const char *node;
};
static void gen_node_metrics(const neu_metrics_t *metrics, struct context *ctx)
static void gen_node_metrics(neu_metrics_t *metrics, struct context *ctx)
{
if (ctx->node[0]) {
neu_node_metrics_t *n = NULL;
+2
View File
@@ -43,6 +43,8 @@ void handle_read(nng_aio *aio)
cmd.driver = req->node;
cmd.group = req->group;
cmd.name = req->name;
cmd.desc = req->desc;
cmd.sync = req->sync;
req->node = NULL;
req->group = NULL;
+191 -187
View File
@@ -39,11 +39,12 @@
#include "plugin.h"
#include "storage.h"
static int adapter_trans_data(enum neu_event_io_type type, int fd,
void *usr_data);
static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data);
static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
void *data);
static void *adapter_consumer(void *arg);
static int adapter_trans_data(enum neu_event_io_type type, int fd,
void *usr_data);
static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data);
static int adapter_command(neu_adapter_t *adapter, neu_reqresp_head_t header,
void *data);
static int adapter_response(neu_adapter_t *adapter, neu_reqresp_head_t *header,
void *data);
static int adapter_responseto(neu_adapter_t * adapter,
@@ -58,6 +59,13 @@ static int adapter_update_metric(neu_adapter_t *adapter,
inline static void reply(neu_adapter_t *adapter, neu_reqresp_head_t *header,
void *data);
int validate_gtags(neu_adapter_t *adapter, neu_req_add_gtag_t *cmd,
neu_resp_add_tag_t *resp);
int try_add_gtags(neu_adapter_t *adapter, neu_req_add_gtag_t *cmd,
neu_resp_add_tag_t *resp);
int add_gtags(neu_adapter_t *adapter, neu_req_add_gtag_t *cmd,
neu_resp_add_tag_t *resp);
static const adapter_callbacks_t callback_funs = {
.command = adapter_command,
.response = adapter_response,
@@ -103,6 +111,36 @@ void neu_adapter_set_error(int error)
create_adapter_error = error;
}
static void *adapter_consumer(void *arg)
{
neu_adapter_t *adapter = (neu_adapter_t *) arg;
while (1) {
neu_msg_t * msg = NULL;
uint32_t n = adapter_msg_q_pop(adapter->msg_q, &msg);
neu_reqresp_head_t *header = neu_msg_get_header(msg);
nlog_debug("adapter(%s) recv msg from: %s %p, type: %s, %u",
adapter->name, header->sender, header->ctx,
neu_reqresp_type_string(header->type), n);
adapter->module->intf_funs->request(
adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
neu_msg_free(msg);
}
return NULL;
}
static inline zlog_category_t *get_log_category(const char *node)
{
char name[NEU_NODE_NAME_LEN] = { 0 };
// replace path separators
for (int i = 0; node[i]; ++i) {
name[i] = ('/' == node[i] || '\\' == node[i]) ? '_' : node[i];
}
return zlog_get_category(name);
}
neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load)
{
int rv = 0;
@@ -166,6 +204,9 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load)
break;
case NEU_NA_TYPE_NDRIVER:
case NEU_NA_TYPE_APP: {
adapter->msg_q = adapter_msg_q_new(adapter->name, 1024);
pthread_create(&adapter->consumer_tid, NULL, adapter_consumer,
(void *) adapter);
while (true) {
uint16_t port = neu_manager_get_port();
struct sockaddr_in local = {
@@ -201,7 +242,7 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info, bool load)
common->adapter = adapter;
common->adapter_callbacks = &adapter->cb_funs;
common->link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
common->log = zlog_get_category(adapter->name);
common->log = get_log_category(adapter->name);
strcpy(common->name, adapter->name);
zlog_level_switch(common->log, default_log_level);
@@ -272,7 +313,7 @@ int neu_adapter_rename(neu_adapter_t *adapter, const char *new_name)
return NEU_ERR_EINTERNAL;
}
zlog_category_t *log = zlog_get_category(name);
zlog_category_t *log = get_log_category(name);
if (NULL == log) {
free(name);
free(old_name);
@@ -348,22 +389,19 @@ static int adapter_register_metric(neu_adapter_t *adapter, const char *name,
uint64_t init)
{
if (NULL == adapter->metrics) {
adapter->metrics = calloc(1, sizeof(*adapter->metrics));
adapter->metrics =
neu_node_metrics_new(adapter, adapter->module->type, adapter->name);
if (NULL == adapter->metrics) {
return -1;
}
adapter->metrics->type = adapter->module->type;
adapter->metrics->name = adapter->name;
adapter->metrics->adapter = adapter;
neu_metrics_add_node(adapter);
}
if (0 > neu_metric_entries_add(&adapter->metrics->entries, name, help, type,
init)) {
if (0 >
neu_node_metrics_add(adapter->metrics, NULL, name, help, type, init)) {
return -1;
}
neu_metrics_register_entry(name, help, type);
return 0;
}
@@ -371,47 +409,17 @@ static int adapter_update_metric(neu_adapter_t *adapter,
const char *metric_name, uint64_t n,
const char *group)
{
neu_metric_entry_t *entry = NULL;
if (NULL == adapter->metrics) {
return -1;
}
if (NULL == group) {
HASH_FIND_STR(adapter->metrics->entries, metric_name, entry);
} else if (NULL != adapter->metrics->group_metrics) {
neu_group_metrics_t *g = NULL;
HASH_FIND_STR(adapter->metrics->group_metrics, group, g);
if (NULL != g) {
HASH_FIND_STR(g->entries, metric_name, entry);
}
}
if (NULL == entry) {
return -1;
}
if (NEU_METRIC_TYPE_COUNTER == entry->type) {
entry->value += n;
} else {
entry->value = n;
}
return 0;
return neu_node_metrics_update(adapter->metrics, group, metric_name, n);
}
static void adapter_reset_metrics(neu_adapter_t *adapter)
static inline void adapter_reset_metrics(neu_adapter_t *adapter)
{
neu_metric_entry_t *entry = NULL;
if (NULL == adapter->metrics) {
return;
}
HASH_LOOP(hh, adapter->metrics->entries, entry) { entry->value = 0; }
neu_group_metrics_t *g = NULL;
HASH_LOOP(hh, adapter->metrics->group_metrics, g)
{
HASH_LOOP(hh, g->entries, entry) { entry->value = 0; }
if (NULL != adapter->metrics) {
neu_node_metrics_reset(adapter->metrics);
}
}
@@ -598,11 +606,18 @@ static int adapter_trans_data(enum neu_event_io_type type, int fd,
return 0;
}
if (header->type == NEU_REQRESP_TRANS_DATA) {
if (adapter_msg_q_push(adapter->msg_q, msg) < 0) {
nlog_warn("adapter: %s trans data msg q is full, drop msg",
adapter->name);
neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
neu_msg_free(msg);
}
return 0;
}
adapter->module->intf_funs->request(
adapter->plugin, (neu_reqresp_head_t *) header, &header[1]);
if (header->type == NEU_REQRESP_TRANS_DATA) {
neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
}
neu_msg_free(msg);
return 0;
}
@@ -790,11 +805,13 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
neu_resp_get_node_state_t *resp =
(neu_resp_get_node_state_t *) &header[1];
neu_metric_entry_t *e = NULL;
if (NULL != adapter->metrics) {
pthread_mutex_lock(&adapter->metrics->lock);
neu_metric_entry_t *e = NULL;
HASH_FIND_STR(adapter->metrics->entries, NEU_METRIC_LAST_RTT_MS, e);
resp->rtt = NULL != e ? e->value : 0;
pthread_mutex_unlock(&adapter->metrics->lock);
}
resp->rtt = NULL != e ? e->value : 0;
resp->state = neu_adapter_get_state(adapter);
header->type = NEU_RESP_GET_NODE_STATE;
neu_msg_exchange(header);
@@ -1033,65 +1050,26 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_REQ_ADD_GTAG: {
neu_req_add_gtag_t *cmd = (neu_req_add_gtag_t *) &header[1];
neu_resp_add_tag_t resp = { 0 };
if (adapter->module->type == NEU_NA_TYPE_DRIVER) {
for (int i = 0; i < cmd->n_group; i++) {
for (int j = 0; j < cmd->groups[i].n_tag; j++) {
int ret = neu_adapter_driver_validate_tag(
(neu_adapter_driver_t *) adapter, cmd->groups[i].group,
&cmd->groups[i].tags[j]);
if (ret == 0) {
resp.index += 1;
} else {
resp.error = ret;
resp.index = 0;
i = cmd->n_group;
break;
}
}
}
} else {
if (adapter->module->type != NEU_NA_TYPE_DRIVER) {
resp.error = NEU_ERR_GROUP_NOT_ALLOW;
}
if (resp.index > 0) {
for (int i = 0; i < cmd->n_group; i++) {
int ret = neu_adapter_driver_try_add_tag(
(neu_adapter_driver_t *) adapter, cmd->groups[i].group,
cmd->groups[i].tags, cmd->groups[i].n_tag);
if (ret != 0) {
resp.index = 0;
resp.error = ret;
break;
} else if (neu_adapter_driver_new_group_count(
(neu_adapter_driver_t *) adapter, cmd) +
neu_adapter_driver_group_count(
(neu_adapter_driver_t *) adapter) >
NEU_GROUP_MAX_PER_NODE) {
resp.error = NEU_ERR_GROUP_MAX_GROUPS;
} else {
if (validate_gtags(adapter, cmd, &resp) == 0 &&
try_add_gtags(adapter, cmd, &resp) == 0 &&
add_gtags(adapter, cmd, &resp) == 0) {
for (int i = 0; i < cmd->n_group; i++) {
adapter_storage_add_tags(cmd->driver, cmd->groups[i].group,
cmd->groups[i].tags,
cmd->groups[i].n_tag);
}
}
}
if (resp.index > 0) {
for (int i = 0; i < cmd->n_group; i++) {
for (int j = 0; j < cmd->groups[i].n_tag; j++) {
int ret = neu_adapter_driver_add_tag(
(neu_adapter_driver_t *) adapter, cmd->groups[i].group,
&cmd->groups[i].tags[j], cmd->groups[i].interval);
if (ret != 0) {
neu_adapter_driver_try_del_tag(
(neu_adapter_driver_t *) adapter,
cmd->groups[i].n_tag - j);
resp.index = 0;
i = cmd->n_group;
resp.error = ret;
break;
}
}
}
}
if (resp.index) {
for (int i = 0; i < cmd->n_group; i++) {
adapter_storage_add_tags(cmd->driver, cmd->groups[i].group,
cmd->groups[i].tags,
cmd->groups[i].n_tag);
}
}
for (int i = 0; i < cmd->n_group; i++) {
for (int j = 0; j < cmd->groups[i].n_tag; j++) {
neu_tag_fini(&cmd->groups[i].tags[j]);
@@ -1258,6 +1236,97 @@ static int adapter_loop(enum neu_event_io_type type, int fd, void *usr_data)
return 0;
}
int validate_gtags(neu_adapter_t *adapter, neu_req_add_gtag_t *cmd,
neu_resp_add_tag_t *resp)
{
for (int group_index = 0; group_index < cmd->n_group; group_index++) {
for (int tag_index = 0; tag_index < cmd->groups[group_index].n_tag;
tag_index++) {
int validation_result = neu_adapter_driver_validate_tag(
(neu_adapter_driver_t *) adapter,
cmd->groups[group_index].group,
&cmd->groups[group_index].tags[tag_index]);
if (validation_result == 0) {
resp->index += 1;
} else {
resp->error = validation_result;
resp->index = 0;
return validation_result;
}
}
}
return 0;
}
int try_add_gtags(neu_adapter_t *adapter, neu_req_add_gtag_t *cmd,
neu_resp_add_tag_t *resp)
{
for (int group_index = 0; group_index < cmd->n_group; group_index++) {
int add_result = neu_adapter_driver_try_add_tag(
(neu_adapter_driver_t *) adapter, cmd->groups[group_index].group,
cmd->groups[group_index].tags, cmd->groups[group_index].n_tag);
if (add_result != 0) {
for (int added_groups_count = 0; added_groups_count < group_index;
added_groups_count++) {
neu_adapter_driver_try_del_tag(
(neu_adapter_driver_t *) adapter,
cmd->groups[added_groups_count].n_tag);
}
resp->index = 0;
resp->error = add_result;
return add_result;
}
}
return 0;
}
int add_gtags(neu_adapter_t *adapter, neu_req_add_gtag_t *cmd,
neu_resp_add_tag_t *resp)
{
for (int group_index = 0; group_index < cmd->n_group; group_index++) {
for (int tag_index = 0; tag_index < cmd->groups[group_index].n_tag;
tag_index++) {
int add_tag_result = neu_adapter_driver_add_tag(
(neu_adapter_driver_t *) adapter,
cmd->groups[group_index].group,
&cmd->groups[group_index].tags[tag_index],
cmd->groups[group_index].interval);
if (add_tag_result != 0) {
for (int added_group_index = 0; added_group_index < group_index;
added_group_index++) {
for (int added_tag_index = 0;
added_tag_index < cmd->groups[added_group_index].n_tag;
added_tag_index++) {
neu_adapter_driver_del_tag(
(neu_adapter_driver_t *) adapter,
cmd->groups[added_group_index].group,
cmd->groups[added_group_index]
.tags[added_tag_index]
.name);
}
}
for (int added_tag_index = 0; added_tag_index < tag_index;
added_tag_index++) {
neu_adapter_driver_del_tag(
(neu_adapter_driver_t *) adapter,
cmd->groups[group_index].group,
cmd->groups[group_index].tags[added_tag_index].name);
}
for (int groups_count = 0; groups_count < cmd->n_group;
groups_count++) {
neu_adapter_driver_try_del_tag(
(neu_adapter_driver_t *) adapter,
cmd->groups[groups_count].n_tag);
}
resp->index = 0;
resp->error = add_tag_result;
return add_tag_result;
}
}
}
return 0;
}
void neu_adapter_destroy(neu_adapter_t *adapter)
{
nlog_notice("adapter %s destroy", adapter->name);
@@ -1268,14 +1337,16 @@ void neu_adapter_destroy(neu_adapter_t *adapter)
if (NULL != adapter->metrics) {
neu_metrics_del_node(adapter);
neu_metric_entry_t *e = NULL;
HASH_LOOP(hh, adapter->metrics->entries, e)
{
neu_metrics_unregister_entry(e->name);
}
neu_node_metrics_free(adapter->metrics);
}
if (adapter->consumer_tid != 0) {
pthread_cancel(adapter->consumer_tid);
}
if (adapter->msg_q != NULL) {
adapter_msg_q_free(adapter->msg_q);
}
char *setting = NULL;
if (adapter_load_setting(adapter->name, &setting) != 0) {
remove_logs(adapter->name);
@@ -1460,110 +1531,43 @@ int neu_adapter_register_group_metric(neu_adapter_t *adapter,
const char *help, neu_metric_type_e type,
uint64_t init)
{
neu_group_metrics_t *group_metrics = NULL;
if (NULL == adapter->metrics) {
return -1;
}
if (0 > neu_metrics_register_entry(name, help, type)) {
return -1;
}
HASH_FIND_STR(adapter->metrics->group_metrics, group_name, group_metrics);
if (NULL == group_metrics) {
group_metrics = calloc(1, sizeof(*group_metrics));
if (NULL == group_metrics) {
return -1;
}
group_metrics->name = strdup(group_name);
if (NULL == group_metrics->name) {
free(group_metrics);
return -1;
}
HASH_ADD_STR(adapter->metrics->group_metrics, name, group_metrics);
}
if (0 > neu_metric_entries_add(&group_metrics->entries, name, help, type,
init)) {
return -1;
}
return 0;
return neu_node_metrics_add(adapter->metrics, group_name, name, help, type,
init);
}
int neu_adapter_update_group_metric(neu_adapter_t *adapter,
const char * group_name,
const char *metric_name, uint64_t n)
{
neu_metric_entry_t * entry = NULL;
neu_group_metrics_t *group_metrics = NULL;
if (NULL == adapter->metrics) {
return -1;
}
HASH_FIND_STR(adapter->metrics->group_metrics, group_name, group_metrics);
if (NULL == group_metrics) {
return -1;
}
HASH_FIND_STR(group_metrics->entries, metric_name, entry);
if (NULL == entry) {
return -1;
}
if (NEU_METRIC_TYPE_COUNTER == entry->type) {
entry->value += n;
} else {
entry->value = n;
}
return 0;
return neu_node_metrics_update(adapter->metrics, group_name, metric_name,
n);
}
int neu_adapter_metric_update_group_name(neu_adapter_t *adapter,
const char * group_name,
const char * new_group_name)
{
neu_group_metrics_t *group_metrics = NULL;
if (NULL == adapter->metrics) {
return -1;
}
HASH_FIND_STR(adapter->metrics->group_metrics, group_name, group_metrics);
if (NULL == group_metrics) {
return -1;
}
char *name = strdup(new_group_name);
if (NULL == name) {
return -1;
}
HASH_DEL(adapter->metrics->group_metrics, group_metrics);
free(group_metrics->name);
group_metrics->name = name;
HASH_ADD_STR(adapter->metrics->group_metrics, name, group_metrics);
return 0;
return neu_node_metrics_update_group(adapter->metrics, group_name,
new_group_name);
}
void neu_adapter_del_group_metrics(neu_adapter_t *adapter,
const char * group_name)
{
if (NULL == adapter->metrics) {
return;
}
neu_group_metrics_t *gm = NULL;
HASH_FIND_STR(adapter->metrics->group_metrics, group_name, gm);
if (NULL != gm) {
HASH_DEL(adapter->metrics->group_metrics, gm);
neu_metric_entry_t *e = NULL;
HASH_LOOP(hh, gm->entries, e) { neu_metrics_unregister_entry(e->name); }
neu_group_metrics_free(gm);
if (NULL != adapter->metrics) {
neu_node_metrics_del_group(adapter->metrics, group_name);
}
}
+4
View File
@@ -26,6 +26,7 @@
#include "adapter_info.h"
#include "core/manager.h"
#include "msg_q.h"
struct neu_adapter {
char *name;
@@ -45,6 +46,9 @@ struct neu_adapter {
int control_fd;
int trans_data_fd;
adapter_msg_q_t *msg_q;
pthread_t consumer_tid;
uint16_t trans_data_port;
neu_events_t *events;
+81 -11
View File
@@ -90,6 +90,7 @@ static void read_group(int64_t timestamp, int64_t timeout,
neu_driver_cache_t *cache, const char *group,
UT_array *tags, UT_array *tag_values);
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,
UT_array *tags, UT_array *tag_values);
static void update(neu_adapter_t *adapter, const char *group, const char *tag,
@@ -224,8 +225,9 @@ static void update_im(neu_adapter_t *adapter, const char *group,
data->group = strdup(group);
utarray_new(data->tags, neu_resp_tag_value_meta_icd());
read_report_group(global_timestamp, 0, driver->cache, group, tags,
data->tags);
read_report_group(global_timestamp, 0,
neu_adapter_get_tag_cache_type(&driver->adapter),
driver->cache, group, tags, data->tags);
if (utarray_len(data->tags) > 0) {
group_t *find = NULL;
@@ -418,7 +420,7 @@ void neu_adapter_driver_read_group(neu_adapter_driver_t *driver,
neu_resp_read_group_t resp = { 0 };
neu_group_t * group = g->group;
UT_array * tags = neu_group_get_read_tag(group);
UT_array *tags = neu_group_query_read_tag(group, cmd->name, cmd->desc);
utarray_new(resp.tags, neu_resp_tag_value_meta_icd());
@@ -574,20 +576,62 @@ static void cal_decimal(neu_type_e tag_type, neu_type_e value_type,
{
switch (tag_type) {
case NEU_TYPE_INT8:
case NEU_TYPE_UINT8:
if (value_type == NEU_TYPE_DOUBLE) {
value->i8 = (int8_t) round(value->d64 / decimal);
} else {
value->i8 = (int8_t) round(value->i64 / decimal);
}
break;
case NEU_TYPE_INT16:
case NEU_TYPE_UINT16:
if (value_type == NEU_TYPE_DOUBLE) {
value->i16 = (int16_t) round(value->d64 / decimal);
} else {
value->i16 = (int16_t) round(value->i64 / decimal);
}
break;
case NEU_TYPE_INT32:
case NEU_TYPE_UINT32:
if (value_type == NEU_TYPE_DOUBLE) {
value->i32 = (int32_t) round(value->d64 / decimal);
} else {
value->i32 = (int32_t) round(value->i64 / decimal);
}
break;
case NEU_TYPE_INT64:
case NEU_TYPE_UINT64:
if (value_type == NEU_TYPE_DOUBLE) {
value->i64 = (int64_t) round(value->d64 / decimal);
} else {
value->i64 = (int64_t) round(value->i64 / decimal);
}
break;
case NEU_TYPE_UINT8:
if (value_type == NEU_TYPE_DOUBLE) {
value->u8 = (uint8_t) round(value->d64 / decimal);
} else {
value->u8 = (uint8_t) round(value->u64 / decimal);
}
break;
case NEU_TYPE_UINT16:
case NEU_TYPE_WORD:
if (value_type == NEU_TYPE_DOUBLE) {
value->u16 = (uint16_t) round(value->d64 / decimal);
} else {
value->u16 = (uint16_t) round(value->u64 / decimal);
}
break;
case NEU_TYPE_UINT32:
case NEU_TYPE_LWORD:
if (value_type == NEU_TYPE_DOUBLE) {
value->u32 = (uint32_t) round(value->d64 / decimal);
} else {
value->u32 = (uint32_t) round(value->u64 / decimal);
}
break;
case NEU_TYPE_UINT64:
case NEU_TYPE_DWORD:
if (value_type == NEU_TYPE_INT64 || value_type == NEU_TYPE_LWORD) {
value->u64 = value->u64 / decimal;
} else if (value_type == NEU_TYPE_DOUBLE) {
value->u64 = (uint64_t) round(value->d64 / decimal);
value->d64 = (uint64_t) round(value->d64 / decimal);
} else {
value->u64 = (uint64_t) round(value->u64 / decimal);
}
break;
case NEU_TYPE_FLOAT:
@@ -1063,6 +1107,29 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver, const char *name)
return ret;
}
uint16_t neu_adapter_driver_group_count(neu_adapter_driver_t *driver)
{
u_int16_t num_groups = 0;
if (driver && driver->groups) {
num_groups = HASH_COUNT(driver->groups);
}
return num_groups;
}
uint16_t neu_adapter_driver_new_group_count(neu_adapter_driver_t *driver,
neu_req_add_gtag_t * cmd)
{
uint16_t new_groups_count = 0;
for (int i = 0; i < cmd->n_group; i++) {
group_t *group_in_driver;
HASH_FIND_STR(driver->groups, cmd->groups[i].group, group_in_driver);
if (!group_in_driver) {
new_groups_count++;
}
}
return new_groups_count;
}
static group_t *find_group(neu_adapter_driver_t *driver, const char *name)
{
group_t *find = NULL;
@@ -1445,6 +1512,7 @@ static int report_callback(void *usr_data)
read_report_group(global_timestamp,
neu_group_get_interval(group->group) *
NEU_DRIVER_TAG_CACHE_EXPIRE_TIME,
neu_adapter_get_tag_cache_type(&group->driver->adapter),
group->driver->cache, group->name, tags, data->tags);
if (utarray_len(data->tags) > 0) {
@@ -1618,6 +1686,7 @@ static int read_callback(void *usr_data)
}
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,
UT_array *tags, UT_array *tag_values)
{
@@ -1710,7 +1779,8 @@ static void read_report_group(int64_t timestamp, int64_t timeout,
break;
}
if (!neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC) &&
if (cache_type != NEU_TAG_CACHE_TYPE_NEVER &&
!neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC) &&
(timestamp - value.timestamp) > timeout && timeout > 0) {
if (value.value.type == NEU_TYPE_PTR) {
free(value.value.value.ptr.ptr);
+3
View File
@@ -53,6 +53,9 @@ int neu_adapter_driver_del_group(neu_adapter_driver_t *driver,
int neu_adapter_driver_group_exist(neu_adapter_driver_t *driver,
const char * name);
UT_array *neu_adapter_driver_get_group(neu_adapter_driver_t *driver);
uint16_t neu_adapter_driver_group_count(neu_adapter_driver_t *driver);
uint16_t neu_adapter_driver_new_group_count(neu_adapter_driver_t *driver,
neu_req_add_gtag_t * cmd);
int neu_adapter_driver_try_del_tag(neu_adapter_driver_t *driver, int n_tag);
int neu_adapter_driver_try_add_tag(neu_adapter_driver_t *driver,
+116
View File
@@ -0,0 +1,116 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2024 EMQ Technologies Co., Ltd All rights reserved.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 3 of the License, or (at your option) any later version.
*
* 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. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with this program; if not, write to the Free Software Foundation,
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#include <pthread.h>
#include "utils/log.h"
#include "utils/utextend.h"
#include "msg_q.h"
struct item {
neu_msg_t * msg;
struct item *prev, *next;
};
struct adapter_msg_q {
struct item *list;
uint32_t max;
uint32_t current;
char * name;
pthread_mutex_t mtx;
pthread_cond_t cond;
};
adapter_msg_q_t *adapter_msg_q_new(const char *name, uint32_t size)
{
struct adapter_msg_q *q = calloc(1, sizeof(struct adapter_msg_q));
pthread_mutex_init(&q->mtx, NULL);
pthread_cond_init(&q->cond, NULL);
q->max = size;
q->name = strdup(name);
q->current = 0;
return q;
}
void adapter_msg_q_free(adapter_msg_q_t *q)
{
struct item *tmp = NULL, *elt = NULL;
nlog_warn("app: %s, drop %u msg", q->name, q->current);
pthread_mutex_destroy(&q->mtx);
pthread_cond_destroy(&q->cond);
DL_FOREACH_SAFE(q->list, elt, tmp)
{
DL_DELETE(q->list, elt);
neu_reqresp_head_t *header = neu_msg_get_header(elt->msg);
neu_trans_data_free((neu_reqresp_trans_data_t *) &header[1]);
neu_msg_free(elt->msg);
free(elt);
}
free(q->name);
free(q);
}
int adapter_msg_q_push(adapter_msg_q_t *q, neu_msg_t *msg)
{
int ret = -1;
pthread_mutex_lock(&q->mtx);
if (q->current < q->max) {
q->current += 1;
struct item *elt = calloc(1, sizeof(struct item));
elt->msg = msg;
DL_APPEND(q->list, elt);
ret = 0;
}
pthread_mutex_unlock(&q->mtx);
if (ret == -1) {
nlog_warn("app: %s, msg q is full, %u(%u)", q->name, q->current,
q->max);
} else {
pthread_cond_signal(&q->cond);
}
return ret;
}
uint32_t adapter_msg_q_pop(adapter_msg_q_t *q, neu_msg_t **p_data)
{
uint32_t ret = 0;
pthread_mutex_lock(&q->mtx);
while (q->current == 0) {
pthread_cond_wait(&q->cond, &q->mtx);
}
struct item *elt = DL_LAST(q->list);
if (elt != NULL) {
DL_DELETE(q->list, elt);
*p_data = elt->msg;
free(elt);
q->current -= 1;
ret = q->current;
}
pthread_mutex_unlock(&q->mtx);
return ret;
}
+36
View File
@@ -0,0 +1,36 @@
/**
* NEURON IIoT System for Industry 4.0
* Copyright (C) 2020-2024 EMQ Technologies Co., Ltd All rights reserved.
*
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU Lesser General Public
* License as published by the Free Software Foundation; either
* version 3 of the License, or (at your option) any later version.
*
* 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. See the GNU
* Lesser General Public License for more details.
*
* You should have received a copy of the GNU Lesser General Public License
* along with this program; if not, write to the Free Software Foundation,
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#ifndef ADAPTER_MSG_Q_H
#define ADAPTER_MSG_Q_H
#include <stdint.h>
#include "base/msg_internal.h"
#include "msg.h"
typedef struct adapter_msg_q adapter_msg_q_t;
adapter_msg_q_t *adapter_msg_q_new(const char *name, uint32_t size);
void adapter_msg_q_free(adapter_msg_q_t *q);
int adapter_msg_q_push(adapter_msg_q_t *q, neu_msg_t *msg);
uint32_t adapter_msg_q_pop(adapter_msg_q_t *q, neu_msg_t **p_data);
#endif
+93 -25
View File
@@ -16,6 +16,7 @@
* along with this program; if not, write to the Free Software Foundation,
* Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
**/
#include <pthread.h>
#include <string.h>
#include <sys/time.h>
@@ -38,11 +39,11 @@ struct neu_group {
tag_elem_t *tags;
uint32_t interval;
int64_t timestamp;
int64_t timestamp;
pthread_mutex_t mtx;
};
static UT_array *to_array(tag_elem_t *tags);
static UT_array *to_read_array(tag_elem_t *tags);
static void split_static_array(tag_elem_t *tags, UT_array **static_tags,
UT_array **other_tags);
static void update_timestamp(neu_group_t *group);
@@ -53,6 +54,7 @@ neu_group_t *neu_group_new(const char *name, uint32_t interval)
group->name = strdup(name);
group->interval = interval;
pthread_mutex_init(&group->mtx, NULL);
return group;
}
@@ -61,6 +63,7 @@ void neu_group_destroy(neu_group_t *group)
{
tag_elem_t *el = NULL, *tmp = NULL;
pthread_mutex_lock(&group->mtx);
HASH_ITER(hh, group->tags, el, tmp)
{
HASH_DEL(group->tags, el);
@@ -68,7 +71,9 @@ void neu_group_destroy(neu_group_t *group)
neu_tag_free(el->tag);
free(el);
}
pthread_mutex_unlock(&group->mtx);
pthread_mutex_destroy(&group->mtx);
free(group->name);
free(group);
}
@@ -118,8 +123,10 @@ int neu_group_add_tag(neu_group_t *group, const neu_datatag_t *tag)
{
tag_elem_t *el = NULL;
pthread_mutex_lock(&group->mtx);
HASH_FIND_STR(group->tags, tag->name, el);
if (el != NULL) {
pthread_mutex_unlock(&group->mtx);
return NEU_ERR_TAG_NAME_CONFLICT;
}
@@ -129,6 +136,7 @@ int neu_group_add_tag(neu_group_t *group, const neu_datatag_t *tag)
HASH_ADD_STR(group->tags, name, el);
update_timestamp(group);
pthread_mutex_unlock(&group->mtx);
return 0;
}
@@ -138,6 +146,7 @@ int neu_group_update_tag(neu_group_t *group, const neu_datatag_t *tag)
tag_elem_t *el = NULL;
int ret = NEU_ERR_TAG_NOT_EXIST;
pthread_mutex_lock(&group->mtx);
HASH_FIND_STR(group->tags, tag->name, el);
if (el != NULL) {
neu_tag_copy(el->tag, tag);
@@ -145,6 +154,7 @@ int neu_group_update_tag(neu_group_t *group, const neu_datatag_t *tag)
update_timestamp(group);
ret = NEU_ERR_SUCCESS;
}
pthread_mutex_unlock(&group->mtx);
return ret;
}
@@ -154,6 +164,7 @@ int neu_group_del_tag(neu_group_t *group, const char *tag_name)
tag_elem_t *el = NULL;
int ret = NEU_ERR_TAG_NOT_EXIST;
pthread_mutex_lock(&group->mtx);
HASH_FIND_STR(group->tags, tag_name, el);
if (el != NULL) {
HASH_DEL(group->tags, el);
@@ -164,28 +175,33 @@ int neu_group_del_tag(neu_group_t *group, const char *tag_name)
update_timestamp(group);
ret = NEU_ERR_SUCCESS;
}
pthread_mutex_unlock(&group->mtx);
return ret;
}
UT_array *neu_group_get_tag(const neu_group_t *group)
UT_array *neu_group_get_tag(neu_group_t *group)
{
UT_array *array = NULL;
pthread_mutex_lock(&group->mtx);
array = to_array(group->tags);
pthread_mutex_unlock(&group->mtx);
return array;
}
UT_array *neu_group_query_tag(neu_group_t *group, const char *name)
static inline UT_array *
filter_tags(tag_elem_t *tags,
bool (*predicate)(const neu_datatag_t *, void *data), void *data)
{
tag_elem_t *el = NULL, *tmp = NULL;
UT_array * array = NULL;
utarray_new(array, neu_tag_get_icd());
HASH_ITER(hh, group->tags, el, tmp)
HASH_ITER(hh, tags, el, tmp)
{
if (strstr(el->tag->name, name) != NULL) {
if (predicate(el->tag, data)) {
utarray_push_back(array, el->tag);
}
}
@@ -193,11 +209,79 @@ UT_array *neu_group_query_tag(neu_group_t *group, const char *name)
return array;
}
static inline bool is_readable(const neu_datatag_t *tag, void *data)
{
(void) data;
return neu_tag_attribute_test(tag, NEU_ATTRIBUTE_READ) ||
neu_tag_attribute_test(tag, NEU_ATTRIBUTE_SUBSCRIBE) ||
neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC);
}
static inline bool name_contains(const neu_datatag_t *tag, void *data)
{
const char *name = data;
return strstr(tag->name, name) != NULL ||
(tag->description != NULL && strstr(tag->description, name) != NULL);
}
static inline bool description_contains(const neu_datatag_t *tag, void *data)
{
const char *str = data;
return tag->description && strstr(tag->description, str) != NULL;
}
struct query {
char *name;
char *desc;
};
static inline bool match_query(const neu_datatag_t *tag, void *data)
{
struct query *q = data;
return (!q->name || name_contains(tag, q->name)) &&
(!q->desc || description_contains(tag, q->desc));
}
static inline bool is_readable_and_match_query(const neu_datatag_t *tag,
void * data)
{
return is_readable(tag, NULL) && match_query(tag, data);
}
UT_array *neu_group_query_tag(neu_group_t *group, const char *name)
{
UT_array *array = NULL;
pthread_mutex_lock(&group->mtx);
array = filter_tags(group->tags, name_contains, (void *) name);
pthread_mutex_unlock(&group->mtx);
return array;
}
UT_array *neu_group_query_read_tag(neu_group_t *group, const char *name,
const char *desc)
{
UT_array * array = NULL;
struct query q = {
.name = (char *) name,
.desc = (char *) desc,
};
pthread_mutex_lock(&group->mtx);
array = filter_tags(group->tags, is_readable_and_match_query, &q);
pthread_mutex_unlock(&group->mtx);
return array;
}
UT_array *neu_group_get_read_tag(neu_group_t *group)
{
UT_array *array = NULL;
array = to_read_array(group->tags);
pthread_mutex_lock(&group->mtx);
array = filter_tags(group->tags, is_readable, NULL);
pthread_mutex_unlock(&group->mtx);
return array;
}
@@ -216,10 +300,12 @@ neu_datatag_t *neu_group_find_tag(neu_group_t *group, const char *tag)
tag_elem_t * find = NULL;
neu_datatag_t *result = NULL;
pthread_mutex_lock(&group->mtx);
HASH_FIND_STR(group->tags, tag, find);
if (find != NULL) {
result = neu_tag_dup(find->tag);
}
pthread_mutex_unlock(&group->mtx);
return result;
}
@@ -269,24 +355,6 @@ static UT_array *to_array(tag_elem_t *tags)
return array;
}
static UT_array *to_read_array(tag_elem_t *tags)
{
tag_elem_t *el = NULL, *tmp = NULL;
UT_array * array = NULL;
utarray_new(array, neu_tag_get_icd());
HASH_ITER(hh, tags, el, tmp)
{
if (neu_tag_attribute_test(el->tag, NEU_ATTRIBUTE_READ) ||
neu_tag_attribute_test(el->tag, NEU_ATTRIBUTE_SUBSCRIBE) ||
neu_tag_attribute_test(el->tag, NEU_ATTRIBUTE_STATIC)) {
utarray_push_back(array, el->tag);
}
}
return array;
}
static void split_static_array(tag_elem_t *tags, UT_array **static_tags,
UT_array **other_tags)
{
+3 -1
View File
@@ -38,9 +38,11 @@ int neu_group_update(neu_group_t *group, uint32_t interval);
int neu_group_add_tag(neu_group_t *group, const neu_datatag_t *tag);
int neu_group_update_tag(neu_group_t *group, const neu_datatag_t *tag);
int neu_group_del_tag(neu_group_t *group, const char *tag_name);
UT_array * neu_group_get_tag(const neu_group_t *group);
UT_array * neu_group_get_tag(neu_group_t *group);
UT_array * neu_group_query_tag(neu_group_t *group, const char *name);
UT_array * neu_group_get_read_tag(neu_group_t *group);
UT_array * neu_group_query_read_tag(neu_group_t *group, const char *name,
const char *desc);
uint16_t neu_group_tag_size(const neu_group_t *group);
neu_datatag_t *neu_group_find_tag(neu_group_t *group, const char *tag);
void neu_group_split_static_tags(neu_group_t *group, UT_array **static_tags,
+17 -6
View File
@@ -262,10 +262,19 @@ int neu_metric_entries_add(neu_metric_entry_t **entries, const char *name,
return -1;
}
entry->name = name;
entry->type = type;
entry->help = help;
entry->value = init;
if (NEU_METRIC_TYPE_ROLLING_COUNTER == type) {
// only allocate rolling counter for nonzero time span
if (init > 0 && NULL == (entry->rcnt = neu_rolling_counter_new(init))) {
free(entry);
return -1;
}
} else {
entry->value = init;
}
entry->name = name;
entry->type = type;
entry->help = help;
HASH_ADD_STR(*entries, name, entry);
return 0;
}
@@ -302,9 +311,11 @@ int neu_metrics_register_entry(const char *name, const char *help,
neu_metric_entry_t *e = NULL;
pthread_rwlock_wrlock(&g_metrics_mtx_);
// use `value` field as reference counter, initialize to zero
// and we don't need to allocate rolling counter for register entries
rv = neu_metric_entries_add(&g_metrics_.registered_metrics, name, help,
type, 1);
if (1 == rv) {
type, 0);
if (-1 != rv) {
HASH_FIND_STR(g_metrics_.registered_metrics, name, e);
++e->value;
rv = 0;
-3
View File
@@ -182,9 +182,6 @@ static inline neu_msg_t *neu_msg_new(neu_reqresp_type_e t, void *ctx,
case NEU_REQ_GET_TEMPLATES:
body_size = neu_reqresp_size(NEU_RESP_GET_TEMPLATES);
break;
case NEU_REQ_READ_GROUP:
body_size = neu_reqresp_size(NEU_RESP_READ_GROUP);
break;
default:
body_size = data_size;
}
+7 -8
View File
@@ -1274,30 +1274,29 @@ int neu_conn_stream_consume(neu_conn_t *conn, void *context,
ssize_t ret = neu_conn_recv(conn, conn->buf + conn->offset,
conn->buf_size - conn->offset);
if (ret > 0) {
pthread_mutex_lock(&conn->mtx);
zlog_recv_protocol(conn->param.log, conn->buf + conn->offset, ret);
conn->offset += ret;
neu_protocol_unpack_buf_t protocol_buf = { 0 };
neu_protocol_unpack_buf_init(&protocol_buf, conn->buf, conn->offset);
int used = 0;
while (neu_protocol_unpack_buf_unused_size(&protocol_buf) > 0) {
used = fn(context, &protocol_buf);
int used = fn(context, &protocol_buf);
zlog_debug(conn->param.log, "buf used: %d offset: %d", used,
conn->offset);
if (used == 0 || used == -1) {
if (used == 0) {
break;
} else if (used == -1) {
neu_conn_disconnect(conn);
break;
} else {
pthread_mutex_lock(&conn->mtx);
conn->offset -= used;
memmove(conn->buf, conn->buf + used, conn->offset);
neu_protocol_unpack_buf_init(&protocol_buf, conn->buf,
conn->offset);
pthread_mutex_unlock(&conn->mtx);
}
}
pthread_mutex_unlock(&conn->mtx);
if (used == -1) {
neu_conn_disconnect(conn);
}
}
return ret;
+20 -6
View File
@@ -615,7 +615,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
free(so_tmp_path);
free(schema_tmp_path);
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_LIBRARY_ADD_FAIL;
e.error = NEU_ERR_LIBRARY_UPDATE_FAIL;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
break;
@@ -628,7 +628,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
free(so_tmp_path);
free(schema_tmp_path);
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_LIBRARY_ADD_FAIL;
e.error = NEU_ERR_LIBRARY_UPDATE_FAIL;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
break;
@@ -641,7 +641,21 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
free(so_tmp_path);
free(schema_tmp_path);
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_LIBRARY_ADD_FAIL;
e.error = NEU_ERR_LIBRARY_UPDATE_FAIL;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
break;
}
if (neu_plugin_manager_update(manager->plugin_manager, cmd->library) !=
NEU_ERR_SUCCESS) {
nlog_warn("update library %s fail", cmd->library);
free(cmd->so_file);
free(cmd->schema_file);
free(so_tmp_path);
free(schema_tmp_path);
header->type = NEU_RESP_ERROR;
e.error = NEU_ERR_LIBRARY_UPDATE_FAIL;
strcpy(header->receiver, header->sender);
reply(manager, header, &e);
break;
@@ -986,7 +1000,7 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
manager_storage_del_node(manager, cmd->node);
if (neu_adapter_get_type(adapter) == NEU_NA_TYPE_APP) {
UT_array *subscriptions = neu_subscribe_manager_get(
manager->subscribe_manager, cmd->node);
manager->subscribe_manager, cmd->node, NULL, NULL);
neu_subscribe_manager_unsub_all(manager->subscribe_manager,
cmd->node);
@@ -1151,8 +1165,8 @@ static int manager_loop(enum neu_event_io_type type, int fd, void *usr_data)
case NEU_REQ_GET_SUBSCRIBE_GROUP: {
neu_req_get_subscribe_group_t *cmd =
(neu_req_get_subscribe_group_t *) &header[1];
UT_array *groups =
neu_manager_get_sub_group_deep_copy(manager, cmd->app);
UT_array *groups = neu_manager_get_sub_group_deep_copy(
manager, cmd->app, cmd->driver, cmd->group);
neu_resp_get_subscribe_group_t resp = { .groups = groups };
strcpy(header->receiver, header->sender);
+7 -4
View File
@@ -698,7 +698,6 @@ int neu_manager_subscribe(neu_manager_t *manager, const char *app,
// this is not an elegant solution due to the current architecture
if (params && 0 == strcmp(adapter->module->module_name, "MQTT")) {
neu_json_elem_t elem = { .name = "topic", .t = NEU_JSON_STR };
neu_json_decode((char *) params, 1, &elem);
neu_parse_param(params, NULL, 1, &elem);
if (elem.v.val_str && 0 == strlen(elem.v.val_str)) {
free(elem.v.val_str);
@@ -791,13 +790,17 @@ int neu_manager_unsubscribe(neu_manager_t *manager, const char *app,
UT_array *neu_manager_get_sub_group(neu_manager_t *manager, const char *app)
{
return neu_subscribe_manager_get(manager->subscribe_manager, app);
return neu_subscribe_manager_get(manager->subscribe_manager, app, NULL,
NULL);
}
UT_array *neu_manager_get_sub_group_deep_copy(neu_manager_t *manager,
const char * app)
const char * app,
const char * driver,
const char * group)
{
UT_array *subs = neu_subscribe_manager_get(manager->subscribe_manager, app);
UT_array *subs = neu_subscribe_manager_get(manager->subscribe_manager, app,
driver, group);
utarray_foreach(subs, neu_resp_subscribe_info_t *, sub)
{
+3 -1
View File
@@ -112,7 +112,9 @@ int neu_manager_unsubscribe(neu_manager_t *manager, const char *app,
const char *driver, const char *group);
UT_array *neu_manager_get_sub_group(neu_manager_t *manager, const char *app);
UT_array *neu_manager_get_sub_group_deep_copy(neu_manager_t *manager,
const char * app);
const char * app,
const char * driver,
const char * group);
int neu_manager_add_ndriver_map(neu_manager_t *manager, const char *ndriver,
const char *driver, const char *group);
int neu_manager_del_ndriver_map(neu_manager_t *manager, const char *ndriver,
+1 -1
View File
@@ -386,4 +386,4 @@ UT_array *neu_node_manager_get_state(neu_node_manager_t *mgr)
}
return states;
}
}
+75 -3
View File
@@ -98,7 +98,7 @@ int neu_plugin_manager_add(neu_plugin_manager_t *mgr,
snprintf(lib_path, sizeof(lib_path) - 1, "%s/%s", lib_paths[i],
plugin_lib_name);
handle = dlopen(lib_path, RTLD_NOW | RTLD_NODELETE);
handle = dlopen(lib_path, RTLD_NOW);
if (handle == NULL) {
nlog_warn("failed to open library %s, error: %s", lib_path,
@@ -173,6 +173,78 @@ int neu_plugin_manager_add(neu_plugin_manager_t *mgr,
return NEU_ERR_SUCCESS;
}
int neu_plugin_manager_update(neu_plugin_manager_t *mgr,
const char * plugin_lib_name)
{
char lib_path[256] = { 0 };
void * handle = NULL;
void * module = NULL;
neu_plugin_module_t *pm = NULL;
plugin_entity_t * plugin = NULL;
char lib_paths[3][64] = { 0 };
snprintf(lib_paths[0], sizeof(lib_paths[0]), "%s", g_plugin_dir);
snprintf(lib_paths[1], sizeof(lib_paths[1]), "%s/system", g_plugin_dir);
snprintf(lib_paths[2], sizeof(lib_paths[2]), "%s/custom", g_plugin_dir);
assert(strlen(plugin_lib_name) <= NEU_PLUGIN_LIBRARY_LEN);
for (size_t i = 0; i < sizeof(lib_paths) / sizeof(lib_paths[0]); i++) {
snprintf(lib_path, sizeof(lib_path) - 1, "%s/%s", lib_paths[i],
plugin_lib_name);
handle = dlopen(lib_path, RTLD_NOW);
if (handle == NULL) {
nlog_warn("failed to open library %s, error: %s", lib_path,
dlerror());
} else {
break;
}
}
if (handle == NULL) {
return NEU_ERR_LIBRARY_FAILED_TO_OPEN;
}
module = dlsym(handle, "neu_plugin_module");
if (module == NULL) {
dlclose(handle);
nlog_warn("failed to find neu_plugin_module symbol in %s", lib_path);
return NEU_ERR_LIBRARY_MODULE_INVALID;
}
pm = (neu_plugin_module_t *) module;
HASH_FIND_STR(mgr->plugins, pm->module_name, plugin);
if (plugin == NULL) {
dlclose(handle);
return NEU_ERR_LIBRARY_MODULE_NOT_EXISTS;
}
plugin->version = pm->version;
plugin->display = pm->display;
plugin->type = pm->type;
plugin->kind = pm->kind;
free(plugin->schema);
free(plugin->description);
free(plugin->description_zh);
plugin->schema = strdup(pm->schema);
plugin->description = strdup(pm->module_descr);
plugin->description_zh = strdup(pm->module_descr_zh);
plugin->single = pm->single;
if (plugin->single) {
free(plugin->single_name);
plugin->single_name = strdup(pm->single_name);
}
nlog_notice("update plugin, name: %s, library: %s, kind: %d, type: %d",
plugin->name, plugin->lib_name, plugin->kind, plugin->type);
dlclose(handle);
return NEU_ERR_SUCCESS;
}
int neu_plugin_manager_del(neu_plugin_manager_t *mgr, const char *plugin_name)
{
plugin_entity_t *plugin = NULL;
@@ -322,7 +394,7 @@ int neu_plugin_manager_create_instance(neu_plugin_manager_t * mgr,
snprintf(lib_path, sizeof(lib_path) - 1, "%s/%s", lib_paths[i],
plugin->lib_name);
instance->handle = dlopen(lib_path, RTLD_NOW | RTLD_NODELETE);
instance->handle = dlopen(lib_path, RTLD_NOW);
if (instance->handle == NULL) {
continue;
@@ -387,7 +459,7 @@ bool neu_plugin_manager_create_instance_by_path(neu_plugin_manager_t *mgr,
(void) mgr;
instance->handle = dlopen(plugin_path, RTLD_NOW | RTLD_NODELETE);
instance->handle = dlopen(plugin_path, RTLD_NOW);
if (instance->handle == NULL) {
const char *dl_err = dlerror();
+3
View File
@@ -71,4 +71,7 @@ bool neu_plugin_manager_create_instance_by_path(neu_plugin_manager_t *mgr,
bool neu_plugin_manager_remove_library(neu_plugin_manager_t *mgr,
const char * library);
int neu_plugin_manager_update(neu_plugin_manager_t *mgr,
const char * plugin_lib_name);
#endif
+5 -2
View File
@@ -105,7 +105,8 @@ UT_array *neu_subscribe_manager_find_by_driver(neu_subscribe_mgr_t *mgr,
return apps;
}
UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app)
UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app,
const char *driver, const char *group)
{
sub_elem_t *el = NULL, *tmp = NULL;
UT_array * groups = NULL;
@@ -116,7 +117,9 @@ UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app)
{
utarray_foreach(el->apps, neu_app_subscribe_t *, sub_app)
{
if (strcmp(sub_app->app_name, app) == 0) {
if (strcmp(sub_app->app_name, app) == 0 &&
(!driver || strstr(el->key.driver, driver)) &&
(!group || strstr(el->key.group, group))) {
neu_resp_subscribe_info_t info = { 0 };
strncpy(info.driver, el->key.driver, sizeof(info.driver));
+2 -1
View File
@@ -47,7 +47,8 @@ UT_array *neu_subscribe_manager_find(neu_subscribe_mgr_t *mgr,
const char *driver, const char *group);
UT_array *neu_subscribe_manager_find_by_driver(neu_subscribe_mgr_t *mgr,
const char * driver);
UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app);
UT_array *neu_subscribe_manager_get(neu_subscribe_mgr_t *mgr, const char *app,
const char *driver, const char *group);
size_t neu_subscribe_manager_group_count(const neu_subscribe_mgr_t *mgr,
const char * app);
UT_array *neu_subscribe_manager_get_ndriver_maps(neu_subscribe_mgr_t *mgr,
+25 -12
View File
@@ -66,7 +66,7 @@ struct event_data {
bool use;
};
#define EVENT_SIZE 1024
#define EVENT_SIZE 1400
struct neu_events {
int epoll_fd;
@@ -182,8 +182,12 @@ neu_events_t *neu_event_new(void)
neu_events_t *events = calloc(1, sizeof(struct neu_events));
events->epoll_fd = epoll_create(1);
events->stop = false;
events->n_event = 0;
nlog_notice("create epoll: %d(%d)", events->epoll_fd, errno);
assert(events->epoll_fd > 0);
events->stop = false;
events->n_event = 0;
pthread_mutex_init(&events->mtx, NULL);
pthread_create(&events->thread, NULL, event_loop, events);
@@ -235,6 +239,7 @@ neu_event_timer_t *neu_event_add_timer(neu_events_t * events,
timer_ctx->event_data->usr_data = timer.usr_data;
timer_ctx->event_data->callback.timer = timer.cb;
timer_ctx->event_data->ctx.timer = events->event_datas[index].ctx.timer;
timer_ctx->event_data->index = index;
timer_ctx->value = value;
timer_ctx->fd = timer_fd;
@@ -247,17 +252,17 @@ neu_event_timer_t *neu_event_add_timer(neu_events_t * events,
zlog_notice(neuron,
"add timer, second: %" PRId64 ", millisecond: %" PRId64
", timer: %d in epoll %d, "
"ret: %d",
"ret: %d, index: %d",
timer.second, timer.millisecond, timer_fd, events->epoll_fd,
ret);
ret, index);
return timer_ctx;
}
int neu_event_del_timer(neu_events_t *events, neu_event_timer_t *timer)
{
zlog_notice(neuron, "del timer: %d from epoll: %d", timer->fd,
events->epoll_fd);
zlog_notice(neuron, "del timer: %d from epoll: %d, index: %d", timer->fd,
events->epoll_fd, timer->event_data->index);
timer->stop = true;
epoll_ctl(events->epoll_fd, EPOLL_CTL_DEL, timer->fd, NULL);
@@ -273,8 +278,13 @@ int neu_event_del_timer(neu_events_t *events, neu_event_timer_t *timer)
neu_event_io_t *neu_event_add_io(neu_events_t *events, neu_event_io_param_t io)
{
int ret = 0;
int index = get_free_event(events);
int ret = 0;
int index = get_free_event(events);
nlog_notice("add io, fd: %d, epoll: %d, index: %d", io.fd, events->epoll_fd,
index);
assert(index >= 0);
neu_event_io_t *io_ctx = &events->event_datas[index].ctx.io;
io_ctx->event_data = &events->event_datas[index];
struct epoll_event event = {
@@ -287,13 +297,15 @@ neu_event_io_t *neu_event_add_io(neu_events_t *events, neu_event_io_param_t io)
io_ctx->event_data->usr_data = io.usr_data;
io_ctx->event_data->callback.io = io.cb;
io_ctx->event_data->ctx.io = events->event_datas[index].ctx.io;
io_ctx->event_data->index = index;
io_ctx->fd = io.fd;
ret = epoll_ctl(events->epoll_fd, EPOLL_CTL_ADD, io.fd, &event);
nlog_notice("add io, fd: %d, epoll: %d, ret: %d", io.fd, events->epoll_fd,
ret);
nlog_notice("add io, fd: %d, epoll: %d, ret: %d(%d), index: %d", io.fd,
events->epoll_fd, ret, errno, index);
assert(ret == 0);
return io_ctx;
}
@@ -304,7 +316,8 @@ int neu_event_del_io(neu_events_t *events, neu_event_io_t *io)
return 0;
}
zlog_notice(neuron, "del io: %d from epoll: %d", io->fd, events->epoll_fd);
zlog_notice(neuron, "del io: %d from epoll: %d, index: %d", io->fd,
events->epoll_fd, io->event_data->index);
epoll_ctl(events->epoll_fd, EPOLL_CTL_DEL, io->fd, NULL);
free_event(events, io->event_data->index);
+2 -2
View File
@@ -125,9 +125,9 @@ static int neuron_run(const neu_cli_args_t *args)
int rv = 0;
signal(SIGINT, sig_handler);
signal(SIGTERM, sig_handler);
signal(SIGABRT, sig_handler);
signal(SIGSEGV, sig_handler);
signal(SIGTERM, sig_handler);
signal(SIGKILL, sig_handler);
// try to enable core dump
rl.rlim_cur = rl.rlim_max = RLIM_INFINITY;
+30
View File
@@ -430,6 +430,11 @@ int neu_json_decode_read_req(char *buf, neu_json_read_req_t **result)
.t = NEU_JSON_BOOL,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
},
{
.name = "query",
.t = NEU_JSON_OBJECT,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
},
};
ret = neu_json_decode_by_json(json_obj, NEU_JSON_ELEM_SIZE(req_elems),
req_elems);
@@ -441,11 +446,36 @@ int neu_json_decode_read_req(char *buf, neu_json_read_req_t **result)
req->group = req_elems[1].v.val_str;
req->sync = req_elems[2].v.val_bool;
neu_json_elem_t query_elems[] = {
{
.name = "name",
.t = NEU_JSON_STR,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
},
{
.name = "description",
.t = NEU_JSON_STR,
.attribute = NEU_JSON_ATTRIBUTE_OPTIONAL,
},
};
if (req_elems[3].v.val_object) {
ret = neu_json_decode_by_json(req_elems[3].v.val_object,
NEU_JSON_ELEM_SIZE(query_elems),
query_elems);
if (ret != 0) {
goto error;
}
req->name = query_elems[0].v.val_str;
req->desc = query_elems[1].v.val_str;
}
*result = req;
neu_json_decode_free(json_obj);
return ret;
error:
free(query_elems[0].v.val_str);
free(query_elems[1].v.val_str);
free(req_elems[0].v.val_str);
free(req_elems[1].v.val_str);
free(req);
+1 -7
View File
@@ -467,17 +467,11 @@ int neu_json_decode_gtag_json(void *json_obj, neu_json_gtag_t *gtag_p)
gtag_p->n_tag = arr.len;
gtag_p->tags = arr.tags;
goto decode_exit;
return 0;
decode_fail:
free(gtag_elems[0].v.val_str);
return -1;
decode_exit:
if (json_obj != NULL) {
neu_json_decode_free(json_obj);
}
return 0;
}
void neu_json_decode_gtag_fini(neu_json_gtag_t *gtag)
+12 -10
View File
@@ -287,7 +287,6 @@ int neu_http_response(nng_aio *aio, neu_err_code_e code, char *content)
case NEU_ERR_LICENSE_EXPIRED:
case NEU_ERR_LICENSE_DISABLED:
case NEU_ERR_LICENSE_MAX_NODES:
case NEU_ERR_LICENSE_MAX_TAGS:
case NEU_ERR_LICENSE_TOKEN_NOT_MATCH:
case NEU_ERR_GROUP_ALREADY_SUBSCRIBED:
case NEU_ERR_PLUGIN_TAG_TYPE_MISMATCH:
@@ -341,12 +340,24 @@ int neu_http_response(nng_aio *aio, neu_err_code_e code, char *content)
case NEU_ERR_PLUGIN_NAME_TOO_LONG:
case NEU_ERR_PLUGIN_NOT_SUPPORT_TEMPLATE:
case NEU_ERR_LIBRARY_ADD_FAIL:
case NEU_ERR_LIBRARY_UPDATE_FAIL:
case NEU_ERR_LIBRARY_MODULE_NOT_EXISTS:
case NEU_ERR_LIBRARY_MODULE_KIND_NOT_SUPPORT:
case NEU_ERR_LIBRARY_MODULE_VERSION_NOT_MATCH:
case NEU_ERR_LIBRARY_NAME_NOT_CONFORM:
case NEU_ERR_LIBRARY_CLIB_NOT_MATCH:
case NEU_ERR_LIBRARY_ARCH_NOT_SUPPORT:
case NEU_ERR_TAG_ATTRIBUTE_NOT_SUPPORT:
case NEU_ERR_TAG_TYPE_NOT_SUPPORT:
case NEU_ERR_TAG_ADDRESS_FORMAT_INVALID:
case NEU_ERR_TAG_DESCRIPTION_TOO_LONG:
case NEU_ERR_TAG_PRECISION_INVALID:
case NEU_ERR_TAG_NAME_TOO_LONG:
case NEU_ERR_TAG_ADDRESS_TOO_LONG:
case NEU_ERR_GROUP_MAX_GROUPS:
case NEU_ERR_LICENSE_MAX_TAGS:
case NEU_ERR_LICENSE_BAD_CLOCK:
case NEU_ERR_LICENSE_MODULE_INVALID:
status = NNG_HTTP_STATUS_BAD_REQUEST;
break;
case NEU_ERR_FILE_NOT_EXIST:
@@ -382,15 +393,6 @@ int neu_http_response(nng_aio *aio, neu_err_code_e code, char *content)
case NEU_ERR_LIBRARY_SYSTEM_NOT_ALLOW_DEL:
status = NNG_HTTP_STATUS_CONFLICT;
break;
case NEU_ERR_TAG_ATTRIBUTE_NOT_SUPPORT:
case NEU_ERR_TAG_TYPE_NOT_SUPPORT:
case NEU_ERR_TAG_ADDRESS_FORMAT_INVALID:
case NEU_ERR_TAG_DESCRIPTION_TOO_LONG:
case NEU_ERR_TAG_PRECISION_INVALID:
case NEU_ERR_TAG_NAME_TOO_LONG:
case NEU_ERR_TAG_ADDRESS_TOO_LONG:
status = NNG_HTTP_STATUS_PARTIAL_CONTENT;
break;
default:
if (code >= NEU_ERR_PLUGIN_ERROR_START &&
code <= NEU_ERR_PLUGIN_ERROR_END) {
+62 -12
View File
@@ -68,15 +68,15 @@ coil_bit_5 = [{"name": "coil_bit_5", "address": "1!000010",
"attribute": config.NEU_TAG_ATTRIBUTE_RW, "type": config.NEU_TYPE_BIT}]
input_bit_1 = [{"name": "input_bit_1", "address": "1!100001",
"attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT}]
"attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT, "description": "bit_1"}]
input_bit_2 = [{"name": "input_bit_2", "address": "1!100002",
"attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT}]
"attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT, "description": "bit_2"}]
input_bit_3 = [{"name": "input_bit_3", "address": "1!100003",
"attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT}]
"attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT, "description": "bit_3"}]
input_bit_4 = [{"name": "input_bit_4", "address": "1!100005",
"attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT}]
"attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT, "description": "bit_4"}]
input_bit_5 = [{"name": "input_bit_5", "address": "1!100010",
"attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT}]
"attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT, "description": "bit_5"}]
input_register_bit = [{"name": "input_register_bit", "address": "1!30101.15",
"attribute": config.NEU_TAG_ATTRIBUTE_READ, "type": config.NEU_TYPE_BIT}]
@@ -116,6 +116,8 @@ hold_string_s = [{"name": "hold_string_s", "address": "1!400870.2",
hold_int16_decimal = [{"name": "hold_int16_decimal", "address": "1!40103",
"attribute": config.NEU_TAG_ATTRIBUTE_RW, "type": config.NEU_TYPE_INT16, "decimal": 0.1}]
hold_int16_decimal_neg = [{"name": "int16_decimal_neg", "address": "1!40140",
"attribute": config.NEU_TAG_ATTRIBUTE_RW, "type": config.NEU_TYPE_INT16, "decimal": -1}]
hold_uint16_decimal = [{"name": "hold_uint16_decimal", "address": "1!400105",
"attribute": config.NEU_TAG_ATTRIBUTE_RW, "type": config.NEU_TYPE_UINT16, "decimal": 0.2}]
hold_int32_decimal = [{"name": "hold_int32_decimal", "address": "1!400199",
@@ -124,6 +126,8 @@ hold_uint32_decimal = [{"name": "hold_uint32_decimal", "address": "1!400115",
"attribute": config.NEU_TAG_ATTRIBUTE_RW, "type": config.NEU_TYPE_UINT32, "decimal": 0.3}]
hold_float_decimal = [{"name": "hold_float_decimal", "address": "1!400119",
"attribute": config.NEU_TAG_ATTRIBUTE_RW, "type": config.NEU_TYPE_FLOAT, "decimal": 0.5}]
hold_float_decimal_neg = [{"name": "float_decimal_neg", "address": "1!400150",
"attribute": config.NEU_TAG_ATTRIBUTE_RW, "type": config.NEU_TYPE_FLOAT, "decimal": -1}]
hold_double_decimal = [{"name": "hold_double_decimal", "address": "1!4000127",
"attribute": config.NEU_TAG_ATTRIBUTE_RW, "type": config.NEU_TYPE_DOUBLE, "decimal": 0.6}]
@@ -378,27 +382,27 @@ class TestModbus:
"attribute": config.NEU_TAG_ATTRIBUTE_RW, "type": config.NEU_TYPE_BYTES}]
response = api.add_tags(node=param[0], group='group', tags=wrong_tag1)
assert 200 == response.status_code
assert 400 == response.status_code
assert error.NEU_ERR_TAG_ATTRIBUTE_NOT_SUPPORT == response.json()[
'error']
response = api.add_tags(node=param[0], group='group', tags=wrong_tag2)
assert 200 == response.status_code
assert 400 == response.status_code
assert error.NEU_ERR_TAG_TYPE_NOT_SUPPORT == response.json()[
'error']
response = api.add_tags(node=param[0], group='group', tags=wrong_tag3)
assert 200 == response.status_code
assert 400 == response.status_code
assert error.NEU_ERR_TAG_ADDRESS_FORMAT_INVALID == response.json()[
'error']
response = api.add_tags(node=param[0], group='group', tags=wrong_tag4)
assert 200 == response.status_code
assert 400 == response.status_code
assert error.NEU_ERR_TAG_ADDRESS_FORMAT_INVALID == response.json()[
'error']
response = api.add_tags(node=param[0], group='group', tags=wrong_tag5)
assert 200 == response.status_code
assert 400 == response.status_code
assert error.NEU_ERR_TAG_ADDRESS_FORMAT_INVALID == response.json()[
'error']
@@ -490,6 +494,40 @@ class TestModbus:
assert 0 == api.read_tag(
node=param[0], group='group', tag=input_bit_5[0]['name'])
@description(given="created modbus node and tags", when="read tags fuzz", then="should return correct result")
def test_read_tags_fuzz(self, param):
response = api.read_tags(
node=param[0], group='group', query={"name": "input_bit"})
assert 200 == response.status_code
assert 5 == len(response.json()["tags"])
response = api.read_tags(
node=param[0], group='group', query={"name": "input_bit_1"})
assert 200 == response.status_code
assert 1 == len(response.json()["tags"])
response = api.read_tags(
node=param[0], group='group', query={"description": "bit"})
assert 200 == response.status_code
assert 5 == len(response.json()["tags"])
response = api.read_tags(
node=param[0], group='group', query={"description": "bit_1"})
assert 200 == response.status_code
assert 1 == len(response.json()["tags"])
response = api.read_tags(
node=param[0], group='group', query={"name": "input_bit", "description": "bit"})
assert 200 == response.status_code
assert 5 == len(response.json()["tags"])
response = api.read_tags(
node=param[0], group='group', query={"name": "input_bit_1", "description": "bit_1"})
assert 200 == response.status_code
assert 1 == len(response.json()["tags"])
@description(given="created modbus node and tags", when="read tags synchronously", then="read fail")
def test_read_tags_sync(self, param):
assert error.NEU_ERR_PLUGIN_NOT_SUPPORT_READ_SYNC == api.read_tag_error(
@@ -567,7 +605,7 @@ class TestModbus:
up_tag[0]['type'] = config.NEU_TYPE_BIT
response = api.update_tags(
node=param[0], group='group', tags=up_tag)
assert 200 == response.status_code
assert 400 == response.status_code
assert error.NEU_ERR_TAG_ATTRIBUTE_NOT_SUPPORT == response.json()[
'error']
@@ -633,6 +671,8 @@ class TestModbus:
def test_add_decimal_tag(self, param):
api.add_tags_check(
node=param[0], group='group', tags=hold_int16_decimal)
api.add_tags_check(
node=param[0], group='group', tags=hold_int16_decimal_neg)
api.add_tags_check(
node=param[0], group='group', tags=hold_uint16_decimal)
api.add_tags_check(
@@ -641,6 +681,8 @@ class TestModbus:
node=param[0], group='group', tags=hold_uint32_decimal)
api.add_tags_check(
node=param[0], group='group', tags=hold_float_decimal)
api.add_tags_check(
node=param[0], group='group', tags=hold_float_decimal_neg)
api.add_tags_check(
node=param[0], group='group', tags=hold_double_decimal)
@@ -648,6 +690,8 @@ class TestModbus:
def test_read_write_decimal_tag(self, param):
api.write_tag_check(
node=param[0], group='group', tag=hold_int16_decimal[0]['name'], value=11)
api.write_tag_check(
node=param[0], group='group', tag=hold_int16_decimal_neg[0]['name'], value=345)
api.write_tag_check(
node=param[0], group='group', tag=hold_uint16_decimal[0]['name'], value=23)
api.write_tag_check(
@@ -656,12 +700,16 @@ class TestModbus:
node=param[0], group='group', tag=hold_uint32_decimal[0]['name'], value=123)
api.write_tag_check(
node=param[0], group='group', tag=hold_float_decimal[0]['name'], value=121.314)
api.write_tag_check(
node=param[0], group='group', tag=hold_float_decimal_neg[0]['name'], value=345.6)
api.write_tag_check(
node=param[0], group='group', tag=hold_double_decimal[0]['name'], value=513.11)
time.sleep(0.3)
assert 11 == api.read_tag(
node=param[0], group='group', tag=hold_int16_decimal[0]['name'])
assert 345 == api.read_tag(
node=param[0], group='group', tag=hold_int16_decimal_neg[0]['name'])
assert 23 == api.read_tag(
node=param[0], group='group', tag=hold_uint16_decimal[0]['name'])
assert 123 == api.read_tag(
@@ -670,6 +718,8 @@ class TestModbus:
node=param[0], group='group', tag=hold_uint32_decimal[0]['name'])
compare_float(121.314, api.read_tag(
node=param[0], group='group', tag=hold_float_decimal[0]['name']))
compare_float(345.6, api.read_tag(
node=param[0], group='group', tag=hold_float_decimal[0]['name']))
assert 513.11 == api.read_tag(
node=param[0], group='group', tag=hold_double_decimal[0]['name'])
@@ -929,7 +979,7 @@ class TestModbus:
node=param[0], group='group', tag=hold_string_l1[0]['name'], value='a')
api.write_tag_check(
node=param[0], group='group', tag=hold_string_l2[0]['name'], value='b')
time.sleep(0.3)
time.sleep(0.5)
assert 'a' == api.read_tag(
node=param[0], group='group', tag=hold_string_l1[0]['name'])
assert 'b' == api.read_tag(
+97
View File
@@ -188,6 +188,103 @@ class TestHttp:
assert len(config_data['settings']) == 1
assert config_data['settings'][0]['node'] == 'mqtt'
@description(given="running neuron", when="get subscribe info", then="should success")
def test_get_subscribe_info(self):
app = 'MQTT-1'
params = {'topic':'/neuron/test'}
driver_1 = 'modbus-01'
driver_2 = 'modbus-02'
group_1 = 'group-01'
group_2 = 'group-02'
try:
api.add_node_check(app, 'MQTT')
api.add_node_check(driver_1, 'Modbus TCP')
api.add_node_check(driver_2, 'Modbus TCP')
api.add_group_check(driver_1, group_1, 1000)
api.add_group_check(driver_1, group_2, 1000)
api.add_group_check(driver_2, group_1, 1000)
api.add_group_check(driver_2, group_2, 1000)
api.subscribe_group_check(app, driver_1, group_1, params=params)
api.subscribe_group_check(app, driver_1, group_2, params=params)
api.subscribe_group_check(app, driver_2, group_1, params=params)
api.subscribe_group_check(app, driver_2, group_2, params=params)
# match all drivers
response = api.get_subscribe_group(app, driver='modbus')
assert 200 == response.status_code
groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
assert len(groups) == 4
assert groups[0]['driver'] == driver_1
assert groups[0]['group'] == group_1
assert groups[1]['driver'] == driver_1
assert groups[1]['group'] == group_2
assert groups[2]['driver'] == driver_2
assert groups[2]['group'] == group_1
assert groups[3]['driver'] == driver_2
assert groups[3]['group'] == group_2
# match one driver
response = api.get_subscribe_group(app, driver=driver_1)
assert 200 == response.status_code
groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
assert len(groups) == 2
assert groups[0]['driver'] == driver_1
assert groups[0]['group'] == group_1
assert groups[1]['driver'] == driver_1
assert groups[1]['group'] == group_2
# match none driver
response = api.get_subscribe_group(app, driver='none')
assert 200 == response.status_code
assert len(response.json()['groups']) == 0
# match all groups
response = api.get_subscribe_group(app, group='group')
assert 200 == response.status_code
groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
assert len(groups) == 4
assert groups[0]['driver'] == driver_1
assert groups[0]['group'] == group_1
assert groups[1]['driver'] == driver_1
assert groups[1]['group'] == group_2
assert groups[2]['driver'] == driver_2
assert groups[2]['group'] == group_1
assert groups[3]['driver'] == driver_2
assert groups[3]['group'] == group_2
# match one group
response = api.get_subscribe_group(app, group=group_1)
assert 200 == response.status_code
groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
assert len(groups) == 2
assert groups[0]['driver'] == driver_1
assert groups[0]['group'] == group_1
assert groups[1]['driver'] == driver_2
assert groups[1]['group'] == group_1
# match none group
response = api.get_subscribe_group(app, group='none')
assert 200 == response.status_code
assert len(response.json()['groups']) == 0
# match driver and group
response = api.get_subscribe_group(app, driver=driver_1, group=group_1)
assert 200 == response.status_code
groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
assert len(groups) == 1
assert groups[0]['driver'] == driver_1
assert groups[0]['group'] == group_1
response = api.get_subscribe_group(app, driver=driver_2, group=group_2)
assert 200 == response.status_code
groups = sorted(response.json()['groups'], key=lambda g: (g['driver'], g['group']))
assert len(groups) == 1
assert groups[0]['driver'] == driver_2
assert groups[0]['group'] == group_2
finally:
api.del_node(app)
api.del_node(driver_1)
api.del_node(driver_2)
@description(given="running neuron", when="get mqtt shcema", then="success")
def test_get_plugin_schema(self):
response = api.get_plugin_schema(plugin='mqtt')
+16 -1
View File
@@ -92,8 +92,23 @@ class TestMetrics:
"send_msg_errors_total": (0, {}),
"recv_msgs_total": (0, {}),
"cached_msgs": (0, {}),
"last_5s_trans_data_msgs": (0, {}),
"last_30s_trans_data_msgs": (0, {}),
"last_60s_trans_data_msgs": (0, {}),
"last_5s_send_bytes": (0, {}),
"last_30s_send_bytes": (0, {}),
"last_60s_send_bytes": (0, {}),
"last_5s_recv_bytes": (0, {}),
"last_30s_recv_bytes": (0, {}),
"last_60s_recv_bytes": (0, {}),
"last_5s_recv_msgs": (0, {}),
"last_30s_recv_msgs": (0, {}),
"last_60s_recv_msgs": (0, {}),
"last_60s_disconnections": (0, {}),
"last_600s_disconnections": (0, {}),
"last_1800s_disconnections": (0, {}),
}
assert_metrics(resp.content.decode('utf-8'), expected_metrics)
@description(given="neuron is started",
+10 -3
View File
@@ -145,12 +145,15 @@ def get_tags(node, group):
return requests.get(url=config.BASE_URL + "/api/v2/tags", headers={"Authorization": config.default_jwt}, params={"node": node, "group": group})
def read_tags(node, group, sync=False):
return requests.post(url=config.BASE_URL + "/api/v2/read", headers={"Authorization": config.default_jwt}, json={"node": node, "group": group, "sync": sync})
def read_tags(node, group, sync=False, query=None):
body = {"node": node, "group": group, "sync": sync}
if query:
body["query"] = query
return requests.post(url=config.BASE_URL + "/api/v2/read", headers={"Authorization": config.default_jwt}, json=body)
def read_tag(node, group, tag, sync=False):
response = read_tags(node, group, sync)
response = read_tags(node, group, sync, query={"name": tag})
assert 200 == response.status_code
x = list(filter(lambda x: x['name'] == tag,
response.json()['tags']))
@@ -219,6 +222,10 @@ def get_plugins():
return requests.get(url=config.BASE_URL + "/api/v2/plugin", headers={"Authorization": config.default_jwt})
def get_subscribe_group(app, driver=None, group=None):
return requests.get(url=config.BASE_URL + "/api/v2/subscribe", headers={"Authorization": config.default_jwt}, params={"app": app, "driver": driver, "group": group})
@gen_check
def add_template(name, plugin, groups=[]):
return requests.post(url=config.BASE_URL + "/api/v2/template", json={"name": name, "plugin": plugin, "groups": groups}, headers={"Authorization": config.default_jwt})
+18
View File
@@ -228,6 +228,24 @@ class TestPlugin:
assert 400 == response.status_code
assert NEU_ERR_BODY_IS_WRONG == response.json().get("error")
@description(given="incorrect configuration", when="update plugin", then="success")
def test_09_update_plugin_success0(self):
test_data = TestPlugin.test_data
plugin_data = {}
plugin_data['library'] = test_data['library_c1']
plugin_data['so_file'] = test_data['so_file_c1_2']
plugin_data['schema_file'] = test_data['schema_file']
response = api.updata_plugin(
library_name=plugin_data['library'], so_file=plugin_data['so_file'], schema_file=plugin_data['schema_file'])
assert 200 == response.status_code
assert NEU_ERR_SUCCESS == response.json().get("error")
response = api.get_plugins()
assert 200 == response.status_code
for plugin in response.json().get("plugins"):
if plugin['name'] == 'c1':
assert plugin['version'] == '2.7.99'
@description(given="incorrect configuration", when="update plugin", then="fail")
def test_10_update_plugin_fail3(self):
test_data = TestPlugin.test_data
+131 -3
View File
@@ -279,7 +279,7 @@ class TestLog:
]
)
assert 200 == response.status_code
assert 400 == response.status_code
assert NEU_ERR_TAG_TYPE_NOT_SUPPORT == response.json()['error']
response = api.get_group()
@@ -342,7 +342,7 @@ class TestLog:
]
)
assert 200 == response.status_code
assert 409 == response.status_code
assert NEU_ERR_TAG_NAME_CONFLICT == response.json()['error']
response = api.get_group()
@@ -355,4 +355,132 @@ class TestLog:
assert 2 == response.json()['groups'][1]['tag_count']
assert 'group_3' == response.json()['groups'][2]['group']
assert 3000 == response.json()['groups'][2]['interval']
assert 2 == response.json()['groups'][2]['tag_count']
assert 2 == response.json()['groups'][2]['tag_count']
@description(given="multiple groups including same name tags in the first group", when="adding", then="add failed")
def test_adding_same_name_gtags_first_group(self):
response = api.add_gtags(
node='modbus-tcp-tag-test',
groups=[
{
"group": "group_f",
"interval": 3000,
"tags": [
{
"name": "tag1",
"address": "1!400001",
"attribute": 3,
"type": 3,
"decimal": 0.01
},
{
"name": "tag1",
"address": "1!400002",
"attribute": 3,
"type": 9,
"precision": 3
}
]
},
{
"group": "group_s",
"interval": 3000,
"tags": [
{
"name": "tag1",
"address": "1!400001",
"attribute": 3,
"type": 9,
"precision": 3
},
{
"name": "tag2",
"address": "1!400002",
"attribute": 3,
"type": 9,
"precision": 3
}
]
}
]
)
assert 409 == response.status_code
assert NEU_ERR_TAG_NAME_CONFLICT == response.json()['error']
@description(given="multiple groups including same name tags in the second group", when="adding", then="add failed")
def test_adding_same_name_gtags_second_group(self):
response = api.add_gtags(
node='modbus-tcp-tag-test',
groups=[
{
"group": "group_f",
"interval": 3000,
"tags": [
{
"name": "tag1",
"address": "1!400001",
"attribute": 3,
"type": 3,
"decimal": 0.01
},
{
"name": "tag2",
"address": "1!400002",
"attribute": 3,
"type": 9,
"precision": 3
}
]
},
{
"group": "group_s",
"interval": 3000,
"tags": [
{
"name": "tag1",
"address": "1!400001",
"attribute": 3,
"type": 9,
"precision": 3
},
{
"name": "tag1",
"address": "1!400002",
"attribute": 3,
"type": 9,
"precision": 3
}
]
}
]
)
assert 409 == response.status_code
assert NEU_ERR_TAG_NAME_CONFLICT == response.json()['error']
@description(given="tag name length exceeds 128", when="adding", then="add failed")
def test_adding_tag_name_exceeds_128(self):
response = api.add_gtags(
node='modbus-tcp-tag-test',
groups=[{"group": "group","interval": 3000,
"tags": [{
"name": "Lt7Pa5Q0tKJN2gL3WiYJLnaFzqBq5q5xyJ1vYHxyf2j0IZR7y3RBYxXGQBHwV7dEzD7lIcF1yX9aR3vH8leYxXxHUILk2eVqQBlaQLaROZt9yZ9xQNxKkfy5mySHb79",
"address": "1!400001",
"attribute": 3,
"type": 3,
"decimal": 0.01}]}])
assert 200 == response.status_code
assert NEU_ERR_SUCCESS == response.json()['error']
response = api.add_gtags(
node='modbus-tcp-tag-test',
groups=[{"group": "group","interval": 3000,
"tags": [{
"name": "Lt7Pa5Q0tKJN2gL3WiYJLnaFzqBq5q5xyJ1vYHxyf2j0IZR7y3RBYxXGQBHwV7dEzD7lIcF1yX9aR3vH8leYxXxHUILk2eVqQBlaQLaROZt9yZ9xQNxKkfy5mySHb79Y8",
"address": "1!400001",
"attribute": 3,
"type": 3,
"decimal": 0.01}]}])
assert 400 == response.status_code
assert NEU_ERR_TAG_NAME_TOO_LONG == response.json()['error']
+1 -1
View File
@@ -63,7 +63,7 @@ static const neu_plugin_intf_funs_t plugin_intf_funs = {
};
const neu_plugin_module_t neu_plugin_module = {
.version = NEURON_PLUGIN_VER_1_0,
.version = NEU_VERSION(2, 7, 99),
.schema = "c1",
.module_name = "c1",
.module_descr = "c1 v2",
+9 -1
View File
@@ -51,10 +51,18 @@ target_include_directories(modbus_test PRIVATE
${CMAKE_SOURCE_DIR}/plugins/modbus)
target_link_libraries(modbus_test neuron-base gtest_main gtest pthread zlog)
add_executable(rolling_counter_test rolling_counter_test.cc)
target_include_directories(rolling_counter_test PRIVATE
${CMAKE_SOURCE_DIR}/src
${CMAKE_SOURCE_DIR}/include
)
target_link_libraries(rolling_counter_test neuron-base gtest_main gtest)
include(GoogleTest)
gtest_discover_tests(json_test)
gtest_discover_tests(http_test)
gtest_discover_tests(jwt_test)
gtest_discover_tests(base64_test)
gtest_discover_tests(tag_sort_test)
gtest_discover_tests(modbus_test)
gtest_discover_tests(modbus_test)
gtest_discover_tests(rolling_counter_test)
+45
View File
@@ -0,0 +1,45 @@
#include <gtest/gtest.h>
#include "utils/log.h"
#include "utils/rolling_counter.h"
zlog_category_t *neuron = NULL;
TEST(RollingCounterTest, neu_rolling_counter_inc)
{
uint64_t ts = 0;
neu_rolling_counter_t *counter = neu_rolling_counter_new(4000);
EXPECT_NE(nullptr, counter);
for (int i = 1; i <= 4; ++i) {
ts += 1000;
EXPECT_EQ(i, neu_rolling_counter_inc(counter, ts, 1));
}
// wrap around
for (int i = 1; i <= 4; ++i) {
ts += 1000;
EXPECT_EQ(4, neu_rolling_counter_inc(counter, ts, 1));
}
// increment without time stamp update
EXPECT_EQ(4, neu_rolling_counter_value(counter));
for (int i = 1; i <= 10; ++i) {
EXPECT_EQ(4 + i, neu_rolling_counter_inc(counter, ts, 1));
}
EXPECT_EQ(4 + 10, neu_rolling_counter_value(counter));
// stale value will be clean
ts += 4000;
EXPECT_EQ(0, neu_rolling_counter_inc(counter, ts, 0));
neu_rolling_counter_free(counter);
}
int main(int argc, char **argv)
{
zlog_init("./config/dev.conf");
neuron = zlog_get_category("neuron");
::testing::InitGoogleTest(&argc, argv);
return RUN_ALL_TESTS();
}
+1 -1
View File
@@ -1 +1 @@
2.7.0-dev
2.7.4