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

Merge pull request #887 from fengzeroz/main

Browse Source 
improve(timestamp): remove update timestamp from adapter
This commit is contained in:
fengzeroz
2022-11-04 15:01:15 +08:00
committed by GitHub
parent 1e28dbf6ed bab2a82469
commit 89c614eda5
12 changed files with 47 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files
include/neuron/plugin.h
+2 -1
View File
@@ -34,6 +34,8 @@ extern "C" {
#define NEURON_PLUGIN_VER_1_0 100
#define NEURON_PLUGIN_VER_2_0 200
static volatile int64_t global_timestamp;
typedef struct neu_plugin_common {
uint32_t magic;
@@ -44,7 +46,6 @@ typedef struct neu_plugin_common {
neu_node_link_state_e link_state;
uint16_t tag_size;
uint32_t tag_all_size;
int64_t timestamp;
zlog_category_t *log;
} neu_plugin_common_t;
include/neuron/utils/time.h
+2 -2
View File
@@ -27,11 +27,11 @@ extern "C" {
#include <stdlib.h>
#include <sys/time.h>
static inline uint64_t neu_time_ms()
static inline int64_t neu_time_ms()
{
struct timeval tv = { 0 };
gettimeofday(&tv, NULL);
return tv.tv_sec * 1000 + tv.tv_usec / 1000;
return (int64_t) tv.tv_sec * 1000 + (int64_t) tv.tv_usec / 1000;
}
#ifdef __cplusplus
plugins/ekuiper/json_rw.c
+1 -1
View File
@@ -198,7 +198,7 @@ int json_encode_read_resp(void *json_object, void *param)
json_read_resp_header_t header = { .group_name = trans_data->group,
.node_name = trans_data->driver,
.timestamp = plugin->common.timestamp };
.timestamp = global_timestamp };
ret = json_encode_read_resp_header(json_object, &header);
if (0 != ret) {
plugins/mqtt/command/heartbeat.c
+2 -1
View File
@@ -27,9 +27,10 @@
char *command_heartbeat_generate(neu_plugin_t *plugin, UT_array *states)
{
(void) plugin;
char * version = NEURON_VERSION;
neu_json_states_head_t header = { .version = version,
.timpstamp = plugin->common.timestamp };
.timpstamp = global_timestamp };
neu_json_states_t json = { 0 };
char * json_str = NULL;
json.n_state = utarray_len(states);
plugins/mqtt/command/read_write.c
+1 -1
View File
@@ -169,7 +169,7 @@ char *command_rw_read_periodic_response(neu_plugin_t * plugin,
char * json_str = NULL;
neu_json_read_periodic_t header = { .group = (char *) data->group,
.node = (char *) data->driver,
.timestamp = plugin->common.timestamp };
.timestamp = global_timestamp };
neu_json_read_resp_t json = { 0 };
wrap_read_response_json(tags, len, &json);
src/adapter/adapter.c
-24
View File
@@ -54,7 +54,6 @@ static int adapter_update_metric(neu_adapter_t *adapter,
const char *group);
inline static void reply(neu_adapter_t *adapter, neu_reqresp_head_t *header,
void *data);
static int update_timestamp(void *usr_data);
static int level_check(void *usr_data);
static const adapter_callbacks_t callback_funs = {
@@ -87,12 +86,6 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
int rv = 0;
neu_adapter_t * adapter = NULL;
neu_event_io_param_t param = { 0 };
neu_event_timer_param_t timer_param = {
.second = 0,
.millisecond = 10,
.cb = update_timestamp,
};
neu_event_timer_param_t timer_level = {
.second = 30,
.millisecond = 0,
@@ -179,8 +172,6 @@ neu_adapter_t *neu_adapter_create(neu_adapter_info_t *info)
timer_level.usr_data = (void *) adapter;
adapter->timer_lev = neu_event_add_timer(adapter->events, timer_level);
timer_param.usr_data = (void *) adapter;
adapter->timer = neu_event_add_timer(adapter->events, timer_param);
nlog_info("Success to create adapter: %s", adapter->name);
@@ -792,7 +783,6 @@ int neu_adapter_uninit(neu_adapter_t *adapter)
neu_adapter_driver_destroy((neu_adapter_driver_t *) adapter);
}
neu_event_del_timer(adapter->events, adapter->timer);
neu_event_del_timer(adapter->events, adapter->timer_lev);
nlog_info("Stop the adapter(%s)", adapter->name);
@@ -1049,20 +1039,6 @@ static int level_check(void *usr_data)
return 0;
}
static int update_timestamp(void *usr_data)
{
neu_adapter_t *adapter = (neu_adapter_t *) usr_data;
struct timeval tv = { 0 };
uint64_t timestamp = 0;
gettimeofday(&tv, NULL);
timestamp = (int64_t) tv.tv_sec * 1000 + (int64_t) tv.tv_usec / 1000;
adapter->timestamp = timestamp;
neu_plugin_to_plugin_common(adapter->plugin)->timestamp = timestamp;
return 0;
}
void *neu_msg_gen(neu_reqresp_head_t *header, void *data)
{
nng_msg *msg = NULL;
src/adapter/adapter_internal.h
+3 -6
View File
@@ -34,8 +34,6 @@ struct neu_adapter {
char *name;
char *setting;
int64_t timestamp;
neu_node_running_state_e state;
adapter_callbacks_t cb_funs;
@@ -47,10 +45,9 @@ struct neu_adapter {
nng_socket sock;
nng_dialer dialer;
neu_events_t * events;
neu_event_io_t * nng_io;
neu_event_timer_t *timer;
int recv_fd;
neu_events_t * events;
neu_event_io_t *nng_io;
int recv_fd;
neu_event_timer_t *timer_lev;
int64_t timestamp_lev;
src/adapter/driver/driver.c
+8 -8
View File
@@ -97,7 +97,7 @@ static void update(neu_adapter_t *adapter, const char *group, const char *tag,
utarray_foreach(tags, neu_datatag_t *, t)
{
neu_driver_cache_update(driver->cache, group, t->name,
driver->adapter.timestamp, value);
global_timestamp, value);
}
driver->adapter.cb_funs.update_metric(&driver->adapter,
NEU_METRIC_TAG_READS_TOTAL,
@@ -108,8 +108,8 @@ static void update(neu_adapter_t *adapter, const char *group, const char *tag,
utarray_free(tags);
}
} else {
neu_driver_cache_update(driver->cache, group, tag,
driver->adapter.timestamp, value);
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(
@@ -119,7 +119,7 @@ static void update(neu_adapter_t *adapter, const char *group, const char *tag,
nlog_debug(
"update driver: %s, group: %s, tag: %s, type: %s, timestamp: %" PRId64,
driver->adapter.name, group, tag, neu_type_string(value.type),
driver->adapter.timestamp);
global_timestamp);
}
neu_adapter_driver_t *neu_adapter_driver_create()
@@ -212,7 +212,7 @@ void neu_adapter_driver_read_group(neu_adapter_driver_t *driver,
}
} else {
read_group(driver->adapter.timestamp,
read_group(global_timestamp,
neu_group_get_interval(group) *
NEU_DRIVER_TAG_CACHE_EXPIRE_TIME,
driver->cache, cmd->group, tags, resp.tags);
@@ -729,7 +729,7 @@ static int report_callback(void *usr_data)
strcpy(data->driver, group->driver->adapter.name);
strcpy(data->group, group->name);
data->n_tag = read_report_group(group->driver->adapter.timestamp,
data->n_tag = read_report_group(global_timestamp,
neu_group_get_interval(group->group) *
NEU_DRIVER_TAG_CACHE_EXPIRE_TIME,
group->driver->cache, group->name, tags,
@@ -802,12 +802,12 @@ static int read_callback(void *usr_data)
}
if (group->grp.tags != NULL) {
int64_t spend = group->driver->adapter.timestamp;
int64_t spend = global_timestamp;
group->driver->adapter.module->intf_funs->driver.group_timer(
group->driver->adapter.plugin, &group->grp);
spend = group->driver->adapter.timestamp - spend;
spend = global_timestamp - spend;
nlog_info("%s-%s timer: %" PRId64, group->driver->adapter.name,
group->name, spend);
src/base/neu_plugin_common.c
-2
View File
@@ -32,8 +32,6 @@ void neu_plugin_common_init(neu_plugin_common_t *common)
common->magic = NEU_PLUGIN_MAGIC_NUMBER;
common->link_state = NEU_NODE_LINK_STATE_DISCONNECTED;
common->tag_size = 0;
common->timestamp = time(NULL);
common->timestamp *= 1000; // don't need much precision at init
}
bool neu_plugin_common_check(neu_plugin_t *plugin)
src/core/manager.c
+20
View File
@@ -27,6 +27,7 @@
#include "event/event.h"
#include "persist/persist.h"
#include "utils/log.h"
#include "utils/time.h"
#include "adapter.h"
#include "adapter/adapter_internal.h"
@@ -56,6 +57,7 @@ inline static void forward_msg_dup(neu_manager_t *manager, nng_msg *msg,
inline static void forward_msg(neu_manager_t *manager, nng_msg *msg,
const char *ndoe);
static void start_static_adapter(neu_manager_t *manager, const char *name);
static int update_timestamp(void *usr_data);
static int report_nodes_state(void *usr_data);
static void start_single_adapter(neu_manager_t *manager, const char *name,
const char *plugin_name, bool display);
@@ -68,6 +70,12 @@ neu_manager_t *neu_manager_create()
.usr_data = (void *) manager,
.cb = manager_loop,
};
neu_event_timer_param_t timestamp_timer_param = {
.second = 0,
.millisecond = 10,
.cb = update_timestamp,
};
neu_event_timer_param_t timer_param = {
.second = 1,
.millisecond = 0,
@@ -130,6 +138,10 @@ neu_manager_t *neu_manager_create()
timer_param.usr_data = (void *) manager;
manager->timer = neu_event_add_timer(manager->events, timer_param);
timestamp_timer_param.usr_data = (void *) manager;
manager->timer_timestamp =
neu_event_add_timer(manager->events, timestamp_timer_param);
return manager;
}
@@ -142,6 +154,7 @@ void neu_manager_destroy(neu_manager_t *manager)
neu_event_del_timer(manager->events, manager->timer);
neu_event_del_timer(manager->events, manager->timer_lev);
neu_event_del_timer(manager->events, manager->timer_timestamp);
strcpy(header.sender, "manager");
utarray_foreach(pipes, nng_pipe *, pipe)
{
@@ -756,3 +769,10 @@ static int report_nodes_state(void *usr_data)
return 0;
}
static int update_timestamp(void *usr_data)
{
(void) usr_data;
global_timestamp = neu_time_ms();
return 0;
}
src/core/manager_internal.h
+6 -4
View File
@@ -32,16 +32,18 @@
#include "subscribe.h"
typedef struct neu_manager {
nng_socket socket;
neu_events_t * events;
neu_event_io_t * loop;
neu_event_timer_t *timer;
nng_socket socket;
neu_events_t * events;
neu_event_io_t *loop;
neu_plugin_manager_t *plugin_manager;
neu_node_manager_t * node_manager;
neu_subscribe_mgr_t * subscribe_manager;
neu_sub_msg_mgr_t * sub_msg_manager;
neu_event_timer_t *timer;
neu_event_timer_t *timer_timestamp;
neu_event_timer_t *timer_lev;
int64_t timestamp_lev_manager;
} neu_manager_t;
src/main.c
+2
View File
@@ -24,6 +24,7 @@
#include "core/manager.h"
#include "utils/log.h"
#include "utils/time.h"
#include "argparse.h"
#include "daemon.h"
@@ -88,6 +89,7 @@ int main(int argc, char *argv[])
pid_t pid = 0;
neu_cli_args_t args = { 0 };
global_timestamp = neu_time_ms();
neu_cli_args_init(&args, argc, argv);
disable_jwt = args.disable_auth;