Fixed datadog agent test

Signed-off-by: Stephen Wakely <fungus.humungus@gmail.com>

src/sinks/datadog/metrics/encoder.rs

@@ -333,6 +333,8 @@ impl DatadogMetricsEncoder {
         self.state.written += n;
 
         let raw_bytes_written = self.state.written;
+        let byte_size = self.state.byte_size.clone();
+
         // Consume the encoder state so we can do our final checks and return the necessary data.
         let state = self.reset_state();
         let payload = state
@@ -357,7 +359,7 @@ impl DatadogMetricsEncoder {
         if recommended_splits == 1 {
             // "One" split means no splits needed: our payload didn't exceed either of the limits.
             Ok((
-                EncodeResult::compressed(payload, raw_bytes_written, self.state.byte_size.clone()),
+                EncodeResult::compressed(payload, raw_bytes_written, byte_size),
                 processed,
             ))
         } else {

src/sinks/datadog/metrics/integration_tests.rs

@@ -4,7 +4,10 @@ use futures::{channel::mpsc::Receiver, stream, StreamExt};
 use hyper::StatusCode;
 use indoc::indoc;
 use rand::{thread_rng, Rng};
-use vector_core::event::{BatchNotifier, BatchStatus, Event, Metric, MetricKind, MetricValue};
+use vector_core::{
+    config::{init_telemetry, Tags, Telemetry},
+    event::{BatchNotifier, BatchStatus, Event, Metric, MetricKind, MetricValue},
+};
 
 use super::DatadogMetricsConfig;
 use crate::{
@@ -207,5 +210,15 @@ async fn real_endpoint() {
 
 #[tokio::test]
 async fn data_volume_tags() {
+    init_telemetry(
+        Telemetry {
+            tags: Tags {
+                emit_service: true,
+                emit_source: true,
+            },
+        },
+        true,
+    );
+
     assert_data_volume_sink_compliance(&DATA_VOLUME_SINK_TAGS, async { run_sink().await }).await;
 }