| 1 | // SPDX-License-Identifier: GPL-3.0-or-later |
| 2 | |
| 3 | #ifndef NETDATA_STREAM_RECEIVER_INTERNALS_H |
| 4 | #define NETDATA_STREAM_RECEIVER_INTERNALS_H |
| 5 | |
| 6 | #include "libnetdata/libnetdata.h" |
| 7 | |
| 8 | #ifdef NETDATA_LOG_STREAM_RECEIVER |
| 9 | #include "stream-traffic-types.h" |
| 10 | struct receiver_state; |
| 11 | void stream_receiver_log_payload(struct receiver_state *rpt, const char *payload, STREAM_TRAFFIC_TYPE type, bool inbound); |
| 12 | #else |
| 13 | #define stream_receiver_log_payload(s, payload, type, inbound) debug_dummy() |
| 14 | #endif |
| 15 | |
| 16 | #include "stream.h" |
| 17 | #include "stream-thread.h" |
| 18 | #include "stream-conf.h" |
| 19 | #include "database/rrd.h" |
| 20 | #include "plugins.d/plugins_d.h" |
| 21 | |
| 22 | struct parser; |
| 23 | |
| 24 | struct receiver_state { |
| 25 | RRDHOST *host; |
| 26 | ND_SOCK sock; |
| 27 | int16_t hops; |
| 28 | int32_t utc_offset; |
| 29 | STREAM_CAPABILITIES capabilities; |
| 30 | char *key; |
| 31 | char *hostname; |
| 32 | char *registry_hostname; |
| 33 | char *machine_guid; |
| 34 | char *os; |
| 35 | char *timezone; // Unused? |
| 36 | char *abbrev_timezone; |
| 37 | char *remote_ip; // Duplicated in pluginsd |
| 38 | char *remote_port; // Duplicated in pluginsd |
| 39 | char *program_name; // Duplicated in pluginsd |
| 40 | char *program_version; |
| 41 | struct rrdhost_system_info *system_info; |
| 42 | time_t connected_since_s; |
| 43 | |
| 44 | struct { |
| 45 | // The parser pointer is safe to read and use, only when having the host receiver lock. |
| 46 | // Without this lock, the data pointed by the pointer may vanish randomly. |
| 47 | // Also, since the receiver sets it when it starts, it should be read with |
| 48 | // an atomic read. |
| 49 | struct parser *parser; |
| 50 | struct plugind cd; |
| 51 | |
| 52 | // compressed data input |
| 53 | struct { |
| 54 | bool enabled; |
| 55 | size_t start; |
| 56 | size_t used; |
| 57 | size_t size; |
| 58 | char *buf; |
| 59 | struct decompressor_state decompressor; |
| 60 | } compressed; |
| 61 | |
| 62 | // uncompressed data input (either directly or via the decompressor) |
| 63 | struct buffered_reader uncompressed; |
| 64 | |
| 65 | // a single line of input (composed via uncompressed buffer input) |
| 66 | BUFFER *line_buffer; |
| 67 | |
| 68 | struct { |
| 69 | SPINLOCK spinlock; |
| 70 | struct stream_opcode msg; |
| 71 | uint32_t msg_slot; |
| 72 | STREAM_CIRCULAR_BUFFER *scb; |
| 73 | } send_to_child; |
| 74 | |
| 75 | nd_poll_event_t wanted; |
| 76 | usec_t last_traffic_ut; |
| 77 | struct pollfd_meta meta; |
| 78 | } thread; |
| 79 | |
| 80 | struct { |
| 81 | uint32_t last_counter_sum; // copy from the host, to detect progress |
| 82 | usec_t last_progress_ut; // last time we found some progress (monotonic) |
| 83 | usec_t last_checked_ut; // last time we checked for stalled progress (monotonic) |
| 84 | |
| 85 | time_t first_time_s; |
| 86 | } replication; |
| 87 | |
| 88 | struct { |
| 89 | bool shutdown; // signal the streaming parser to exit |
| 90 | STREAM_HANDSHAKE reason; |
| 91 | } exit; |
| 92 | |
| 93 | struct stream_receiver_config config; |
| 94 | |
| 95 | #ifdef NETDATA_LOG_STREAM_RECEIVER |
| 96 | struct { |
| 97 | struct timespec first_call; |
| 98 | SPINLOCK spinlock; |
| 99 | FILE *fp; |
| 100 | } log; |
| 101 | #endif |
| 102 | }; |
| 103 | |
| 104 | typedef enum { |
| 105 | RRDHOST_SET_RECEIVER_OK, // attached |
| 106 | RRDHOST_SET_RECEIVER_ALREADY_ATTACHED, // another receiver already attached |
| 107 | RRDHOST_SET_RECEIVER_CLEANUP_BUSY, // obsolete-all cleanup is running; caller should answer BUSY_TRY_LATER |
| 108 | } RRDHOST_SET_RECEIVER_RESULT; |
| 109 | |
| 110 | RRDHOST_SET_RECEIVER_RESULT rrdhost_set_receiver(RRDHOST *host, struct receiver_state *rpt); |
| 111 | void rrdhost_clear_receiver(struct receiver_state *rpt, STREAM_HANDSHAKE reason); |
| 112 | void stream_receiver_log_status(struct receiver_state *rpt, const char *msg, STREAM_HANDSHAKE reason, ND_LOG_FIELD_PRIORITY priority); |
| 113 | |
| 114 | void stream_receiver_free(struct receiver_state *rpt); |
| 115 | bool stream_receiver_signal_to_stop_and_wait(RRDHOST *host, STREAM_HANDSHAKE reason); |
| 116 | |
| 117 | void stream_receiver_send_opcode(struct receiver_state *rpt, struct stream_opcode msg); |
| 118 | void stream_receiver_handle_op(struct stream_thread *sth, struct receiver_state *rpt, struct stream_opcode *msg); |
| 119 | |
| 120 | void stream_receiver_check_all_nodes_from_poll(struct stream_thread *sth, usec_t now_ut); |
| 121 | void stream_receiver_replication_check_from_poll(struct stream_thread *sth, usec_t now_ut); |
| 122 | |
| 123 | #endif //NETDATA_STREAM_RECEIVER_INTERNALS_H |