diff options
author | Costa Tsaousis <costa@netdata.cloud> | 2023-06-08 16:33:22 +0300 |
---|---|---|
committer | GitHub <noreply@github.com> | 2023-06-08 16:33:22 +0300 |
commit | 80d83b7bd1eca5872ed3ac5c34eb8bcb5fbd56e8 (patch) | |
tree | 8afd42a59cc6b114b26fdd3e38b137db99357154 /streaming | |
parent | 028e26a194f5421432b577ee67afed8b66c0f6b7 (diff) |
api v2 nodes for streaming statuses (#15162)
* api v2 nodes for streaming statuses
* remove test
* move parts of the output
* in api/v2/data return 5 values per point when aggregation=percentage and raw option is given; return final values when aggregation=percentage is not the final grouping
Diffstat (limited to 'streaming')
-rw-r--r-- | streaming/receiver.c | 68 | ||||
-rw-r--r-- | streaming/replication.c | 11 | ||||
-rw-r--r-- | streaming/rrdpush.c | 37 | ||||
-rw-r--r-- | streaming/rrdpush.h | 30 | ||||
-rw-r--r-- | streaming/sender.c | 170 |
5 files changed, 285 insertions, 31 deletions
diff --git a/streaming/receiver.c b/streaming/receiver.c index 2b72486fd9..a89fdf04a3 100644 --- a/streaming/receiver.c +++ b/streaming/receiver.c @@ -423,6 +423,58 @@ static void rrdpush_receiver_replication_reset(RRDHOST *host) { rrdhost_receiver_replicating_charts_zero(host); } +void rrdhost_receiver_to_json(BUFFER *wb, RRDHOST *host, const char *key, time_t now __maybe_unused) { + size_t receiver_hops = host->system_info ? host->system_info->hops : (host == localhost) ? 0 : 1; + + netdata_mutex_lock(&host->receiver_lock); + + buffer_json_member_add_object(wb, key); + buffer_json_member_add_uint64(wb, "hops", receiver_hops); + + bool online = host == localhost || !rrdhost_flag_check(host, RRDHOST_FLAG_ORPHAN | RRDHOST_FLAG_RRDPUSH_RECEIVER_DISCONNECTED); + buffer_json_member_add_boolean(wb, "online", online); + + if(host->child_connect_time || host->child_disconnected_time) { + time_t since = MAX(host->child_connect_time, host->child_disconnected_time); + buffer_json_member_add_time_t(wb, "since", since); + buffer_json_member_add_time_t(wb, "age", now - since); + } + + if(!online && host->rrdpush_last_receiver_exit_reason) + buffer_json_member_add_string(wb, "reason", host->rrdpush_last_receiver_exit_reason); + + if(host != localhost && host->receiver) { + buffer_json_member_add_object(wb, "replication"); + { + size_t instances = rrdhost_receiver_replicating_charts(host); + buffer_json_member_add_boolean(wb, "in_progress", instances); + buffer_json_member_add_double(wb, "completion", host->rrdpush_receiver_replication_percent); + buffer_json_member_add_uint64(wb, "instances", instances); + } + buffer_json_object_close(wb); // replication + + buffer_json_member_add_object(wb, "source"); + { + + char buf[1024 + 1]; + SOCKET_PEERS peers = socket_peers(host->receiver->fd); + bool ssl = SSL_connection(&host->receiver->ssl); + + snprintfz(buf, 1024, "[%s]:%d%s", peers.local.ip, peers.local.port, ssl ? ":SSL" : ""); + buffer_json_member_add_string(wb, "local", buf); + + snprintfz(buf, 1024, "[%s]:%d%s", peers.peer.ip, peers.peer.port, ssl ? ":SSL" : ""); + buffer_json_member_add_string(wb, "remote", buf); + + stream_capabilities_to_json_array(wb, host->receiver->capabilities, "capabilities"); + } + buffer_json_object_close(wb); // source + } + buffer_json_object_close(wb); // collection + + netdata_mutex_unlock(&host->receiver_lock); +} + static bool rrdhost_set_receiver(RRDHOST *host, struct receiver_state *rpt) { bool signal_rrdcontext = false; bool set_this = false; @@ -496,6 +548,7 @@ static void rrdhost_clear_receiver(struct receiver_state *rpt) { rrdhost_flag_set(host, RRDHOST_FLAG_ORPHAN); host->receiver = NULL; + host->rrdpush_last_receiver_exit_reason = rpt->exit.reason; } netdata_mutex_unlock(&host->receiver_lock); @@ -546,6 +599,18 @@ bool stop_streaming_receiver(RRDHOST *host, const char *reason) { return ret; } +static void rrdpush_send_error_on_taken_over_connection(struct receiver_state *rpt, const char *msg) { + send_timeout( +#ifdef ENABLE_HTTPS + &rpt->ssl, +#endif + rpt->fd, + (char *)msg, + strlen(msg), + 0, + 5); +} + void rrdpush_receive_log_status(struct receiver_state *rpt, const char *msg, const char *status) { log_stream_connection(rpt->client_ip, rpt->client_port, @@ -677,11 +742,13 @@ static void rrdpush_receive(struct receiver_state *rpt) if(!host) { rrdpush_receive_log_status(rpt, "failed to find/create host structure", "INTERNAL ERROR DROPPING CONNECTION"); + rrdpush_send_error_on_taken_over_connection(rpt, START_STREAMING_ERROR_INTERNAL_ERROR); goto cleanup; } if (unlikely(rrdhost_flag_check(host, RRDHOST_FLAG_PENDING_CONTEXT_LOAD))) { rrdpush_receive_log_status(rpt, "host is initializing", "INITIALIZATION IN PROGRESS RETRY LATER"); + rrdpush_send_error_on_taken_over_connection(rpt, START_STREAMING_ERROR_INITIALIZATION); goto cleanup; } @@ -690,6 +757,7 @@ static void rrdpush_receive(struct receiver_state *rpt) if(!rrdhost_set_receiver(host, rpt)) { rrdpush_receive_log_status(rpt, "host is already served by another receiver", "DUPLICATE RECEIVER DROPPING CONNECTION"); + rrdpush_send_error_on_taken_over_connection(rpt, START_STREAMING_ERROR_ALREADY_STREAMING); goto cleanup; } } diff --git a/streaming/replication.c b/streaming/replication.c index a50913a1a4..c6fafc357b 100644 --- a/streaming/replication.c +++ b/streaming/replication.c @@ -274,6 +274,12 @@ static void replication_query_finalize(BUFFER *wb, struct replication_query *q, replication_queries.queries_finished += queries; replication_queries.points_read += q->points_read; replication_queries.points_generated += q->points_generated; + + if(q->st && q->st->rrdhost->sender) { + struct sender_state *s = q->st->rrdhost->sender; + s->replication.latest_completed_before_t = q->query.before; + } + netdata_spinlock_unlock(&replication_queries.spinlock); } @@ -644,7 +650,7 @@ bool replication_response_execute_and_finalize(struct replication_query *q, size buffer_fast_strcat(wb, "\n", 1); worker_is_busy(WORKER_JOB_BUFFER_COMMIT); - sender_commit(host->sender, wb); + sender_commit(host->sender, wb, STREAM_TRAFFIC_TYPE_REPLICATION); worker_is_busy(WORKER_JOB_CLEANUP); if(enable_streaming) { @@ -1466,6 +1472,9 @@ void replication_add_request(struct sender_state *sender, const char *chart_id, .not_indexed_preprocessing = false, }; + if(!sender->replication.oldest_request_after_t || rq.after < sender->replication.oldest_request_after_t) + sender->replication.oldest_request_after_t = rq.after; + if(start_streaming && rrdpush_sender_get_buffer_used_percent(sender) <= STREAMING_START_MAX_SENDER_BUFFER_PERCENTAGE_ALLOWED) replication_execute_request(&rq, false); diff --git a/streaming/rrdpush.c b/streaming/rrdpush.c index a113fbae05..c481871cc7 100644 --- a/streaming/rrdpush.c +++ b/streaming/rrdpush.c @@ -375,7 +375,7 @@ bool rrdset_push_chart_definition_now(RRDSET *st) { BUFFER *wb = sender_start(host->sender); rrdpush_send_chart_definition(wb, st); - sender_commit(host->sender, wb); + sender_commit(host->sender, wb, STREAM_TRAFFIC_TYPE_METADATA); sender_thread_buffer_free(); return true; @@ -443,7 +443,7 @@ void rrdset_push_metrics_finished(RRDSET_STREAM_BUFFER *rsb, RRDSET *st) { buffer_fast_strcat(rsb->wb, PLUGINSD_KEYWORD_END_V2 "\n", sizeof(PLUGINSD_KEYWORD_END_V2) - 1 + 1); } - sender_commit(st->rrdhost->sender, rsb->wb); + sender_commit(st->rrdhost->sender, rsb->wb, STREAM_TRAFFIC_TYPE_DATA); *rsb = (RRDSET_STREAM_BUFFER){ .wb = NULL, }; } @@ -483,7 +483,7 @@ RRDSET_STREAM_BUFFER rrdset_push_metric_initialize(RRDSET *st, time_t wall_clock if(unlikely(!exposed_upstream)) { BUFFER *wb = sender_start(host->sender); replication_in_progress = rrdpush_send_chart_definition(wb, st); - sender_commit(host->sender, wb); + sender_commit(host->sender, wb, STREAM_TRAFFIC_TYPE_METADATA); } if(replication_in_progress) @@ -514,7 +514,7 @@ void rrdpush_send_host_labels(RRDHOST *host) { rrdlabels_walkthrough_read(host->rrdlabels, send_labels_callback, wb); buffer_sprintf(wb, "OVERWRITE %s\n", "labels"); - sender_commit(host->sender, wb); + sender_commit(host->sender, wb, STREAM_TRAFFIC_TYPE_METADATA); sender_thread_buffer_free(); } @@ -533,7 +533,7 @@ void rrdpush_claimed_id(RRDHOST *host) buffer_sprintf(wb, "CLAIMED_ID %s %s\n", host->machine_guid, (host->aclk_state.claimed_id ? host->aclk_state.claimed_id : "NULL") ); rrdhost_aclk_state_unlock(host); - sender_commit(host->sender, wb); + sender_commit(host->sender, wb, STREAM_TRAFFIC_TYPE_METADATA); sender_thread_buffer_free(); } @@ -706,7 +706,7 @@ int rrdpush_receiver_permission_denied(struct web_client *w) { // we always respond with the same message and error code // to prevent an attacker from gaining info about the error buffer_flush(w->response.data); - buffer_sprintf(w->response.data, "You are not permitted to access this. Check the logs for more info."); + buffer_strcat(w->response.data, START_STREAMING_ERROR_NOT_PERMITTED); return HTTP_RESP_UNAUTHORIZED; } @@ -714,7 +714,7 @@ int rrdpush_receiver_too_busy_now(struct web_client *w) { // we always respond with the same message and error code // to prevent an attacker from gaining info about the error buffer_flush(w->response.data); - buffer_sprintf(w->response.data, "The server is too busy now to accept this request. Try later."); + buffer_strcat(w->response.data, START_STREAMING_ERROR_BUSY_TRY_LATER); return HTTP_RESP_SERVICE_UNAVAILABLE; } @@ -1146,7 +1146,7 @@ int rrdpush_receiver_thread_spawn(struct web_client *w, char *decoded_query_stri // Have not set WEB_CLIENT_FLAG_DONT_CLOSE_SOCKET - caller should clean up buffer_flush(w->response.data); - buffer_strcat(w->response.data, "This GUID is already streaming to this server"); + buffer_strcat(w->response.data, START_STREAMING_ERROR_ALREADY_STREAMING); receiver_state_free(rpt); return HTTP_RESP_CONFLICT; } @@ -1191,15 +1191,18 @@ static struct { { STREAM_HANDSHAKE_OK_V3, "OK_V3" }, { STREAM_HANDSHAKE_OK_V2, "OK_V2" }, { STREAM_HANDSHAKE_OK_V1, "OK_V1" }, - { STREAM_HANDSHAKE_ERROR_BAD_HANDSHAKE, "ERROR_BAD_HANDSHAKE" }, - { STREAM_HANDSHAKE_ERROR_LOCALHOST, "ERROR_LOCALHOST" }, - { STREAM_HANDSHAKE_ERROR_ALREADY_CONNECTED, "ERROR_ALREADY_CONNECTED" }, - { STREAM_HANDSHAKE_ERROR_DENIED, "ERROR_DENIED" }, - { STREAM_HANDSHAKE_ERROR_SEND_TIMEOUT, "ERROR_SEND_TIMEOUT" }, - { STREAM_HANDSHAKE_ERROR_RECEIVE_TIMEOUT, "ERROR_RECEIVE_TIMEOUT" }, - { STREAM_HANDSHAKE_ERROR_INVALID_CERTIFICATE, "ERROR_INVALID_CERTIFICATE" }, - { STREAM_HANDSHAKE_ERROR_SSL_ERROR, "ERROR_SSL_ERROR" }, - { STREAM_HANDSHAKE_ERROR_CANT_CONNECT, "ERROR_CANT_CONNECT" }, + { STREAM_HANDSHAKE_ERROR_BAD_HANDSHAKE, "BAD HANDSHAKE" }, + { STREAM_HANDSHAKE_ERROR_LOCALHOST, "LOCALHOST" }, + { STREAM_HANDSHAKE_ERROR_ALREADY_CONNECTED, "ALREADY CONNECTED" }, + { STREAM_HANDSHAKE_ERROR_DENIED, "DENIED" }, + { STREAM_HANDSHAKE_ERROR_SEND_TIMEOUT, "SEND TIMEOUT" }, + { STREAM_HANDSHAKE_ERROR_RECEIVE_TIMEOUT, "RECEIVE TIMEOUT" }, + { STREAM_HANDSHAKE_ERROR_INVALID_CERTIFICATE, "INVALID CERTIFICATE" }, + { STREAM_HANDSHAKE_ERROR_SSL_ERROR, "SSL ERROR" }, + { STREAM_HANDSHAKE_ERROR_CANT_CONNECT, "CANT CONNECT" }, + { STREAM_HANDSHAKE_BUSY_TRY_LATER, "BUSY TRY LATER" }, + { STREAM_HANDSHAKE_INTERNAL_ERROR, "INTERNAL ERROR" }, + { STREAM_HANDSHAKE_INITIALIZATION, "INITIALIZING" }, { 0, NULL }, }; diff --git a/streaming/rrdpush.h b/streaming/rrdpush.h index cb33694cda..f97c8ddfb3 100644 --- a/streaming/rrdpush.h +++ b/streaming/rrdpush.h @@ -72,6 +72,9 @@ STREAM_CAPABILITIES stream_our_capabilities(); #define START_STREAMING_ERROR_SAME_LOCALHOST "Don't hit me baby, you are trying to stream my localhost back" #define START_STREAMING_ERROR_ALREADY_STREAMING "This GUID is already streaming to this server" #define START_STREAMING_ERROR_NOT_PERMITTED "You are not permitted to access this. Check the logs for more info." +#define START_STREAMING_ERROR_BUSY_TRY_LATER "The server is too busy now to accept this request. Try later." +#define START_STREAMING_ERROR_INTERNAL_ERROR "The server encountered an internal error. Try later." +#define START_STREAMING_ERROR_INITIALIZATION "The server is initializing. Try later." typedef enum { STREAM_HANDSHAKE_OK_V5 = 5, // COMPRESSION @@ -87,12 +90,27 @@ typedef enum { STREAM_HANDSHAKE_ERROR_RECEIVE_TIMEOUT = -6, STREAM_HANDSHAKE_ERROR_INVALID_CERTIFICATE = -7, STREAM_HANDSHAKE_ERROR_SSL_ERROR = -8, - STREAM_HANDSHAKE_ERROR_CANT_CONNECT = -9 + STREAM_HANDSHAKE_ERROR_CANT_CONNECT = -9, + STREAM_HANDSHAKE_BUSY_TRY_LATER = -10, + STREAM_HANDSHAKE_INTERNAL_ERROR = -11, + STREAM_HANDSHAKE_INITIALIZATION = -12, } STREAM_HANDSHAKE; // ---------------------------------------------------------------------------- +typedef enum __attribute__((packed)) { + STREAM_TRAFFIC_TYPE_REPLICATION, + STREAM_TRAFFIC_TYPE_FUNCTIONS, + STREAM_TRAFFIC_TYPE_METADATA, + STREAM_TRAFFIC_TYPE_DATA, + + // terminator + STREAM_TRAFFIC_TYPE_MAX, +} STREAM_TRAFFIC_TYPE; + +// ---------------------------------------------------------------------------- + typedef struct { char *os_name; char *os_id; @@ -148,6 +166,7 @@ struct sender_state { size_t sent_bytes_on_this_connection; size_t send_attempts; time_t last_traffic_seen_t; + time_t last_state_since_t; // the timestamp of the last state (online/offline) change size_t not_connected_loops; // Metrics are collected asynchronously by collector threads calling rrdset_done_push(). This can also trigger // the lazy creation of the sender thread - both cases (buffer access and thread creation) are guarded here. @@ -157,6 +176,8 @@ struct sender_state { int read_len; STREAM_CAPABILITIES capabilities; + size_t sent_bytes_on_this_connection_per_type[STREAM_TRAFFIC_TYPE_MAX]; + int rrdpush_sender_pipe[2]; // collector to sender thread signaling int rrdpush_sender_socket; @@ -176,6 +197,8 @@ struct sender_state { struct { DICTIONARY *requests; // de-duplication of replication requests, per chart + time_t oldest_request_after_t; // the timestamp of the oldest replication request + time_t latest_completed_before_t; // the timestamp of the latest replication request struct { size_t pending_requests; // the currently outstanding replication requests @@ -306,7 +329,7 @@ void rrdpush_destinations_init(RRDHOST *host); void rrdpush_destinations_free(RRDHOST *host); BUFFER *sender_start(struct sender_state *s); -void sender_commit(struct sender_state *s, BUFFER *wb); +void sender_commit(struct sender_state *s, BUFFER *wb, STREAM_TRAFFIC_TYPE type); int rrdpush_init(); bool rrdpush_receiver_needs_dbengine(); int configured_as_parent(); @@ -368,6 +391,9 @@ bool stop_streaming_receiver(RRDHOST *host, const char *reason); void sender_thread_buffer_free(void); +void rrdhost_receiver_to_json(BUFFER *wb, RRDHOST *host, const char *key, time_t now __maybe_unused); +void rrdhost_sender_to_json(BUFFER *wb, RRDHOST *host, const char *key, time_t now __maybe_unused); + #include "replication.h" #endif //NETDATA_RRDPUSH_H diff --git a/streaming/sender.c b/streaming/sender.c index cedf58ce2d..b207039b5e 100644 --- a/streaming/sender.c +++ b/streaming/sender.c @@ -84,7 +84,7 @@ static inline void deactivate_compression(struct sender_state *s) { #define SENDER_BUFFER_ADAPT_TO_TIMES_MAX_SIZE 3 // Collector thread finishing a transmission -void sender_commit(struct sender_state *s, BUFFER *wb) { +void sender_commit(struct sender_state *s, BUFFER *wb, STREAM_TRAFFIC_TYPE type) { if(unlikely(wb != sender_thread_buffer)) fatal("STREAMING: sender is trying to commit a buffer that is not this thread's buffer."); @@ -163,6 +163,8 @@ void sender_commit(struct sender_state *s, BUFFER *wb) { if(cbuffer_add_unsafe(s->buffer, dst, dst_len)) s->flags |= SENDER_FLAG_OVERFLOW; + else + s->sent_bytes_on_this_connection_per_type[type] += dst_len; src = src + size_to_compress; src_len -= size_to_compress; @@ -170,9 +172,13 @@ void sender_commit(struct sender_state *s, BUFFER *wb) { } else if(cbuffer_add_unsafe(s->buffer, src, src_len)) s->flags |= SENDER_FLAG_OVERFLOW; + else + s->sent_bytes_on_this_connection_per_type[type] += src_len; #else if(cbuffer_add_unsafe(s->buffer, src, src_len)) s->flags |= SENDER_FLAG_OVERFLOW; + else + s->sent_bytes_on_this_connection_per_type[type] += src_len; #endif replication_recalculate_buffer_used_ratio_unsafe(s); @@ -204,7 +210,7 @@ void rrdpush_sender_send_this_host_variable_now(RRDHOST *host, const RRDVAR_ACQU if(rrdhost_can_send_definitions_to_parent(host)) { BUFFER *wb = sender_start(host->sender); rrdpush_sender_add_host_variable_to_buffer(wb, rva); - sender_commit(host->sender, wb); + sender_commit(host->sender, wb, STREAM_TRAFFIC_TYPE_METADATA); sender_thread_buffer_free(); } } @@ -233,7 +239,7 @@ static void rrdpush_sender_thread_send_custom_host_variables(RRDHOST *host) { }; int ret = rrdvar_walkthrough_read(host->rrdvars, rrdpush_sender_thread_custom_host_variables_callback, &tmp); (void)ret; - sender_commit(host->sender, wb); + sender_commit(host->sender, wb, STREAM_TRAFFIC_TYPE_METADATA); sender_thread_buffer_free(); debug(D_STREAM, "RRDVAR sent %d VARIABLES", ret); @@ -426,6 +432,33 @@ struct { .worker_job_id = WORKER_SENDER_JOB_DISCONNECT_BAD_HANDSHAKE, .postpone_reconnect_seconds = 1 * 60, // 1 minute }, + { + .response = START_STREAMING_ERROR_BUSY_TRY_LATER, + .length = sizeof(START_STREAMING_ERROR_BUSY_TRY_LATER) - 1, + .version = STREAM_HANDSHAKE_BUSY_TRY_LATER, + .dynamic = false, + .error = "remote server is currently busy, we should try later", + .worker_job_id = WORKER_SENDER_JOB_DISCONNECT_BAD_HANDSHAKE, + .postpone_reconnect_seconds = 2 * 60, // 2 minutes + }, + { + .response = START_STREAMING_ERROR_INTERNAL_ERROR, + .length = sizeof(START_STREAMING_ERROR_INTERNAL_ERROR) - 1, + .version = STREAM_HANDSHAKE_INTERNAL_ERROR, + .dynamic = false, + .error = "remote server is encountered an internal error, we should try later", + .worker_job_id = WORKER_SENDER_JOB_DISCONNECT_BAD_HANDSHAKE, + .postpone_reconnect_seconds = 5 * 60, // 5 minutes + }, + { + .response = START_STREAMING_ERROR_INITIALIZATION, + .length = sizeof(START_STREAMING_ERROR_INITIALIZATION) - 1, + .version = STREAM_HANDSHAKE_INITIALIZATION, + .dynamic = false, + .error = "remote server is initializing, we should try later", + .worker_job_id = WORKER_SENDER_JOB_DISCONNECT_BAD_HANDSHAKE, + .postpone_reconnect_seconds = 2 * 60, // 2 minute + }, // terminator { @@ -750,6 +783,10 @@ static bool attempt_to_connect(struct sender_state *state) { state->send_attempts = 0; + // reset the bytes we have sent for this session + state->sent_bytes_on_this_connection = 0; + memset(state->sent_bytes_on_this_connection_per_type, 0, sizeof(state->sent_bytes_on_this_connection_per_type)); + if(rrdpush_sender_thread_connect_to_parent(state->host, state->default_port, state->timeout, state)) { // reset the buffer, to properly send charts and metrics rrdpush_sender_on_connect(state->host); @@ -760,9 +797,6 @@ static bool attempt_to_connect(struct sender_state *state) // make sure the next reconnection will be immediate state->not_connected_loops = 0; - // reset the bytes we have sent for this session - state->sent_bytes_on_this_connection = 0; - // let the data collection threads know we are ready rrdhost_flag_set(state->host, RRDHOST_FLAG_RRDPUSH_SENDER_CONNECTED); @@ -776,9 +810,6 @@ static bool attempt_to_connect(struct sender_state *state) // increase the failed connections counter state->not_connected_loops++; - // reset the number of bytes sent - state->sent_bytes_on_this_connection = 0; - // slow re-connection on repeating errors usec_t now_ut = now_monotonic_usec(); usec_t end_ut = now_ut + USEC_PER_SEC * state->reconnect_delay; @@ -899,7 +930,7 @@ void stream_execute_function_callback(BUFFER *func_wb, int code, void *data) { buffer_fast_strcat(wb, buffer_tostring(func_wb), buffer_strlen(func_wb)); pluginsd_function_result_end_to_buffer(wb); - sender_commit(s, wb); + sender_commit(s, wb, STREAM_TRAFFIC_TYPE_FUNCTIONS); sender_thread_buffer_free(); internal_error(true, "STREAM %s [send to %s] FUNCTION transaction %s sending back response (%zu bytes, %llu usec).", @@ -1067,6 +1098,120 @@ void rrdpush_signal_sender_to_wake_up(struct sender_state *s) { } } +static NETDATA_DOUBLE rrdhost_sender_replication_completion(RRDHOST *host, time_t now, size_t *instances) { + size_t charts = rrdhost_sender_replicating_charts(host); + NETDATA_DOUBLE completion; + if(!charts || !host->sender->replication.oldest_request_after_t) + completion = 100.0; + else if(!host->sender->replication.latest_completed_before_t || host->sender->replication.latest_completed_before_t < host->sender->replication.oldest_request_after_t) + completion = 0.0; + else { + time_t total = now - host->sender->replication.oldest_request_after_t; + time_t current = host->sender->replication.latest_completed_before_t - host->sender->replication.oldest_request_after_t; + completion = (NETDATA_DOUBLE) current * 100.0 / (NETDATA_DOUBLE) total; + } + + *instances = charts; + + return completion; +} + +void rrdhost_sender_to_json(BUFFER *wb, RRDHOST *host, const char *key, time_t now __maybe_unused) { + bool online = rrdhost_flag_check(host, RRDHOST_FLAG_RRDPUSH_SENDER_CONNECTED); + buffer_json_member_add_object(wb, key); + + if(host->sender) + buffer_json_member_add_uint64(wb, "hops", host->sender->hops); + + buffer_json_member_add_boolean(wb, "online", online); + + if(host->sender && host->sender->last_state_since_t) { + buffer_json_member_add_time_t(wb, "since", host->sender->last_state_since_t); + buffer_json_member_add_time_t(wb, "age", now - host->sender->last_state_since_t); + } + + if(!online && host->sender && host->sender->exit.reason) + buffer_json_member_add_string(wb, "reason", host->sender->exit.reason); + + buffer_json_member_add_object(wb, "replication"); + { + size_t instances; + NETDATA_DOUBLE completion = rrdhost_sender_replication_completion(host, now, &instances); + buffer_json_member_add_boolean(wb, "in_progress", instances); + buffer_json_member_add_double(wb, "completion", completion); + buffer_json_member_add_uint64(wb, "instances", instances); + } + buffer_json_object_close(wb); + + if(host->sender) { + netdata_mutex_lock(&host->sender->mutex); + + buffer_json_member_add_object(wb, "destination"); + { + char buf[1024 + 1]; + if(online && host->sender->rrdpush_sender_socket != -1) { + SOCKET_PEERS peers = socket_peers(host->sender->rrdpush_sender_socket); + bool ssl = SSL_connection(&host->sender->ssl); + + snprintfz(buf, 1024, "[%s]:%d%s", peers.local.ip, peers.local.port, ssl ? ":SSL" : ""); + buffer_json_member_add_string(wb, "local", buf); + + snprintfz(buf, 1024, "[%s]:%d%s", peers.peer.ip, peers.peer.port, ssl ? ":SSL" : ""); + buffer_json_member_add_string(wb, "remote", buf); + + stream_capabilities_to_json_array(wb, online ? host->sender->capabilities : 0, + "capabilities"); + + buffer_json_member_add_object(wb, "traffic"); + { + bool compression = false; +#ifdef ENABLE_COMPRESSION + compression = (stream_has_capability(host->sender, STREAM_CAP_COMPRESSION) && host->sender->compressor); +#endif + buffer_json_member_add_boolean(wb, "compression", compression); + buffer_json_member_add_uint64(wb, "data", host->sender->sent_bytes_on_this_connection_per_type[STREAM_TRAFFIC_TYPE_DATA]); + buffer_json_member_add_uint64(wb, "metadata", host->sender->sent_bytes_on_this_connection_per_type[STREAM_TRAFFIC_TYPE_METADATA]); + buffer_json_member_add_uint64(wb, "functions", host->sender->sent_bytes_on_this_connection_per_type[STREAM_TRAFFIC_TYPE_FUNCTIONS]); + buffer_json_member_add_uint64(wb, "replication", host->sender->sent_bytes_on_this_connection_per_type[STREAM_TRAFFIC_TYPE_REPLICATION]); + } + buffer_json_object_close(wb); // traffic + } + + buffer_json_member_add_array(wb, "candidates"); + struct rrdpush_destinations *d; + for (d = host->destinations; d; d = d->next) { + buffer_json_add_array_item_object(wb); + { + + if (d->ssl) { + snprintfz(buf, 1024, "%s:SSL", string2str(d->destination)); + buffer_json_member_add_string(wb, "destination", buf); + } + else + buffer_json_member_add_string(wb, "destination", string2str(d->destination)); + + buffer_json_member_add_time_t(wb, "last_check", d->last_attempt); + buffer_json_member_add_time_t(wb, "age", now - d->last_attempt); + buffer_json_member_add_string(wb, "last_error", d->last_error); + buffer_json_member_add_string(wb, "last_handshake", + stream_handshake_error_to_string(d->last_handshake)); + buffer_json_member_add_time_t(wb, "next_check", d->postpone_reconnection_until); + buffer_json_member_add_time_t(wb, "next_in", + (d->postpone_reconnection_until > now) ? + d->postpone_reconnection_until - now : 0); + } + buffer_json_object_close(wb); // each candidate + } + buffer_json_array_close(wb); // candidates + } + buffer_json_object_close(wb); // destination + + netdata_mutex_unlock(&host->sender->mutex); + } + + buffer_json_object_close(wb); // streaming +} + static bool rrdhost_set_sender(RRDHOST *host) { if(unlikely(!host->sender)) return false; @@ -1076,6 +1221,8 @@ static bool rrdhost_set_sender(RRDHOST *host) { rrdhost_flag_clear(host, RRDHOST_FLAG_RRDPUSH_SENDER_CONNECTED | RRDHOST_FLAG_RRDPUSH_SENDER_READY_4_METRICS); rrdhost_flag_set(host, RRDHOST_FLAG_RRDPUSH_SENDER_SPAWN); host->sender->tid = gettid(); + host->sender->last_state_since_t = now_realtime_sec(); + host->sender->exit.reason = NULL; ret = true; } netdata_mutex_unlock(&host->sender->mutex); @@ -1091,8 +1238,8 @@ static void rrdhost_clear_sender___while_having_sender_mutex(RRDHOST *host) { if(host->sender->tid == gettid()) { host->sender->tid = 0; host->sender->exit.shutdown = false; - host->sender->exit.reason = NULL; rrdhost_flag_clear(host, RRDHOST_FLAG_RRDPUSH_SENDER_SPAWN | RRDHOST_FLAG_RRDPUSH_SENDER_CONNECTED | RRDHOST_FLAG_RRDPUSH_SENDER_READY_4_METRICS); + host->sender->last_state_since_t = now_realtime_sec(); } rrdpush_reset_destinations_postpone_time(host); @@ -1291,6 +1438,7 @@ void *rrdpush_sender_thread(void *ptr) { now_s = s->last_traffic_seen_t = now_monotonic_sec(); rrdpush_claimed_id(s->host); rrdpush_send_host_labels(s->host); + s->replication.oldest_request_after_t = 0; rrdhost_flag_set(s->host, RRDHOST_FLAG_RRDPUSH_SENDER_READY_4_METRICS); info("STREAM %s [send to %s]: enabling metrics streaming...", rrdhost_hostname(s->host), s->connected_to); |