@cryptotaxi247 / netdata
Code Commits Issues Pulls Sessions CI/CD
master
master
h 129 lines 4.64 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #ifndef NETDATA_ACLK_QUERY_QUEUE_H
4 #define NETDATA_ACLK_QUERY_QUEUE_H
5
6 #include "database/rrd.h"
7 #include "schema-wrappers/schema_wrappers.h"
8
9 #include "aclk_util.h"
10
11 typedef enum {
12 UNKNOWN = 0,
13 HTTP_API_V2,
14 REGISTER_NODE,
15 NODE_STATE_UPDATE,
16 UPDATE_NODE_INFO,
17 ALARM_PROVIDE_CFG,
18 ALARM_SNAPSHOT,
19 UPDATE_NODE_COLLECTORS,
20 CTX_SEND_SNAPSHOT, // Context snapshot to the cloud
21 CTX_SEND_SNAPSHOT_UPD, // Context incremental update to the cloud
22 CTX_CHECKPOINT, // Context checkpoint from the cloud
23 CTX_STOP_STREAMING, // Context stop streaming
24 CREATE_NODE_INSTANCE, // Create node instance on the agent
25 SEND_NODE_INSTANCES, // Send node instances to the cloud
26 ALERT_START_STREAMING, // Start alert streaming from cloud
27 ALERT_CHECKPOINT, // Do an alert version check
28 ACLK_QUERY_TYPE_COUNT // always keep this as last
29 } aclk_query_type_t;
30
31 struct aclk_query_http_api_v2 {
32 char *payload;
33 char *query;
34 };
35
36 struct aclk_bin_payload {
37 char *payload;
38 size_t size;
39 enum aclk_topics topic;
40 const char *msg_name;
41 };
42
43 // ----------------------------------------------------------------------------
44 // Reference-counted completion for safe timed waits
45 // Both waiter and query hold a reference; last one to release frees the structure
46
47 struct aclk_sync_completion {
48 struct completion compl;
49 int32_t refcount;
50 };
51
52 static inline struct aclk_sync_completion *aclk_sync_completion_create(void) {
53 struct aclk_sync_completion *sc = callocz(1, sizeof(*sc));
54 completion_init(&sc->compl);
55 sc->refcount = 2; // One for waiter, one for query
56 return sc;
57 }
58
59 static inline void aclk_sync_completion_release(struct aclk_sync_completion *sc) {
60 if (__atomic_sub_fetch(&sc->refcount, 1, __ATOMIC_ACQ_REL) == 0) {
61 completion_destroy(&sc->compl);
62 freez(sc);
63 }
64 }
65
66 // Called by query processing to signal completion and release query's reference
67 static inline void aclk_sync_completion_signal(struct aclk_sync_completion *sc) {
68 completion_mark_complete(&sc->compl);
69 aclk_sync_completion_release(sc);
70 }
71
72 // Called by waiter - waits with timeout, then releases waiter's reference
73 // Returns true if completed within timeout, false if timed out
74 static inline bool aclk_sync_completion_timedwait(struct aclk_sync_completion *sc, uint64_t timeout_s) {
75 bool result = completion_timedwait_for(&sc->compl, timeout_s);
76 aclk_sync_completion_release(sc);
77 return result;
78 }
79
80 // ----------------------------------------------------------------------------
81
82 typedef struct {
83 aclk_query_type_t type;
84 bool allocated;
85
86 // dedup_id is used to deduplicate queries in the list
87 // if type and dedup_id is the same message is deduplicated
88 // set dedup_id to NULL to never deduplicate the message
89 // set dedup_id to constant (e.g. empty string "") to make
90 // message of this type ever exist only once in the list
91 char *dedup_id;
92 char *callback_topic;
93 char *msg_id;
94 union {
95 char *claim_id;
96 char *machine_guid;
97 };
98
99 struct timeval created_tv;
100 usec_t created;
101 int timeout;
102
103 uint64_t version;
104 union {
105 struct aclk_query_http_api_v2 http_api_v2;
106 struct aclk_bin_payload bin_payload;
107 void *payload;
108 char *node_id;
109 } data;
110 struct aclk_sync_completion *sync_completion;
111 } aclk_query_t;
112
113 aclk_query_t *aclk_query_new(aclk_query_type_t type);
114 void aclk_query_free(aclk_query_t *query);
115
116 void aclk_execute_query(aclk_query_t *query);
117 void aclk_add_job(aclk_query_t *query);
118
119 #define QUEUE_IF_PAYLOAD_PRESENT(query) \
120 do { \
121 if (likely((query)->data.bin_payload.payload)) { \
122 aclk_execute_query(query); \
123 } else { \
124 nd_log(NDLS_DAEMON, NDLP_ERR, "Failed to generate payload"); \
125 aclk_query_free(query); \
126 } \
127 } while (0)
128
129 #endif /* NETDATA_ACLK_QUERY_QUEUE_H */