/src/fluent-bit/plugins/in_opentelemetry/opentelemetry.c
Line | Count | Source (jump to first uncovered line) |
1 | | /* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */ |
2 | | |
3 | | /* Fluent Bit |
4 | | * ========== |
5 | | * Copyright (C) 2015-2022 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_downstream.h> |
23 | | #include <fluent-bit/flb_network.h> |
24 | | #include <fluent-bit/flb_config.h> |
25 | | |
26 | | #include "http_conn.h" |
27 | | #include "opentelemetry.h" |
28 | | #include "opentelemetry_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_opentelemetry_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 http_conn *conn; |
40 | 0 | struct flb_opentelemetry *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", connection->fd); |
53 | |
|
54 | 0 | conn = opentelemetry_conn_add(connection, ctx); |
55 | |
|
56 | 0 | if (conn == NULL) { |
57 | 0 | return -1; |
58 | 0 | } |
59 | | |
60 | 0 | return 0; |
61 | 0 | } |
62 | | |
63 | | static int in_opentelemetry_init(struct flb_input_instance *ins, |
64 | | struct flb_config *config, void *data) |
65 | 0 | { |
66 | 0 | unsigned short int port; |
67 | 0 | int ret; |
68 | 0 | struct flb_opentelemetry *ctx; |
69 | |
|
70 | 0 | (void) data; |
71 | | |
72 | | /* Create context and basic conf */ |
73 | 0 | ctx = opentelemetry_config_create(ins); |
74 | 0 | if (!ctx) { |
75 | 0 | return -1; |
76 | 0 | } |
77 | 0 | ctx->collector_id = -1; |
78 | | |
79 | | /* Populate context with config map defaults and incoming properties */ |
80 | 0 | ret = flb_input_config_map_set(ins, (void *) ctx); |
81 | 0 | if (ret == -1) { |
82 | 0 | flb_plg_error(ctx->ins, "configuration error"); |
83 | 0 | opentelemetry_config_destroy(ctx); |
84 | 0 | return -1; |
85 | 0 | } |
86 | | |
87 | | /* Set the context */ |
88 | 0 | flb_input_set_context(ins, ctx); |
89 | |
|
90 | 0 | port = (unsigned short int) strtoul(ctx->tcp_port, NULL, 10); |
91 | |
|
92 | 0 | ctx->downstream = flb_downstream_create(FLB_TRANSPORT_TCP, |
93 | 0 | ins->flags, |
94 | 0 | ctx->listen, |
95 | 0 | port, |
96 | 0 | ins->tls, |
97 | 0 | config, |
98 | 0 | &ins->net_setup); |
99 | |
|
100 | 0 | if (ctx->downstream == NULL) { |
101 | 0 | flb_plg_error(ctx->ins, |
102 | 0 | "could not initialize downstream on %s:%s. Aborting", |
103 | 0 | ctx->listen, ctx->tcp_port); |
104 | |
|
105 | 0 | opentelemetry_config_destroy(ctx); |
106 | |
|
107 | 0 | return -1; |
108 | 0 | } |
109 | | |
110 | 0 | flb_plg_info(ctx->ins, "listening on %s:%s", ctx->listen, ctx->tcp_port); |
111 | |
|
112 | 0 | ctx->evl = config->evl; |
113 | |
|
114 | 0 | if (ctx->successful_response_code != 200 && |
115 | 0 | ctx->successful_response_code != 201 && |
116 | 0 | ctx->successful_response_code != 204) { |
117 | 0 | flb_plg_error(ctx->ins, "%d is not supported response code. Use default 201", |
118 | 0 | ctx->successful_response_code); |
119 | 0 | ctx->successful_response_code = 201; |
120 | 0 | } |
121 | | |
122 | | /* Collect upon data available on the standard input */ |
123 | 0 | ret = flb_input_set_collector_socket(ins, |
124 | 0 | in_opentelemetry_collect, |
125 | 0 | ctx->downstream->server_fd, |
126 | 0 | config); |
127 | 0 | if (ret == -1) { |
128 | 0 | flb_plg_error(ctx->ins, "Could not set collector for IN_TCP input plugin"); |
129 | 0 | opentelemetry_config_destroy(ctx); |
130 | 0 | return -1; |
131 | 0 | } |
132 | | |
133 | 0 | ctx->collector_id = ret; |
134 | |
|
135 | 0 | return 0; |
136 | 0 | } |
137 | | |
138 | | static int in_opentelemetry_exit(void *data, struct flb_config *config) |
139 | 0 | { |
140 | 0 | struct flb_opentelemetry *ctx; |
141 | |
|
142 | 0 | (void) config; |
143 | |
|
144 | 0 | ctx = data; |
145 | |
|
146 | 0 | if (ctx != NULL) { |
147 | 0 | opentelemetry_config_destroy(ctx); |
148 | 0 | } |
149 | |
|
150 | 0 | return 0; |
151 | 0 | } |
152 | | |
153 | | /* Configuration properties map */ |
154 | | static struct flb_config_map config_map[] = { |
155 | | { |
156 | | FLB_CONFIG_MAP_SIZE, "buffer_max_size", HTTP_BUFFER_MAX_SIZE, |
157 | | 0, FLB_TRUE, offsetof(struct flb_opentelemetry, buffer_max_size), |
158 | | "" |
159 | | }, |
160 | | |
161 | | { |
162 | | FLB_CONFIG_MAP_SIZE, "buffer_chunk_size", HTTP_BUFFER_CHUNK_SIZE, |
163 | | 0, FLB_TRUE, offsetof(struct flb_opentelemetry, buffer_chunk_size), |
164 | | "" |
165 | | }, |
166 | | |
167 | | { |
168 | | FLB_CONFIG_MAP_STR, "tag_key", NULL, |
169 | | 0, FLB_TRUE, offsetof(struct flb_opentelemetry, tag_key), |
170 | | "" |
171 | | }, |
172 | | { |
173 | | FLB_CONFIG_MAP_INT, "successful_response_code", "201", |
174 | | 0, FLB_TRUE, offsetof(struct flb_opentelemetry, successful_response_code), |
175 | | "Set successful response code. 200, 201 and 204 are supported." |
176 | | }, |
177 | | { |
178 | | FLB_CONFIG_MAP_BOOL, "raw_traces", "false", |
179 | | 0, FLB_TRUE, offsetof(struct flb_opentelemetry, raw_traces), |
180 | | "Forward traces without processing" |
181 | | }, |
182 | | |
183 | | /* EOF */ |
184 | | {0} |
185 | | }; |
186 | | |
187 | | /* Plugin reference */ |
188 | | struct flb_input_plugin in_opentelemetry_plugin = { |
189 | | .name = "opentelemetry", |
190 | | .description = "OpenTelemetry", |
191 | | .cb_init = in_opentelemetry_init, |
192 | | .cb_pre_run = NULL, |
193 | | .cb_collect = in_opentelemetry_collect, |
194 | | .cb_flush_buf = NULL, |
195 | | .cb_pause = NULL, |
196 | | .cb_resume = NULL, |
197 | | .cb_exit = in_opentelemetry_exit, |
198 | | .config_map = config_map, |
199 | | .flags = FLB_INPUT_NET_SERVER | FLB_IO_OPT_TLS |
200 | | }; |