From 7e878e8dbb7f74288cba6271239b0ede30a0b331 Mon Sep 17 00:00:00 2001
From: a1exsh <a1exsh@users.noreply.github.com>
Date: Thu, 5 Oct 2023 13:16:48 +0200
Subject: [PATCH] Add `number_of_events` tag to `publish_events` spans (#1553)

To allow the publishers to learn about their batch size distribution.
---
 .../main/java/org/zalando/nakadi/service/TracingService.java | 4 ++++
 .../nakadi/service/publishing/BinaryEventPublisher.java      | 5 ++++-
 .../zalando/nakadi/service/publishing/EventPublisher.java    | 3 +++
 3 files changed, 11 insertions(+), 1 deletion(-)

diff --git a/core-common/src/main/java/org/zalando/nakadi/service/TracingService.java b/core-common/src/main/java/org/zalando/nakadi/service/TracingService.java
index 4d2572553a..b7e9291f30 100644
--- a/core-common/src/main/java/org/zalando/nakadi/service/TracingService.java
+++ b/core-common/src/main/java/org/zalando/nakadi/service/TracingService.java
@@ -85,6 +85,10 @@ public static Span setTag(final String key, final String value) {
         return getActiveSpan().setTag(key, value);
     }
 
+    public static Span setTag(final String key, final Number value) {
+        return getActiveSpan().setTag(key, value);
+    }
+
     public static Span setErrorFlag() {
         return setErrorFlag(getActiveSpan());
     }
diff --git a/core-services/src/main/java/org/zalando/nakadi/service/publishing/BinaryEventPublisher.java b/core-services/src/main/java/org/zalando/nakadi/service/publishing/BinaryEventPublisher.java
index 643e126535..1212598e61 100644
--- a/core-services/src/main/java/org/zalando/nakadi/service/publishing/BinaryEventPublisher.java
+++ b/core-services/src/main/java/org/zalando/nakadi/service/publishing/BinaryEventPublisher.java
@@ -82,6 +82,9 @@ private List<NakadiRecordResult> processInternal(final EventType eventType,
                                                      final List<NakadiRecord> records,
                                                      final List<Check> checks,
                                                      final Map<HeaderTag, String> consumerTags) {
+
+        TracingService.setTag("number_of_events", records.size());
+
         for (final Check check : checks) {
             final List<NakadiRecordResult> res = check.execute(eventType, records);
             if (res != null && !res.isEmpty()) {
@@ -147,4 +150,4 @@ public List<NakadiRecordResult> delete(final List<NakadiRecord> events, final Ev
     public List<NakadiRecordResult> publishInternal(final EventType eventType, final List<NakadiRecord> events) {
         return processInternal(eventType, events, internalPublishingChecks, null);
     }
-}
\ No newline at end of file
+}
diff --git a/core-services/src/main/java/org/zalando/nakadi/service/publishing/EventPublisher.java b/core-services/src/main/java/org/zalando/nakadi/service/publishing/EventPublisher.java
index 5be8b4464b..af9bc3c6ea 100644
--- a/core-services/src/main/java/org/zalando/nakadi/service/publishing/EventPublisher.java
+++ b/core-services/src/main/java/org/zalando/nakadi/service/publishing/EventPublisher.java
@@ -137,6 +137,9 @@ EventPublishResult processInternal(final String events,
 
         Closeable publishingCloser = null;
         final List<BatchItem> batch = BatchFactory.from(events);
+
+        TracingService.setTag("number_of_events", batch.size());
+
         try {
             publishingCloser = timelineSync.workWithEventType(eventTypeName, nakadiSettings.getTimelineWaitTimeoutMs());