KIP 714 with compression support #4721

anchitj · 2024-05-17T06:31:06Z

No description provided.

…data Response.

* WIP:Push telemetry is being scheduled now. * Push * Compliation Fix * Working * Use UUID in PUSH * Remove fprintf * Changes * Fix CONFIGURATION.md * Fix size * Update s2i bounds --------- Co-authored-by: Milind L <[email protected]>

* Add broker selection and client termination * Address review comments

* Serialise metrics using nanopb * Move nanopb and opentelemetry inside src * Add metrics.options file * Remove unused includes * Style fix * Fix formatting * Skip copyright check * Add nanopb and opentelemetry in windows vcxproj * Include headers directories in CMAKE * Use flexver with PushTelemetry * Fix memory leaks * Change import path * Use rd_bool_t everywhere * Fix librdkafka.vcxproj * Use rd_bool_t * PR Feedback * Add nanopb license * Include opentelemtry license

* Support for delta temporality * Style fix * Fix bugs * Fix memory leaks and formatting * Fixes * PR Feedback

* Add telemetry encode and decode unit tests * Style fix * Improve test * PR Feedback

* Add max telemetry bytes * Clear telemetry_max_bytes * PR comments

…master_merge

…master_merge_with_compression

emasab

First comments

src/rdkafka_telemetry_decode.c

src/rdkafka_int.h

src/rdkafka_request.c

src/rdkafka_telemetry_decode.c

src/rdkafka_telemetry.c

src/rdkafka.h

src/rdkafka_telemetry_decode.c

tests/0009-mock_cluster.c

…master_merge_with_compression

emasab

First the segfault to fix

src/rdkafka_telemetry_encode.c

emasab · 2024-06-27T12:28:27Z

src/rdkafka_telemetry_encode.h

+    *rd_kafka_telemetry_metric_value_calculator_t)(rd_kafka_t *,
+                                                   rd_kafka_broker_t *);


Let's pass the timestamp so we use a uniform timestamp across all metrics.
Also please add the param names even if not required they're useful for clarity.

Suggested change

*rd_kafka_telemetry_metric_value_calculator_t)(rd_kafka_t *,

rd_kafka_broker_t *);

typedef rd_kafka_telemetry_metric_value_t (

*rd_kafka_telemetry_metric_value_calculator_t)(

rd_kafka_t *rk,

rd_kafka_broker_t *rkb_selected,

rd_ts_t now_nanos);

emasab · 2024-06-27T12:29:35Z

src/rdkafka_telemetry_encode.c

+        if (rk->rk_type == RD_KAFKA_PRODUCER &&
+            (metric_idx ==
+                 RD_KAFKA_TELEMETRY_METRIC_PRODUCER_NODE_REQUEST_LATENCY_AVG ||
+             metric_idx ==
+                 RD_KAFKA_TELEMETRY_METRIC_PRODUCER_NODE_REQUEST_LATENCY_MAX)) {
+                return true;
+        }
+        if (rk->rk_type == RD_KAFKA_CONSUMER &&
+            (metric_idx ==
+                 RD_KAFKA_TELEMETRY_METRIC_CONSUMER_NODE_REQUEST_LATENCY_AVG ||
+             metric_idx ==
+                 RD_KAFKA_TELEMETRY_METRIC_CONSUMER_NODE_REQUEST_LATENCY_MAX)) {
+                return true;
+        }
+        return false;


This can be added to the metric definition

Suggested change

if (rk->rk_type == RD_KAFKA_PRODUCER &&

(metric_idx ==

RD_KAFKA_TELEMETRY_METRIC_PRODUCER_NODE_REQUEST_LATENCY_AVG ||

metric_idx ==

RD_KAFKA_TELEMETRY_METRIC_PRODUCER_NODE_REQUEST_LATENCY_MAX)) {

return true;

}

if (rk->rk_type == RD_KAFKA_CONSUMER &&

(metric_idx ==

RD_KAFKA_TELEMETRY_METRIC_CONSUMER_NODE_REQUEST_LATENCY_AVG ||

metric_idx ==

RD_KAFKA_TELEMETRY_METRIC_CONSUMER_NODE_REQUEST_LATENCY_MAX)) {

return true;

}

return false;

return info[metric_idx].is_per_broker;

Updated and also fixed the bug at the initial loop where total_metrics_count is being calculated. We were passing the loop index instead of metrics_to_encode[i]

src/rdkafka_telemetry_encode.c

emasab · 2024-06-27T12:58:57Z

src/rdkafka_telemetry_encode.c

+
+        total.doubleValue = 0;
+        TAILQ_FOREACH(rkb, &rk->rk_brokers, rkb_link) {
+                ts_last = rkb->rkb_c_historic.ts_last;


ts_last shoud be outside the loop and taken from rk:

Suggested change

ts_last = rkb->rkb_c_historic.ts_last;

ts_last = rk->rk_telemetry.rk_historic_c.ts_last;

src/rdkafka_telemetry_encode.c

emasab · 2024-06-27T13:57:09Z

src/rdkafka_telemetry_encode.c

+                            (cnt_diff * RDKAFKA_TELEMETRY_NS_TO_MS_FACTOR);
+                }
+        }
+        avg_throttle.intValue = sum_value / broker_count;


double_value instead of intValue, same for queue_time_avg. For the latter you have to divide by THREE_ORDERS_MAGNITUDE to go from µs to ms

Suggested change

avg_throttle.intValue = sum_value / broker_count;

avg_throttle.double_value = avg;

src/rdkafka_telemetry_encode.c

emasab · 2024-06-27T14:39:49Z

src/rdkafka_telemetry_encode.c

+
+        TAILQ_FOREACH(rkb, &rk->rk_brokers, rkb_link) {
+                rkb->rkb_c_historic.assigned_partitions = rkb->rkb_toppar_cnt;
+                rkb->rkb_c_historic.connects = rkb->rkb_c.connects.val;
+                rkb->rkb_c_historic.ts_last  = rd_uclock() * 1000;
+                rkb->rkb_c_historic.connects = rkb->rkb_c.connects.val;
+                /* Only ra_v is being used to keep track of the metrics */
+                rkb->rkb_c_historic.rkb_avg_rtt.ra_v = rkb->rkb_avg_rtt.ra_v;
+                rd_atomic32_set(&rkb->rkb_avg_rtt.ra_v.maxv_reset, 1);
+                rkb->rkb_c_historic.rkb_avg_throttle.ra_v =
+                    rkb->rkb_avg_throttle.ra_v;
+                rd_atomic32_set(&rkb->rkb_avg_throttle.ra_v.maxv_reset, 1);
+                rkb->rkb_c_historic.rkb_avg_outbuf_latency.ra_v =
+                    rkb->rkb_avg_outbuf_latency.ra_v;
+                rd_atomic32_set(&rkb->rkb_avg_outbuf_latency.ra_v.maxv_reset,
+                                1);
+        }


Here it just needs to set the ts_last in rk and the historic connect count.

Suggested change

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

rkb->rkb_c_historic.assigned_partitions = rkb->rkb_toppar_cnt;

rkb->rkb_c_historic.connects = rkb->rkb_c.connects.val;

rkb->rkb_c_historic.ts_last = rd_uclock() * 1000;

rkb->rkb_c_historic.connects = rkb->rkb_c.connects.val;

/* Only ra_v is being used to keep track of the metrics */

rkb->rkb_c_historic.rkb_avg_rtt.ra_v = rkb->rkb_avg_rtt.ra_v;

rd_atomic32_set(&rkb->rkb_avg_rtt.ra_v.maxv_reset, 1);

rkb->rkb_c_historic.rkb_avg_throttle.ra_v =

rkb->rkb_avg_throttle.ra_v;

rd_atomic32_set(&rkb->rkb_avg_throttle.ra_v.maxv_reset, 1);

rkb->rkb_c_historic.rkb_avg_outbuf_latency.ra_v =

rkb->rkb_avg_outbuf_latency.ra_v;

rd_atomic32_set(&rkb->rkb_avg_outbuf_latency.ra_v.maxv_reset,

1);

}

rk->rk_telemetry.rk_historic_c.ts_last = now_ns;

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

rkb->rkb_telemetry.rkb_historic_c.connects =

rd_atomic32_get(&rkb->rkb_c.connects);

}

emasab · 2024-06-27T14:49:44Z

src/rdkafka_telemetry_encode.c

+            NULL;
+        rd_ts_t now_ns = rd_uclock() * 1000;
+
+        int resource_attributes_count =


First thing it needs to do is to rollover the dedicated avgs, the ones used by the stats cb cannot be reused because they can be reset by the corresponding callback.
They must be incremented in the same places, for example:

rd_avg_add( &rkb->rkb_telemetry.rd_avg_current.rkb_avg_outbuf_latency, rkbuf->rkbuf_ts_sent - rkbuf->rkbuf_ts_enq);

Suggested change

int resource_attributes_count =

if (rk->rk_telemetry.rk_historic_c.ts_start == 0) {

first_push = rd_true;

rk->rk_telemetry.rk_historic_c.ts_start = now_ns;

rk->rk_telemetry.rk_historic_c.ts_last = now_ns;

}

i = 0;

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

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

&rkb->rkb_telemetry.rd_avg_current.rkb_avg_rtt);

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_rollover(

&rkb->rkb_telemetry.rd_avg_rollover.rkb_avg_throttle,

&rkb->rkb_telemetry.rd_avg_current.rkb_avg_throttle);

if (first_push) {

rkb->rkb_telemetry.rkb_historic_c.connects =

rd_atomic32_get(&rkb->rkb_c.connects);

}

i++;

}

int resource_attributes_count =

src/rdkafka_telemetry_encode.c

emasab · 2024-06-27T14:56:26Z

src/rdkafka_telemetry_encode.c

+        for (i = 0; i < metrics_to_encode_count; i++) {
+                if (is_per_broker_metric(rk, (int)i)) {
+                        total_metrics_count += rk->rk_broker_cnt.val - 1;
+                }
+        }


This goes inside the read lock, just before Serializing metrics

src/rdkafka_telemetry_encode.c

emasab

Comments about struct changes, compression functions and ApiVersion

emasab · 2024-06-28T15:39:43Z

src/rdkafka_int.h

+                /* TODO: Use rd_list_t to store the metrics */
+                int *matched_metrics;
+                size_t matched_metrics_cnt;
+        } rk_telemetry;


Here you can store ts_start and ts_last

Suggested change

} rk_telemetry;

struct {

rd_ts_t ts_last; /**< Timestamp of last push */

rd_ts_t ts_start; /**< Timestamp from when collection

* started */

} rk_historic_c;

} rk_telemetry;

emasab · 2024-06-28T15:46:40Z

src/rdkafka_broker.h

+        struct {
+                rd_ts_t ts_last;  /**< Timestamp of last push */
+                rd_ts_t ts_start; /**< Timestamp from when collection started */
+                int32_t assigned_partitions;     /**< Current number of assigned
+                                                   partitions. */
+                int32_t connects;                /**< Connection attempts,
+                                                  *   successful or not. */
+                rd_avg_t rkb_avg_rtt;            /* Current RTT period */
+                rd_avg_t rkb_avg_throttle;       /* Current throttle period */
+                rd_avg_t rkb_avg_outbuf_latency; /**< Current latency
+                                                  *   between buf_enq0
+                                                  *   and writing to socket
+                                                  */
+
+        } rkb_c_historic;


This is the struct I'm referring to in the other comments, only connects is left in historic_c for each broker and we have two sets of avg for increasing the avg and the rollover.

Suggested change

struct {

rd_ts_t ts_last; /**< Timestamp of last push */

rd_ts_t ts_start; /**< Timestamp from when collection started */

int32_t assigned_partitions; /**< Current number of assigned

partitions. */

int32_t connects; /**< Connection attempts,

* successful or not. */

rd_avg_t rkb_avg_rtt; /* Current RTT period */

rd_avg_t rkb_avg_throttle; /* Current throttle period */

rd_avg_t rkb_avg_outbuf_latency; /**< Current latency

* between buf_enq0

* and writing to socket

*/

} rkb_c_historic;

struct {

struct {

int32_t connects; /**< Connection attempts,

* successful or not. */

} rkb_historic_c;

struct {

rd_avg_t rkb_avg_rtt; /* Current RTT avg */

rd_avg_t rkb_avg_throttle; /* Current throttle avg */

rd_avg_t

rkb_avg_outbuf_latency; /**< Current latency

* between buf_enq0

* and writing to socket

*/

} rd_avg_current;

struct {

rd_avg_t rkb_avg_rtt; /**< Rolled over RTT avg */

rd_avg_t rkb_avg_throttle; /**< Rolled over throttle avg */

rd_avg_t

rkb_avg_outbuf_latency; /**< Rolled over outbuf

* latency avg */

} rd_avg_rollover;

} rkb_telemetry;

emasab · 2024-06-28T15:49:07Z

src/rdkafka_broker.c

-                                             int16_t maxver,
-                                             int *featuresp) {
+static int16_t
+rd_kafka_broker_ApiVersion_supported_implementation(rd_kafka_broker_t *rkb,


This can be rd_kafka_broker_ApiVersion_supported0, the function that takes all the parameters, without changing the naming convention for now.

Suggested change

rd_kafka_broker_ApiVersion_supported_implementation(rd_kafka_broker_t *rkb,

rd_kafka_broker_ApiVersion_supported0(rd_kafka_broker_t *rkb,

emasab · 2024-06-28T15:53:48Z

src/rdkafka_int.h

+ * @enum Telemetry States
+ */
+typedef enum {
+        /**< Initial state, awaiting telemetry broker to be assigned */


It's /** , because /**< is used when the documentation comes after the field

Suggested change

/**< Initial state, awaiting telemetry broker to be assigned */

/** Initial state, awaiting telemetry broker to be assigned */

emasab · 2024-06-28T16:01:31Z

src/rdkafka_int.h

+        /**< Client is being terminated and last PushTelemetry is scheduled to
+           send */


Adding an asterisk at the beginning of the line is not required by Doxygen format but it helps readability

Suggested change

/**< Client is being terminated and last PushTelemetry is scheduled to

send */

/**< Client is being terminated and last PushTelemetry is scheduled to

* send */

emasab · 2024-06-28T16:34:14Z

src/rdkafka_telemetry_encode.c

+                rd_kafka_dbg(rk, TELEMETRY, "RD_KAFKA_TELEMETRY_METRICS_INFO",
+                             "Encoding failed: %s", PB_GET_ERROR(&stream));
+                rd_free(buffer);
+                free_metrics(metrics, metric_names, data_points,
+                             datapoint_attributes_key_values,
+                             total_metrics_count);
+                free_resource_attributes(resource_attributes_key_values,
+                                         resource_attributes_struct,
+                                         resource_attributes_count);
+
+                return NULL;


Once you do rd_buf_t changes and add the label:

Suggested change

rd_kafka_dbg(rk, TELEMETRY, "RD_KAFKA_TELEMETRY_METRICS_INFO",

"Encoding failed: %s", PB_GET_ERROR(&stream));

rd_free(buffer);

free_metrics(metrics, metric_names, data_points,

datapoint_attributes_key_values,

total_metrics_count);

free_resource_attributes(resource_attributes_key_values,

resource_attributes_struct,

resource_attributes_count);

return NULL;

rd_kafka_dbg(rk, TELEMETRY, "PUSH", "Encoding failed: %s",

PB_GET_ERROR(&stream));

rd_buf_destroy_free(rbuf);

goto fail;

emasab · 2024-06-28T16:37:46Z

src/rdkafka_telemetry_encode.c

+        free_metrics(metrics, metric_names, data_points,
+                     datapoint_attributes_key_values, total_metrics_count);
+        free_resource_attributes(resource_attributes_key_values,
+                                 resource_attributes_struct,
+                                 resource_attributes_count);


Move this after reset_historical_metrics with a rd_kafka_rdunlock

Suggested change

free_metrics(metrics, metric_names, data_points,

datapoint_attributes_key_values, total_metrics_count);

free_resource_attributes(resource_attributes_key_values,

resource_attributes_struct,

resource_attributes_count);

free_metrics(metrics, metric_names, data_points,

datapoint_attributes_key_values, total_metrics_count);

free_resource_attributes(resource_attributes_key_values,

resource_attributes_struct,

resource_attributes_count);

rd_kafka_rdunlock(rk);

return rbuf;

fail:

free_metrics(metrics, metric_names, data_points,

datapoint_attributes_key_values, total_metrics_count);

free_resource_attributes(resource_attributes_key_values,

resource_attributes_struct,

resource_attributes_count);

rd_kafka_rdunlock(rk);

return NULL;

emasab · 2024-06-28T16:44:56Z

src/rdkafka_telemetry.c

+        size_t metrics_payload_size = 0, compressed_metrics_payload_size = 0;
+        void *metrics_payload =
+                 rd_kafka_telemetry_encode_metrics(rk, &metrics_payload_size),
+             *compressed_metrics_payload        = NULL;
+        rd_kafka_compression_t compression_used = RD_KAFKA_COMPRESSION_NONE;
+
+        if (rk->rk_telemetry.accepted_compression_types_cnt != 0) {
+                compression_used = rd_kafka_push_telemetry_payload_compress(
+                    rk, rkb, metrics_payload, metrics_payload_size,
+                    &compressed_metrics_payload,
+                    &compressed_metrics_payload_size);
+        } else {
+                rd_kafka_dbg(rk, TELEMETRY, "PUSHSENT",
+                             "No compression types accepted, sending "
+                             "uncompressed payload");
+                compressed_metrics_payload      = metrics_payload;
+                metrics_payload                 = NULL;
+                compressed_metrics_payload_size = metrics_payload_size;
+        }


We can pass the rd_buf_t for compression here.

Suggested change

size_t metrics_payload_size = 0, compressed_metrics_payload_size = 0;

void *metrics_payload =

rd_kafka_telemetry_encode_metrics(rk, &metrics_payload_size),

*compressed_metrics_payload = NULL;

rd_kafka_compression_t compression_used = RD_KAFKA_COMPRESSION_NONE;

if (rk->rk_telemetry.accepted_compression_types_cnt != 0) {

compression_used = rd_kafka_push_telemetry_payload_compress(

rk, rkb, metrics_payload, metrics_payload_size,

&compressed_metrics_payload,

&compressed_metrics_payload_size);

} else {

rd_kafka_dbg(rk, TELEMETRY, "PUSHSENT",

"No compression types accepted, sending "

"uncompressed payload");

compressed_metrics_payload = metrics_payload;

metrics_payload = NULL;

compressed_metrics_payload_size = metrics_payload_size;

}

rd_buf_t *metrics_payload = rd_kafka_telemetry_encode_metrics(rk);

size_t compressed_metrics_payload_size = 0;

void *compressed_metrics_payload = NULL;

rd_kafka_compression_t compression_used =

rd_kafka_push_telemetry_payload_compress(

rk, rkb, metrics_payload, &compressed_metrics_payload,

&compressed_metrics_payload_size);

if (compressed_metrics_payload_size >

(size_t)rk->rk_telemetry.telemetry_max_bytes) {

rd_kafka_log(rk, LOG_WARNING, "TELEMETRY",

"Metrics payload size %" PRIdsz

" exceeds telemetry_max_bytes %" PRId32

"specified by the broker.",

compressed_metrics_payload_size,

rk->rk_telemetry.telemetry_max_bytes);

}

emasab · 2024-06-28T16:46:37Z

src/rdkafka_telemetry.c

+            0, RD_KAFKA_REPLYQ(rk->rk_ops, 0), rd_kafka_handle_PushTelemetry,
+            NULL);
+
+        rd_free(compressed_metrics_payload);


We can free the rd_buf_t here and in case it was compressed, the compressed payload.

Suggested change

rd_free(compressed_metrics_payload);

rd_buf_destroy_free(metrics_payload);

if (compression_used != RD_KAFKA_COMPRESSION_NONE)

rd_free(compressed_metrics_payload);

emasab · 2024-06-28T16:54:26Z

src/rdkafka_telemetry.c

+static rd_kafka_compression_t
+rd_kafka_push_telemetry_payload_compress(rd_kafka_t *rk,
+                                         rd_kafka_broker_t *rkb,
+                                         void *payload,
+                                         size_t payload_len,
+                                         void **compressed_payload,
+                                         size_t *compressed_payload_size) {
+        rd_kafka_compression_t compression_used = RD_KAFKA_COMPRESSION_NONE;
+        int i, r = -1;
+
+        for (i = 0; i < (int)rk->rk_telemetry.accepted_compression_types_cnt;
+             i++) {
+                rd_kafka_compression_t compression_type =
+                    rk->rk_telemetry.accepted_compression_types[i];
+
+                switch (compression_type) {
+#if WITH_ZLIB
+                case RD_KAFKA_COMPRESSION_GZIP:
+                        r = rd_kafka_compress_gzip(rkb, payload, payload_len,
+                                                   compressed_payload,
+                                                   compressed_payload_size);
+                        compression_used = RD_KAFKA_COMPRESSION_GZIP;
+                        break;
+#endif
+                case RD_KAFKA_COMPRESSION_LZ4:
+                        // TODO: Using 0 for compression level for now.
+                        r = rd_kafka_lz4_compress_direct(
+                            rkb, 0, payload, payload_len, compressed_payload,
+                            compressed_payload_size);
+                        compression_used = RD_KAFKA_COMPRESSION_LZ4;
+                        break;
+#if WITH_ZSTD
+                case RD_KAFKA_COMPRESSION_ZSTD:
+                        // TODO: Using 0 for compression level for now.
+                        r = rd_kafka_zstd_compress_direct(
+                            rkb, 0, payload, payload_len, compressed_payload,
+                            compressed_payload_size);
+                        compression_used = RD_KAFKA_COMPRESSION_ZSTD;
+                        break;
+#endif
+#if WITH_SNAPPY
+                case RD_KAFKA_COMPRESSION_SNAPPY:
+                        r = rd_kafka_compress_snappy(rkb, payload, payload_len,
+                                                     compressed_payload,
+                                                     compressed_payload_size);
+                        compression_used = RD_KAFKA_COMPRESSION_SNAPPY;
+                        break;
+#endif
+                default:
+                        break;
+                }
+                if (compression_used != RD_KAFKA_COMPRESSION_NONE && r == 0) {
+                        rd_kafka_dbg(
+                            rk, TELEMETRY, "PUSH",
+                            "Compressed payload of size %" PRIusz " to %" PRIusz
+                            " using compression type "
+                            "%s",
+                            payload_len, *compressed_payload_size,
+                            rd_kafka_compression2str(compression_used));
+                        rd_free(payload);
+                        return compression_used;
+                }
+        }
+        if (compression_used != RD_KAFKA_COMPRESSION_NONE && r == -1) {
+                rd_kafka_dbg(rk, TELEMETRY, "PUSHERR",
+                             "Failed to compress payload with available "
+                             "compression types");
+        }
+        rd_kafka_dbg(rk, TELEMETRY, "PUSH", "Sending uncompressed payload");
+
+        *compressed_payload      = payload;
+        *compressed_payload_size = payload_len;
+        return RD_KAFKA_COMPRESSION_NONE;
+}


You can use existing compression functions here or set the pointers to payload->rbuf_wpos->seg_p and payload->rbuf_wpos->seg_of in case no compression was applied.

Suggested change

static rd_kafka_compression_t

rd_kafka_push_telemetry_payload_compress(rd_kafka_t *rk,

rd_kafka_broker_t *rkb,

void *payload,

size_t payload_len,

void **compressed_payload,

size_t *compressed_payload_size) {

rd_kafka_compression_t compression_used = RD_KAFKA_COMPRESSION_NONE;

int i, r = -1;

for (i = 0; i < (int)rk->rk_telemetry.accepted_compression_types_cnt;

i++) {

rd_kafka_compression_t compression_type =

rk->rk_telemetry.accepted_compression_types[i];

switch (compression_type) {

#if WITH_ZLIB

case RD_KAFKA_COMPRESSION_GZIP:

r = rd_kafka_compress_gzip(rkb, payload, payload_len,

compressed_payload,

compressed_payload_size);

compression_used = RD_KAFKA_COMPRESSION_GZIP;

break;

#endif

case RD_KAFKA_COMPRESSION_LZ4:

// TODO: Using 0 for compression level for now.

r = rd_kafka_lz4_compress_direct(

rkb, 0, payload, payload_len, compressed_payload,

compressed_payload_size);

compression_used = RD_KAFKA_COMPRESSION_LZ4;

break;

#if WITH_ZSTD

case RD_KAFKA_COMPRESSION_ZSTD:

// TODO: Using 0 for compression level for now.

r = rd_kafka_zstd_compress_direct(

rkb, 0, payload, payload_len, compressed_payload,

compressed_payload_size);

compression_used = RD_KAFKA_COMPRESSION_ZSTD;

break;

#endif

#if WITH_SNAPPY

case RD_KAFKA_COMPRESSION_SNAPPY:

r = rd_kafka_compress_snappy(rkb, payload, payload_len,

compressed_payload,

compressed_payload_size);

compression_used = RD_KAFKA_COMPRESSION_SNAPPY;

break;

#endif

default:

break;

}

if (compression_used != RD_KAFKA_COMPRESSION_NONE && r == 0) {

rd_kafka_dbg(

rk, TELEMETRY, "PUSH",

"Compressed payload of size %" PRIusz " to %" PRIusz

" using compression type "

"%s",

payload_len, *compressed_payload_size,

rd_kafka_compression2str(compression_used));

rd_free(payload);

return compression_used;

}

}

if (compression_used != RD_KAFKA_COMPRESSION_NONE && r == -1) {

rd_kafka_dbg(rk, TELEMETRY, "PUSHERR",

"Failed to compress payload with available "

"compression types");

}

rd_kafka_dbg(rk, TELEMETRY, "PUSH", "Sending uncompressed payload");

*compressed_payload = payload;

*compressed_payload_size = payload_len;

return RD_KAFKA_COMPRESSION_NONE;

}

static rd_kafka_compression_t

rd_kafka_push_telemetry_payload_compress(rd_kafka_t *rk,

rd_kafka_broker_t *rkb,

rd_buf_t *payload,

void **compressed_payload,

size_t *compressed_payload_size) {

rd_kafka_compression_t compression_used = RD_KAFKA_COMPRESSION_NONE;

rd_slice_t payload_slice;

size_t i;

rd_kafka_resp_err_t r = RD_KAFKA_RESP_ERR_NO_ERROR;

rd_slice_init_full(&payload_slice, payload);

for (i = 0; i < rk->rk_telemetry.accepted_compression_types_cnt; i++) {

rd_kafka_compression_t compression_type =

rk->rk_telemetry.accepted_compression_types[i];

switch (compression_type) {

#if WITH_ZLIB

case RD_KAFKA_COMPRESSION_GZIP:

/* TODO: Using 0 for compression level for now. */

r = rd_kafka_gzip_compress(rkb, 0, &payload_slice,

compressed_payload,

compressed_payload_size);

compression_used = RD_KAFKA_COMPRESSION_GZIP;

break;

#endif

case RD_KAFKA_COMPRESSION_LZ4:

/* TODO: Using 0 for compression level for now. */

r = rd_kafka_lz4_compress(

rkb, rd_true, 0, &payload_slice, compressed_payload,

compressed_payload_size);

compression_used = RD_KAFKA_COMPRESSION_LZ4;

break;

#if WITH_ZSTD

case RD_KAFKA_COMPRESSION_ZSTD:

/* TODO: Using 0 for compression level for now. */

r = rd_kafka_zstd_compress(rkb, 0, &payload_slice,

compressed_payload,

compressed_payload_size);

compression_used = RD_KAFKA_COMPRESSION_ZSTD;

break;

#endif

#if WITH_SNAPPY

case RD_KAFKA_COMPRESSION_SNAPPY:

r = rd_kafka_snappy_compress(rkb, &payload_slice,

compressed_payload,

compressed_payload_size);

compression_used = RD_KAFKA_COMPRESSION_SNAPPY;

break;

#endif

default:

break;

}

if (compression_used != RD_KAFKA_COMPRESSION_NONE &&

r == RD_KAFKA_RESP_ERR_NO_ERROR) {

rd_kafka_dbg(

rk, TELEMETRY, "PUSH",

"Compressed payload of size %" PRIusz " to %" PRIusz

" using compression type "

"%s",

payload->rbuf_size, *compressed_payload_size,

rd_kafka_compression2str(compression_used));

return compression_used;

}

}

if (compression_used != RD_KAFKA_COMPRESSION_NONE &&

r != RD_KAFKA_RESP_ERR_NO_ERROR) {

rd_kafka_dbg(rk, TELEMETRY, "PUSH",

"Failed to compress payload with available "

"compression types");

}

rd_kafka_dbg(rk, TELEMETRY, "PUSH", "Sending uncompressed payload");

*compressed_payload = payload->rbuf_wpos->seg_p;

*compressed_payload_size = payload->rbuf_wpos->seg_of;

return RD_KAFKA_COMPRESSION_NONE;

}

milindl and others added 30 commits July 25, 2023 12:06

Add boilerplate for the entire flow (#4365)

3f0eed3

Added new data type UUID for topic id. Added topic id support in Meta…

56fa55a

…data Response.

Merge Get and Push Protocol (#4377)

9546d71

* WIP:Push telemetry is being scheduled now. * Push * Compliation Fix * Working * Use UUID in PUSH * Remove fprintf * Changes * Fix CONFIGURATION.md * Fix size * Update s2i bounds --------- Co-authored-by: Milind L <[email protected]>

Clang format fixes

e11aa51

Fix memory leaks (#4378)

ca6a8e5

Add broker selection and client termination [KIP-714] (#4382)

ef07a95

* Add broker selection and client termination * Address review comments

Handle delta temporality and some code refactoring (#4410)

5c943ba

* Support for delta temporality * Style fix * Fix bugs * Fix memory leaks and formatting * Fixes * PR Feedback

Use rd_kafka_compression_t (#4419)

9e3cf26

Add method to get requests from mock cluster

21f7072

Fix to ns

f1c4fc4

Add capacity in mock broker to change PushIntervalMs

7fdbfbe

Fix memory leak

e6a9d66

Fix jitter: should not be decimal

e0b4f3e

Add test for telemetry using mock broker

d7e4b47

Address comments

a614d2c

Unit tests push telemetry encode decode (#4440)

1216b39

* Add telemetry encode and decode unit tests * Style fix * Improve test * PR Feedback

Initial code for metrics compression

4ed6abd

Style and some bug fixes

3d66f64

Little refactoring

88dcfc1

Add max telemetry bytes (#4507)

81913fe

* Add max telemetry bytes * Clear telemetry_max_bytes * PR comments

Add other producer metrics

ccc1a26

More changes

b6c04ac

Add enable.metrics.push config

94545fe

Fix bugs

aa8a668

Refactor

8c38f8e

Fix warnings

3987849

Fix UT

4d1d192

Nullify after free

2f6afa2

Fix bugs

eff9351

anchitj added 15 commits December 11, 2023 17:19

Add new broker errors

89cfa2a

Memory leak fixes

8ca308b

PR feedback

40edc31

Style fix

e6a88cb

Immediate schedule getTelemetry for unknown subscription id

d541a2b

Rebase fix

6abdf47

Use rd_uclock

b92899b

Master merge

ccf2ba9

Master merge

c30a432

Add more mock test

8d7e70b

Master merge

c815cb5

Style fix

e516730

Merge branch 'master' into dev_kip_714_mock_broker_integration_tests_…

3cb7286

…master_merge

Merge master

d15171b

Merge compression

9bf7966

anchitj mentioned this pull request May 17, 2024

KIP-714 #4602

Closed

anchitj requested a review from emasab May 17, 2024 07:01

anchitj mentioned this pull request May 28, 2024

Distributed tracing story with Confluent Kafka client confluentinc/confluent-kafka-dotnet#1269

Closed

Merge branch 'master' into dev_kip_714_mock_broker_integration_tests_…

5c17295

…master_merge_with_compression

anchitj requested a review from a team as a code owner June 21, 2024 13:41

emasab requested changes Jun 22, 2024

View reviewed changes

anchitj added 2 commits June 25, 2024 12:34

PR feedback first phase

9a0c0c2

Merge branch 'master' into dev_kip_714_mock_broker_integration_tests_…

604d18f

…master_merge_with_compression

anchitj force-pushed the dev_kip_714_mock_broker_integration_tests_master_merge_with_compression branch from 3accc47 to 604d18f Compare June 25, 2024 07:05

Minor fixes

ae64b34

emasab requested changes Jun 27, 2024

View reviewed changes

src/rdkafka_telemetry_encode.c Outdated Show resolved Hide resolved

src/rdkafka_telemetry_encode.c Outdated Show resolved Hide resolved

src/rdkafka_telemetry_encode.c Outdated Show resolved Hide resolved

emasab requested changes Jun 27, 2024

View reviewed changes

Segfault fixes

4f2d428

emasab requested changes Jun 28, 2024

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KIP 714 with compression support #4721

KIP 714 with compression support #4721

anchitj commented May 17, 2024

emasab left a comment

emasab left a comment

emasab Jun 27, 2024

emasab Jun 27, 2024

anchitj Jun 28, 2024

emasab Jun 27, 2024

emasab Jun 27, 2024 •

edited

Loading

emasab Jun 27, 2024

emasab Jun 27, 2024

emasab Jun 27, 2024

anchitj Jun 28, 2024

emasab left a comment

emasab Jun 28, 2024 •

edited

Loading

emasab Jun 28, 2024

emasab Jun 28, 2024

emasab Jun 28, 2024

emasab Jun 28, 2024

emasab Jun 28, 2024

emasab Jun 28, 2024

emasab Jun 28, 2024

emasab Jun 28, 2024

emasab Jun 28, 2024

		rd_kafka_telemetry_metric_value_calculator_t)(rd_kafka_t ,
		rd_kafka_broker_t *);

	ts_last = rkb->rkb_c_historic.ts_last;
	ts_last = rk->rk_telemetry.rk_historic_c.ts_last;

	avg_throttle.intValue = sum_value / broker_count;
	avg_throttle.double_value = avg;

-        int resource_attributes_count =
+        if (rk->rk_telemetry.rk_historic_c.ts_start == 0) {
+                first_push                              = rd_true;
+                rk->rk_telemetry.rk_historic_c.ts_start = now_ns;
+                rk->rk_telemetry.rk_historic_c.ts_last  = now_ns;
+        }
+        i = 0;
+        TAILQ_FOREACH(rkb, &rk->rk_brokers, rkb_link) {
+                rd_avg_rollover(&rkb->rkb_telemetry.rd_avg_rollover.rkb_avg_rtt,
+                                &rkb->rkb_telemetry.rd_avg_current.rkb_avg_rtt);
+                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_rollover(
+                    &rkb->rkb_telemetry.rd_avg_rollover.rkb_avg_throttle,
+                    &rkb->rkb_telemetry.rd_avg_current.rkb_avg_throttle);
+                if (first_push) {
+                        rkb->rkb_telemetry.rkb_historic_c.connects =
+                            rd_atomic32_get(&rkb->rkb_c.connects);
+                }
+                i++;
+        }
+        int resource_attributes_count =

	rd_kafka_broker_ApiVersion_supported_implementation(rd_kafka_broker_t *rkb,
	rd_kafka_broker_ApiVersion_supported0(rd_kafka_broker_t *rkb,

	/*< Initial state, awaiting telemetry broker to be assigned /
	/** Initial state, awaiting telemetry broker to be assigned */

		/**< Client is being terminated and last PushTelemetry is scheduled to
		send */

KIP 714 with compression support #4721

Are you sure you want to change the base?

KIP 714 with compression support #4721

Conversation

anchitj commented May 17, 2024

emasab left a comment

Choose a reason for hiding this comment

emasab left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

emasab Jun 27, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

emasab left a comment

Choose a reason for hiding this comment

emasab Jun 28, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

emasab Jun 27, 2024 •

edited

Loading

emasab Jun 28, 2024 •

edited

Loading