GithubStar/neuron
1
0
Fork 0
mirror of https://github.com/emqx/neuron.git synced 2026-10-18 13:41:22 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat(driver): new interface for updating tags, tags can carry metadata

Browse Source
This commit is contained in:
fengzero
2023-07-21 08:28:51 +00:00
parent bb70cbf860
commit 7ea2bbf35e
5 changed files with 99 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/adapter.h
+3
View File
@@ -791,6 +791,9 @@ typedef struct adapter_callbacks {
struct {
void (*update)(neu_adapter_t *adapter, const char *group,
const char *tag, neu_dvalue_t value);
void (*update_with_meta)(neu_adapter_t *adapter, const char *group,
const char *tag, neu_dvalue_t value,
neu_tag_meta_t *metas, int n_meta);
void (*write_response)(neu_adapter_t *adapter, void *req,
int error);
void (*update_im)(neu_adapter_t *adapter, const char *group,
include/neuron/tag.h
+9 -2
View File
@@ -23,11 +23,13 @@ config_ **/
#include <stdbool.h>
#include <stdint.h>
#include "define.h"
#include "type.h"
#include "utils/utextend.h"
#include "json/json.h"
#define NEU_TAG_META_SIZE 20
#define NEU_TAG_META_LENGTH 20
#define NEU_TAG_META_SIZE 3
typedef enum {
NEU_ATTRIBUTE_READ = 1,
@@ -84,9 +86,14 @@ typedef struct {
double decimal;
char * description;
neu_datatag_addr_option_u option;
uint8_t meta[NEU_TAG_META_SIZE];
uint8_t meta[NEU_TAG_META_LENGTH];
} neu_datatag_t;
typedef struct neu_tag_meta {
char name[NEU_TAG_NAME_LEN];
neu_dvalue_t value;
} neu_tag_meta_t;
UT_icd *neu_tag_get_icd();
neu_datatag_t *neu_tag_dup(const neu_datatag_t *tag);
src/adapter/driver/cache.c
+24 -1
View File
@@ -42,6 +42,8 @@ struct elem {
neu_dvalue_t value;
neu_tag_meta_t metas[NEU_TAG_META_SIZE];
tkey_t key;
UT_hash_handle hh;
};
@@ -125,7 +127,8 @@ void neu_driver_cache_add(neu_driver_cache_t *cache, const char *group,
void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group,
const char *tag, int64_t timestamp,
neu_dvalue_t value)
neu_dvalue_t value, neu_tag_meta_t *metas,
int n_meta)
{
struct elem *elem = NULL;
tkey_t key = to_key(group, tag);
@@ -187,6 +190,10 @@ void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group,
elem->value.type = value.type;
elem->value.value = value.value;
memset(elem->metas, 0, sizeof(neu_tag_meta_t) * NEU_TAG_META_SIZE);
for (int i = 0; i < n_meta; i++) {
memcpy(&elem->metas[i], &metas[i], sizeof(neu_tag_meta_t));
}
}
nng_mtx_unlock(cache->mtx);
@@ -241,6 +248,14 @@ int neu_driver_cache_get(neu_driver_cache_t *cache, const char *group,
break;
}
for (int i = 0; i < NEU_TAG_META_SIZE; i++) {
if (elem->metas[i].name != NULL &&
strlen(elem->metas[i].name) > 0) {
memcpy(&value->metas[i], &elem->metas[i],
sizeof(neu_tag_meta_t));
}
}
ret = 0;
}
@@ -299,6 +314,14 @@ int neu_driver_cache_get_changed(neu_driver_cache_t *cache, const char *group,
break;
}
for (int i = 0; i < NEU_TAG_META_SIZE; i++) {
if (elem->metas[i].name != NULL &&
strlen(elem->metas[i].name) > 0) {
memcpy(&value->metas[i], &elem->metas[i],
sizeof(neu_tag_meta_t));
}
}
if (elem->value.type != NEU_TYPE_ERROR) {
elem->changed = false;
}
src/adapter/driver/cache.h
+5 -3
View File
@@ -33,14 +33,16 @@ void neu_driver_cache_add(neu_driver_cache_t *cache, const char *group,
const char *tag, neu_dvalue_t value);
void neu_driver_cache_update(neu_driver_cache_t *cache, const char *group,
const char *tag, int64_t timestamp,
neu_dvalue_t value);
neu_dvalue_t value, neu_tag_meta_t *metas,
int n_meta);
void neu_driver_cache_del(neu_driver_cache_t *cache, const char *group,
const char *tag);
typedef struct {
neu_dvalue_t value;
int64_t timestamp;
neu_dvalue_t value;
int64_t timestamp;
neu_tag_meta_t metas[NEU_TAG_META_SIZE];
} neu_driver_cache_value_t;
int neu_driver_cache_get(neu_driver_cache_t *cache, const char *group,
src/adapter/driver/driver.c
+58 -44
View File
@@ -85,6 +85,11 @@ static int read_report_group(int64_t timestamp, int64_t timeout,
UT_array *tags, neu_resp_tag_value_t *datas);
static void update(neu_adapter_t *adapter, const char *group, const char *tag,
neu_dvalue_t value);
static void update_im(neu_adapter_t *adapter, const char *group,
const char *tag, neu_dvalue_t value);
static void update_with_meta(neu_adapter_t *adapter, const char *group,
const char *tag, neu_dvalue_t value,
neu_tag_meta_t *metas, int n_meta);
static void write_response(neu_adapter_t *adapter, void *r, neu_error error);
static group_t *find_group(neu_adapter_driver_t *driver, const char *name);
static void store_write_tag(group_t *group, to_be_write_tag_t *tag);
@@ -100,6 +105,49 @@ static void write_response(neu_adapter_t *adapter, void *r, neu_error error)
free(req);
}
static void update_with_meta(neu_adapter_t *adapter, const char *group,
const char *tag, neu_dvalue_t value,
neu_tag_meta_t *metas, int n_meta)
{
neu_adapter_driver_t *driver = (neu_adapter_driver_t *) adapter;
if (value.type == NEU_TYPE_ERROR && tag == NULL) {
group_t *g = find_group(driver, group);
if (g != NULL) {
UT_array *tags = neu_group_get_read_tag(g->group);
uint64_t err_count = 0;
utarray_foreach(tags, neu_datatag_t *, t)
{
if (neu_tag_attribute_test(t, NEU_ATTRIBUTE_STATIC)) {
continue;
}
neu_driver_cache_update(driver->cache, group, t->name,
global_timestamp, value, NULL, 0);
++err_count;
}
driver->adapter.cb_funs.update_metric(
&driver->adapter, NEU_METRIC_TAG_READS_TOTAL, err_count, NULL);
driver->adapter.cb_funs.update_metric(
&driver->adapter, NEU_METRIC_TAG_READ_ERRORS_TOTAL, err_count,
NULL);
utarray_free(tags);
}
} else {
neu_driver_cache_update(driver->cache, group, tag, global_timestamp,
value, metas, n_meta);
driver->adapter.cb_funs.update_metric(
&driver->adapter, NEU_METRIC_TAG_READS_TOTAL, 1, NULL);
driver->adapter.cb_funs.update_metric(
&driver->adapter, NEU_METRIC_TAG_READ_ERRORS_TOTAL,
NEU_TYPE_ERROR == value.type, NULL);
}
nlog_info(
"update driver: %s, group: %s, tag: %s, type: %s, timestamp: %" PRId64,
driver->adapter.name, group, tag, neu_type_string(value.type),
global_timestamp);
}
static void update_im(neu_adapter_t *adapter, const char *group,
const char *tag, neu_dvalue_t value)
{
@@ -112,7 +160,8 @@ static void update_im(neu_adapter_t *adapter, const char *group,
}
UT_array *tags = neu_adapter_driver_get_ptag(driver, group, tag);
neu_driver_cache_update(driver->cache, group, tag, global_timestamp, value);
neu_driver_cache_update(driver->cache, group, tag, global_timestamp, value,
NULL, 0);
driver->adapter.cb_funs.update_metric(&driver->adapter,
NEU_METRIC_TAG_READS_TOTAL, 1, NULL);
neu_datatag_t *first = utarray_front(tags);
@@ -155,54 +204,19 @@ static void update_im(neu_adapter_t *adapter, const char *group,
static void update(neu_adapter_t *adapter, const char *group, const char *tag,
neu_dvalue_t value)
{
neu_adapter_driver_t *driver = (neu_adapter_driver_t *) adapter;
if (value.type == NEU_TYPE_ERROR && tag == NULL) {
group_t *g = find_group(driver, group);
if (g != NULL) {
UT_array *tags = neu_group_get_read_tag(g->group);
uint64_t err_count = 0;
utarray_foreach(tags, neu_datatag_t *, t)
{
if (neu_tag_attribute_test(t, NEU_ATTRIBUTE_STATIC)) {
continue;
}
neu_driver_cache_update(driver->cache, group, t->name,
global_timestamp, value);
++err_count;
}
driver->adapter.cb_funs.update_metric(
&driver->adapter, NEU_METRIC_TAG_READS_TOTAL, err_count, NULL);
driver->adapter.cb_funs.update_metric(
&driver->adapter, NEU_METRIC_TAG_READ_ERRORS_TOTAL, err_count,
NULL);
utarray_free(tags);
}
} else {
neu_driver_cache_update(driver->cache, group, tag, global_timestamp,
value);
driver->adapter.cb_funs.update_metric(
&driver->adapter, NEU_METRIC_TAG_READS_TOTAL, 1, NULL);
driver->adapter.cb_funs.update_metric(
&driver->adapter, NEU_METRIC_TAG_READ_ERRORS_TOTAL,
NEU_TYPE_ERROR == value.type, NULL);
}
nlog_info(
"update driver: %s, group: %s, tag: %s, type: %s, timestamp: %" PRId64,
driver->adapter.name, group, tag, neu_type_string(value.type),
global_timestamp);
update_with_meta(adapter, group, tag, value, NULL, 0);
}
neu_adapter_driver_t *neu_adapter_driver_create()
{
neu_adapter_driver_t *driver = calloc(1, sizeof(neu_adapter_driver_t));
driver->cache = neu_driver_cache_new();
driver->driver_events = neu_event_new();
driver->adapter.cb_funs.driver.update = update;
driver->adapter.cb_funs.driver.write_response = write_response;
driver->adapter.cb_funs.driver.update_im = update_im;
driver->cache = neu_driver_cache_new();
driver->driver_events = neu_event_new();
driver->adapter.cb_funs.driver.update = update;
driver->adapter.cb_funs.driver.write_response = write_response;
driver->adapter.cb_funs.driver.update_im = update_im;
driver->adapter.cb_funs.driver.update_with_meta = update_with_meta;
return driver;
}
@@ -610,7 +624,7 @@ void neu_adapter_driver_write_tag(neu_adapter_driver_t *driver,
if (neu_tag_attribute_test(tag, NEU_ATTRIBUTE_STATIC)) {
neu_driver_cache_update(g->driver->cache, g->name, tag->name,
global_timestamp, cmd->value);
global_timestamp, cmd->value, NULL, 0);
neu_tag_set_static_value(tag, &cmd->value.value);
neu_group_update_tag(g->group, tag);
adapter_storage_update_tag_value(cmd->driver, cmd->group, tag);