Add number_of_events tag to publish_events spans (#1553)

To allow the publishers to learn about their batch size distribution.

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());
     }

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);
     }
-}
+}

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());