/src/fluent-bit/lib/cmetrics/src/cmt_decode_statsd.c

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

/*  CMetrics
 *  ========
 *  Copyright 2021-2024 The CMetrics 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 <float.h> /* for DBL_EPSILON */
#include <cmetrics/cmetrics.h>
#include <cmetrics/cmt_metric.h>
#include <cmetrics/cmt_map.h>
#include <cmetrics/cmt_counter.h>
#include <cmetrics/cmt_summary.h>
#include <cmetrics/cmt_histogram.h>
#include <cmetrics/cmt_gauge.h>
#include <cmetrics/cmt_untyped.h>
#include <cmetrics/cmt_decode_statsd.h>
#include <cmetrics/cmt_compat.h>

static struct cmt_map_label *create_map_label(char *caption, size_t length)
{
    struct cmt_map_label *map_label;

    map_label = calloc(1, sizeof(struct cmt_map_label));
    if (!map_label) {
        return NULL;
    }

    if (map_label != NULL) {
        if (caption != NULL) {
            if (length == 0) {
                length = strlen(caption);
            }

            map_label->name = cfl_sds_create_len(caption, length);

            if (map_label->name == NULL) {
                cmt_errno();

                free(map_label);

                map_label = NULL;
            }
        }
    }

    return map_label;
}

static int append_new_map_label_key(struct cmt_map *map, char *name)
{
    struct cmt_map_label *label;

    label = create_map_label(name, 0);

    if (label == NULL) {
        return CMT_DECODE_STATSD_ALLOCATION_ERROR;
    }

    cfl_list_add(&label->_head, &map->label_keys);
    map->label_count++;

    return CMT_DECODE_STATSD_SUCCESS;
}

static int append_new_metric_label_value(struct cmt_metric *metric, char *name, size_t length)
{
    struct cmt_map_label *label;

    label = create_map_label(name, length);

    if (label == NULL) {
        return CMT_DECODE_STATSD_ALLOCATION_ERROR;
    }

    cfl_list_add(&label->_head, &metric->labels);

    return CMT_DECODE_STATSD_SUCCESS;
}

static int is_incremental(char *str)
{
    return (*str == '+' || *str == '-');
}

static int decode_labels(struct cmt *cmt,
                         struct cmt_map *map,
                         struct cmt_metric *metric,
                         char *labels, int incremental)
{
    void                 **value_index_list;
    size_t                 map_label_index;
    size_t                 map_label_count;
    struct cfl_list       *label_iterator;
    struct cmt_map_label  *current_label;
    size_t                 label_index;
    int                    label_found;
    char                  *label_kv, *colon;
    cfl_sds_t              label_k = NULL, label_v = NULL, tmp = NULL;
    int                    result;
    struct cfl_list *head = NULL;
    struct cfl_list *kvs = NULL;
    struct cfl_split_entry *cur = NULL;

    result = CMT_DECODE_STATSD_SUCCESS;

    value_index_list = calloc(128, sizeof(void *));

    if (value_index_list == NULL) {
        return CMT_DECODE_STATSD_ALLOCATION_ERROR;
    }

    label_found = CMT_FALSE;
    label_index = 0;

    if (incremental) {
        label_k = cfl_sds_create("incremental");
        if (label_k != NULL) {
            result = append_new_map_label_key(map, label_k);
            cfl_sds_destroy(label_k);

            if (result == CMT_DECODE_STATSD_SUCCESS) {
                tmp = (void *) cfl_sds_create("true");
                if (tmp != NULL) {
                    value_index_list[label_index] = tmp;
                }
            }
        }
    }

    if (labels != NULL) {
        kvs = cfl_utils_split(labels, ',', -1 );
        if (kvs == NULL) {
            goto split_error;
        }

        cfl_list_foreach(head, kvs) {
            cur = cfl_list_entry(head, struct cfl_split_entry, _head);
            label_kv = cur->value;

            /* skip tags without a value */
            colon = strchr(label_kv, ':');
            if (colon == NULL) {
                continue;
            }
            label_k = cfl_sds_create_len(label_kv, colon - label_kv);
            if (label_k == NULL) {
                for (label_index = 0 ; label_index < 128 ; label_index++) {
                    if (value_index_list[label_index] != NULL) {
                        cfl_sds_destroy(value_index_list[label_index]);
                    }
                }

                free(value_index_list);

                if (kvs != NULL) {
                    cfl_utils_split_free(kvs);
                }

                return CMT_DECODE_STATSD_INVALID_TAG_FORMAT_ERROR;
            }
            label_v = cfl_sds_create_len(colon + 1, strlen(label_kv) - strlen(label_k) - 1);
            if (label_v == NULL) {
                cfl_sds_destroy(label_k);

                for (label_index = 0 ; label_index < 128 ; label_index++) {
                    if (value_index_list[label_index] != NULL) {
                        cfl_sds_destroy(value_index_list[label_index]);
                    }
                }

                free(value_index_list);

                if (kvs != NULL) {
                    cfl_utils_split_free(kvs);
                }

                return CMT_DECODE_STATSD_INVALID_TAG_FORMAT_ERROR;
            }

            /* look up the key position from the start for every tag */
            label_found = CMT_FALSE;
            label_index = 0;

            cfl_list_foreach(label_iterator, &map->label_keys) {
                current_label = cfl_list_entry(label_iterator, struct cmt_map_label, _head);

                if (strcmp(current_label->name, label_k) == 0) {
                    label_found = CMT_TRUE;

                    break;
                }

                label_index++;
            }

            if (label_index > 127) {
                cfl_sds_destroy(label_k);
                cfl_sds_destroy(label_v);

                for (label_index = 0 ; label_index < 128 ; label_index++) {
                    if (value_index_list[label_index] != NULL) {
                        cfl_sds_destroy(value_index_list[label_index]);
                    }
                }

                free(value_index_list);

                if (kvs != NULL) {
                    cfl_utils_split_free(kvs);
                }

                return CMT_DECODE_STATSD_INVALID_ARGUMENT_ERROR;
            }

            if (label_found == CMT_FALSE) {
                result = append_new_map_label_key(map, label_k);
            }

            if (result == CMT_DECODE_STATSD_SUCCESS) {
                /* a repeated key replaces the previous value */
                if (value_index_list[label_index] != NULL) {
                    cfl_sds_destroy(value_index_list[label_index]);
                }

                value_index_list[label_index] = (void *) cfl_sds_create_len(label_v,
                                                                            cfl_sds_len(label_v));
            }

            cfl_sds_destroy(label_k);
            cfl_sds_destroy(label_v);
        }
    }

split_error: /* Nop for adding labels */

    map_label_count = cfl_list_size(&map->label_keys);

    for (map_label_index = 0 ;
         result == CMT_DECODE_STATSD_SUCCESS &&
         map_label_index < map_label_count ;
         map_label_index++) {

        if (value_index_list[map_label_index] != NULL) {
            label_v = (char *) value_index_list[map_label_index];
            result = append_new_metric_label_value(metric, label_v, 0);
        }
    }

    for (label_index = 0 ; label_index < 128 ; label_index++) {
        if (value_index_list[label_index] != NULL) {
            cfl_sds_destroy(value_index_list[label_index]);
        }
    }

    free(value_index_list);

    if (kvs != NULL) {
        cfl_utils_split_free(kvs);
    }

    return result;
}

static int decode_numerical_message(struct cmt *cmt,
                                    struct cmt_map *map,
                                    struct cmt_statsd_message *m)
{
    struct cmt_metric *metric;
    int                result;
    uint64_t           ts;
    int                incremental = 0;

    ts = cfl_time_now();

    result = CMT_DECODE_STATSD_SUCCESS;

    metric = calloc(1, sizeof(struct cmt_metric));

    if (metric == NULL) {
        return CMT_DECODE_STATSD_ALLOCATION_ERROR;
    }

    cfl_list_init(&metric->labels);

    incremental = is_incremental(m->value);

    result = decode_labels(cmt,
                           map,
                           metric,
                           m->labels,
                           incremental);

    if (result) {
        destroy_label_list(&metric->labels);

        free(metric);
    }
    else {
        cfl_list_add(&metric->_head, &map->metrics);
    }

    if (result == CMT_DECODE_STATSD_SUCCESS) {
        if ((m->sample_rate - 0.0) > DBL_EPSILON &&
            (1.0 - m->sample_rate) > DBL_EPSILON) {
            cmt_metric_set(metric, ts, strtod(m->value, NULL) / m->sample_rate);
        }
        else {
            cmt_metric_set(metric, ts, strtod(m->value, NULL));
        }
    }

    return result;
}

static int decode_counter_entry(struct cmt *cmt,
                                void *instance,
                                struct cmt_statsd_message *m)
{
    struct cmt_counter *counter;
    int                 result;

    result = CMT_DECODE_STATSD_SUCCESS;

    counter = (struct cmt_counter *) instance;

    counter->map->metric_static_set = 0;

    result = decode_numerical_message(cmt,
                                      counter->map,
                                      m);

    return result;
}

static int decode_gauge_entry(struct cmt *cmt,
                              void *instance,
                              struct cmt_statsd_message *m)
{
    struct cmt_gauge *gauge;
    int               result;

    result = CMT_DECODE_STATSD_SUCCESS;

    gauge = (struct cmt_gauge *) instance;

    gauge->map->metric_static_set = 0;

    result = decode_numerical_message(cmt,
                                      gauge->map,
                                      m);

    return result;
}

static int decode_untyped_entry(struct cmt *cmt,
                                void *instance,
                                struct cmt_statsd_message *m)
{
    struct cmt_untyped *untyped;
    int                 result;

    result = CMT_DECODE_STATSD_SUCCESS;

    untyped = (struct cmt_untyped *) instance;

    untyped->map->metric_static_set = 0;

    result = decode_numerical_message(cmt,
                                      untyped->map,
                                      m);

    return result;
}

static int decode_statsd_message(struct cmt *cmt,
                                 struct cmt_statsd_message *m,
                                 int flags)
{
    char *metric_name = NULL;
    char *metric_subsystem   = NULL;
    char *metric_namespace   = NULL;
    char *metric_description = NULL;
    void *instance;
    int   result;

    result = CMT_DECODE_STATSD_SUCCESS;

    metric_description = "-";
    metric_name = cfl_sds_create_len(m->bucket, m->bucket_len);
    if (metric_name == NULL) {
        return CMT_DECODE_STATSD_ALLOCATION_ERROR;
    }
    metric_namespace = "";
    metric_subsystem = "";

    switch (m->type) {
    case CMT_DECODE_STATSD_TYPE_COUNTER:
        instance = cmt_counter_create(cmt,
                                      metric_namespace,
                                      metric_subsystem,
                                      metric_name,
                                      metric_description,
                                      0, NULL);

        if (instance == NULL) {
            cfl_sds_destroy(metric_name);
            return CMT_DECODE_STATSD_ALLOCATION_ERROR;
        }

        result = decode_counter_entry(cmt, instance, m);

        if (result) {
            cmt_counter_destroy(instance);
        }
        break;
    case CMT_DECODE_STATSD_TYPE_GAUGE:
        instance = cmt_gauge_create(cmt,
                                    metric_namespace,
                                    metric_subsystem,
                                    metric_name,
                                    metric_description,
                                    0, NULL);

        if (instance == NULL) {
            cfl_sds_destroy(metric_name);
            return CMT_DECODE_STATSD_ALLOCATION_ERROR;
        }

        result = decode_gauge_entry(cmt, instance, m);

        if (result) {
            cmt_gauge_destroy(instance);
        }
        break;
    case CMT_DECODE_STATSD_TYPE_SET:
        /* Set type will be translated as an untyped */
        instance = cmt_untyped_create(cmt,
                                      metric_namespace,
                                      metric_subsystem,
                                      metric_name,
                                      metric_description,
                                      0, NULL);

        if (instance == NULL) {
            cfl_sds_destroy(metric_name);
            return CMT_DECODE_STATSD_ALLOCATION_ERROR;
        }

        result = decode_untyped_entry(cmt, instance, m);

        if (result) {
            cmt_untyped_destroy(instance);
        }
        break;
    case CMT_DECODE_STATSD_TYPE_TIMER:
        /* TODO: Add histogram observer */
        if (flags & CMT_DECODE_STATSD_GAUGE_OBSERVER) {
            instance = cmt_gauge_create(cmt,
                                        metric_namespace,
                                        metric_subsystem,
                                        metric_name,
                                        metric_description,
                                        0, NULL);

            if (instance == NULL) {
                cfl_sds_destroy(metric_name);
                return CMT_DECODE_STATSD_ALLOCATION_ERROR;
            }

            result = decode_gauge_entry(cmt, instance, m);

            if (result) {
                cmt_gauge_destroy(instance);
            }
        }
        break;
    default:
        result = CMT_DECODE_STATSD_UNSUPPORTED_METRIC_TYPE;
        break;
    }

    cfl_sds_destroy(metric_name);

    return result;
}

static int cmt_get_statsd_type(char *str)
{
    switch (*str) {
    case 'g':
        return CMT_DECODE_STATSD_TYPE_GAUGE;
    case 's':
        return CMT_DECODE_STATSD_TYPE_SET;
    case 'c':
        return CMT_DECODE_STATSD_TYPE_COUNTER;
    case 'm':
        if (*(str + 1) == 's') {
            return CMT_DECODE_STATSD_TYPE_TIMER;
        }
    }
    return CMT_DECODE_STATSD_TYPE_COUNTER;
}

static int statsd_process_line(struct cmt *cmt, char *line, int flags)
{
    char *colon = NULL, *bar = NULL, *atmark = NULL, *labels = NULL;
    struct cmt_statsd_message m = {0};

    /*
     * bucket:value|type|@sample_rate|#key1:value1,key2:value2,...
     * ------
     */
    colon = strchr(line, ':');
    if (colon == NULL) {
        return CMT_DECODE_STATSD_INVALID_ARGUMENT_ERROR;
    }
    m.bucket = line;
    m.bucket_len = (colon - line);

    /*
     * bucket:value|type|@sample_rate|#key1:value1,key2:value2,...
     *              ----
     */
    bar = strchr(colon + 1, '|');
    if (bar == NULL) {
        return CMT_DECODE_STATSD_INVALID_ARGUMENT_ERROR;
    }
    m.type = cmt_get_statsd_type(bar + 1);

    /*
     * bucket:value|type|@sample_rate|#key1:value1,key2:value2,...
     *        -----
     */
    m.value = colon + 1;
    m.value_len = (bar - colon - 1);

    /*
     * bucket:value|type|@sample_rate|#key1:value1,key2:value2,...
     *                   ------------
     */
    atmark = strstr(bar + 1, "|@");
    if (atmark == NULL || atof(atmark + 2) == 0) {
        m.sample_rate = 1.0;
    }
    else {
        m.sample_rate = atof(atmark + 2);
    }

    /*
     * bucket:value|type|@sample_rate|#key1:value1,key2:value2,...
     *                                ------------
     */
    labels = strstr(bar + 1, "|#");
    if (labels != NULL) {
        m.labels = labels + 2;
    }

    return decode_statsd_message(cmt, &m, flags);
}

static int decode_metrics_lines(struct cmt *cmt,
                                char *in_buf, size_t in_size,
                                int flags)
{
    int ret = CMT_DECODE_STATSD_SUCCESS;
    struct cfl_list *head = NULL;
    struct cfl_list *kvs = NULL;
    struct cfl_split_entry *cur = NULL;

    kvs = cfl_utils_split(in_buf, '\n', -1 );
    if (kvs == NULL) {
        goto split_error;
    }

    cfl_list_foreach(head, kvs) {
        cur = cfl_list_entry(head, struct cfl_split_entry, _head);
        /* StatsD format always has | at least one, skip anything else */
        if (strstr(cur->value, "|") == NULL) {
            continue;
        }

        ret = statsd_process_line(cmt, cur->value, flags);
        if (ret != CMT_DECODE_STATSD_SUCCESS) {
            ret = CMT_DECODE_STATSD_DECODE_ERROR;

            break;
        }
    }

    if (kvs != NULL) {
        cfl_utils_split_free(kvs);
    }

    return ret;

split_error:
    return -1;
}

int cmt_decode_statsd_create(struct cmt **out_cmt, char *in_buf, size_t in_size, int flags)
{
    int         result = CMT_DECODE_STATSD_INVALID_ARGUMENT_ERROR;
    struct cmt *cmt    = NULL;

    cmt = cmt_create();

    if (cmt == NULL) {
        return CMT_DECODE_STATSD_ALLOCATION_ERROR;
    }

    result = decode_metrics_lines(cmt, in_buf, in_size, flags);
    if (result != CMT_DECODE_STATSD_SUCCESS) {
        cmt_destroy(cmt);
        result = CMT_DECODE_STATSD_DECODE_ERROR;

        return result;
    }

    *out_cmt = cmt;

    return result;
}

void cmt_decode_statsd_destroy(struct cmt *cmt)
{
    cmt_destroy(cmt);
}

Coverage Report

Created: 2026-09-28 07:36

Line	Count	Source
1		/* -- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -- */
2
3		/* CMetrics
4		* ========
5		* Copyright 2021-2024 The CMetrics 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 <float.h> /* for DBL_EPSILON */
21		#include <cmetrics/cmetrics.h>
22		#include <cmetrics/cmt_metric.h>
23		#include <cmetrics/cmt_map.h>
24		#include <cmetrics/cmt_counter.h>
25		#include <cmetrics/cmt_summary.h>
26		#include <cmetrics/cmt_histogram.h>
27		#include <cmetrics/cmt_gauge.h>
28		#include <cmetrics/cmt_untyped.h>
29		#include <cmetrics/cmt_decode_statsd.h>
30		#include <cmetrics/cmt_compat.h>
31
32		static struct cmt_map_label create_map_label(char caption, size_t length)
33	0	{
34	0	struct cmt_map_label *map_label;
35
36	0	map_label = calloc(1, sizeof(struct cmt_map_label));
37	0	if (!map_label) {
38	0	return NULL;
39	0	}
40
41	0	if (map_label != NULL) {
42	0	if (caption != NULL) {
43	0	if (length == 0) {
44	0	length = strlen(caption);
45	0	}
46
47	0	map_label->name = cfl_sds_create_len(caption, length);
48
49	0	if (map_label->name == NULL) {
50	0	cmt_errno();
51
52	0	free(map_label);
53
54	0	map_label = NULL;
55	0	}
56	0	}
57	0	}
58
59	0	return map_label;
60	0	}
61
62		static int append_new_map_label_key(struct cmt_map map, char name)
63	0	{
64	0	struct cmt_map_label *label;
65
66	0	label = create_map_label(name, 0);
67
68	0	if (label == NULL) {
69	0	return CMT_DECODE_STATSD_ALLOCATION_ERROR;
70	0	}
71
72	0	cfl_list_add(&label->_head, &map->label_keys);
73	0	map->label_count++;
74
75	0	return CMT_DECODE_STATSD_SUCCESS;
76	0	}
77
78		static int append_new_metric_label_value(struct cmt_metric metric, char name, size_t length)
79	0	{
80	0	struct cmt_map_label *label;
81
82	0	label = create_map_label(name, length);
83
84	0	if (label == NULL) {
85	0	return CMT_DECODE_STATSD_ALLOCATION_ERROR;
86	0	}
87
88	0	cfl_list_add(&label->_head, &metric->labels);
89
90	0	return CMT_DECODE_STATSD_SUCCESS;
91	0	}
92
93		static int is_incremental(char *str)
94	0	{
95	0	return (str == '+' \|\| str == '-');
96	0	}
97
98		static int decode_labels(struct cmt *cmt,
99		struct cmt_map *map,
100		struct cmt_metric *metric,
101		char *labels, int incremental)
102	0	{
103	0	void **value_index_list;
104	0	size_t map_label_index;
105	0	size_t map_label_count;
106	0	struct cfl_list *label_iterator;
107	0	struct cmt_map_label *current_label;
108	0	size_t label_index;
109	0	int label_found;
110	0	char label_kv, colon;
111	0	cfl_sds_t label_k = NULL, label_v = NULL, tmp = NULL;
112	0	int result;
113	0	struct cfl_list *head = NULL;
114	0	struct cfl_list *kvs = NULL;
115	0	struct cfl_split_entry *cur = NULL;
116
117	0	result = CMT_DECODE_STATSD_SUCCESS;
118
119	0	value_index_list = calloc(128, sizeof(void *));
120
121	0	if (value_index_list == NULL) {
122	0	return CMT_DECODE_STATSD_ALLOCATION_ERROR;
123	0	}
124
125	0	label_found = CMT_FALSE;
126	0	label_index = 0;
127
128	0	if (incremental) {
129	0	label_k = cfl_sds_create("incremental");
130	0	if (label_k != NULL) {
131	0	result = append_new_map_label_key(map, label_k);
132	0	cfl_sds_destroy(label_k);
133
134	0	if (result == CMT_DECODE_STATSD_SUCCESS) {
135	0	tmp = (void *) cfl_sds_create("true");
136	0	if (tmp != NULL) {
137	0	value_index_list[label_index] = tmp;
138	0	}
139	0	}
140	0	}
141	0	}
142
143	0	if (labels != NULL) {
144	0	kvs = cfl_utils_split(labels, ',', -1 );
145	0	if (kvs == NULL) {
146	0	goto split_error;
147	0	}
148
149	0	cfl_list_foreach(head, kvs) {
150	0	cur = cfl_list_entry(head, struct cfl_split_entry, _head);
151	0	label_kv = cur->value;
152
153		/* skip tags without a value */
154	0	colon = strchr(label_kv, ':');
155	0	if (colon == NULL) {
156	0	continue;
157	0	}
158	0	label_k = cfl_sds_create_len(label_kv, colon - label_kv);
159	0	if (label_k == NULL) {
160	0	for (label_index = 0 ; label_index < 128 ; label_index++) {
161	0	if (value_index_list[label_index] != NULL) {
162	0	cfl_sds_destroy(value_index_list[label_index]);
163	0	}
164	0	}
165
166	0	free(value_index_list);
167
168	0	if (kvs != NULL) {
169	0	cfl_utils_split_free(kvs);
170	0	}
171
172	0	return CMT_DECODE_STATSD_INVALID_TAG_FORMAT_ERROR;
173	0	}
174	0	label_v = cfl_sds_create_len(colon + 1, strlen(label_kv) - strlen(label_k) - 1);
175	0	if (label_v == NULL) {
176	0	cfl_sds_destroy(label_k);
177
178	0	for (label_index = 0 ; label_index < 128 ; label_index++) {
179	0	if (value_index_list[label_index] != NULL) {
180	0	cfl_sds_destroy(value_index_list[label_index]);
181	0	}
182	0	}
183
184	0	free(value_index_list);
185
186	0	if (kvs != NULL) {
187	0	cfl_utils_split_free(kvs);
188	0	}
189
190	0	return CMT_DECODE_STATSD_INVALID_TAG_FORMAT_ERROR;
191	0	}
192
193		/* look up the key position from the start for every tag */
194	0	label_found = CMT_FALSE;
195	0	label_index = 0;
196
197	0	cfl_list_foreach(label_iterator, &map->label_keys) {
198	0	current_label = cfl_list_entry(label_iterator, struct cmt_map_label, _head);
199
200	0	if (strcmp(current_label->name, label_k) == 0) {
201	0	label_found = CMT_TRUE;
202
203	0	break;
204	0	}
205
206	0	label_index++;
207	0	}
208
209	0	if (label_index > 127) {
210	0	cfl_sds_destroy(label_k);
211	0	cfl_sds_destroy(label_v);
212
213	0	for (label_index = 0 ; label_index < 128 ; label_index++) {
214	0	if (value_index_list[label_index] != NULL) {
215	0	cfl_sds_destroy(value_index_list[label_index]);
216	0	}
217	0	}
218
219	0	free(value_index_list);
220
221	0	if (kvs != NULL) {
222	0	cfl_utils_split_free(kvs);
223	0	}
224
225	0	return CMT_DECODE_STATSD_INVALID_ARGUMENT_ERROR;
226	0	}
227
228	0	if (label_found == CMT_FALSE) {
229	0	result = append_new_map_label_key(map, label_k);
230	0	}
231
232	0	if (result == CMT_DECODE_STATSD_SUCCESS) {
233		/* a repeated key replaces the previous value */
234	0	if (value_index_list[label_index] != NULL) {
235	0	cfl_sds_destroy(value_index_list[label_index]);
236	0	}
237
238	0	value_index_list[label_index] = (void *) cfl_sds_create_len(label_v,
239	0	cfl_sds_len(label_v));
240	0	}
241
242	0	cfl_sds_destroy(label_k);
243	0	cfl_sds_destroy(label_v);
244	0	}
245	0	}
246
247	0	split_error: /* Nop for adding labels */
248
249	0	map_label_count = cfl_list_size(&map->label_keys);
250
251	0	for (map_label_index = 0 ;
252	0	result == CMT_DECODE_STATSD_SUCCESS &&
253	0	map_label_index < map_label_count ;
254	0	map_label_index++) {
255
256	0	if (value_index_list[map_label_index] != NULL) {
257	0	label_v = (char *) value_index_list[map_label_index];
258	0	result = append_new_metric_label_value(metric, label_v, 0);
259	0	}
260	0	}
261
262	0	for (label_index = 0 ; label_index < 128 ; label_index++) {
263	0	if (value_index_list[label_index] != NULL) {
264	0	cfl_sds_destroy(value_index_list[label_index]);
265	0	}
266	0	}
267
268	0	free(value_index_list);
269
270	0	if (kvs != NULL) {
271	0	cfl_utils_split_free(kvs);
272	0	}
273
274	0	return result;
275	0	}
276
277		static int decode_numerical_message(struct cmt *cmt,
278		struct cmt_map *map,
279		struct cmt_statsd_message *m)
280	0	{
281	0	struct cmt_metric *metric;
282	0	int result;
283	0	uint64_t ts;
284	0	int incremental = 0;
285
286	0	ts = cfl_time_now();
287
288	0	result = CMT_DECODE_STATSD_SUCCESS;
289
290	0	metric = calloc(1, sizeof(struct cmt_metric));
291
292	0	if (metric == NULL) {
293	0	return CMT_DECODE_STATSD_ALLOCATION_ERROR;
294	0	}
295
296	0	cfl_list_init(&metric->labels);
297
298	0	incremental = is_incremental(m->value);
299
300	0	result = decode_labels(cmt,
301	0	map,
302	0	metric,
303	0	m->labels,
304	0	incremental);
305
306	0	if (result) {
307	0	destroy_label_list(&metric->labels);
308
309	0	free(metric);
310	0	}
311	0	else {
312	0	cfl_list_add(&metric->_head, &map->metrics);
313	0	}
314
315	0	if (result == CMT_DECODE_STATSD_SUCCESS) {
316	0	if ((m->sample_rate - 0.0) > DBL_EPSILON &&
317	0	(1.0 - m->sample_rate) > DBL_EPSILON) {
318	0	cmt_metric_set(metric, ts, strtod(m->value, NULL) / m->sample_rate);
319	0	}
320	0	else {
321	0	cmt_metric_set(metric, ts, strtod(m->value, NULL));
322	0	}
323	0	}
324
325	0	return result;
326	0	}
327
328		static int decode_counter_entry(struct cmt *cmt,
329		void *instance,
330		struct cmt_statsd_message *m)
331	0	{
332	0	struct cmt_counter *counter;
333	0	int result;
334
335	0	result = CMT_DECODE_STATSD_SUCCESS;
336
337	0	counter = (struct cmt_counter *) instance;
338
339	0	counter->map->metric_static_set = 0;
340
341	0	result = decode_numerical_message(cmt,
342	0	counter->map,
343	0	m);
344
345	0	return result;
346	0	}
347
348		static int decode_gauge_entry(struct cmt *cmt,
349		void *instance,
350		struct cmt_statsd_message *m)
351	0	{
352	0	struct cmt_gauge *gauge;
353	0	int result;
354
355	0	result = CMT_DECODE_STATSD_SUCCESS;
356
357	0	gauge = (struct cmt_gauge *) instance;
358
359	0	gauge->map->metric_static_set = 0;
360
361	0	result = decode_numerical_message(cmt,
362	0	gauge->map,
363	0	m);
364
365	0	return result;
366	0	}
367
368		static int decode_untyped_entry(struct cmt *cmt,
369		void *instance,
370		struct cmt_statsd_message *m)
371	0	{
372	0	struct cmt_untyped *untyped;
373	0	int result;
374
375	0	result = CMT_DECODE_STATSD_SUCCESS;
376
377	0	untyped = (struct cmt_untyped *) instance;
378
379	0	untyped->map->metric_static_set = 0;
380
381	0	result = decode_numerical_message(cmt,
382	0	untyped->map,
383	0	m);
384
385	0	return result;
386	0	}
387
388		static int decode_statsd_message(struct cmt *cmt,
389		struct cmt_statsd_message *m,
390		int flags)
391	0	{
392	0	char *metric_name = NULL;
393	0	char *metric_subsystem = NULL;
394	0	char *metric_namespace = NULL;
395	0	char *metric_description = NULL;
396	0	void *instance;
397	0	int result;
398
399	0	result = CMT_DECODE_STATSD_SUCCESS;
400
401	0	metric_description = "-";
402	0	metric_name = cfl_sds_create_len(m->bucket, m->bucket_len);
403	0	if (metric_name == NULL) {
404	0	return CMT_DECODE_STATSD_ALLOCATION_ERROR;
405	0	}
406	0	metric_namespace = "";
407	0	metric_subsystem = "";
408
409	0	switch (m->type) {
410	0	case CMT_DECODE_STATSD_TYPE_COUNTER:
411	0	instance = cmt_counter_create(cmt,
412	0	metric_namespace,
413	0	metric_subsystem,
414	0	metric_name,
415	0	metric_description,
416	0	0, NULL);
417
418	0	if (instance == NULL) {
419	0	cfl_sds_destroy(metric_name);
420	0	return CMT_DECODE_STATSD_ALLOCATION_ERROR;
421	0	}
422
423	0	result = decode_counter_entry(cmt, instance, m);
424
425	0	if (result) {
426	0	cmt_counter_destroy(instance);
427	0	}
428	0	break;
429	0	case CMT_DECODE_STATSD_TYPE_GAUGE:
430	0	instance = cmt_gauge_create(cmt,
431	0	metric_namespace,
432	0	metric_subsystem,
433	0	metric_name,
434	0	metric_description,
435	0	0, NULL);
436
437	0	if (instance == NULL) {
438	0	cfl_sds_destroy(metric_name);
439	0	return CMT_DECODE_STATSD_ALLOCATION_ERROR;
440	0	}
441
442	0	result = decode_gauge_entry(cmt, instance, m);
443
444	0	if (result) {
445	0	cmt_gauge_destroy(instance);
446	0	}
447	0	break;
448	0	case CMT_DECODE_STATSD_TYPE_SET:
449		/* Set type will be translated as an untyped */
450	0	instance = cmt_untyped_create(cmt,
451	0	metric_namespace,
452	0	metric_subsystem,
453	0	metric_name,
454	0	metric_description,
455	0	0, NULL);
456
457	0	if (instance == NULL) {
458	0	cfl_sds_destroy(metric_name);
459	0	return CMT_DECODE_STATSD_ALLOCATION_ERROR;
460	0	}
461
462	0	result = decode_untyped_entry(cmt, instance, m);
463
464	0	if (result) {
465	0	cmt_untyped_destroy(instance);
466	0	}
467	0	break;
468	0	case CMT_DECODE_STATSD_TYPE_TIMER:
469		/* TODO: Add histogram observer */
470	0	if (flags & CMT_DECODE_STATSD_GAUGE_OBSERVER) {
471	0	instance = cmt_gauge_create(cmt,
472	0	metric_namespace,
473	0	metric_subsystem,
474	0	metric_name,
475	0	metric_description,
476	0	0, NULL);
477
478	0	if (instance == NULL) {
479	0	cfl_sds_destroy(metric_name);
480	0	return CMT_DECODE_STATSD_ALLOCATION_ERROR;
481	0	}
482
483	0	result = decode_gauge_entry(cmt, instance, m);
484
485	0	if (result) {
486	0	cmt_gauge_destroy(instance);
487	0	}
488	0	}
489	0	break;
490	0	default:
491	0	result = CMT_DECODE_STATSD_UNSUPPORTED_METRIC_TYPE;
492	0	break;
493	0	}
494
495	0	cfl_sds_destroy(metric_name);
496
497	0	return result;
498	0	}
499
500		static int cmt_get_statsd_type(char *str)
501	0	{
502	0	switch (*str) {
503	0	case 'g':
504	0	return CMT_DECODE_STATSD_TYPE_GAUGE;
505	0	case 's':
506	0	return CMT_DECODE_STATSD_TYPE_SET;
507	0	case 'c':
508	0	return CMT_DECODE_STATSD_TYPE_COUNTER;
509	0	case 'm':
510	0	if (*(str + 1) == 's') {
511	0	return CMT_DECODE_STATSD_TYPE_TIMER;
512	0	}
513	0	}
514	0	return CMT_DECODE_STATSD_TYPE_COUNTER;
515	0	}
516
517		static int statsd_process_line(struct cmt cmt, char line, int flags)
518	0	{
519	0	char colon = NULL, bar = NULL, atmark = NULL, labels = NULL;
520	0	struct cmt_statsd_message m = {0};
521
522		/*
523		* bucket:value\|type\|@sample_rate\|#key1:value1,key2:value2,...
524		* ------
525		*/
526	0	colon = strchr(line, ':');
527	0	if (colon == NULL) {
528	0	return CMT_DECODE_STATSD_INVALID_ARGUMENT_ERROR;
529	0	}
530	0	m.bucket = line;
531	0	m.bucket_len = (colon - line);
532
533		/*
534		* bucket:value\|type\|@sample_rate\|#key1:value1,key2:value2,...
535		* ----
536		*/
537	0	bar = strchr(colon + 1, '\|');
538	0	if (bar == NULL) {
539	0	return CMT_DECODE_STATSD_INVALID_ARGUMENT_ERROR;
540	0	}
541	0	m.type = cmt_get_statsd_type(bar + 1);
542
543		/*
544		* bucket:value\|type\|@sample_rate\|#key1:value1,key2:value2,...
545		* -----
546		*/
547	0	m.value = colon + 1;
548	0	m.value_len = (bar - colon - 1);
549
550		/*
551		* bucket:value\|type\|@sample_rate\|#key1:value1,key2:value2,...
552		* ------------
553		*/
554	0	atmark = strstr(bar + 1, "\|@");
555	0	if (atmark == NULL \|\| atof(atmark + 2) == 0) {
556	0	m.sample_rate = 1.0;
557	0	}
558	0	else {
559	0	m.sample_rate = atof(atmark + 2);
560	0	}
561
562		/*
563		* bucket:value\|type\|@sample_rate\|#key1:value1,key2:value2,...
564		* ------------
565		*/
566	0	labels = strstr(bar + 1, "\|#");
567	0	if (labels != NULL) {
568	0	m.labels = labels + 2;
569	0	}
570
571	0	return decode_statsd_message(cmt, &m, flags);
572	0	}
573
574		static int decode_metrics_lines(struct cmt *cmt,
575		char *in_buf, size_t in_size,
576		int flags)
577	0	{
578	0	int ret = CMT_DECODE_STATSD_SUCCESS;
579	0	struct cfl_list *head = NULL;
580	0	struct cfl_list *kvs = NULL;
581	0	struct cfl_split_entry *cur = NULL;
582
583	0	kvs = cfl_utils_split(in_buf, '\n', -1 );
584	0	if (kvs == NULL) {
585	0	goto split_error;
586	0	}
587
588	0	cfl_list_foreach(head, kvs) {
589	0	cur = cfl_list_entry(head, struct cfl_split_entry, _head);
590		/* StatsD format always has \| at least one, skip anything else */
591	0	if (strstr(cur->value, "\|") == NULL) {
592	0	continue;
593	0	}
594
595	0	ret = statsd_process_line(cmt, cur->value, flags);
596	0	if (ret != CMT_DECODE_STATSD_SUCCESS) {
597	0	ret = CMT_DECODE_STATSD_DECODE_ERROR;
598
599	0	break;
600	0	}
601	0	}
602
603	0	if (kvs != NULL) {
604	0	cfl_utils_split_free(kvs);
605	0	}
606
607	0	return ret;
608
609	0	split_error:
610	0	return -1;
611	0	}
612
613		int cmt_decode_statsd_create(struct cmt *out_cmt, char in_buf, size_t in_size, int flags)
614	0	{
615	0	int result = CMT_DECODE_STATSD_INVALID_ARGUMENT_ERROR;
616	0	struct cmt *cmt = NULL;
617
618	0	cmt = cmt_create();
619
620	0	if (cmt == NULL) {
621	0	return CMT_DECODE_STATSD_ALLOCATION_ERROR;
622	0	}
623
624	0	result = decode_metrics_lines(cmt, in_buf, in_size, flags);
625	0	if (result != CMT_DECODE_STATSD_SUCCESS) {
626	0	cmt_destroy(cmt);
627	0	result = CMT_DECODE_STATSD_DECODE_ERROR;
628
629	0	return result;
630	0	}
631
632	0	*out_cmt = cmt;
633
634	0	return result;
635	0	}
636
637		void cmt_decode_statsd_destroy(struct cmt *cmt)
638	0	{
639	0	cmt_destroy(cmt);
640	0	}