/*

 * librdkafka - Apache Kafka C library

 *

 * Copyright (c) 2023, Confluent Inc.

 * All rights reserved.

 *

 * Redistribution and use in source and binary forms, with or without

 * modification, are permitted provided that the following conditions are met:

 *

 * 1. Redistributions of source code must retain the above copyright notice,

 *    this list of conditions and the following disclaimer.

 * 2. Redistributions in binary form must reproduce the above copyright notice,

 *    this list of conditions and the following disclaimer in the documentation

 *    and/or other materials provided with the distribution.

 *

 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"

 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE

 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE

 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE

 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR

 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF

 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS

 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN

 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)

 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE

 * POSSIBILITY OF SUCH DAMAGE.

 */

#include "rdkafka_telemetry_encode.h"

#include "nanopb/pb_encode.h"

#include "opentelemetry/metrics.pb.h"

#define THREE_ORDERS_MAGNITUDE 1000

typedef struct {

        opentelemetry_proto_metrics_v1_Metric **metrics;

        size_t count;

} rd_kafka_telemetry_metrics_repeated_t;

typedef struct {

        opentelemetry_proto_common_v1_KeyValue **key_values;

        size_t count;

} rd_kafka_telemetry_key_values_repeated_t;

#define calculate_avg(_avg_, _scale_factor_)                                   \

        ((_avg_).ra_v.avg / (double)_scale_factor_)

#define calculate_max(_avg_, _scale_factor_)                                   \

        RD_CEIL_INTEGER_DIVISION((_avg_).ra_v.maxv, _scale_factor_)

#define brokers_avg(_rk_, _avg_name_, _scale_factor_, _metric_)                \

        do {                                                                   \

                rd_kafka_broker_t *_rkb_;                                      \

                double avg = 0;                                                \

                int count  = 0;                                                \

                TAILQ_FOREACH(_rkb_, &(_rk_)->rk_brokers, rkb_link) {          \

                        rd_avg_t *rd_avg_rollover =                            \

                            &_rkb_->rkb_telemetry.rd_avg_rollover._avg_name_;  \

                        if (rd_avg_rollover->ra_v.cnt) {                       \

                                avg = (avg * count +                           \

                                       rd_avg_rollover->ra_v.sum) /            \

                                      (double)(count +                         \

                                               rd_avg_rollover->ra_v.cnt);     \

                                count += rd_avg_rollover->ra_v.cnt;            \

                        }                                                      \

                }                                                              \

                if (_scale_factor_ > 1)                                        \

                        (_metric_).double_value = avg / _scale_factor_;        \

                else                                                           \

                        (_metric_).double_value = avg;                         \

        } while (0)

#define brokers_max(_rk_, _avg_name_, _scale_factor_, _metric_)                \

        do {                                                                   \

                rd_kafka_broker_t *_rkb_;                                      \

                _metric_.int_value = 0;                                        \

                TAILQ_FOREACH(_rkb_, &(_rk_)->rk_brokers, rkb_link) {          \

                        _metric_.int_value =                                   \

                            RD_MAX(_metric_.int_value,                         \

                                   _rkb_->rkb_telemetry.rd_avg_rollover        \

                                       ._avg_name_.ra_v.maxv);                 \

                }                                                              \

                if (_scale_factor_ > 1)                                        \

                        (_metric_).int_value = RD_CEIL_INTEGER_DIVISION(       \

                            (_metric_).int_value, _scale_factor_);             \

        } while (0)

static rd_kafka_telemetry_metric_value_t

calculate_connection_creation_total(rd_kafka_t *rk,

                                    rd_kafka_broker_t *rkb_selected,

                                    rd_ts_t now_ns) {

        rd_kafka_telemetry_metric_value_t total;

        rd_kafka_broker_t *rkb;

        total.int_value = 0;

        TAILQ_FOREACH(rkb, &rk->rk_brokers, rkb_link) {

                const int32_t connects = rd_atomic32_get(&rkb->rkb_c.connects);

                if (!rk->rk_telemetry.delta_temporality)

                        total.int_value += connects;

                else

                        total.int_value +=

                            connects -

                            rkb->rkb_telemetry.rkb_historic_c.connects;

        }

        return total;

}

static rd_kafka_telemetry_metric_value_t

calculate_connection_creation_rate(rd_kafka_t *rk,

                                   rd_kafka_broker_t *rkb_selected,

                                   rd_ts_t now_ns) {

        rd_kafka_telemetry_metric_value_t total;

        rd_kafka_broker_t *rkb;

        rd_ts_t ts_last = rk->rk_telemetry.rk_historic_c.ts_last;

        total.double_value = 0;

        TAILQ_FOREACH(rkb, &rk->rk_brokers, rkb_link) {

                total.double_value +=

                    rd_atomic32_get(&rkb->rkb_c.connects) -

                    rkb->rkb_telemetry.rkb_historic_c.connects;

        }

        double seconds = (now_ns - ts_last) / 1e9;

        if (seconds > 1.0)

                total.double_value /= seconds;

        return total;

}

static rd_kafka_telemetry_metric_value_t

calculate_broker_avg_rtt(rd_kafka_t *rk,

                         rd_kafka_broker_t *rkb_selected,

                         rd_ts_t now_ns) {

        rd_kafka_telemetry_metric_value_t avg_rtt = RD_ZERO_INIT;

        avg_rtt.double_value                      = calculate_avg(

            rkb_selected->rkb_telemetry.rd_avg_rollover.rkb_avg_rtt,

            THREE_ORDERS_MAGNITUDE);

        return avg_rtt;

}

static rd_kafka_telemetry_metric_value_t

calculate_broker_max_rtt(rd_kafka_t *rk,

                         rd_kafka_broker_t *rkb_selected,

                         rd_ts_t now_ns) {

        rd_kafka_telemetry_metric_value_t max_rtt = RD_ZERO_INIT;

        max_rtt.int_value                         = calculate_max(

            rkb_selected->rkb_telemetry.rd_avg_rollover.rkb_avg_rtt,

            THREE_ORDERS_MAGNITUDE);

        return max_rtt;

}

static rd_kafka_telemetry_metric_value_t

calculate_produce_latency_avg(rd_kafka_t *rk,

                              rd_kafka_broker_t *rkb_selected,

                              rd_ts_t now_ns) {

        rd_kafka_telemetry_metric_value_t avg_rtt = RD_ZERO_INIT;

        brokers_avg(rk, rkb_avg_produce_latency, THREE_ORDERS_MAGNITUDE,

                    avg_rtt);

        return avg_rtt;

}

static rd_kafka_telemetry_metric_value_t

calculate_produce_latency_max(rd_kafka_t *rk,

                              rd_kafka_broker_t *rkb_selected,

                              rd_ts_t now_ns) {

        rd_kafka_telemetry_metric_value_t max_rtt = RD_ZERO_INIT;

        brokers_max(rk, rkb_avg_produce_latency, THREE_ORDERS_MAGNITUDE,

                    max_rtt);

        return max_rtt;

}

static rd_kafka_telemetry_metric_value_t

calculate_throttle_avg(rd_kafka_t *rk,

                       rd_kafka_broker_t *rkb_selected,

                       rd_ts_t now_ns) {

        rd_kafka_telemetry_metric_value_t avg_throttle;

        brokers_avg(rk, rkb_avg_throttle, 1, avg_throttle);

        return avg_throttle;

}

static rd_kafka_telemetry_metric_value_t

calculate_throttle_max(rd_kafka_t *rk,

                       rd_kafka_broker_t *rkb_selected,

                       rd_ts_t now_ns) {

        rd_kafka_telemetry_metric_value_t max_throttle;

        brokers_max(rk, rkb_avg_throttle, 1, max_throttle);

        return max_throttle;

}

static rd_kafka_telemetry_metric_value_t

calculate_queue_time_avg(rd_kafka_t *rk,

                         rd_kafka_broker_t *rkb_selected,

                         rd_ts_t now_ns) {

        rd_kafka_telemetry_metric_value_t avg_queue_time;

        brokers_avg(rk, rkb_avg_outbuf_latency, THREE_ORDERS_MAGNITUDE,

                    avg_queue_time);

        return avg_queue_time;

}

static rd_kafka_telemetry_metric_value_t

calculate_queue_time_max(rd_kafka_t *rk,

                         rd_kafka_broker_t *rkb_selected,

                         rd_ts_t now_ns) {

        rd_kafka_telemetry_metric_value_t max_queue_time;

        brokers_max(rk, rkb_avg_outbuf_latency, THREE_ORDERS_MAGNITUDE,

                    max_queue_time);

        return max_queue_time;

}

static rd_kafka_telemetry_metric_value_t

calculate_consumer_assigned_partitions(rd_kafka_t *rk,

                                       rd_kafka_broker_t *rkb_selected,

                                       rd_ts_t now_ns) {

        rd_kafka_telemetry_metric_value_t assigned_partitions;

        assigned_partitions.int_value =

            rk->rk_cgrp ? rk->rk_cgrp->rkcg_c.assignment_size : 0;

        return assigned_partitions;

}

static rd_kafka_telemetry_metric_value_t

calculate_consumer_rebalance_latency_avg(rd_kafka_t *rk,

                                         rd_kafka_broker_t *rkb_selected,

                                         rd_ts_t now_ns) {

        rd_kafka_telemetry_metric_value_t avg_rebalance_time;

        avg_rebalance_time.double_value = calculate_avg(

            rk->rk_telemetry.rd_avg_rollover.rk_avg_rebalance_latency,

            THREE_ORDERS_MAGNITUDE);

        return avg_rebalance_time;

}

static rd_kafka_telemetry_metric_value_t

calculate_consumer_rebalance_latency_max(rd_kafka_t *rk,

                                         rd_kafka_broker_t *rkb_selected,

                                         rd_ts_t now_ns) {

        rd_kafka_telemetry_metric_value_t max_rebalance_time;

        max_rebalance_time.int_value = calculate_max(

            rk->rk_telemetry.rd_avg_rollover.rk_avg_rebalance_latency,

            THREE_ORDERS_MAGNITUDE);

        return max_rebalance_time;

}

static rd_kafka_telemetry_metric_value_t

calculate_consumer_rebalance_latency_total(rd_kafka_t *rk,

                                           rd_kafka_broker_t *rkb_selected,

                                           rd_ts_t now_ns) {

        rd_kafka_telemetry_metric_value_t total_rebalance_time;

        total_rebalance_time.int_value = RD_CEIL_INTEGER_DIVISION(

            rk->rk_telemetry.rd_avg_rollover.rk_avg_rebalance_latency.ra_v.sum,

            THREE_ORDERS_MAGNITUDE);

        if (!rk->rk_telemetry.delta_temporality) {

                total_rebalance_time.int_value +=

                    rk->rk_telemetry.rk_historic_c.rebalance_latency_total;

        }

        return total_rebalance_time;

}

static rd_kafka_telemetry_metric_value_t

calculate_consumer_fetch_latency_avg(rd_kafka_t *rk,

                                     rd_kafka_broker_t *rkb_selected,

                                     rd_ts_t now_ns) {

        rd_kafka_telemetry_metric_value_t avg_fetch_time;

        brokers_avg(rk, rkb_avg_fetch_latency, THREE_ORDERS_MAGNITUDE,

                    avg_fetch_time);

        return avg_fetch_time;

}

static rd_kafka_telemetry_metric_value_t

calculate_consumer_fetch_latency_max(rd_kafka_t *rk,

                                     rd_kafka_broker_t *rkb_selected,

                                     rd_ts_t now_ns) {

        rd_kafka_telemetry_metric_value_t max_fetch_time;

        brokers_max(rk, rkb_avg_fetch_latency, THREE_ORDERS_MAGNITUDE,

                    max_fetch_time);

        return max_fetch_time;

}

static rd_kafka_telemetry_metric_value_t

calculate_consumer_poll_idle_ratio_avg(rd_kafka_t *rk,

                                       rd_kafka_broker_t *rkb_selected,

                                       rd_ts_t now_ns) {

        rd_kafka_telemetry_metric_value_t avg_poll_idle_avg;

        avg_poll_idle_avg.double_value = calculate_avg(

            rk->rk_telemetry.rd_avg_rollover.rk_avg_poll_idle_ratio, 1e6);

        return avg_poll_idle_avg;

}

static rd_kafka_telemetry_metric_value_t

calculate_consumer_commit_latency_avg(rd_kafka_t *rk,

                                      rd_kafka_broker_t *rkb_selected,

                                      rd_ts_t now_ns) {

        rd_kafka_telemetry_metric_value_t avg_commit_time;

        avg_commit_time.double_value = calculate_avg(

            rk->rk_telemetry.rd_avg_rollover.rk_avg_commit_latency,

            THREE_ORDERS_MAGNITUDE);

        return avg_commit_time;

}

static rd_kafka_telemetry_metric_value_t

calculate_consumer_commit_latency_max(rd_kafka_t *rk,

                                      rd_kafka_broker_t *rkb_selected,

                                      rd_ts_t now_ns) {

        rd_kafka_telemetry_metric_value_t max_commit_time;

        max_commit_time.int_value = calculate_max(

            rk->rk_telemetry.rd_avg_rollover.rk_avg_commit_latency,

            THREE_ORDERS_MAGNITUDE);

        return max_commit_time;

}

static void reset_historical_metrics(rd_kafka_t *rk, rd_ts_t now_ns) {

        rd_kafka_broker_t *rkb;

        rk->rk_telemetry.rk_historic_c.ts_last = now_ns;

        rk->rk_telemetry.rk_historic_c.rebalance_latency_total +=

            RD_CEIL_INTEGER_DIVISION(rk->rk_telemetry.rd_avg_rollover

                                         .rk_avg_rebalance_latency.ra_v.sum,

                                     THREE_ORDERS_MAGNITUDE);

        TAILQ_FOREACH(rkb, &rk->rk_brokers, rkb_link) {

                rkb->rkb_telemetry.rkb_historic_c.connects =

                    rd_atomic32_get(&rkb->rkb_c.connects);

        }

}

static const rd_kafka_telemetry_metric_value_calculator_t

    PRODUCER_METRIC_VALUE_CALCULATORS[RD_KAFKA_TELEMETRY_PRODUCER_METRIC__CNT] =

        {

            [RD_KAFKA_TELEMETRY_METRIC_PRODUCER_CONNECTION_CREATION_RATE] =

                &calculate_connection_creation_rate,

            [RD_KAFKA_TELEMETRY_METRIC_PRODUCER_CONNECTION_CREATION_TOTAL] =

                &calculate_connection_creation_total,

            [RD_KAFKA_TELEMETRY_METRIC_PRODUCER_NODE_REQUEST_LATENCY_AVG] =

                &calculate_broker_avg_rtt,

            [RD_KAFKA_TELEMETRY_METRIC_PRODUCER_NODE_REQUEST_LATENCY_MAX] =

                &calculate_broker_max_rtt,

            [RD_KAFKA_TELEMETRY_METRIC_PRODUCER_PRODUCE_THROTTLE_TIME_AVG] =

                &calculate_throttle_avg,

            [RD_KAFKA_TELEMETRY_METRIC_PRODUCER_PRODUCE_THROTTLE_TIME_MAX] =

                &calculate_throttle_max,

            [RD_KAFKA_TELEMETRY_METRIC_PRODUCER_RECORD_QUEUE_TIME_AVG] =

                &calculate_queue_time_avg,

            [RD_KAFKA_TELEMETRY_METRIC_PRODUCER_RECORD_QUEUE_TIME_MAX] =

                &calculate_queue_time_max,

            [RD_KAFKA_TELEMETRY_METRIC_PRODUCER_PRODUCE_LATENCY_AVG] =

                &calculate_produce_latency_avg,

            [RD_KAFKA_TELEMETRY_METRIC_PRODUCER_PRODUCE_LATENCY_MAX] =

                &calculate_produce_latency_max,

};

static const rd_kafka_telemetry_metric_value_calculator_t

    CONSUMER_METRIC_VALUE_CALCULATORS[RD_KAFKA_TELEMETRY_CONSUMER_METRIC__CNT] = {

        [RD_KAFKA_TELEMETRY_METRIC_CONSUMER_CONNECTION_CREATION_RATE] =

            &calculate_connection_creation_rate,

        [RD_KAFKA_TELEMETRY_METRIC_CONSUMER_CONNECTION_CREATION_TOTAL] =

            &calculate_connection_creation_total,

        [RD_KAFKA_TELEMETRY_METRIC_CONSUMER_NODE_REQUEST_LATENCY_AVG] =

            &calculate_broker_avg_rtt,

        [RD_KAFKA_TELEMETRY_METRIC_CONSUMER_NODE_REQUEST_LATENCY_MAX] =

            &calculate_broker_max_rtt,

        [RD_KAFKA_TELEMETRY_METRIC_CONSUMER_COORDINATOR_ASSIGNED_PARTITIONS] =

            &calculate_consumer_assigned_partitions,

        [RD_KAFKA_TELEMETRY_METRIC_CONSUMER_COORDINATOR_REBALANCE_LATENCY_AVG] =

            &calculate_consumer_rebalance_latency_avg,

        [RD_KAFKA_TELEMETRY_METRIC_CONSUMER_COORDINATOR_REBALANCE_LATENCY_MAX] =

            &calculate_consumer_rebalance_latency_max,

        [RD_KAFKA_TELEMETRY_METRIC_CONSUMER_COORDINATOR_REBALANCE_LATENCY_TOTAL] =

            &calculate_consumer_rebalance_latency_total,

        [RD_KAFKA_TELEMETRY_METRIC_CONSUMER_FETCH_MANAGER_FETCH_LATENCY_AVG] =

            &calculate_consumer_fetch_latency_avg,

        [RD_KAFKA_TELEMETRY_METRIC_CONSUMER_FETCH_MANAGER_FETCH_LATENCY_MAX] =

            &calculate_consumer_fetch_latency_max,

        [RD_KAFKA_TELEMETRY_METRIC_CONSUMER_POLL_IDLE_RATIO_AVG] =

            &calculate_consumer_poll_idle_ratio_avg,

        [RD_KAFKA_TELEMETRY_METRIC_CONSUMER_COORDINATOR_COMMIT_LATENCY_AVG] =

            &calculate_consumer_commit_latency_avg,

        [RD_KAFKA_TELEMETRY_METRIC_CONSUMER_COORDINATOR_COMMIT_LATENCY_MAX] =

            &calculate_consumer_commit_latency_max,

};

static const char *get_client_rack(const rd_kafka_t *rk) {

        return rk->rk_conf.client_rack &&

                       RD_KAFKAP_STR_LEN(rk->rk_conf.client_rack)

                   ? (const char *)rk->rk_conf.client_rack->str

                   : NULL;

}

static const char *get_group_id(const rd_kafka_t *rk) {

        return rk->rk_conf.group_id_str ? (const char *)rk->rk_conf.group_id_str

                                        : NULL;

}

static const char *get_group_instance_id(const rd_kafka_t *rk) {

        return rk->rk_conf.group_instance_id

                   ? (const char *)rk->rk_conf.group_instance_id

                   : NULL;

}

static const char *get_member_id(const rd_kafka_t *rk) {

        return rk->rk_cgrp && rk->rk_cgrp->rkcg_member_id &&

                       rk->rk_cgrp->rkcg_member_id->len > 0

                   ? (const char *)rk->rk_cgrp->rkcg_member_id->str

                   : NULL;

}

static const char *get_transactional_id(const rd_kafka_t *rk) {

        return rk->rk_conf.eos.transactional_id

                   ? (const char *)rk->rk_conf.eos.transactional_id

                   : NULL;

}

static const rd_kafka_telemetry_attribute_config_t producer_attributes[] = {

    {"client_rack", get_client_rack},

    {"transactional_id", get_transactional_id},

};

static const rd_kafka_telemetry_attribute_config_t consumer_attributes[] = {

    {"client_rack", get_client_rack},

    {"group_id", get_group_id},

    {"group_instance_id", get_group_instance_id},

    {"member_id", get_member_id},

};

static int

count_attributes(rd_kafka_t *rk,

                 const rd_kafka_telemetry_attribute_config_t *configs,

                 int config_count) {

        int count = 0, i;

        for (i = 0; i < config_count; ++i) {

                if (configs[i].getValue(rk)) {

                        count++;

                }

        }

        return count;

}

static void set_attributes(rd_kafka_t *rk,

                           rd_kafka_telemetry_resource_attribute_t *attributes,

                           const rd_kafka_telemetry_attribute_config_t *configs,

                           int config_count) {

        int attr_idx = 0, i;

        for (i = 0; i < config_count; ++i) {

                const char *value = configs[i].getValue(rk);

                if (value) {

                        attributes[attr_idx].name  = configs[i].name;

                        attributes[attr_idx].value = value;

                        attr_idx++;

                }

        }

}

static int

resource_attributes(rd_kafka_t *rk,

                    rd_kafka_telemetry_resource_attribute_t **attributes) {

        int count = 0;

        const rd_kafka_telemetry_attribute_config_t *configs;

        int config_count;

        if (rk->rk_type == RD_KAFKA_PRODUCER) {

                configs      = producer_attributes;

                config_count = RD_ARRAY_SIZE(producer_attributes);

        } else if (rk->rk_type == RD_KAFKA_CONSUMER) {

                configs      = consumer_attributes;

                config_count = RD_ARRAY_SIZE(consumer_attributes);

        } else {

                *attributes = NULL;

                return 0;

        }

        count = count_attributes(rk, configs, config_count);

        if (count == 0) {

                *attributes = NULL;

                return 0;

        }

        *attributes =

            rd_malloc(sizeof(rd_kafka_telemetry_resource_attribute_t) * count);

        set_attributes(rk, *attributes, configs, config_count);

        return count;

}

static bool

encode_string(pb_ostream_t *stream, const pb_field_t *field, void *const *arg) {

        if (!pb_encode_tag_for_field(stream, field))

                return false;

        return pb_encode_string(stream, (uint8_t *)(*arg), strlen(*arg));

}

// TODO: Update to handle multiple data points.

static bool encode_number_data_point(pb_ostream_t *stream,

                                     const pb_field_t *field,

                                     void *const *arg) {

        opentelemetry_proto_metrics_v1_NumberDataPoint *data_point =

            (opentelemetry_proto_metrics_v1_NumberDataPoint *)*arg;

        if (!pb_encode_tag_for_field(stream, field))

                return false;

        return pb_encode_submessage(

            stream, opentelemetry_proto_metrics_v1_NumberDataPoint_fields,

            data_point);

}

static bool

encode_metric(pb_ostream_t *stream, const pb_field_t *field, void *const *arg) {

        rd_kafka_telemetry_metrics_repeated_t *metricArr =

            (rd_kafka_telemetry_metrics_repeated_t *)*arg;

        size_t i;

        for (i = 0; i < metricArr->count; i++) {

                opentelemetry_proto_metrics_v1_Metric *metric =

                    metricArr->metrics[i];

                if (!pb_encode_tag_for_field(stream, field))

                        return false;

                if (!pb_encode_submessage(

                        stream, opentelemetry_proto_metrics_v1_Metric_fields,

                        metric))

                        return false;

        }

        return true;

}

static bool encode_scope_metrics(pb_ostream_t *stream,

                                 const pb_field_t *field,

                                 void *const *arg) {

        opentelemetry_proto_metrics_v1_ScopeMetrics *scope_metrics =

            (opentelemetry_proto_metrics_v1_ScopeMetrics *)*arg;

        if (!pb_encode_tag_for_field(stream, field))

                return false;

        return pb_encode_submessage(

            stream, opentelemetry_proto_metrics_v1_ScopeMetrics_fields,

            scope_metrics);

}

static bool encode_resource_metrics(pb_ostream_t *stream,

                                    const pb_field_t *field,

                                    void *const *arg) {

        opentelemetry_proto_metrics_v1_ResourceMetrics *resource_metrics =

            (opentelemetry_proto_metrics_v1_ResourceMetrics *)*arg;

        if (!pb_encode_tag_for_field(stream, field))

                return false;

        return pb_encode_submessage(

            stream, opentelemetry_proto_metrics_v1_ResourceMetrics_fields,

            resource_metrics);

}

static bool encode_key_value(pb_ostream_t *stream,

                             const pb_field_t *field,

                             void *const *arg) {

        if (!pb_encode_tag_for_field(stream, field))

                return false;

        opentelemetry_proto_common_v1_KeyValue *key_value =

            (opentelemetry_proto_common_v1_KeyValue *)*arg;

        return pb_encode_submessage(

            stream, opentelemetry_proto_common_v1_KeyValue_fields, key_value);

}

static bool encode_key_values(pb_ostream_t *stream,

                              const pb_field_t *field,

                              void *const *arg) {

        rd_kafka_telemetry_key_values_repeated_t *kv_arr =

            (rd_kafka_telemetry_key_values_repeated_t *)*arg;

        size_t i;

        for (i = 0; i < kv_arr->count; i++) {

                opentelemetry_proto_common_v1_KeyValue *kv =

                    kv_arr->key_values[i];

                if (!pb_encode_tag_for_field(stream, field))

                        return false;

                if (!pb_encode_submessage(

                        stream, opentelemetry_proto_common_v1_KeyValue_fields,

                        kv))

                        return false;

        }

        return true;

}

static void free_metrics(

    opentelemetry_proto_metrics_v1_Metric **metrics,

    char **metric_names,

    opentelemetry_proto_metrics_v1_NumberDataPoint **data_points,

    opentelemetry_proto_common_v1_KeyValue *datapoint_attributes_key_values,

    size_t count) {

        size_t i;

        for (i = 0; i < count; i++) {

                rd_free(data_points[i]);

                rd_free(metric_names[i]);

                rd_free(metrics[i]);

        }

        rd_free(data_points);

        rd_free(metric_names);

        rd_free(metrics);

        rd_free(datapoint_attributes_key_values);

}

static void free_resource_attributes(

    opentelemetry_proto_common_v1_KeyValue **resource_attributes_key_values,

    rd_kafka_telemetry_resource_attribute_t *resource_attributes_struct,

    size_t count) {

        size_t i;

        if (count == 0)

                return;

        for (i = 0; i < count; i++)

                rd_free(resource_attributes_key_values[i]);

        rd_free(resource_attributes_struct);

        rd_free(resource_attributes_key_values);

}

static void serialize_Metric(

    rd_kafka_t *rk,

    rd_kafka_broker_t *rkb,

    const rd_kafka_telemetry_metric_info_t *info,

    opentelemetry_proto_metrics_v1_Metric **metric,

    opentelemetry_proto_metrics_v1_NumberDataPoint **data_point,

    opentelemetry_proto_common_v1_KeyValue *data_point_attribute,

    rd_kafka_telemetry_metric_value_calculator_t metric_value_calculator,

    char **metric_name,

    bool is_per_broker,

    rd_ts_t now_ns) {

        rd_ts_t ts_last  = rk->rk_telemetry.rk_historic_c.ts_last,

                ts_start = rk->rk_telemetry.rk_historic_c.ts_start;

        size_t metric_name_len;

        if (info->is_int) {

                (*data_point)->which_value =

                    opentelemetry_proto_metrics_v1_NumberDataPoint_as_int_tag;

                (*data_point)->value.as_int =

                    metric_value_calculator(rk, rkb, now_ns).int_value;

        } else {

                (*data_point)->which_value =

                    opentelemetry_proto_metrics_v1_NumberDataPoint_as_double_tag;

                (*data_point)->value.as_double =

                    metric_value_calculator(rk, rkb, now_ns).double_value;

        }

        (*data_point)->time_unix_nano = now_ns;

        if (info->type == RD_KAFKA_TELEMETRY_METRIC_TYPE_GAUGE ||

            (info->type == RD_KAFKA_TELEMETRY_METRIC_TYPE_SUM &&

             rk->rk_telemetry.delta_temporality))

                (*data_point)->start_time_unix_nano = ts_last;

        else

                (*data_point)->start_time_unix_nano = ts_start;

        if (is_per_broker) {

                data_point_attribute->key.funcs.encode = &encode_string;

                data_point_attribute->key.arg =

                    RD_KAFKA_TELEMETRY_METRIC_NODE_ID_ATTRIBUTE;

                data_point_attribute->has_value = true;

                data_point_attribute->value.which_value =

                    opentelemetry_proto_common_v1_AnyValue_int_value_tag;

                rd_kafka_broker_lock(rkb);

                data_point_attribute->value.value.int_value = rkb->rkb_nodeid;

                rd_kafka_broker_unlock(rkb);

                (*data_point)->attributes.funcs.encode = &encode_key_value;

                (*data_point)->attributes.arg          = data_point_attribute;

        }

        switch (info->type) {

        case RD_KAFKA_TELEMETRY_METRIC_TYPE_SUM: {

                (*metric)->which_data =

                    opentelemetry_proto_metrics_v1_Metric_sum_tag;

                (*metric)->data.sum.data_points.funcs.encode =

                    &encode_number_data_point;

                (*metric)->data.sum.data_points.arg = *data_point;

                (*metric)->data.sum.aggregation_temporality =

                    rk->rk_telemetry.delta_temporality

                        ? opentelemetry_proto_metrics_v1_AggregationTemporality_AGGREGATION_TEMPORALITY_DELTA

                        : opentelemetry_proto_metrics_v1_AggregationTemporality_AGGREGATION_TEMPORALITY_CUMULATIVE;

                (*metric)->data.sum.is_monotonic = true;

                break;

        }

        case RD_KAFKA_TELEMETRY_METRIC_TYPE_GAUGE: {

                (*metric)->which_data =

                    opentelemetry_proto_metrics_v1_Metric_gauge_tag;

                (*metric)->data.gauge.data_points.funcs.encode =

                    &encode_number_data_point;

                (*metric)->data.gauge.data_points.arg = *data_point;

                break;

        }

        default:

                rd_assert(!"Unknown metric type");

                break;

        }

        (*metric)->description.funcs.encode = &encode_string;

        (*metric)->description.arg          = (void *)info->description;

        metric_name_len =

            strlen(RD_KAFKA_TELEMETRY_METRIC_PREFIX) + strlen(info->name) + 1;

        *metric_name = rd_calloc(1, metric_name_len);

        rd_snprintf(*metric_name, metric_name_len, "%s%s",

                    RD_KAFKA_TELEMETRY_METRIC_PREFIX, info->name);

        (*metric)->name.funcs.encode = &encode_string;

        (*metric)->name.arg          = *metric_name;

        /* Skipping unit as Java client does the same */

}

/**

 * @brief Encodes the metrics to opentelemetry_proto_metrics_v1_MetricsData and

 * returns the serialized data. Currently only supports encoding of connection

 * creation total by default

 *

 * @locks none

 * @locks_acquired rd_kafka_rdlock()

 * @locality main thread

 */

rd_buf_t *rd_kafka_telemetry_encode_metrics(rd_kafka_t *rk) {

        rd_buf_t *rbuf = NULL;

        rd_kafka_broker_t *rkb;

        size_t message_size;

        void *buffer = NULL;

        pb_ostream_t stream;

        bool status;

        char **metric_names;

        const int *metrics_to_encode = rk->rk_telemetry.matched_metrics;

        const size_t metrics_to_encode_count =

            rk->rk_telemetry.matched_metrics_cnt;

        const rd_kafka_telemetry_metric_info_t *info =

            RD_KAFKA_TELEMETRY_METRIC_INFO(rk);

        size_t total_metrics_count = metrics_to_encode_count;

        size_t i, metric_idx = 0;

        if (!metrics_to_encode_count)

                return rd_buf_new(1, 1);

        opentelemetry_proto_metrics_v1_MetricsData metrics_data =

            opentelemetry_proto_metrics_v1_MetricsData_init_zero;

        opentelemetry_proto_metrics_v1_ResourceMetrics resource_metrics =

            opentelemetry_proto_metrics_v1_ResourceMetrics_init_zero;

        opentelemetry_proto_metrics_v1_Metric **metrics;

        opentelemetry_proto_common_v1_KeyValue *

            *resource_attributes_key_values = NULL;

        opentelemetry_proto_common_v1_KeyValue

            *datapoint_attributes_key_values = NULL;

        opentelemetry_proto_metrics_v1_NumberDataPoint **data_points;

        rd_kafka_telemetry_metrics_repeated_t metrics_repeated;

        rd_kafka_telemetry_key_values_repeated_t resource_attributes_repeated;

        rd_kafka_telemetry_resource_attribute_t *resource_attributes_struct =

            NULL;

        rd_ts_t now_ns = rd_uclock() * 1000;

        rd_kafka_rdlock(rk);

        for (i = 0; i < metrics_to_encode_count; i++) {

                if (info[metrics_to_encode[i]].is_per_broker) {

                        total_metrics_count += rk->rk_broker_cnt.val - 1;

                }

        }

        rd_kafka_dbg(rk, TELEMETRY, "PUSH", "Serializing metrics");

        TAILQ_FOREACH(rkb, &rk->rk_brokers, rkb_link) {

                rd_avg_destroy(&rkb->rkb_telemetry.rd_avg_rollover.rkb_avg_rtt);

                rd_avg_rollover(&rkb->rkb_telemetry.rd_avg_rollover.rkb_avg_rtt,

                                &rkb->rkb_telemetry.rd_avg_current.rkb_avg_rtt);

                rd_avg_destroy(

                    &rkb->rkb_telemetry.rd_avg_rollover.rkb_avg_outbuf_latency);

                rd_avg_rollover(

                    &rkb->rkb_telemetry.rd_avg_rollover.rkb_avg_outbuf_latency,

                    &rkb->rkb_telemetry.rd_avg_current.rkb_avg_outbuf_latency);

                rd_avg_destroy(

                    &rkb->rkb_telemetry.rd_avg_rollover.rkb_avg_throttle);

                rd_avg_rollover(

                    &rkb->rkb_telemetry.rd_avg_rollover.rkb_avg_throttle,

                    &rkb->rkb_telemetry.rd_avg_current.rkb_avg_throttle);

                if (rk->rk_type == RD_KAFKA_CONSUMER) {

                        rd_avg_destroy(&rkb->rkb_telemetry.rd_avg_rollover

                                            .rkb_avg_fetch_latency);

                        rd_avg_rollover(&rkb->rkb_telemetry.rd_avg_rollover

                                             .rkb_avg_fetch_latency,

                                        &rkb->rkb_telemetry.rd_avg_current

                                             .rkb_avg_fetch_latency);

                } else if (rk->rk_type == RD_KAFKA_PRODUCER) {

                        rd_avg_destroy(&rkb->rkb_telemetry.rd_avg_rollover

                                            .rkb_avg_produce_latency);

                        rd_avg_rollover(&rkb->rkb_telemetry.rd_avg_rollover

                                             .rkb_avg_produce_latency,

                                        &rkb->rkb_telemetry.rd_avg_current

                                             .rkb_avg_produce_latency);

                }

        }

        if (rk->rk_type == RD_KAFKA_CONSUMER) {

                rd_avg_destroy(

                    &rk->rk_telemetry.rd_avg_rollover.rk_avg_poll_idle_ratio);

                rd_avg_rollover(

                    &rk->rk_telemetry.rd_avg_rollover.rk_avg_poll_idle_ratio,

                    &rk->rk_telemetry.rd_avg_current.rk_avg_poll_idle_ratio);

                rd_avg_destroy(

                    &rk->rk_telemetry.rd_avg_rollover.rk_avg_rebalance_latency);

                rd_avg_rollover(

                    &rk->rk_telemetry.rd_avg_rollover.rk_avg_rebalance_latency,

                    &rk->rk_telemetry.rd_avg_current.rk_avg_rebalance_latency);

                rd_avg_destroy(

                    &rk->rk_telemetry.rd_avg_rollover.rk_avg_commit_latency);

                rd_avg_rollover(

                    &rk->rk_telemetry.rd_avg_rollover.rk_avg_commit_latency,

                    &rk->rk_telemetry.rd_avg_current.rk_avg_commit_latency);

        }

        int resource_attributes_count =

            resource_attributes(rk, &resource_attributes_struct);

        rd_kafka_dbg(rk, TELEMETRY, "PUSH", "Resource attributes count: %d",

                     resource_attributes_count);

        if (resource_attributes_count > 0) {

                resource_attributes_key_values =

                    rd_malloc(sizeof(opentelemetry_proto_common_v1_KeyValue *) *

                              resource_attributes_count);

                int ind;

                for (ind = 0; ind < resource_attributes_count; ++ind) {

                        resource_attributes_key_values[ind] = rd_calloc(

                            1, sizeof(opentelemetry_proto_common_v1_KeyValue));

                        resource_attributes_key_values[ind]->key.funcs.encode =

                            &encode_string;

                        resource_attributes_key_values[ind]->key.arg =

                            (void *)resource_attributes_struct[ind].name;

                        resource_attributes_key_values[ind]->has_value = true;

                        resource_attributes_key_values[ind]->value.which_value =

                            opentelemetry_proto_common_v1_AnyValue_string_value_tag;

                        resource_attributes_key_values[ind]

                            ->value.value.string_value.funcs.encode =

                            &encode_string;

                        resource_attributes_key_values[ind]

                            ->value.value.string_value.arg =

                            (void *)resource_attributes_struct[ind].value;

                }

                resource_attributes_repeated.key_values =

                    resource_attributes_key_values;

                resource_attributes_repeated.count = resource_attributes_count;

                resource_metrics.has_resource      = true;

                resource_metrics.resource.attributes.funcs.encode =

                    &encode_key_values;

                resource_metrics.resource.attributes.arg =

                    &resource_attributes_repeated;

        }

        opentelemetry_proto_metrics_v1_ScopeMetrics scope_metrics =

            opentelemetry_proto_metrics_v1_ScopeMetrics_init_zero;

        opentelemetry_proto_common_v1_InstrumentationScope

            instrumentation_scope =

                opentelemetry_proto_common_v1_InstrumentationScope_init_zero;

        instrumentation_scope.name.funcs.encode    = &encode_string;

        instrumentation_scope.name.arg             = (void *)rd_kafka_name(rk);

        instrumentation_scope.version.funcs.encode = &encode_string;

        instrumentation_scope.version.arg = (void *)rd_kafka_version_str();

        scope_metrics.has_scope = true;

        scope_metrics.scope     = instrumentation_scope;

        metrics = rd_malloc(sizeof(opentelemetry_proto_metrics_v1_Metric *) *

                            total_metrics_count);

        data_points =

            rd_malloc(sizeof(opentelemetry_proto_metrics_v1_NumberDataPoint *) *

                      total_metrics_count);

        datapoint_attributes_key_values =

            rd_malloc(sizeof(opentelemetry_proto_common_v1_KeyValue) *

                      total_metrics_count);

        metric_names = rd_malloc(sizeof(char *) * total_metrics_count);

        rd_kafka_dbg(rk, TELEMETRY, "PUSH",

                     "Total metrics to be encoded count: %" PRIusz,

                     total_metrics_count);

        for (i = 0; i < metrics_to_encode_count; i++) {

                rd_kafka_telemetry_metric_value_calculator_t

                    metric_value_calculator =

                        (rk->rk_type == RD_KAFKA_PRODUCER)

                            ? PRODUCER_METRIC_VALUE_CALCULATORS

                                  [metrics_to_encode[i]]

                            : CONSUMER_METRIC_VALUE_CALCULATORS

                                  [metrics_to_encode[i]];

                if (info[metrics_to_encode[i]].is_per_broker) {

                        rd_kafka_broker_t *rkb;

                        TAILQ_FOREACH(rkb, &rk->rk_brokers, rkb_link) {

                                metrics[metric_idx] = rd_calloc(

                                    1,

                                    sizeof(

                                        opentelemetry_proto_metrics_v1_Metric));

                                data_points[metric_idx] = rd_calloc(

                                    1,

                                    sizeof(

                                        opentelemetry_proto_metrics_v1_NumberDataPoint));

                                serialize_Metric(

                                    rk, rkb, &info[metrics_to_encode[i]],

                                    &metrics[metric_idx],

                                    &data_points[metric_idx],

                                    &datapoint_attributes_key_values

                                        [metric_idx],