@cryptotaxi247 / netdata
Code Commits Issues Pulls Sessions CI/CD
master
master
c 233 lines 8.56 KB
Raw
1 // SPDX-License-Identifier: GPL-3.0-or-later
2
3 #include "exporting_engine.h"
4 #include "graphite/graphite.h"
5 #include "json/json.h"
6 #include "opentsdb/opentsdb.h"
7
8 #ifdef ENABLE_PROMETHEUS_REMOTE_WRITE
9 #include "prometheus/remote_write/remote_write.h"
10 #endif
11
12 #if HAVE_KINESIS
13 #include "aws_kinesis/aws_kinesis.h"
14 #endif
15
16 #ifdef ENABLE_EXPORTING_PUBSUB
17 #include "pubsub/pubsub.h"
18 #endif
19
20 #ifdef HAVE_MONGOC
21 #include "mongodb/mongodb.h"
22 #endif
23
24 /**
25 * Initialize connectors
26 *
27 * @param engine an engine data structure.
28 * @return Returns 0 on success, 1 on failure.
29 */
30 int init_connectors(struct engine *engine)
31 {
32 engine->now = now_realtime_sec();
33
34 for (struct instance *instance = engine->instance_root; instance; instance = instance->next) {
35 instance->index = engine->instance_num++;
36 instance->after = engine->now;
37
38 // Set thread tag and initialize connector based on type
39 switch (instance->config.type) {
40 case EXPORTING_CONNECTOR_TYPE_GRAPHITE:
41 instance->config.thread_tag = "EXPGRPH";
42 if (init_graphite_instance(instance) != 0)
43 return 1;
44 break;
45 case EXPORTING_CONNECTOR_TYPE_GRAPHITE_HTTP:
46 instance->config.thread_tag = "EXPGRPH";
47 if (init_graphite_instance(instance) != 0)
48 return 1;
49 break;
50 case EXPORTING_CONNECTOR_TYPE_JSON:
51 instance->config.thread_tag = "EXPJSON";
52 if (init_json_instance(instance) != 0)
53 return 1;
54 break;
55 case EXPORTING_CONNECTOR_TYPE_JSON_HTTP:
56 instance->config.thread_tag = "EXPJSON";
57 if (init_json_http_instance(instance) != 0)
58 return 1;
59 break;
60 case EXPORTING_CONNECTOR_TYPE_OPENTSDB:
61 instance->config.thread_tag = "EXPTSDB";
62 if (init_opentsdb_telnet_instance(instance) != 0)
63 return 1;
64 break;
65 case EXPORTING_CONNECTOR_TYPE_OPENTSDB_HTTP:
66 instance->config.thread_tag = "EXPTSDB";
67 if (init_opentsdb_http_instance(instance) != 0)
68 return 1;
69 break;
70 case EXPORTING_CONNECTOR_TYPE_PROMETHEUS_REMOTE_WRITE:
71 instance->config.thread_tag = "EXPPRW";
72 #ifdef ENABLE_PROMETHEUS_REMOTE_WRITE
73 if (init_prometheus_remote_write_instance(instance) != 0)
74 return 1;
75 #endif
76 break;
77 case EXPORTING_CONNECTOR_TYPE_KINESIS:
78 instance->config.thread_tag = "EXPKINS";
79 #if HAVE_KINESIS
80 if (init_aws_kinesis_instance(instance) != 0)
81 return 1;
82 #endif
83 break;
84 case EXPORTING_CONNECTOR_TYPE_PUBSUB:
85 instance->config.thread_tag = "EXPPUBS";
86 #if ENABLE_EXPORTING_PUBSUB
87 if (init_pubsub_instance(instance) != 0)
88 return 1;
89 #endif
90 break;
91 case EXPORTING_CONNECTOR_TYPE_MONGODB:
92 instance->config.thread_tag = "EXPMNG";
93 #ifdef HAVE_MONGOC
94 if (init_mongodb_instance(instance) != 0)
95 return 1;
96 #endif
97 break;
98 default:
99 instance->config.thread_tag = "EXPCON";
100 netdata_log_error("EXPORTING: unknown exporting connector type");
101 return 1;
102 }
103
104 // dispatch the instance worker thread
105 char threadname[ND_THREAD_TAG_MAX + 1];
106 snprintfz(threadname, ND_THREAD_TAG_MAX, "%s[%zu]", instance->config.thread_tag, instance->index);
107
108 instance->thread = nd_thread_create(threadname, NETDATA_THREAD_OPTION_DEFAULT, instance->worker, instance);
109 if (!instance->thread) {
110 netdata_log_error("EXPORTING: cannot create thread worker for instance %s", instance->config.name);
111 instance->exited = 1;
112 return 1;
113 }
114 }
115
116 return 0;
117 }
118
119 // TODO: use a base64 encoder from a library
120 static size_t base64_encode(unsigned char *input, size_t input_size, char *output, size_t output_size)
121 {
122 uint32_t value;
123 static char lookup[] = "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
124 "abcdefghijklmnopqrstuvwxyz"
125 "0123456789+/";
126 if ((input_size / 3 + 1) * 4 >= output_size) {
127 netdata_log_error("Output buffer for encoding size=%zu is not large enough for %zu-bytes input", output_size, input_size);
128 return 0;
129 }
130 size_t count = 0;
131 while (input_size >= 3) {
132 value = ((input[0] << 16) + (input[1] << 8) + input[2]) & 0xffffff;
133 output[0] = lookup[value >> 18];
134 output[1] = lookup[(value >> 12) & 0x3f];
135 output[2] = lookup[(value >> 6) & 0x3f];
136 output[3] = lookup[value & 0x3f];
137 //netdata_log_error("Base-64 encode (%04x) -> %c %c %c %c\n", value, output[0], output[1], output[2], output[3]);
138 output += 4;
139 input += 3;
140 input_size -= 3;
141 count += 4;
142 }
143 switch (input_size) {
144 case 2:
145 value = (input[0] << 10) + (input[1] << 2);
146 output[0] = lookup[(value >> 12) & 0x3f];
147 output[1] = lookup[(value >> 6) & 0x3f];
148 output[2] = lookup[value & 0x3f];
149 output[3] = '=';
150 //netdata_log_error("Base-64 encode (%06x) -> %c %c %c %c\n", (value>>2)&0xffff, output[0], output[1], output[2], output[3]);
151 count += 4;
152 output[4] = '\0';
153 break;
154 case 1:
155 value = input[0] << 4;
156 output[0] = lookup[(value >> 6) & 0x3f];
157 output[1] = lookup[value & 0x3f];
158 output[2] = '=';
159 output[3] = '=';
160 //netdata_log_error("Base-64 encode (%06x) -> %c %c %c %c\n", value, output[0], output[1], output[2], output[3]);
161 count += 4;
162 output[4] = '\0';
163 break;
164 case 0:
165 output[0] = '\0';
166 break;
167 }
168
169 return count;
170 }
171
172 /**
173 * Initialize a ring buffer and credentials for a simple connector
174 *
175 * @param instance an instance data structure.
176 */
177 void simple_connector_init(struct instance *instance)
178 {
179 struct simple_connector_data *connector_specific_data =
180 (struct simple_connector_data *)instance->connector_specific_data;
181
182 if (connector_specific_data->first_buffer)
183 return;
184
185 // Initialize the active buffers that will be used for sending data
186 connector_specific_data->header = buffer_create(0, &netdata_buffers_statistics.buffers_exporters);
187 connector_specific_data->buffer = buffer_create(0, &netdata_buffers_statistics.buffers_exporters);
188
189 // create a ring buffer with all buffers initialized
190 struct simple_connector_buffer *first_buffer = NULL;
191
192 if (instance->config.buffer_on_failures < 1)
193 instance->config.buffer_on_failures = 1;
194
195 for (int i = 0; i < instance->config.buffer_on_failures; i++) {
196 struct simple_connector_buffer *current_buffer = callocz(1, sizeof(struct simple_connector_buffer));
197
198 // Initialize both header and buffer for each ring buffer entry
199 // This ensures we never have NULL pointers during buffer swapping
200 current_buffer->header = buffer_create(0, &netdata_buffers_statistics.buffers_exporters);
201 current_buffer->buffer = buffer_create(0, &netdata_buffers_statistics.buffers_exporters);
202
203 if (!connector_specific_data->first_buffer)
204 first_buffer = current_buffer;
205 else
206 current_buffer->next = connector_specific_data->first_buffer;
207
208 connector_specific_data->first_buffer = current_buffer;
209 }
210
211 first_buffer->next = connector_specific_data->first_buffer;
212 connector_specific_data->last_buffer = connector_specific_data->first_buffer;
213
214 if (*instance->config.username || *instance->config.password) {
215 BUFFER *auth_string = buffer_create(0, &netdata_buffers_statistics.buffers_exporters);
216
217 buffer_sprintf(auth_string, "%s:%s", instance->config.username, instance->config.password);
218
219 size_t encoded_size = (buffer_strlen(auth_string) / 3 + 1) * 4 + 1;
220 char *encoded_credentials = callocz(1, encoded_size);
221
222 base64_encode((unsigned char*)buffer_tostring(auth_string), buffer_strlen(auth_string), encoded_credentials, encoded_size);
223
224 buffer_flush(auth_string);
225 buffer_sprintf(auth_string, "Authorization: Basic %s\n", encoded_credentials);
226
227 freez(encoded_credentials);
228
229 connector_specific_data->auth_string = strdupz(buffer_tostring(auth_string));
230
231 buffer_free(auth_string);
232 }
233 }