/src/fluent-bit/plugins/in_http/http_config.c

Source
/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */

/*  Fluent Bit
 *  ==========
 *  Copyright (C) 2015-2026 The Fluent Bit Authors
 *
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

#include <fluent-bit/flb_input_plugin.h>
#include <fluent-bit/flb_oauth2_jwt.h>

#include "http.h"
#include "http_config.h"
#include "http_conn.h"
#include "http_config.h"

struct flb_http *http_config_create(struct flb_input_instance *ins)
{
    struct mk_list            *header_iterator;
    struct flb_slist_entry    *header_value;
    struct flb_slist_entry    *header_name;
    struct flb_config_map_val *header_pair;
    char                       port[8];
    int                        ret;
    struct flb_http           *ctx;

    ctx = flb_calloc(1, sizeof(struct flb_http));
    if (!ctx) {
        flb_errno();
        return NULL;
    }
    ctx->ins = ins;
    mk_list_init(&ctx->connections);

    ctx->oauth2_cfg.jwks_refresh_interval = 300;

    /* Load the config map */
    ret = flb_input_config_map_set(ins, (void *) ctx);
    if (ret == -1) {
        flb_free(ctx);
        return NULL;
    }

    /* Apply OAuth2 JWT config map properties if any */
    if (ins->oauth2_jwt_config_map && mk_list_size(&ins->oauth2_jwt_properties) > 0) {
        ret = flb_config_map_set(&ins->oauth2_jwt_properties, ins->oauth2_jwt_config_map,
                                 &ctx->oauth2_cfg);
        if (ret == -1) {
            flb_free(ctx);
            return NULL;
        }
    }

    /* Listen interface (if not set, defaults to 0.0.0.0:9880) */
    flb_input_net_default_listener("0.0.0.0", 9880, ins);

    ctx->listen = flb_strdup(ins->host.listen);
    snprintf(port, sizeof(port) - 1, "%d", ins->host.port);
    ctx->tcp_port = flb_strdup(port);

    /* HTTP Server specifics */
    ctx->server = flb_calloc(1, sizeof(struct mk_server));
    if (!ctx->server) {
        flb_errno();
        http_config_destroy(ctx);
        return NULL;
    }

    ctx->server->keep_alive = MK_TRUE;

    /* monkey detects server->workers == 0 as the server not being initialized at the
     * moment so we want to make sure that it stays that way!
     */

    ret = flb_log_event_encoder_init(&ctx->log_encoder,
                                     FLB_LOG_EVENT_FORMAT_DEFAULT);

    if (ret != FLB_EVENT_ENCODER_SUCCESS) {
        flb_plg_error(ctx->ins, "error initializing event encoder : %d", ret);
        http_config_destroy(ctx);
        return NULL;
    }

    ctx->success_headers_str = flb_sds_create_size(1);

    if (ctx->success_headers_str == NULL) {
        http_config_destroy(ctx);
        return NULL;
    }

    /* Create record accessor for tag_key if specified */
    if (ctx->tag_key) {
        ctx->ra_tag_key = flb_ra_create(ctx->tag_key, FLB_TRUE);
        if (!ctx->ra_tag_key) {
            flb_plg_error(ctx->ins, "invalid record accessor pattern for tag_key: %s", ctx->tag_key);
            http_config_destroy(ctx);
            return NULL;
        }
    }

    flb_config_map_foreach(header_iterator, header_pair, ctx->success_headers) {
        header_name = mk_list_entry_first(header_pair->val.list,
                                          struct flb_slist_entry,
                                          _head);

        header_value = mk_list_entry_last(header_pair->val.list,
                                          struct flb_slist_entry,
                                          _head);

        ret = flb_sds_cat_safe(&ctx->success_headers_str,
                               header_name->str,
                               flb_sds_len(header_name->str));

        if (ret == 0) {
            ret = flb_sds_cat_safe(&ctx->success_headers_str,
                                   ": ",
                                   2);
        }

        if (ret == 0) {
            ret = flb_sds_cat_safe(&ctx->success_headers_str,
                                   header_value->str,
                                   flb_sds_len(header_value->str));
        }

        if (ret == 0) {
            ret = flb_sds_cat_safe(&ctx->success_headers_str,
                                   "\r\n",
                                   2);
        }

        if (ret != 0) {
            http_config_destroy(ctx);

            return NULL;
        }
    }

    return ctx;
}

int http_config_destroy(struct flb_http *ctx)
{
    if (ctx->ra_tag_key) {
        flb_ra_destroy(ctx->ra_tag_key);
    }

    /* release all connections */
    http_conn_release_all(ctx);

    flb_log_event_encoder_destroy(&ctx->log_encoder);

    if (ctx->collector_id != -1) {
        flb_input_collector_delete(ctx->collector_id, ctx->ins);

        ctx->collector_id = -1;
    }

    if (ctx->downstream != NULL) {
        flb_downstream_destroy(ctx->downstream);
    }

    if (ctx->server) {
        flb_free(ctx->server);
    }

    if (ctx->enable_http2) {
        flb_http_server_destroy(&ctx->http_server);
    }

    if (ctx->success_headers_str != NULL) {
        flb_sds_destroy(ctx->success_headers_str);
    }

    if (ctx->oauth2_ctx) {
        flb_oauth2_jwt_context_destroy(ctx->oauth2_ctx);
        ctx->oauth2_ctx = NULL;
        ctx->oauth2_cfg.issuer = NULL;
        ctx->oauth2_cfg.jwks_url = NULL;
        ctx->oauth2_cfg.allowed_audience = NULL;
    }
    else {
        if (ctx->oauth2_cfg.issuer) {
            flb_sds_destroy(ctx->oauth2_cfg.issuer);
        }

        if (ctx->oauth2_cfg.jwks_url) {
            flb_sds_destroy(ctx->oauth2_cfg.jwks_url);
        }

        if (ctx->oauth2_cfg.allowed_audience) {
            flb_sds_destroy(ctx->oauth2_cfg.allowed_audience);
        }
    }


    flb_free(ctx->listen);
    flb_free(ctx->tcp_port);
    flb_free(ctx);
    return 0;
}

Coverage Report

Created: 2026-03-09 07:04

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		#include <fluent-bit/flb_input_plugin.h>
21		#include <fluent-bit/flb_oauth2_jwt.h>
22
23		#include "http.h"
24		#include "http_config.h"
25		#include "http_conn.h"
26		#include "http_config.h"
27
28		struct flb_http http_config_create(struct flb_input_instance ins)
29	0	{
30	0	struct mk_list *header_iterator;
31	0	struct flb_slist_entry *header_value;
32	0	struct flb_slist_entry *header_name;
33	0	struct flb_config_map_val *header_pair;
34	0	char port[8];
35	0	int ret;
36	0	struct flb_http *ctx;
37
38	0	ctx = flb_calloc(1, sizeof(struct flb_http));
39	0	if (!ctx) {
40	0	flb_errno();
41	0	return NULL;
42	0	}
43	0	ctx->ins = ins;
44	0	mk_list_init(&ctx->connections);
45
46	0	ctx->oauth2_cfg.jwks_refresh_interval = 300;
47
48		/* Load the config map */
49	0	ret = flb_input_config_map_set(ins, (void *) ctx);
50	0	if (ret == -1) {
51	0	flb_free(ctx);
52	0	return NULL;
53	0	}
54
55		/* Apply OAuth2 JWT config map properties if any */
56	0	if (ins->oauth2_jwt_config_map && mk_list_size(&ins->oauth2_jwt_properties) > 0) {
57	0	ret = flb_config_map_set(&ins->oauth2_jwt_properties, ins->oauth2_jwt_config_map,
58	0	&ctx->oauth2_cfg);
59	0	if (ret == -1) {
60	0	flb_free(ctx);
61	0	return NULL;
62	0	}
63	0	}
64
65		/* Listen interface (if not set, defaults to 0.0.0.0:9880) */
66	0	flb_input_net_default_listener("0.0.0.0", 9880, ins);
67
68	0	ctx->listen = flb_strdup(ins->host.listen);
69	0	snprintf(port, sizeof(port) - 1, "%d", ins->host.port);
70	0	ctx->tcp_port = flb_strdup(port);
71
72		/* HTTP Server specifics */
73	0	ctx->server = flb_calloc(1, sizeof(struct mk_server));
74	0	if (!ctx->server) {
75	0	flb_errno();
76	0	http_config_destroy(ctx);
77	0	return NULL;
78	0	}
79
80	0	ctx->server->keep_alive = MK_TRUE;
81
82		/* monkey detects server->workers == 0 as the server not being initialized at the
83		* moment so we want to make sure that it stays that way!
84		*/
85
86	0	ret = flb_log_event_encoder_init(&ctx->log_encoder,
87	0	FLB_LOG_EVENT_FORMAT_DEFAULT);
88
89	0	if (ret != FLB_EVENT_ENCODER_SUCCESS) {
90	0	flb_plg_error(ctx->ins, "error initializing event encoder : %d", ret);
91	0	http_config_destroy(ctx);
92	0	return NULL;
93	0	}
94
95	0	ctx->success_headers_str = flb_sds_create_size(1);
96
97	0	if (ctx->success_headers_str == NULL) {
98	0	http_config_destroy(ctx);
99	0	return NULL;
100	0	}
101
102		/* Create record accessor for tag_key if specified */
103	0	if (ctx->tag_key) {
104	0	ctx->ra_tag_key = flb_ra_create(ctx->tag_key, FLB_TRUE);
105	0	if (!ctx->ra_tag_key) {
106	0	flb_plg_error(ctx->ins, "invalid record accessor pattern for tag_key: %s", ctx->tag_key);
107	0	http_config_destroy(ctx);
108	0	return NULL;
109	0	}
110	0	}
111
112	0	flb_config_map_foreach(header_iterator, header_pair, ctx->success_headers) {
113	0	header_name = mk_list_entry_first(header_pair->val.list,
114	0	struct flb_slist_entry,
115	0	_head);
116
117	0	header_value = mk_list_entry_last(header_pair->val.list,
118	0	struct flb_slist_entry,
119	0	_head);
120
121	0	ret = flb_sds_cat_safe(&ctx->success_headers_str,
122	0	header_name->str,
123	0	flb_sds_len(header_name->str));
124
125	0	if (ret == 0) {
126	0	ret = flb_sds_cat_safe(&ctx->success_headers_str,
127	0	": ",
128	0	2);
129	0	}
130
131	0	if (ret == 0) {
132	0	ret = flb_sds_cat_safe(&ctx->success_headers_str,
133	0	header_value->str,
134	0	flb_sds_len(header_value->str));
135	0	}
136
137	0	if (ret == 0) {
138	0	ret = flb_sds_cat_safe(&ctx->success_headers_str,
139	0	"\r\n",
140	0	2);
141	0	}
142
143	0	if (ret != 0) {
144	0	http_config_destroy(ctx);
145
146	0	return NULL;
147	0	}
148	0	}
149
150	0	return ctx;
151	0	}
152
153		int http_config_destroy(struct flb_http *ctx)
154	0	{
155	0	if (ctx->ra_tag_key) {
156	0	flb_ra_destroy(ctx->ra_tag_key);
157	0	}
158
159		/* release all connections */
160	0	http_conn_release_all(ctx);
161
162	0	flb_log_event_encoder_destroy(&ctx->log_encoder);
163
164	0	if (ctx->collector_id != -1) {
165	0	flb_input_collector_delete(ctx->collector_id, ctx->ins);
166
167	0	ctx->collector_id = -1;
168	0	}
169
170	0	if (ctx->downstream != NULL) {
171	0	flb_downstream_destroy(ctx->downstream);
172	0	}
173
174	0	if (ctx->server) {
175	0	flb_free(ctx->server);
176	0	}
177
178	0	if (ctx->enable_http2) {
179	0	flb_http_server_destroy(&ctx->http_server);
180	0	}
181
182	0	if (ctx->success_headers_str != NULL) {
183	0	flb_sds_destroy(ctx->success_headers_str);
184	0	}
185
186	0	if (ctx->oauth2_ctx) {
187	0	flb_oauth2_jwt_context_destroy(ctx->oauth2_ctx);
188	0	ctx->oauth2_ctx = NULL;
189	0	ctx->oauth2_cfg.issuer = NULL;
190	0	ctx->oauth2_cfg.jwks_url = NULL;
191	0	ctx->oauth2_cfg.allowed_audience = NULL;
192	0	}
193	0	else {
194	0	if (ctx->oauth2_cfg.issuer) {
195	0	flb_sds_destroy(ctx->oauth2_cfg.issuer);
196	0	}
197
198	0	if (ctx->oauth2_cfg.jwks_url) {
199	0	flb_sds_destroy(ctx->oauth2_cfg.jwks_url);
200	0	}
201
202	0	if (ctx->oauth2_cfg.allowed_audience) {
203	0	flb_sds_destroy(ctx->oauth2_cfg.allowed_audience);
204	0	}
205	0	}
206
207
208	0	flb_free(ctx->listen);
209	0	flb_free(ctx->tcp_port);
210	0	flb_free(ctx);
211	0	return 0;
212	0	}