/src/fluent-bit/plugins/in_splunk/splunk.c
Line | Count | Source |
1 | | /* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ |
2 | | |
3 | | /* Fluent Bit |
4 | | * ========== |
5 | | * Copyright (C) 2015-2026 The Fluent Bit Authors |
6 | | * |
7 | | * Licensed under the Apache License, Version 2.0 (the "License"); |
8 | | * you may not use this file except in compliance with the License. |
9 | | * You may obtain a copy of the License at |
10 | | * |
11 | | * http://www.apache.org/licenses/LICENSE-2.0 |
12 | | * |
13 | | * Unless required by applicable law or agreed to in writing, software |
14 | | * distributed under the License is distributed on an "AS IS" BASIS, |
15 | | * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
16 | | * See the License for the specific language governing permissions and |
17 | | * limitations under the License. |
18 | | */ |
19 | | |
20 | | |
21 | | #include <fluent-bit/flb_input_plugin.h> |
22 | | #include <fluent-bit/flb_network.h> |
23 | | #include <fluent-bit/flb_config.h> |
24 | | |
25 | | #include "splunk.h" |
26 | | #include "splunk_conn.h" |
27 | | #include "splunk_prot.h" |
28 | | #include "splunk_config.h" |
29 | | |
30 | | /* |
31 | | * For a server event, the collection event means a new client have arrived, we |
32 | | * accept the connection and create a new TCP instance which will wait for |
33 | | * JSON map messages. |
34 | | */ |
35 | | static int in_splunk_collect(struct flb_input_instance *ins, |
36 | | struct flb_config *config, void *in_context) |
37 | 0 | { |
38 | 0 | struct flb_connection *connection; |
39 | 0 | struct splunk_conn *conn; |
40 | 0 | struct flb_splunk *ctx; |
41 | |
|
42 | 0 | ctx = in_context; |
43 | |
|
44 | 0 | connection = flb_downstream_conn_get(ctx->downstream); |
45 | |
|
46 | 0 | if (connection == NULL) { |
47 | 0 | flb_plg_error(ctx->ins, "could not accept new connection"); |
48 | |
|
49 | 0 | return -1; |
50 | 0 | } |
51 | | |
52 | 0 | flb_plg_trace(ctx->ins, "new TCP connection arrived FD=%i", |
53 | 0 | connection->fd); |
54 | |
|
55 | 0 | conn = splunk_conn_add(connection, ctx); |
56 | |
|
57 | 0 | if (conn == NULL) { |
58 | 0 | flb_downstream_conn_release(connection); |
59 | |
|
60 | 0 | return -1; |
61 | 0 | } |
62 | | |
63 | 0 | return 0; |
64 | 0 | } |
65 | | |
66 | | static int in_splunk_init(struct flb_input_instance *ins, |
67 | | struct flb_config *config, void *data) |
68 | 0 | { |
69 | 0 | unsigned short int port; |
70 | 0 | int ret; |
71 | 0 | struct flb_splunk *ctx; |
72 | |
|
73 | 0 | (void) data; |
74 | | |
75 | | /* Create context and basic conf */ |
76 | 0 | ctx = splunk_config_create(ins); |
77 | 0 | if (!ctx) { |
78 | 0 | return -1; |
79 | 0 | } |
80 | | |
81 | 0 | ctx->collector_id = -1; |
82 | | |
83 | | /* Populate context with config map defaults and incoming properties */ |
84 | 0 | ret = flb_input_config_map_set(ins, (void *) ctx); |
85 | 0 | if (ret == -1) { |
86 | 0 | flb_plg_error(ctx->ins, "configuration error"); |
87 | 0 | splunk_config_destroy(ctx); |
88 | 0 | return -1; |
89 | 0 | } |
90 | | |
91 | | /* Set the context */ |
92 | 0 | flb_input_set_context(ins, ctx); |
93 | |
|
94 | 0 | port = (unsigned short int) strtoul(ctx->tcp_port, NULL, 10); |
95 | | |
96 | |
|
97 | 0 | if (ctx->enable_http2) { |
98 | 0 | ret = flb_http_server_init(&ctx->http_server, |
99 | 0 | HTTP_PROTOCOL_VERSION_AUTODETECT, |
100 | 0 | (FLB_HTTP_SERVER_FLAG_KEEPALIVE | FLB_HTTP_SERVER_FLAG_AUTO_INFLATE), |
101 | 0 | NULL, |
102 | 0 | ins->host.listen, |
103 | 0 | ins->host.port, |
104 | 0 | ins->tls, |
105 | 0 | ins->flags, |
106 | 0 | &ins->net_setup, |
107 | 0 | flb_input_event_loop_get(ins), |
108 | 0 | ins->config, |
109 | 0 | (void *) ctx); |
110 | |
|
111 | 0 | if (ret != 0) { |
112 | 0 | flb_plg_error(ctx->ins, |
113 | 0 | "could not initialize http server on %s:%u. Aborting", |
114 | 0 | ins->host.listen, ins->host.port); |
115 | |
|
116 | 0 | splunk_config_destroy(ctx); |
117 | |
|
118 | 0 | return -1; |
119 | 0 | } |
120 | | |
121 | 0 | ret = flb_http_server_start(&ctx->http_server); |
122 | |
|
123 | 0 | if (ret != 0) { |
124 | 0 | flb_plg_error(ctx->ins, |
125 | 0 | "could not start http server on %s:%u. Aborting", |
126 | 0 | ins->host.listen, ins->host.port); |
127 | |
|
128 | 0 | splunk_config_destroy(ctx); |
129 | |
|
130 | 0 | return -1; |
131 | 0 | } |
132 | | |
133 | 0 | flb_http_server_set_buffer_max_size(&ctx->http_server, ctx->buffer_max_size); |
134 | |
|
135 | 0 | ctx->http_server.request_callback = splunk_prot_handle_ng; |
136 | |
|
137 | 0 | flb_input_downstream_set(ctx->http_server.downstream, ctx->ins); |
138 | |
|
139 | 0 | flb_plg_info(ctx->ins, "listening on %s:%u", |
140 | 0 | ins->host.listen, ins->host.port); |
141 | 0 | } |
142 | 0 | else { |
143 | 0 | ctx->downstream = flb_downstream_create(FLB_TRANSPORT_TCP, |
144 | 0 | ins->flags, |
145 | 0 | ctx->listen, |
146 | 0 | port, |
147 | 0 | ins->tls, |
148 | 0 | config, |
149 | 0 | &ins->net_setup); |
150 | |
|
151 | 0 | if (ctx->downstream == NULL) { |
152 | 0 | flb_plg_error(ctx->ins, |
153 | 0 | "could not initialize downstream on %s:%s. Aborting", |
154 | 0 | ctx->listen, ctx->tcp_port); |
155 | |
|
156 | 0 | splunk_config_destroy(ctx); |
157 | |
|
158 | 0 | return -1; |
159 | 0 | } |
160 | | |
161 | 0 | flb_input_downstream_set(ctx->downstream, ctx->ins); |
162 | |
|
163 | 0 | flb_plg_info(ctx->ins, "listening on %s:%s", ctx->listen, ctx->tcp_port); |
164 | | |
165 | | /* Collect upon data available on the standard input */ |
166 | 0 | ret = flb_input_set_collector_socket(ins, |
167 | 0 | in_splunk_collect, |
168 | 0 | ctx->downstream->server_fd, |
169 | 0 | config); |
170 | 0 | if (ret == -1) { |
171 | 0 | flb_plg_error(ctx->ins, "Could not set collector for IN_TCP input plugin"); |
172 | 0 | splunk_config_destroy(ctx); |
173 | |
|
174 | 0 | return -1; |
175 | 0 | } |
176 | | |
177 | 0 | ctx->collector_id = ret; |
178 | 0 | } |
179 | | |
180 | | |
181 | 0 | return 0; |
182 | 0 | } |
183 | | |
184 | | static int in_splunk_exit(void *data, struct flb_config *config) |
185 | 0 | { |
186 | 0 | struct flb_splunk *ctx; |
187 | |
|
188 | 0 | (void) config; |
189 | |
|
190 | 0 | ctx = data; |
191 | |
|
192 | 0 | if (ctx != NULL) { |
193 | 0 | splunk_config_destroy(ctx); |
194 | 0 | } |
195 | |
|
196 | 0 | return 0; |
197 | 0 | } |
198 | | |
199 | | |
200 | | static void in_splunk_pause(void *data, struct flb_config *config) |
201 | 0 | { |
202 | 0 | struct flb_splunk *ctx = data; |
203 | |
|
204 | 0 | flb_input_collector_pause(ctx->collector_id, ctx->ins); |
205 | |
|
206 | 0 | } |
207 | | |
208 | | static void in_splunk_resume(void *data, struct flb_config *config) |
209 | 0 | { |
210 | 0 | struct flb_splunk *ctx = data; |
211 | |
|
212 | 0 | flb_input_collector_resume(ctx->collector_id, ctx->ins); |
213 | 0 | } |
214 | | |
215 | | /* Configuration properties map */ |
216 | | static struct flb_config_map config_map[] = { |
217 | | { |
218 | | FLB_CONFIG_MAP_BOOL, "http2", "true", |
219 | | 0, FLB_TRUE, offsetof(struct flb_splunk, enable_http2), |
220 | | "Enable HTTP/2 support" |
221 | | }, |
222 | | |
223 | | { |
224 | | FLB_CONFIG_MAP_SIZE, "buffer_max_size", HTTP_BUFFER_MAX_SIZE, |
225 | | 0, FLB_TRUE, offsetof(struct flb_splunk, buffer_max_size), |
226 | | "Set the maximum size of buffer" |
227 | | }, |
228 | | |
229 | | { |
230 | | FLB_CONFIG_MAP_SIZE, "buffer_chunk_size", HTTP_BUFFER_CHUNK_SIZE, |
231 | | 0, FLB_TRUE, offsetof(struct flb_splunk, buffer_chunk_size), |
232 | | "Set the initial buffer size to store incoming data" |
233 | | }, |
234 | | |
235 | | { |
236 | | FLB_CONFIG_MAP_SLIST_1, "success_header", NULL, |
237 | | FLB_CONFIG_MAP_MULT, FLB_TRUE, offsetof(struct flb_splunk, success_headers), |
238 | | "Add an HTTP header key/value pair on success. Multiple headers can be set" |
239 | | }, |
240 | | |
241 | | { |
242 | | FLB_CONFIG_MAP_STR, "splunk_token", NULL, |
243 | | 0, FLB_FALSE, 0, |
244 | | "Set valid Splunk HEC tokens for the requests" |
245 | | }, |
246 | | |
247 | | { |
248 | | FLB_CONFIG_MAP_BOOL, "store_token_in_metadata", "true", |
249 | | 0, FLB_TRUE, offsetof(struct flb_splunk, store_token_in_metadata), |
250 | | "Store Splunk HEC tokens in metadata. If set as false, they will be stored into records." |
251 | | }, |
252 | | |
253 | | { |
254 | | FLB_CONFIG_MAP_STR, "splunk_token_key", "@splunk_token", |
255 | | 0, FLB_TRUE, offsetof(struct flb_splunk, store_token_key), |
256 | | "Set a record key for storing Splunk HEC token for the request" |
257 | | }, |
258 | | |
259 | | { |
260 | | FLB_CONFIG_MAP_STR, "tag_key", NULL, |
261 | | 0, FLB_TRUE, offsetof(struct flb_splunk, tag_key), |
262 | | "Set a record key to specify the tag of the record" |
263 | | }, |
264 | | { |
265 | | FLB_CONFIG_MAP_BOOL, "add_remote_addr", "false", |
266 | | 0, FLB_TRUE, offsetof(struct flb_splunk, add_remote_addr), |
267 | | "Inject a remote address using the X-Forwarded-For header or connection address" |
268 | | }, |
269 | | { |
270 | | FLB_CONFIG_MAP_STR, "remote_addr_key", "remote_addr", |
271 | | 0, FLB_TRUE, offsetof(struct flb_splunk, remote_addr_key), |
272 | | "Set a record key for storing the remote address" |
273 | | }, |
274 | | |
275 | | |
276 | | /* EOF */ |
277 | | {0} |
278 | | }; |
279 | | |
280 | | /* Plugin reference */ |
281 | | struct flb_input_plugin in_splunk_plugin = { |
282 | | .name = "splunk", |
283 | | .description = "Input plugin for Splunk HEC payloads", |
284 | | .cb_init = in_splunk_init, |
285 | | .cb_pre_run = NULL, |
286 | | .cb_collect = in_splunk_collect, |
287 | | .cb_flush_buf = NULL, |
288 | | .cb_pause = in_splunk_pause, |
289 | | .cb_resume = in_splunk_resume, |
290 | | .cb_exit = in_splunk_exit, |
291 | | .config_map = config_map, |
292 | | .flags = FLB_INPUT_NET_SERVER | FLB_IO_OPT_TLS |
293 | | }; |