GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-29 07:06:32 +08:00
Code Issues Packages Projects Releases Wiki Activity

tag quality & timestamp

Browse Source
This commit is contained in:
fengzero
2026-10-23 03:46:49 +00:00
parent 73f0f6d653
commit 10f3b194bb
8 changed files with 237 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/type.h
+12 -3
View File
@@ -340,9 +340,17 @@ static inline char *neu_value_str(neu_type_e type, neu_value_u value)
}
typedef struct {
neu_type_e type;
neu_value_u value;
uint8_t precision;
int64_t timestamp;
int32_t quality;
bool has_timestamp;
bool has_quality;
} neu_tag_sample_attrs_t;
typedef struct {
neu_type_e type;
neu_value_u value;
uint8_t precision;
neu_tag_sample_attrs_t attrs;
} neu_dvalue_t;
static inline bool neu_dvalue_is_array(const neu_dvalue_t *dvalue)
@@ -508,6 +516,7 @@ static inline void neu_copy_dvalue(const neu_dvalue_t *src, neu_dvalue_t *dst)
dst->type = src->type;
dst->precision = src->precision;
dst->value = src->value;
dst->attrs = src->attrs;
// value need deep copy
switch (src->type) {
src/adapter/driver/cache.c
+31 -10
View File
@@ -169,6 +169,7 @@ void neu_driver_cache_add(neu_driver_cache_t *cache, const char *group,
elem->timestamp = 0;
elem->changed = false;
elem->value = value;
memset(&elem->value_old.attrs, 0, sizeof(elem->value_old.attrs));
pthread_mutex_unlock(&cache->mtx);
}
@@ -229,6 +230,13 @@ bool neu_driver_cache_update_change(neu_driver_cache_t *cache,
tkey_t key = to_key(group, tag);
bool tag_changed = false;
if (!value.attrs.has_timestamp) {
value.attrs.timestamp = 0;
}
if (!value.attrs.has_quality) {
value.attrs.quality = 0;
}
pthread_mutex_lock(&cache->mtx);
HASH_FIND(hh, cache->table, &key, sizeof(tkey_t), elem);
if (elem != NULL) {
@@ -483,9 +491,8 @@ bool neu_driver_cache_update_change(neu_driver_cache_t *cache,
}
case NEU_TYPE_FLOAT:
if (elem->value_old.precision == 0) {
elem->changed =
elem->value_old.value.f32 != value.value.f32;
tag_changed = elem->changed;
tag_changed = elem->value_old.value.f32 != value.value.f32;
elem->changed |= tag_changed;
} else {
if (fabs(elem->value_old.value.f32 - value.value.f32) >
pow(0.1, elem->value_old.precision)) {
@@ -496,9 +503,8 @@ bool neu_driver_cache_update_change(neu_driver_cache_t *cache,
break;
case NEU_TYPE_DOUBLE:
if (elem->value_old.precision == 0) {
elem->changed =
elem->value_old.value.d64 != value.value.d64;
tag_changed = elem->changed;
tag_changed = elem->value_old.value.d64 != value.value.d64;
elem->changed |= tag_changed;
} else {
if (fabs(elem->value_old.value.d64 - value.value.d64) >
pow(0.1, elem->value_old.precision)) {
@@ -728,8 +734,8 @@ bool neu_driver_cache_update_change(neu_driver_cache_t *cache,
}
case NEU_TYPE_FLOAT:
if (elem->value.precision == 0) {
elem->changed = elem->value.value.f32 != value.value.f32;
tag_changed = elem->changed;
tag_changed = elem->value.value.f32 != value.value.f32;
elem->changed |= tag_changed;
} else {
if (fabs(elem->value.value.f32 - value.value.f32) >
pow(0.1, elem->value.precision)) {
@@ -740,8 +746,8 @@ bool neu_driver_cache_update_change(neu_driver_cache_t *cache,
break;
case NEU_TYPE_DOUBLE:
if (elem->value.precision == 0) {
elem->changed = elem->value.value.d64 != value.value.d64;
tag_changed = elem->changed;
tag_changed = elem->value.value.d64 != value.value.d64;
elem->changed |= tag_changed;
} else {
if (fabs(elem->value.value.d64 - value.value.d64) >
pow(0.1, elem->value.precision)) {
@@ -758,13 +764,26 @@ bool neu_driver_cache_update_change(neu_driver_cache_t *cache,
}
}
const neu_tag_sample_attrs_t *previous_attrs =
sub_filter_err && elem->value.type == NEU_TYPE_ERROR
? &elem->value_old.attrs
: &elem->value.attrs;
if (previous_attrs->has_quality != value.attrs.has_quality ||
(value.attrs.has_quality &&
previous_attrs->quality != value.attrs.quality)) {
elem->changed = true;
tag_changed = true;
}
if (sub_filter_err && value.type != NEU_TYPE_ERROR) {
elem->value_old.type = value.type;
elem->value_old.value = value.value;
elem->value_old.precision = value.precision;
elem->value_old.attrs = value.attrs;
}
error_not_report:
elem->value.attrs = value.attrs;
if (change) {
elem->changed = true;
@@ -887,6 +906,7 @@ int neu_driver_cache_meta_get(neu_driver_cache_t *cache, const char *group,
if (elem != NULL) {
value->timestamp = elem->timestamp;
value->value.attrs = elem->value.attrs;
value->value.type = elem->value.type;
value->value.precision = elem->value.precision;
@@ -1061,6 +1081,7 @@ int neu_driver_cache_meta_get_changed(neu_driver_cache_t *cache,
if (elem != NULL && elem->changed) {
value->timestamp = elem->timestamp;
value->value.attrs = elem->value.attrs;
value->value.type = elem->value.type;
value->value.precision = elem->value.precision;
src/adapter/driver/cache.h
+1 -1
View File
@@ -22,7 +22,7 @@
#include <stdint.h>
#include "type.h"
#include "tag.h"
typedef struct neu_driver_cache neu_driver_cache_t;
src/adapter/driver/driver.c
+6
View File
@@ -2851,6 +2851,8 @@ static void read_report_group(int64_t timestamp, int64_t timeout,
tag_value.datatag.bias = tag->bias;
tag_value.value.attrs = value.value.attrs;
if (value.value.type == NEU_TYPE_ERROR) {
tag_value.value = value.value;
@@ -3041,6 +3043,8 @@ static void read_group(int64_t timestamp, int64_t timeout,
continue;
}
tag_value.value.attrs = value.value.attrs;
if (value.value.type == NEU_TYPE_ERROR) {
tag_value.value = value.value;
utarray_push_back(tag_values, &tag_value);
@@ -3236,6 +3240,8 @@ static void read_group_paginate(int64_t timestamp, int64_t timeout,
continue;
}
tag_value.value.attrs = value.value.attrs;
if (value.value.type == NEU_TYPE_ERROR) {
tag_value.value = value.value;
utarray_push_back(tag_values, &tag_value);
tests/ft/plugin/test_plugin.py
+1 -1
View File
@@ -244,7 +244,7 @@ class TestPlugin:
assert 200 == response.status_code
for plugin in response.json().get("plugins"):
if plugin['name'] == 'c1':
assert plugin['version'] == '2.15.99'
assert plugin['version'] == '2.16.99'
@description(given="incorrect configuration", when="update plugin", then="fail")
def test_10_update_plugin_fail3(self):
tests/plugins/c1/c1_2.c
+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 = NEU_VERSION(2, 16, 99),
.version = NEU_VERSION(NEU_VERSION_MAJOR, NEU_VERSION_MINOR, 99),
.schema = "c1",
.module_name = "c1",
.module_descr = "c1 v2",
tests/ut/CMakeLists.txt
+9
View File
@@ -77,6 +77,14 @@ target_include_directories(common_test PRIVATE
)
target_link_libraries(common_test neuron-base gtest_main gtest)
add_executable(cache_test cache_test.cc
${CMAKE_SOURCE_DIR}/src/adapter/driver/cache.c)
target_include_directories(cache_test PRIVATE
${CMAKE_SOURCE_DIR}/src
${CMAKE_SOURCE_DIR}/include
)
target_link_libraries(cache_test gtest_main gtest pthread jansson m)
file(COPY ${CMAKE_SOURCE_DIR}/tests/ut/serverBMS_3_test.cid DESTINATION ${UT_DIRECTORY}/config)
add_executable(cid_test cid_test.cc)
target_include_directories(cid_test PRIVATE
@@ -112,6 +120,7 @@ gtest_discover_tests(async_queue_test)
gtest_discover_tests(rolling_counter_test)
gtest_discover_tests(mqtt_client_test)
gtest_discover_tests(common_test)
gtest_discover_tests(cache_test)
gtest_discover_tests(cid_test)
gtest_discover_tests(mqtt_schema_test)
gtest_discover_tests(ede_test)
tests/ut/cache_test.cc
+176
View File
@@ -0,0 +1,176 @@
#include <gtest/gtest.h>
extern "C" {
#include "adapter/driver/cache.h"
bool sub_filter_err = false;
}
class DriverCacheSample : public testing::Test {
protected:
void SetUp() override
{
cache = neu_driver_cache_new();
neu_dvalue_t initial = {};
initial.type = NEU_TYPE_ERROR;
neu_driver_cache_add(cache, "group", "tag", initial);
}
void TearDown() override
{
neu_driver_cache_destroy(cache);
sub_filter_err = false;
}
neu_dvalue_t sample()
{
neu_dvalue_t value = {};
value.type = NEU_TYPE_DOUBLE;
value.value.d64 = 1.25;
value.attrs.has_timestamp = true;
value.attrs.timestamp = 100;
value.attrs.has_quality = true;
value.attrs.quality = 192;
return value;
}
bool update(neu_dvalue_t value)
{
return neu_driver_cache_update_change(cache, "group", "tag", 200,
value, NULL, 0, false);
}
int changed()
{
return neu_driver_cache_meta_get_changed(cache, "group", "tag", &out,
&metas, &n_meta);
}
neu_driver_cache_t *cache = NULL;
neu_driver_cache_value_t out = {};
neu_tag_meta_t *metas = NULL;
int n_meta = 0;
};
TEST_F(DriverCacheSample, existing_update_preserves_optional_attrs)
{
for (int flags = 0; flags < 4; ++flags) {
neu_dvalue_t value = sample();
value.attrs.has_timestamp = flags & 1;
value.attrs.has_quality = flags & 2;
neu_driver_cache_update(cache, "group", "tag", 200, value, NULL, 0);
ASSERT_EQ(0, neu_driver_cache_meta_get(cache, "group", "tag", &out,
&metas, &n_meta));
EXPECT_EQ(200, out.timestamp);
EXPECT_EQ(value.attrs.has_timestamp, out.value.attrs.has_timestamp);
EXPECT_EQ(value.attrs.has_quality, out.value.attrs.has_quality);
EXPECT_EQ(flags & 1 ? 100 : 0, out.value.attrs.timestamp);
EXPECT_EQ(flags & 2 ? 192 : 0, out.value.attrs.quality);
}
}
TEST_F(DriverCacheSample, timestamp_alone_does_not_trigger_change)
{
neu_dvalue_t value = sample();
ASSERT_TRUE(update(value));
ASSERT_EQ(0, changed());
value.attrs.timestamp++;
EXPECT_FALSE(update(value));
EXPECT_NE(0, changed());
ASSERT_EQ(0, neu_driver_cache_meta_get(cache, "group", "tag", &out,
&metas, &n_meta));
EXPECT_EQ(value.attrs.timestamp, out.value.attrs.timestamp);
value.attrs.has_timestamp = false;
EXPECT_FALSE(update(value));
EXPECT_NE(0, changed());
}
TEST_F(DriverCacheSample, quality_change_stays_pending_until_read)
{
for (neu_type_e type : { NEU_TYPE_FLOAT, NEU_TYPE_DOUBLE }) {
neu_dvalue_t value = sample();
value.type = type;
if (type == NEU_TYPE_FLOAT) {
value.value.f32 = 1.25f;
}
ASSERT_TRUE(update(value));
ASSERT_EQ(0, changed());
value.attrs.quality = 0;
EXPECT_TRUE(update(value));
value.attrs.timestamp++;
EXPECT_FALSE(update(value));
ASSERT_EQ(0, changed());
EXPECT_TRUE(out.value.attrs.has_quality);
EXPECT_EQ(0, out.value.attrs.quality);
EXPECT_NE(0, changed());
}
}
TEST_F(DriverCacheSample, absent_attrs_clear_previous_sample)
{
neu_dvalue_t value = sample();
ASSERT_TRUE(update(value));
ASSERT_EQ(0, changed());
value.attrs = {};
EXPECT_TRUE(update(value));
ASSERT_EQ(0, changed());
EXPECT_FALSE(out.value.attrs.has_timestamp);
EXPECT_FALSE(out.value.attrs.has_quality);
EXPECT_EQ(0, out.value.attrs.timestamp);
EXPECT_EQ(0, out.value.attrs.quality);
EXPECT_FALSE(update(value));
EXPECT_NE(0, changed());
}
TEST_F(DriverCacheSample, forced_update_preserves_attrs)
{
neu_dvalue_t value = sample();
ASSERT_TRUE(update(value));
ASSERT_EQ(0, changed());
neu_driver_cache_update_change(cache, "group", "tag", 300, value, NULL, 0,
true);
ASSERT_EQ(0, changed());
EXPECT_EQ(300, out.timestamp);
EXPECT_EQ(value.attrs.timestamp, out.value.attrs.timestamp);
EXPECT_EQ(value.attrs.quality, out.value.attrs.quality);
}
TEST_F(DriverCacheSample, filtered_error_compares_last_good_quality)
{
sub_filter_err = true;
neu_dvalue_t value = sample();
ASSERT_TRUE(update(value));
ASSERT_EQ(0, changed());
neu_dvalue_t error = {};
error.type = NEU_TYPE_ERROR;
error.value.i32 = 1;
EXPECT_FALSE(update(error));
EXPECT_NE(0, changed());
EXPECT_FALSE(update(value));
EXPECT_NE(0, changed());
EXPECT_FALSE(update(error));
value.attrs.quality = 0;
EXPECT_TRUE(update(value));
ASSERT_EQ(0, changed());
EXPECT_TRUE(out.value.attrs.has_quality);
EXPECT_EQ(0, out.value.attrs.quality);
}
TEST_F(DriverCacheSample, copy_dvalue_preserves_attrs)
{
neu_dvalue_t src = sample();
neu_dvalue_t dst = {};
neu_copy_dvalue(&src, &dst);
EXPECT_EQ(src.attrs.has_timestamp, dst.attrs.has_timestamp);
EXPECT_EQ(src.attrs.timestamp, dst.attrs.timestamp);
EXPECT_EQ(src.attrs.has_quality, dst.attrs.has_quality);
EXPECT_EQ(src.attrs.quality, dst.attrs.quality);
}