diff options
Diffstat (limited to 'aclk/schema-wrappers')
-rw-r--r-- | aclk/schema-wrappers/chart_config.cc | 105 | ||||
-rw-r--r-- | aclk/schema-wrappers/chart_config.h | 50 | ||||
-rw-r--r-- | aclk/schema-wrappers/chart_stream.cc | 337 | ||||
-rw-r--r-- | aclk/schema-wrappers/chart_stream.h | 121 | ||||
-rw-r--r-- | aclk/schema-wrappers/proto_2_json.cc | 16 | ||||
-rw-r--r-- | aclk/schema-wrappers/schema_wrappers.h | 2 |
6 files changed, 0 insertions, 631 deletions
diff --git a/aclk/schema-wrappers/chart_config.cc b/aclk/schema-wrappers/chart_config.cc deleted file mode 100644 index 87e34e0dfc..0000000000 --- a/aclk/schema-wrappers/chart_config.cc +++ /dev/null @@ -1,105 +0,0 @@ -#include "chart_config.h" - -#include "proto/chart/v1/config.pb.h" - -#include "libnetdata/libnetdata.h" - -#include "schema_wrapper_utils.h" - -void destroy_update_chart_config(struct update_chart_config *cfg) -{ - freez(cfg->claim_id); - freez(cfg->node_id); - freez(cfg->hashes); -} - -void destroy_chart_config_updated(struct chart_config_updated *cfg) -{ - freez(cfg->type); - freez(cfg->family); - freez(cfg->context); - freez(cfg->title); - freez(cfg->plugin); - freez(cfg->module); - freez(cfg->units); - freez(cfg->config_hash); -} - -struct update_chart_config parse_update_chart_config(const char *data, size_t len) -{ - chart::v1::UpdateChartConfigs cfgs; - update_chart_config res; - memset(&res, 0, sizeof(res)); - - if (!cfgs.ParseFromArray(data, len)) - return res; - - res.claim_id = strdupz(cfgs.claim_id().c_str()); - res.node_id = strdupz(cfgs.node_id().c_str()); - - // to not do bazillion tiny allocations for individual strings - // we calculate how much memory we will need for all of them - // and allocate at once - int hash_count = cfgs.config_hashes_size(); - size_t total_strlen = 0; - for (int i = 0; i < hash_count; i++) - total_strlen += cfgs.config_hashes(i).length(); - total_strlen += hash_count; //null bytes - - res.hashes = (char**)callocz( 1, - (hash_count+1) * sizeof(char*) + //char * array incl. terminating NULL at the end - total_strlen //strings themselves incl. 1 null byte each - ); - - char* dest = ((char*)res.hashes) + (hash_count + 1 /* NULL ptr */) * sizeof(char *); - // now copy them strings - // null bytes handled by callocz - for (int i = 0; i < hash_count; i++) { - strcpy(dest, cfgs.config_hashes(i).c_str()); - res.hashes[i] = dest; - dest += strlen(dest) + 1 /* end string null */; - } - - return res; -} - -char *generate_chart_configs_updated(size_t *len, const struct chart_config_updated *config_list, int list_size) -{ - chart::v1::ChartConfigsUpdated configs; - for (int i = 0; i < list_size; i++) { - chart::v1::ChartConfigUpdated *config = configs.add_configs(); - config->set_type(config_list[i].type); - if (config_list[i].family) - config->set_family(config_list[i].family); - config->set_context(config_list[i].context); - config->set_title(config_list[i].title); - config->set_priority(config_list[i].priority); - config->set_plugin(config_list[i].plugin); - - if (config_list[i].module) - config->set_module(config_list[i].module); - - switch (config_list[i].chart_type) { - case RRDSET_TYPE_LINE: - config->set_chart_type(chart::v1::LINE); - break; - case RRDSET_TYPE_AREA: - config->set_chart_type(chart::v1::AREA); - break; - case RRDSET_TYPE_STACKED: - config->set_chart_type(chart::v1::STACKED); - break; - default: - return NULL; - } - - config->set_units(config_list[i].units); - config->set_config_hash(config_list[i].config_hash); - } - - *len = PROTO_COMPAT_MSG_SIZE(configs); - char *bin = (char*)mallocz(*len); - configs.SerializeToArray(bin, *len); - - return bin; -} diff --git a/aclk/schema-wrappers/chart_config.h b/aclk/schema-wrappers/chart_config.h deleted file mode 100644 index f08f76b618..0000000000 --- a/aclk/schema-wrappers/chart_config.h +++ /dev/null @@ -1,50 +0,0 @@ -// SPDX-License-Identifier: GPL-3.0-or-later - -#ifndef ACLK_SCHEMA_WRAPPER_CHART_CONFIG_H -#define ACLK_SCHEMA_WRAPPER_CHART_CONFIG_H - -#include <stdlib.h> - -#include "database/rrd.h" - -#ifdef __cplusplus -extern "C" { -#endif - -struct update_chart_config { - char *claim_id; - char *node_id; - char **hashes; -}; - -enum chart_config_chart_type { - LINE, - AREA, - STACKED -}; - -struct chart_config_updated { - char *type; - char *family; - char *context; - char *title; - uint64_t priority; - char *plugin; - char *module; - RRDSET_TYPE chart_type; - char *units; - char *config_hash; -}; - -void destroy_update_chart_config(struct update_chart_config *cfg); -void destroy_chart_config_updated(struct chart_config_updated *cfg); - -struct update_chart_config parse_update_chart_config(const char *data, size_t len); - -char *generate_chart_configs_updated(size_t *len, const struct chart_config_updated *config_list, int list_size); - -#ifdef __cplusplus -} -#endif - -#endif /* ACLK_SCHEMA_WRAPPER_CHART_CONFIG_H */ diff --git a/aclk/schema-wrappers/chart_stream.cc b/aclk/schema-wrappers/chart_stream.cc deleted file mode 100644 index 54c9407586..0000000000 --- a/aclk/schema-wrappers/chart_stream.cc +++ /dev/null @@ -1,337 +0,0 @@ -// SPDX-License-Identifier: GPL-3.0-or-later - -#include "aclk/aclk_util.h" - -#include "proto/chart/v1/stream.pb.h" -#include "chart_stream.h" - -#include "schema_wrapper_utils.h" - -#include <sys/time.h> -#include <stdlib.h> - -stream_charts_and_dims_t parse_stream_charts_and_dims(const char *data, size_t len) -{ - chart::v1::StreamChartsAndDimensions msg; - stream_charts_and_dims_t res; - memset(&res, 0, sizeof(res)); - - if (!msg.ParseFromArray(data, len)) - return res; - - res.node_id = strdup(msg.node_id().c_str()); - res.claim_id = strdup(msg.claim_id().c_str()); - res.seq_id = msg.sequence_id(); - res.batch_id = msg.batch_id(); - set_timeval_from_google_timestamp(msg.seq_id_created_at(), &res.seq_id_created_at); - - return res; -} - -chart_and_dim_ack_t parse_chart_and_dimensions_ack(const char *data, size_t len) -{ - chart::v1::ChartsAndDimensionsAck msg; - chart_and_dim_ack_t res = { .claim_id = NULL, .node_id = NULL, .last_seq_id = 0 }; - - if (!msg.ParseFromArray(data, len)) - return res; - - res.node_id = strdup(msg.node_id().c_str()); - res.claim_id = strdup(msg.claim_id().c_str()); - res.last_seq_id = msg.last_sequence_id(); - - return res; -} - -char *generate_reset_chart_messages(size_t *len, chart_reset_t reset) -{ - chart::v1::ResetChartMessages msg; - - msg.set_claim_id(reset.claim_id); - msg.set_node_id(reset.node_id); - switch (reset.reason) { - case DB_EMPTY: - msg.set_reason(chart::v1::ResetReason::DB_EMPTY); - break; - case SEQ_ID_NOT_EXISTS: - msg.set_reason(chart::v1::ResetReason::SEQ_ID_NOT_EXISTS); - break; - case TIMESTAMP_MISMATCH: - msg.set_reason(chart::v1::ResetReason::TIMESTAMP_MISMATCH); - break; - default: - return NULL; - } - - *len = PROTO_COMPAT_MSG_SIZE(msg); - char *bin = (char*)malloc(*len); - if (bin) - msg.SerializeToArray(bin, *len); - - return bin; -} - -void chart_instance_updated_destroy(struct chart_instance_updated *instance) -{ - freez((char*)instance->id); - freez((char*)instance->claim_id); - - rrdlabels_destroy(instance->chart_labels); - - freez((char*)instance->config_hash); -} - -static int set_chart_instance_updated(chart::v1::ChartInstanceUpdated *chart, const struct chart_instance_updated *update) -{ - google::protobuf::Map<std::string, std::string> *map; - aclk_lib::v1::ACLKMessagePosition *pos; - - chart->set_id(update->id); - chart->set_claim_id(update->claim_id); - chart->set_node_id(update->node_id); - chart->set_name(update->name); - - map = chart->mutable_chart_labels(); - rrdlabels_walkthrough_read(update->chart_labels, label_add_to_map_callback, map); - - switch (update->memory_mode) { - case RRD_MEMORY_MODE_NONE: - chart->set_memory_mode(chart::v1::NONE); - break; - case RRD_MEMORY_MODE_RAM: - chart->set_memory_mode(chart::v1::RAM); - break; - case RRD_MEMORY_MODE_MAP: - chart->set_memory_mode(chart::v1::MAP); - break; - case RRD_MEMORY_MODE_SAVE: - chart->set_memory_mode(chart::v1::SAVE); - break; - case RRD_MEMORY_MODE_ALLOC: - chart->set_memory_mode(chart::v1::ALLOC); - break; - case RRD_MEMORY_MODE_DBENGINE: - chart->set_memory_mode(chart::v1::DB_ENGINE); - break; - default: - return 1; - break; - } - - chart->set_update_every_interval(update->update_every); - chart->set_config_hash(update->config_hash); - - pos = chart->mutable_position(); - pos->set_sequence_id(update->position.sequence_id); - pos->set_previous_sequence_id(update->position.previous_sequence_id); - set_google_timestamp_from_timeval(update->position.seq_id_creation_time, pos->mutable_seq_id_created_at()); - - return 0; -} - -static int set_chart_dim_updated(chart::v1::ChartDimensionUpdated *dim, const struct chart_dimension_updated *c_dim) -{ - aclk_lib::v1::ACLKMessagePosition *pos; - - dim->set_id(c_dim->id); - dim->set_chart_id(c_dim->chart_id); - dim->set_node_id(c_dim->node_id); - dim->set_claim_id(c_dim->claim_id); - dim->set_name(c_dim->name); - - set_google_timestamp_from_timeval(c_dim->created_at, dim->mutable_created_at()); - set_google_timestamp_from_timeval(c_dim->last_timestamp, dim->mutable_last_timestamp()); - - pos = dim->mutable_position(); - pos->set_sequence_id(c_dim->position.sequence_id); - pos->set_previous_sequence_id(c_dim->position.previous_sequence_id); - set_google_timestamp_from_timeval(c_dim->position.seq_id_creation_time, pos->mutable_seq_id_created_at()); - - return 0; -} - -char *generate_charts_and_dimensions_updated(size_t *len, char **payloads, size_t *payload_sizes, int *is_dim, struct aclk_message_position *new_positions, uint64_t batch_id) -{ - chart::v1::ChartsAndDimensionsUpdated msg; - chart::v1::ChartInstanceUpdated db_chart; - chart::v1::ChartDimensionUpdated db_dim; - aclk_lib::v1::ACLKMessagePosition *pos; - - msg.set_batch_id(batch_id); - - for (int i = 0; payloads[i]; i++) { - if (is_dim[i]) { - if (!db_dim.ParseFromArray(payloads[i], payload_sizes[i])) { - error("[ACLK] Could not parse chart::v1::chart_dimension_updated"); - return NULL; - } - - pos = db_dim.mutable_position(); - pos->set_sequence_id(new_positions[i].sequence_id); - pos->set_previous_sequence_id(new_positions[i].previous_sequence_id); - set_google_timestamp_from_timeval(new_positions[i].seq_id_creation_time, pos->mutable_seq_id_created_at()); - - chart::v1::ChartDimensionUpdated *dim = msg.add_dimensions(); - *dim = db_dim; - } else { - if (!db_chart.ParseFromArray(payloads[i], payload_sizes[i])) { - error("[ACLK] Could not parse chart::v1::ChartInstanceUpdated"); - return NULL; - } - - pos = db_chart.mutable_position(); - pos->set_sequence_id(new_positions[i].sequence_id); - pos->set_previous_sequence_id(new_positions[i].previous_sequence_id); - set_google_timestamp_from_timeval(new_positions[i].seq_id_creation_time, pos->mutable_seq_id_created_at()); - - chart::v1::ChartInstanceUpdated *chart = msg.add_charts(); - *chart = db_chart; - } - } - - *len = PROTO_COMPAT_MSG_SIZE(msg); - char *bin = (char*)mallocz(*len); - msg.SerializeToArray(bin, *len); - - return bin; -} - -char *generate_charts_updated(size_t *len, char **payloads, size_t *payload_sizes, struct aclk_message_position *new_positions) -{ - chart::v1::ChartsAndDimensionsUpdated msg; - - msg.set_batch_id(chart_batch_id); - - for (int i = 0; payloads[i]; i++) { - chart::v1::ChartInstanceUpdated db_msg; - chart::v1::ChartInstanceUpdated *chart; - aclk_lib::v1::ACLKMessagePosition *pos; - - if (!db_msg.ParseFromArray(payloads[i], payload_sizes[i])) { - error("[ACLK] Could not parse chart::v1::ChartInstanceUpdated"); - return NULL; - } - - pos = db_msg.mutable_position(); - pos->set_sequence_id(new_positions[i].sequence_id); - pos->set_previous_sequence_id(new_positions[i].previous_sequence_id); - set_google_timestamp_from_timeval(new_positions[i].seq_id_creation_time, pos->mutable_seq_id_created_at()); - - chart = msg.add_charts(); - *chart = db_msg; - } - - *len = PROTO_COMPAT_MSG_SIZE(msg); - char *bin = (char*)mallocz(*len); - msg.SerializeToArray(bin, *len); - - return bin; -} - -char *generate_chart_dimensions_updated(size_t *len, char **payloads, size_t *payload_sizes, struct aclk_message_position *new_positions) -{ - chart::v1::ChartsAndDimensionsUpdated msg; - - msg.set_batch_id(chart_batch_id); - - for (int i = 0; payloads[i]; i++) { - chart::v1::ChartDimensionUpdated db_msg; - chart::v1::ChartDimensionUpdated *dim; - aclk_lib::v1::ACLKMessagePosition *pos; - - if (!db_msg.ParseFromArray(payloads[i], payload_sizes[i])) { - error("[ACLK] Could not parse chart::v1::chart_dimension_updated"); - return NULL; - } - - pos = db_msg.mutable_position(); - pos->set_sequence_id(new_positions[i].sequence_id); - pos->set_previous_sequence_id(new_positions[i].previous_sequence_id); - set_google_timestamp_from_timeval(new_positions[i].seq_id_creation_time, pos->mutable_seq_id_created_at()); - - dim = msg.add_dimensions(); - *dim = db_msg; - } - - *len = PROTO_COMPAT_MSG_SIZE(msg); - char *bin = (char*)mallocz(*len); - msg.SerializeToArray(bin, *len); - - return bin; -} - -char *generate_chart_instance_updated(size_t *len, const struct chart_instance_updated *update) -{ - chart::v1::ChartInstanceUpdated *chart = new chart::v1::ChartInstanceUpdated(); - - if (set_chart_instance_updated(chart, update)) - return NULL; - - *len = PROTO_COMPAT_MSG_SIZE_PTR(chart); - char *bin = (char*)mallocz(*len); - chart->SerializeToArray(bin, *len); - - delete chart; - return bin; -} - -char *generate_chart_dimension_updated(size_t *len, const struct chart_dimension_updated *dim) -{ - chart::v1::ChartDimensionUpdated *proto_dim = new chart::v1::ChartDimensionUpdated(); - - if (set_chart_dim_updated(proto_dim, dim)) - return NULL; - - *len = PROTO_COMPAT_MSG_SIZE_PTR(proto_dim); - char *bin = (char*)mallocz(*len); - proto_dim->SerializeToArray(bin, *len); - - delete proto_dim; - return bin; -} - -using namespace google::protobuf; - -char *generate_retention_updated(size_t *len, struct retention_updated *data) -{ - chart::v1::RetentionUpdated msg; - - msg.set_claim_id(data->claim_id); - msg.set_node_id(data->node_id); - - switch (data->memory_mode) { - case RRD_MEMORY_MODE_NONE: - msg.set_memory_mode(chart::v1::NONE); - break; - case RRD_MEMORY_MODE_RAM: - msg.set_memory_mode(chart::v1::RAM); - break; - case RRD_MEMORY_MODE_MAP: - msg.set_memory_mode(chart::v1::MAP); - break; - case RRD_MEMORY_MODE_SAVE: - msg.set_memory_mode(chart::v1::SAVE); - break; - case RRD_MEMORY_MODE_ALLOC: - msg.set_memory_mode(chart::v1::ALLOC); - break; - case RRD_MEMORY_MODE_DBENGINE: - msg.set_memory_mode(chart::v1::DB_ENGINE); - break; - default: - return NULL; - } - - for (int i = 0; i < data->interval_duration_count; i++) { - Map<uint32, uint32> *map = msg.mutable_interval_durations(); - map->insert({data->interval_durations[i].update_every, data->interval_durations[i].retention}); - } - - set_google_timestamp_from_timeval(data->rotation_timestamp, msg.mutable_rotation_timestamp()); - - *len = PROTO_COMPAT_MSG_SIZE(msg); - char *bin = (char*)mallocz(*len); - msg.SerializeToArray(bin, *len); - - return bin; -} diff --git a/aclk/schema-wrappers/chart_stream.h b/aclk/schema-wrappers/chart_stream.h deleted file mode 100644 index 9048668683..0000000000 --- a/aclk/schema-wrappers/chart_stream.h +++ /dev/null @@ -1,121 +0,0 @@ -// SPDX-License-Identifier: GPL-3.0-or-later - -#ifndef ACLK_SCHEMA_WRAPPER_CHART_STREAM_H -#define ACLK_SCHEMA_WRAPPER_CHART_STREAM_H - -#ifdef __cplusplus -extern "C" { -#endif - -#include "database/rrd.h" - -typedef struct { - char* claim_id; - char* node_id; - - uint64_t seq_id; - uint64_t batch_id; - - struct timeval seq_id_created_at; -} stream_charts_and_dims_t; - -stream_charts_and_dims_t parse_stream_charts_and_dims(const char *data, size_t len); - -typedef struct { - char* claim_id; - char* node_id; - - uint64_t last_seq_id; -} chart_and_dim_ack_t; - -chart_and_dim_ack_t parse_chart_and_dimensions_ack(const char *data, size_t len); - -enum chart_reset_reason { - DB_EMPTY, - SEQ_ID_NOT_EXISTS, - TIMESTAMP_MISMATCH -}; - -typedef struct { - char *claim_id; - char *node_id; - - enum chart_reset_reason reason; -} chart_reset_t; - -char *generate_reset_chart_messages(size_t *len, const chart_reset_t reset); - -struct aclk_message_position { - uint64_t sequence_id; - struct timeval seq_id_creation_time; - uint64_t previous_sequence_id; -}; - -struct chart_instance_updated { - const char *id; - const char *claim_id; - const char *node_id; - const char *name; - - DICTIONARY *chart_labels; - - RRD_MEMORY_MODE memory_mode; - - uint32_t update_every; - const char * config_hash; - - struct aclk_message_position position; -}; - -void chart_instance_updated_destroy(struct chart_instance_updated *instance); - -struct chart_dimension_updated { - const char *id; - const char *chart_id; - const char *node_id; - const char *claim_id; - const char *name; - struct timeval created_at; - struct timeval last_timestamp; - struct aclk_message_position position; -}; - -typedef struct { - struct chart_instance_updated *charts; - uint16_t chart_count; - - struct chart_dimension_updated *dims; - uint16_t dim_count; - - uint64_t batch_id; -} charts_and_dims_updated_t; - -struct interval_duration { - uint32_t update_every; - uint32_t retention; -}; - -struct retention_updated { - char *claim_id; - char *node_id; - - RRD_MEMORY_MODE memory_mode; - - struct interval_duration *interval_durations; - int interval_duration_count; - - struct timeval rotation_timestamp; -}; - -char *generate_charts_and_dimensions_updated(size_t *len, char **payloads, size_t *payload_sizes, int *is_dim, struct aclk_message_position *new_positions, uint64_t batch_id); -char *generate_charts_updated(size_t *len, char **payloads, size_t *payload_sizes, struct aclk_message_position *new_positions); -char *generate_chart_instance_updated(size_t *len, const struct chart_instance_updated *update); -char *generate_chart_dimensions_updated(size_t *len, char **payloads, size_t *payload_sizes, struct aclk_message_position *new_positions); -char *generate_chart_dimension_updated(size_t *len, const struct chart_dimension_updated *dim); -char *generate_retention_updated(size_t *len, struct retention_updated *data); - -#ifdef __cplusplus -} -#endif - -#endif /* ACLK_SCHEMA_WRAPPER_CHART_STREAM_H */ diff --git a/aclk/schema-wrappers/proto_2_json.cc b/aclk/schema-wrappers/proto_2_json.cc index 0e473eb6c4..8853b2e083 100644 --- a/aclk/schema-wrappers/proto_2_json.cc +++ b/aclk/schema-wrappers/proto_2_json.cc @@ -4,8 +4,6 @@ #include "proto/alarm/v1/config.pb.h" #include "proto/alarm/v1/stream.pb.h" #include "proto/aclk/v1/lib.pb.h" -#include "proto/chart/v1/config.pb.h" -#include "proto/chart/v1/stream.pb.h" #include "proto/agent/v1/connection.pb.h" #include "proto/agent/v1/disconnect.pb.h" #include "proto/nodeinstance/connection/v1/connection.pb.h" @@ -29,14 +27,6 @@ static google::protobuf::Message *msg_name_to_protomsg(const char *msgname) return new nodeinstance::v1::UpdateNodeInstanceConnection; if (!strcmp(msgname, "CreateNodeInstance")) return new nodeinstance::create::v1::CreateNodeInstance; - if (!strcmp(msgname, "ChartsAndDimensionsUpdated")) - return new chart::v1::ChartsAndDimensionsUpdated; - if (!strcmp(msgname, "ChartConfigsUpdated")) - return new chart::v1::ChartConfigsUpdated; - if (!strcmp(msgname, "ResetChartMessages")) - return new chart::v1::ResetChartMessages; - if (!strcmp(msgname, "RetentionUpdated")) - return new chart::v1::RetentionUpdated; if (!strcmp(msgname, "UpdateNodeInfo")) return new nodeinstance::info::v1::UpdateNodeInfo; if (!strcmp(msgname, "AlarmLogHealth")) @@ -59,12 +49,6 @@ static google::protobuf::Message *msg_name_to_protomsg(const char *msgname) return new nodeinstance::create::v1::CreateNodeInstanceResult; if (!strcmp(msgname, "SendNodeInstances")) return new agent::v1::SendNodeInstances; - if (!strcmp(msgname, "StreamChartsAndDimensions")) - return new chart::v1::StreamChartsAndDimensions; - if (!strcmp(msgname, "ChartsAndDimensionsAck")) - return new chart::v1::ChartsAndDimensionsAck; - if (!strcmp(msgname, "UpdateChartConfigs")) - return new chart::v1::UpdateChartConfigs; if (!strcmp(msgname, "StartAlarmStreaming")) return new alarms::v1::StartAlarmStreaming; if (!strcmp(msgname, "SendAlarmLogHealth")) diff --git a/aclk/schema-wrappers/schema_wrappers.h b/aclk/schema-wrappers/schema_wrappers.h index 26412cacc7..a96f7ea7ac 100644 --- a/aclk/schema-wrappers/schema_wrappers.h +++ b/aclk/schema-wrappers/schema_wrappers.h @@ -8,8 +8,6 @@ #include "connection.h" #include "node_connection.h" #include "node_creation.h" -#include "chart_config.h" -#include "chart_stream.h" #include "alarm_config.h" #include "alarm_stream.h" #include "node_info.h" |