opentelemetry: JSON Metrics and Exp Histograms

include/fluent-bit/flb_opentelemetry.h

@@ -21,6 +21,7 @@
 #define FLB_OPENTELEMETRY_H
 
 #include <fluent-bit/flb_log_event_encoder.h>
+#include <cfl/cfl.h>
 #include <msgpack.h>
 #include <stdint.h>
 
@@ -173,6 +174,9 @@ int flb_opentelemetry_logs_json_to_msgpack(struct flb_log_event_encoder *encoder
                                            const char *logs_body_key,
                                            int *error_status);
 
+int flb_opentelemetry_metrics_json_to_cmt(struct cfl_list *context_list,
+                                          const char *body, size_t len);
+
 struct ctrace *flb_opentelemetry_json_traces_to_ctrace(const char *body, size_t len,
                                                        int *error_status);
 
@@ -207,6 +211,12 @@ int flb_otel_utils_json_payload_append_converted_kvlist(struct flb_log_event_enc
                                                         int target_field,
                                                         msgpack_object *object);
 
+struct cfl_variant *flb_otel_utils_msgpack_object_to_cfl_variant(
+                        msgpack_object *object);
+
+int flb_otel_utils_clone_kvlist_from_otlp_json_array(struct cfl_kvlist *target,
+                                                      msgpack_object *attributes_object);
+
 int flb_otel_utils_hex_to_id(const char *str, int len, unsigned char *out_buf, int out_size);
 
 uint64_t flb_otel_utils_convert_string_number_to_u64(char *str, size_t len);

lib/cmetrics/CMakeLists.txt

@@ -4,9 +4,9 @@ set(CMAKE_POSITION_INDEPENDENT_CODE ON)
 set(CMAKE_EXPORT_COMPILE_COMMANDS ON)
 
 # CMetrics Version
-set(CMT_VERSION_MAJOR  1)
+set(CMT_VERSION_MAJOR  2)
 set(CMT_VERSION_MINOR  0)
-set(CMT_VERSION_PATCH  7)
+set(CMT_VERSION_PATCH  2)
 set(CMT_VERSION_STR "${CMT_VERSION_MAJOR}.${CMT_VERSION_MINOR}.${CMT_VERSION_PATCH}")
 
 # Include helpers

lib/cmetrics/include/cmetrics/cmetrics.h

@@ -28,6 +28,7 @@
 #define CMT_HISTOGRAM 2
 #define CMT_SUMMARY   3
 #define CMT_UNTYPED   4
+#define CMT_EXP_HISTOGRAM 5
 
 #define CMT_AGGREGATION_TYPE_UNSPECIFIED 0
 #define CMT_AGGREGATION_TYPE_DELTA       1
@@ -65,6 +66,7 @@ struct cmt {
     struct cfl_list counters;
     struct cfl_list gauges;
     struct cfl_list histograms;
+    struct cfl_list exp_histograms;
     struct cfl_list summaries;
     struct cfl_list untypeds;
 

lib/cmetrics/include/cmetrics/cmt_cat.h

@@ -26,6 +26,7 @@ struct cmt_counter;
 struct cmt_gauge;
 struct cmt_untyped;
 struct cmt_histogram;
+struct cmt_exp_histogram;
 struct cmt_summary;
 
 int cmt_cat_copy_label_keys(struct cmt_map *map, char **out);
@@ -34,6 +35,7 @@ int cmt_cat_counter(struct cmt *cmt, struct cmt_counter *counter, struct cmt_map
 int cmt_cat_gauge(struct cmt *cmt, struct cmt_gauge *gauge, struct cmt_map *filtered_map);
 int cmt_cat_untyped(struct cmt *cmt, struct cmt_untyped *untyped, struct cmt_map *filtered_map);
 int cmt_cat_histogram(struct cmt *cmt, struct cmt_histogram *histogram, struct cmt_map *filtered_map);
+int cmt_cat_exp_histogram(struct cmt *cmt, struct cmt_exp_histogram *exp_histogram, struct cmt_map *filtered_map);
 int cmt_cat_summary(struct cmt *cmt, struct cmt_summary *summary, struct cmt_map *filtered_map);
 int cmt_cat(struct cmt *dst, struct cmt *src);
 

lib/cmetrics/include/cmetrics/cmt_decode_msgpack.h

@@ -55,6 +55,7 @@ struct cmt_msgpack_decode_context {
     uint64_t           *summary_quantiles;
     size_t             summary_quantiles_count;
     int                aggregation_type;
+    int                metric_value_type_set;
 };
 
 int cmt_decode_msgpack_create(struct cmt **out_cmt, char *in_buf, size_t in_size, 

lib/cmetrics/include/cmetrics/cmt_encode_opentelemetry.h

@@ -34,6 +34,8 @@
 struct cmt_opentelemetry_context
 {
     size_t                                          resource_index;
+    size_t                                          scope_metrics_count;
+    Opentelemetry__Proto__Metrics__V1__ScopeMetrics **scope_metrics_list;
     Opentelemetry__Proto__Metrics__V1__MetricsData *metrics_data;
     struct cmt                                     *cmt;
 };

lib/cmetrics/include/cmetrics/cmt_exp_histogram.h

@@ -0,0 +1,64 @@
+/* -*- Mode: C; tab-width: 4; indent-tabs-mode: nil; c-basic-offset: 4 -*- */
+
+/*  CMetrics
+ *  ========
+ *  Copyright 2021-2022 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.
+ */
+
+#ifndef CMT_EXP_HISTOGRAM_H
+#define CMT_EXP_HISTOGRAM_H
+
+#include <cmetrics/cmetrics.h>
+#include <cmetrics/cmt_opts.h>
+#include <cmetrics/cmt_metric.h>
+
+struct cmt_exp_histogram {
+    struct cmt_opts opts;
+    struct cmt_map *map;
+    struct cfl_list _head;
+    struct cmt *cmt;
+    int aggregation_type;
+};
+
+struct cmt_exp_histogram *cmt_exp_histogram_create(struct cmt *cmt,
+                                                   char *ns, char *subsystem,
+                                                   char *name, char *help,
+                                                   int label_count, char **label_keys);
+
+int cmt_exp_histogram_set_default(struct cmt_exp_histogram *exp_histogram,
+                                  uint64_t timestamp,
+                                  int32_t scale,
+                                  uint64_t zero_count,
+                                  double zero_threshold,
+                                  int32_t positive_offset,
+                                  size_t positive_bucket_count,
+                                  uint64_t *positive_bucket_counts,
+                                  int32_t negative_offset,
+                                  size_t negative_bucket_count,
+                                  uint64_t *negative_bucket_counts,
+                                  int sum_set,
+                                  double sum,
+                                  uint64_t count,
+                                  int labels_count, char **label_vals);
+
+int cmt_exp_histogram_destroy(struct cmt_exp_histogram *exp_histogram);
+
+int cmt_exp_histogram_to_explicit(struct cmt_metric *metric,
+                                  double **upper_bounds,
+                                  size_t *upper_bounds_count,
+                                  uint64_t **bucket_counts,
+                                  size_t *bucket_count);
+
+#endif

lib/cmetrics/include/cmetrics/cmt_metric.h

@@ -21,16 +21,40 @@
 #define CMT_METRIC_H
 
 #include <cmetrics/cmetrics.h>
+#include <stdint.h>
+
+enum cmt_metric_value_type {
+    CMT_METRIC_VALUE_DOUBLE = 0,
+    CMT_METRIC_VALUE_INT64  = 1,
+    CMT_METRIC_VALUE_UINT64 = 2
+};
 
 struct cmt_metric {
     /* counters and gauges */
     uint64_t val;
+    uint64_t value_type;
+    uint64_t val_int64;
+    uint64_t val_uint64;
 
     /* histogram */
     uint64_t *hist_buckets;
     uint64_t hist_count;
     uint64_t hist_sum;
 
+    /* exponential histogram */
+    int exp_hist_sum_set;
+    int32_t exp_hist_scale;
+    uint64_t exp_hist_zero_count;
+    double exp_hist_zero_threshold;
+    int32_t exp_hist_positive_offset;
+    uint64_t *exp_hist_positive_buckets;
+    size_t exp_hist_positive_count;
+    int32_t exp_hist_negative_offset;
+    uint64_t *exp_hist_negative_buckets;
+    size_t exp_hist_negative_count;
+    uint64_t exp_hist_count;
+    uint64_t exp_hist_sum;
+
     /* summary */
     int sum_quantiles_set;     /* specify if quantive values has been set */
     uint64_t *sum_quantiles;   /* 0, 0.25, 0.5, 0.75 and 1 */
@@ -46,11 +70,21 @@ struct cmt_metric {
 };
 
 void cmt_metric_set(struct cmt_metric *metric, uint64_t timestamp, double val);
+void cmt_metric_set_double(struct cmt_metric *metric, uint64_t timestamp, double val);
+void cmt_metric_set_int64(struct cmt_metric *metric, uint64_t timestamp, int64_t val);
+void cmt_metric_set_uint64(struct cmt_metric *metric, uint64_t timestamp, uint64_t val);
 void cmt_metric_inc(struct cmt_metric *metric, uint64_t timestamp);
 void cmt_metric_dec(struct cmt_metric *metric, uint64_t timestamp);
 void cmt_metric_add(struct cmt_metric *metric, uint64_t timestamp, double val);
 void cmt_metric_sub(struct cmt_metric *metric, uint64_t timestamp, double val);
 double cmt_metric_get_value(struct cmt_metric *metric);
+int cmt_metric_get_value_type(struct cmt_metric *metric);
+int64_t cmt_metric_get_int64_value(struct cmt_metric *metric);
+uint64_t cmt_metric_get_uint64_value(struct cmt_metric *metric);
+void cmt_metric_get_value_snapshot(struct cmt_metric *metric,
+                                   int *out_type,
+                                   int64_t *out_int64,
+                                   uint64_t *out_uint64);
 uint64_t cmt_metric_get_timestamp(struct cmt_metric *metric);
 
 void cmt_metric_hist_inc(struct cmt_metric *metric, uint64_t timestamp,

lib/cmetrics/include/cmetrics/cmt_mpack_utils.h

@@ -33,6 +33,7 @@ struct cmt_mpack_map_entry_callback_t {
 };
 
 int cmt_mpack_consume_double_tag(mpack_reader_t *reader, double *output_buffer);
+int cmt_mpack_consume_int_tag(mpack_reader_t *reader, int64_t *output_buffer);
 int cmt_mpack_consume_uint_tag(mpack_reader_t *reader, uint64_t *output_buffer);
 int cmt_mpack_consume_string_tag(mpack_reader_t *reader, cfl_sds_t *output_buffer);
 int cmt_mpack_unpack_map(mpack_reader_t *reader, 

lib/cmetrics/include/cmetrics/cmt_variant_utils.h

@@ -88,6 +88,14 @@ static inline int pack_cfl_variant_int64(mpack_writer_t *writer,
     return 0;
 }
 
+static inline int pack_cfl_variant_uint64(mpack_writer_t *writer,
+                                          uint64_t value)
+{
+    mpack_write_u64(writer, value);
+
+    return 0;
+}
+
 static inline int pack_cfl_variant_double(mpack_writer_t *writer,
                                           double value)
 {
@@ -96,6 +104,13 @@ static inline int pack_cfl_variant_double(mpack_writer_t *writer,
     return 0;
 }
 
+static inline int pack_cfl_variant_null(mpack_writer_t *writer)
+{
+    mpack_write_nil(writer);
+
+    return 0;
+}
+
 static inline int pack_cfl_variant_array(mpack_writer_t *writer,
                                          struct cfl_array *array)
 {
@@ -169,9 +184,15 @@ static inline int pack_cfl_variant(mpack_writer_t *writer,
     else if (value->type == CFL_VARIANT_INT) {
         result = pack_cfl_variant_int64(writer, value->data.as_int64);
     }
+    else if (value->type == CFL_VARIANT_UINT) {
+        result = pack_cfl_variant_uint64(writer, value->data.as_uint64);
+    }
     else if (value->type == CFL_VARIANT_DOUBLE) {
         result = pack_cfl_variant_double(writer, value->data.as_double);
     }
+    else if (value->type == CFL_VARIANT_NULL) {
+        result = pack_cfl_variant_null(writer);
+    }
     else if (value->type == CFL_VARIANT_ARRAY) {
         result = pack_cfl_variant_array(writer, value->data.as_array);
     }
@@ -552,6 +573,27 @@ static inline int unpack_cfl_variant_double(mpack_reader_t *reader,
     return 0;
 }
 
+static inline int unpack_cfl_variant_null(mpack_reader_t *reader,
+                                          struct cfl_variant **value)
+{
+    mpack_tag_t tag;
+    int         result;
+
+    result = unpack_cfl_variant_read_tag(reader, &tag, mpack_type_nil);
+
+    if (result != 0) {
+        return result;
+    }
+
+    *value = cfl_variant_create_from_null();
+
+    if (*value == NULL) {
+        return -3;
+    }
+
+    return 0;
+}
+
 static inline int unpack_cfl_variant_array(mpack_reader_t *reader,
                                            struct cfl_variant **value)
 {
@@ -624,6 +666,9 @@ static inline int unpack_cfl_variant(mpack_reader_t *reader,
     else if (value_type == mpack_type_double) {
         result = unpack_cfl_variant_double(reader, value);
     }
+    else if (value_type == mpack_type_nil) {
+        result = unpack_cfl_variant_null(reader, value);
+    }
     else if (value_type == mpack_type_array) {
         result = unpack_cfl_variant_array(reader, value);
     }

lib/cmetrics/src/CMakeLists.txt

@@ -14,6 +14,7 @@ set(src
   cmt_untyped.c
   cmt_summary.c
   cmt_histogram.c
+  cmt_exp_histogram.c
   cmt_metric.c
   cmt_metric_histogram.c
   cmt_map.c
@@ -76,6 +77,9 @@ endif()
 # Static Library
 add_library(cmetrics-static STATIC ${src})
 target_link_libraries(cmetrics-static mpack-static cfl-static fluent-otel-proto)
+if(NOT MSVC)
+  target_link_libraries(cmetrics-static m)
+endif()
 
 # Install Library
 if(MSVC)

lib/cmetrics/src/cmetrics.c

@@ -25,6 +25,7 @@
 #include <cmetrics/cmt_gauge.h>
 #include <cmetrics/cmt_summary.h>
 #include <cmetrics/cmt_histogram.h>
+#include <cmetrics/cmt_exp_histogram.h>
 #include <cmetrics/cmt_untyped.h>
 #include <cmetrics/cmt_atomic.h>
 #include <cmetrics/cmt_compat.h>
@@ -76,6 +77,7 @@ struct cmt *cmt_create()
     cfl_list_init(&cmt->counters);
     cfl_list_init(&cmt->gauges);
     cfl_list_init(&cmt->histograms);
+    cfl_list_init(&cmt->exp_histograms);
     cfl_list_init(&cmt->summaries);
     cfl_list_init(&cmt->untypeds);
 
@@ -94,6 +96,7 @@ void cmt_destroy(struct cmt *cmt)
     struct cmt_gauge *g;
     struct cmt_summary *s;
     struct cmt_histogram *h;
+    struct cmt_exp_histogram *eh;
     struct cmt_untyped *u;
 
     cfl_list_foreach_safe(head, tmp, &cmt->counters) {
@@ -116,6 +119,11 @@ void cmt_destroy(struct cmt *cmt)
         cmt_histogram_destroy(h);
     }
 
+    cfl_list_foreach_safe(head, tmp, &cmt->exp_histograms) {
+        eh = cfl_list_entry(head, struct cmt_exp_histogram, _head);
+        cmt_exp_histogram_destroy(eh);
+    }
+
     cfl_list_foreach_safe(head, tmp, &cmt->untypeds) {
         u = cfl_list_entry(head, struct cmt_untyped, _head);
         cmt_untyped_destroy(u);