/src/fluent-bit/plugins/processor_sampling/sampling_conf.c

Source (jump to first uncovered line)
/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */

/*  Fluent Bit
 *  ==========
 *  Copyright (C) 2015-2025 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_processor_plugin.h>
#include <fluent-bit/flb_kv.h>

#include "sampling.h"

static int sampling_type_lookup(const char *type_str)
{
    if (strcasecmp(type_str, "test") == 0) {
        return SAMPLING_TYPE_TEST;
    }
    else if (strcasecmp(type_str, "probabilistic") == 0) {
        return SAMPLING_TYPE_PROBABILISTIC;
    }
    else if (strcasecmp(type_str, "tail") == 0) {
        return SAMPLING_TYPE_TAIL;
    }

    return -1;
}

static char *sampling_config_type_str(int type)
{
    switch (type) {
    case SAMPLING_TYPE_TEST:
        return "test";
    case SAMPLING_TYPE_PROBABILISTIC:
        return "probabilistic";
    case SAMPLING_TYPE_TAIL:
        return "tail";
    default:
        return "unknown";
    }
}

static struct sampling_plugin *sampling_config_get_plugin(int type)
{
    struct sampling_plugin *plugin = NULL;

    switch (type) {
    /*
        case SAMPLING_TYPE_TEST:
        plugin = &sampling_test_plugin;
        break;
    */
    case SAMPLING_TYPE_PROBABILISTIC:
        plugin = &sampling_probabilistic_plugin;
        break;
    case SAMPLING_TYPE_TAIL:
        plugin = &sampling_tail_plugin;
        break;
    default:
        plugin = NULL;
    }

    return plugin;
}

/* Register properties inside 'rules' into the ctx->plugin_rules_properties list */
 int sampling_config_process_rules(struct flb_config *config, struct sampling *ctx)
{
    int ret;
    char val[1024];
    struct cfl_list *head;
    struct cfl_variant *var;
    struct cfl_kvlist *kv;
    struct cfl_kvpair *pair;
    struct mk_list *map;
    struct flb_kv *kv_entry;

    if (!ctx->sampling_settings) {
        /* no rules have been defined */
        return 0;
    }

    var = ctx->sampling_settings;
    if (var->type != CFL_VARIANT_KVLIST) {
        flb_plg_error(ctx->ins, "rules must be a map");
        return -1;
    }

    kv = var->data.as_kvlist;
    cfl_list_foreach(head, &kv->list) {
        pair = cfl_list_entry(head, struct cfl_kvpair, _head);

        if (pair->val->type != CFL_VARIANT_INT &&
            pair->val->type != CFL_VARIANT_UINT &&
            pair->val->type != CFL_VARIANT_STRING &&
            pair->val->type != CFL_VARIANT_BOOL &&
            pair->val->type != CFL_VARIANT_DOUBLE) {
            flb_plg_error(ctx->ins, "invalid value type for key '%s'", pair->key);
            return -1;
        }

        /*
         * Internal kvlist expects the value to be in string format, convert them from native
         * to it string version. We might need a better interface for this.
         */
        ret = -1;

        if (pair->val->type == CFL_VARIANT_INT) {
            ret = snprintf(val, sizeof(val) - 1, "%ld", pair->val->data.as_int64);
        }
        else if (pair->val->type == CFL_VARIANT_UINT) {
            ret = snprintf(val, sizeof(val) - 1, "%ld", pair->val->data.as_uint64);
        }
        else if (pair->val->type == CFL_VARIANT_DOUBLE) {
            ret = snprintf(val, sizeof(val) - 1, "%f", pair->val->data.as_double);
        }
        else if (pair->val->type == CFL_VARIANT_BOOL) {
            ret = snprintf(val, sizeof(val) - 1, "%s", pair->val->data.as_bool ? "true" : "false");
        }
        else if (pair->val->type == CFL_VARIANT_STRING) {
            ret = snprintf(val, sizeof(val) - 1, "%s", pair->val->data.as_string);
        }
        else {
            flb_plg_error(ctx->ins, "invalid value type for key '%s'", pair->key);
            return -1;
        }

        if (ret <= 0) {
            flb_plg_error(ctx->ins, "failed to convert value to string");
            return -1;
        }

        kv_entry = flb_kv_item_create_len(&ctx->plugin_settings_properties, pair->key, strlen(pair->key), val, ret);
        if (!kv_entry) {
            flb_plg_error(ctx->ins, "failed to create kv entry for rule key '%s'", pair->key);
            return -1;
        }
    }

    map = flb_config_map_create(config, ctx->plugin->config_map);
    if (!map) {
        flb_plg_error(ctx->ins, "failed to create map for plugin rules");
        return -1;
    }
    ctx->plugin_config_map = map;

    ret = flb_config_map_properties_check(ctx->type_str, &ctx->plugin_settings_properties, map);
    if (ret == -1) {
        flb_plg_error(ctx->ins, "failed to validate plugin rules properties");
        return -1;
    }

    return 0;
}

struct sampling *sampling_config_create(struct flb_processor_instance *processor_instance,
                                        struct flb_config *config)
{
    int ret;
    struct sampling *ctx;
    struct sampling_plugin *plugin_context;
    struct sampling_conditions *sampling_conditions;

    ctx = flb_calloc(1, sizeof(struct sampling));
    if (!ctx) {
        flb_errno();
        return NULL;
    }
    ctx->ins = processor_instance;
    ctx->input_ins = flb_processor_get_input_instance(ctx->ins->pu);

    /* config map */
    ret = flb_processor_instance_config_map_set(processor_instance, ctx);
    if (ret == -1) {
        flb_free(ctx);
        return NULL;
    }

    /* sampling type: this is mandatory */
    if (!ctx->type_str) {
        flb_plg_error(processor_instance, "no sampling 'type' defined");
        flb_free(ctx);
        return NULL;
    }

    /* type (int) */
    ret = sampling_type_lookup(ctx->type_str);
    if (ret == -1) {
        flb_plg_error(processor_instance, "unknown sampling type '%s'", ctx->type_str);
        flb_free(ctx);
        return NULL;
    }
    ctx->type = ret;

    plugin_context = sampling_config_get_plugin(ctx->type);
    if (!plugin_context) {
        flb_plg_error(processor_instance, "no plugin context found for sampling type '%s'",
                      sampling_config_type_str(ctx->type));
        flb_free(ctx);
        return NULL;
    }
    ctx->plugin = plugin_context;

    cfl_list_init(&ctx->plugins);
    flb_kv_init(&ctx->plugin_settings_properties);

    /* load conditions */
    if (ctx->conditions) {
        sampling_conditions = sampling_conditions_create(ctx, ctx->conditions);
        if (!sampling_conditions) {
            flb_plg_error(processor_instance, "failed to create conditions");
            flb_free(ctx);
            return NULL;
        }
        ctx->sampling_conditions = sampling_conditions;
    }

    return ctx;
}

void sampling_config_destroy(struct flb_config *config, struct sampling *ctx)
{
    if (!ctx) {
        return;
    }

    if (ctx->sampling_conditions) {
        sampling_conditions_destroy(ctx->sampling_conditions);
    }

    if (ctx->plugin) {
        if (ctx->plugin->cb_exit) {
            ctx->plugin->cb_exit(config, ctx->plugin_context);
        }
    }

    flb_kv_release(&ctx->plugin_settings_properties);

    if (ctx->plugin_config_map) {
        flb_config_map_destroy(ctx->plugin_config_map);
    }

    flb_free(ctx);
}

Coverage Report

Created: 2025-06-24 08:09

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-2025 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_processor_plugin.h>
21		#include <fluent-bit/flb_kv.h>
22
23		#include "sampling.h"
24
25		static int sampling_type_lookup(const char *type_str)
26	0	{
27	0	if (strcasecmp(type_str, "test") == 0) {
28	0	return SAMPLING_TYPE_TEST;
29	0	}
30	0	else if (strcasecmp(type_str, "probabilistic") == 0) {
31	0	return SAMPLING_TYPE_PROBABILISTIC;
32	0	}
33	0	else if (strcasecmp(type_str, "tail") == 0) {
34	0	return SAMPLING_TYPE_TAIL;
35	0	}
36
37	0	return -1;
38	0	}
39
40		static char *sampling_config_type_str(int type)
41	0	{
42	0	switch (type) {
43	0	case SAMPLING_TYPE_TEST:
44	0	return "test";
45	0	case SAMPLING_TYPE_PROBABILISTIC:
46	0	return "probabilistic";
47	0	case SAMPLING_TYPE_TAIL:
48	0	return "tail";
49	0	default:
50	0	return "unknown";
51	0	}
52	0	}
53
54		static struct sampling_plugin *sampling_config_get_plugin(int type)
55	0	{
56	0	struct sampling_plugin *plugin = NULL;
57
58	0	switch (type) {
59		/*
60		case SAMPLING_TYPE_TEST:
61		plugin = &sampling_test_plugin;
62		break;
63		*/
64	0	case SAMPLING_TYPE_PROBABILISTIC:
65	0	plugin = &sampling_probabilistic_plugin;
66	0	break;
67	0	case SAMPLING_TYPE_TAIL:
68	0	plugin = &sampling_tail_plugin;
69	0	break;
70	0	default:
71	0	plugin = NULL;
72	0	}
73
74	0	return plugin;
75	0	}
76
77		/* Register properties inside 'rules' into the ctx->plugin_rules_properties list */
78		int sampling_config_process_rules(struct flb_config config, struct sampling ctx)
79	0	{
80	0	int ret;
81	0	char val[1024];
82	0	struct cfl_list *head;
83	0	struct cfl_variant *var;
84	0	struct cfl_kvlist *kv;
85	0	struct cfl_kvpair *pair;
86	0	struct mk_list *map;
87	0	struct flb_kv *kv_entry;
88
89	0	if (!ctx->sampling_settings) {
90		/* no rules have been defined */
91	0	return 0;
92	0	}
93
94	0	var = ctx->sampling_settings;
95	0	if (var->type != CFL_VARIANT_KVLIST) {
96	0	flb_plg_error(ctx->ins, "rules must be a map");
97	0	return -1;
98	0	}
99
100	0	kv = var->data.as_kvlist;
101	0	cfl_list_foreach(head, &kv->list) {
102	0	pair = cfl_list_entry(head, struct cfl_kvpair, _head);
103
104	0	if (pair->val->type != CFL_VARIANT_INT &&
105	0	pair->val->type != CFL_VARIANT_UINT &&
106	0	pair->val->type != CFL_VARIANT_STRING &&
107	0	pair->val->type != CFL_VARIANT_BOOL &&
108	0	pair->val->type != CFL_VARIANT_DOUBLE) {
109	0	flb_plg_error(ctx->ins, "invalid value type for key '%s'", pair->key);
110	0	return -1;
111	0	}
112
113		/*
114		* Internal kvlist expects the value to be in string format, convert them from native
115		* to it string version. We might need a better interface for this.
116		*/
117	0	ret = -1;
118
119	0	if (pair->val->type == CFL_VARIANT_INT) {
120	0	ret = snprintf(val, sizeof(val) - 1, "%ld", pair->val->data.as_int64);
121	0	}
122	0	else if (pair->val->type == CFL_VARIANT_UINT) {
123	0	ret = snprintf(val, sizeof(val) - 1, "%ld", pair->val->data.as_uint64);
124	0	}
125	0	else if (pair->val->type == CFL_VARIANT_DOUBLE) {
126	0	ret = snprintf(val, sizeof(val) - 1, "%f", pair->val->data.as_double);
127	0	}
128	0	else if (pair->val->type == CFL_VARIANT_BOOL) {
129	0	ret = snprintf(val, sizeof(val) - 1, "%s", pair->val->data.as_bool ? "true" : "false");
130	0	}
131	0	else if (pair->val->type == CFL_VARIANT_STRING) {
132	0	ret = snprintf(val, sizeof(val) - 1, "%s", pair->val->data.as_string);
133	0	}
134	0	else {
135	0	flb_plg_error(ctx->ins, "invalid value type for key '%s'", pair->key);
136	0	return -1;
137	0	}
138
139	0	if (ret <= 0) {
140	0	flb_plg_error(ctx->ins, "failed to convert value to string");
141	0	return -1;
142	0	}
143
144	0	kv_entry = flb_kv_item_create_len(&ctx->plugin_settings_properties, pair->key, strlen(pair->key), val, ret);
145	0	if (!kv_entry) {
146	0	flb_plg_error(ctx->ins, "failed to create kv entry for rule key '%s'", pair->key);
147	0	return -1;
148	0	}
149	0	}
150
151	0	map = flb_config_map_create(config, ctx->plugin->config_map);
152	0	if (!map) {
153	0	flb_plg_error(ctx->ins, "failed to create map for plugin rules");
154	0	return -1;
155	0	}
156	0	ctx->plugin_config_map = map;
157
158	0	ret = flb_config_map_properties_check(ctx->type_str, &ctx->plugin_settings_properties, map);
159	0	if (ret == -1) {
160	0	flb_plg_error(ctx->ins, "failed to validate plugin rules properties");
161	0	return -1;
162	0	}
163
164	0	return 0;
165	0	}
166
167		struct sampling sampling_config_create(struct flb_processor_instance processor_instance,
168		struct flb_config *config)
169	0	{
170	0	int ret;
171	0	struct sampling *ctx;
172	0	struct sampling_plugin *plugin_context;
173	0	struct sampling_conditions *sampling_conditions;
174
175	0	ctx = flb_calloc(1, sizeof(struct sampling));
176	0	if (!ctx) {
177	0	flb_errno();
178	0	return NULL;
179	0	}
180	0	ctx->ins = processor_instance;
181	0	ctx->input_ins = flb_processor_get_input_instance(ctx->ins->pu);
182
183		/* config map */
184	0	ret = flb_processor_instance_config_map_set(processor_instance, ctx);
185	0	if (ret == -1) {
186	0	flb_free(ctx);
187	0	return NULL;
188	0	}
189
190		/* sampling type: this is mandatory */
191	0	if (!ctx->type_str) {
192	0	flb_plg_error(processor_instance, "no sampling 'type' defined");
193	0	flb_free(ctx);
194	0	return NULL;
195	0	}
196
197		/* type (int) */
198	0	ret = sampling_type_lookup(ctx->type_str);
199	0	if (ret == -1) {
200	0	flb_plg_error(processor_instance, "unknown sampling type '%s'", ctx->type_str);
201	0	flb_free(ctx);
202	0	return NULL;
203	0	}
204	0	ctx->type = ret;
205
206	0	plugin_context = sampling_config_get_plugin(ctx->type);
207	0	if (!plugin_context) {
208	0	flb_plg_error(processor_instance, "no plugin context found for sampling type '%s'",
209	0	sampling_config_type_str(ctx->type));
210	0	flb_free(ctx);
211	0	return NULL;
212	0	}
213	0	ctx->plugin = plugin_context;
214
215	0	cfl_list_init(&ctx->plugins);
216	0	flb_kv_init(&ctx->plugin_settings_properties);
217
218		/* load conditions */
219	0	if (ctx->conditions) {
220	0	sampling_conditions = sampling_conditions_create(ctx, ctx->conditions);
221	0	if (!sampling_conditions) {
222	0	flb_plg_error(processor_instance, "failed to create conditions");
223	0	flb_free(ctx);
224	0	return NULL;
225	0	}
226	0	ctx->sampling_conditions = sampling_conditions;
227	0	}
228
229	0	return ctx;
230	0	}
231
232		void sampling_config_destroy(struct flb_config config, struct sampling ctx)
233	0	{
234	0	if (!ctx) {
235	0	return;
236	0	}
237
238	0	if (ctx->sampling_conditions) {
239	0	sampling_conditions_destroy(ctx->sampling_conditions);
240	0	}
241
242	0	if (ctx->plugin) {
243	0	if (ctx->plugin->cb_exit) {
244	0	ctx->plugin->cb_exit(config, ctx->plugin_context);
245	0	}
246	0	}
247
248	0	flb_kv_release(&ctx->plugin_settings_properties);
249
250	0	if (ctx->plugin_config_map) {
251	0	flb_config_map_destroy(ctx->plugin_config_map);
252	0	}
253
254	0	flb_free(ctx);
255	0	}