fix(observability): correct emitted metrics (vectordotdev#17562)

Ref vectordotdev#17465 

There were a couple of outstanding issues from the above PR. 

1. The redis sink was counting the network bytes before actually
creating the bytes, so the count was always zero.
2. The Vector sink was not counting the `JsonSize` of the event. The
comment saying this was not being used was incorrect.

Signed-off-by: Stephen Wakely <[email protected]>

Author: StephenWakely

src/sinks/redis.rs

@@ -290,10 +290,11 @@ fn encode_event(
     transformer.transform(&mut event);
 
     let mut bytes = BytesMut::new();
-    let byte_size = bytes.len();
 
     // Errors are handled by `Encoder`.
     encoder.encode(event, &mut bytes).ok()?;
+
+    let byte_size = bytes.len();
     let value = bytes.freeze();
 
     let event = EncodedEvent::new(RedisKvEntry { key, value }, byte_size, event_byte_size);

src/sinks/vector/sink.rs

@@ -7,7 +7,7 @@ use tower::Service;
 use vector_common::json_size::JsonSize;
 use vector_core::{
     stream::{BatcherSettings, DriverResponse},
-    ByteSizeOf,
+    ByteSizeOf, EstimatedJsonEncodedSizeOf,
 };
 
 use super::service::VectorRequest;
@@ -20,6 +20,7 @@ use crate::{
 /// Data for a single event.
 struct EventData {
     byte_size: usize,
+    json_byte_size: JsonSize,
     finalizers: EventFinalizers,
     wrapper: EventWrapper,
 }
@@ -30,6 +31,7 @@ struct EventCollection {
     pub finalizers: EventFinalizers,
     pub events: Vec<EventWrapper>,
     pub events_byte_size: usize,
+    pub events_json_byte_size: JsonSize,
 }
 
 pub struct VectorSink<S> {
@@ -48,6 +50,7 @@ where
         input
             .map(|mut event| EventData {
                 byte_size: event.size_of(),
+                json_byte_size: event.estimated_json_encoded_size_of(),
                 finalizers: event.take_finalizers(),
                 wrapper: EventWrapper::from(event),
             })
@@ -57,13 +60,14 @@ where
                     event_collection.finalizers.merge(item.finalizers);
                     event_collection.events.push(item.wrapper);
                     event_collection.events_byte_size += item.byte_size;
+                    event_collection.events_json_byte_size += item.json_byte_size;
                 },
             ))
             .map(|event_collection| {
                 let builder = RequestMetadataBuilder::new(
                     event_collection.events.len(),
                     event_collection.events_byte_size,
-                    JsonSize::new(event_collection.events_byte_size), // this is fine as it isn't being used
+                    event_collection.events_json_byte_size,
                 );
 
                 let encoded_events = proto_vector::PushEventsRequest {