Feature work/kafka - Revise byte count metric

src/Agent/NewRelic/Agent/Core/Agent.cs

@@ -416,6 +416,11 @@ public void RecordCountMetric(string metricName, long count = 1)
             _agentHealthReporter.ReportCountMetric(metricName, count);
         }
 
+        public void RecordByteMetric(string metricName, long totalBytes, long? exclusiveBytes = null)
+        {
+            _agentHealthReporter.ReportByteMetric(metricName, totalBytes, exclusiveBytes);
+        }
+
         public void RecordLogMessage(string frameworkName, object logEvent, Func<object, DateTime> getTimestamp, Func<object, object> getLevel, Func<object, string> getLogMessage, Func<object, Exception> getLogException, Func<object, Dictionary<string, object>> getContextData, string spanId, string traceId)
         {
             _agentHealthReporter.ReportLogForwardingFramework(frameworkName);

src/Agent/NewRelic/Agent/Core/AgentHealth/AgentHealthReporter.cs

@@ -112,6 +112,12 @@ public void ReportCountMetric(string metricName, long count)
             var metric = _metricBuilder.TryBuildCountMetric(metricName, count);
             TrySend(metric);
         }
+        public void ReportByteMetric(string metricName, long totalBytes, long? exclusiveBytes)
+        {
+            var metric = _metricBuilder.TryBuildByteMetric(metricName, totalBytes, exclusiveBytes);
+            TrySend(metric);
+        }
+
 
 
         public void ReportDotnetVersion()

src/Agent/NewRelic/Agent/Core/AgentHealth/IAgentHealthReporter.cs

@@ -149,5 +149,6 @@ public interface IAgentHealthReporter : IOutOfBandMetricSource
         void ReportLoggingEventsDropped(int droppedCount);
         void ReportLogForwardingFramework(string logFramework);
         void ReportLogForwardingEnabledWithFramework(string logFramework);
+        void ReportByteMetric(string metricName, long totalBytes, long? exclusiveBytes = null);
     }
 }

src/Agent/NewRelic/Agent/Core/WireModels/IMetricBuilder.cs

@@ -207,5 +207,6 @@ public interface IMetricBuilder
         MetricWireModel TryBuildSupportabilityLoggingEventsDroppedMetric(int droppedCount);
 
         MetricWireModel TryBuildCountMetric(string metricName, long count);
+        MetricWireModel TryBuildByteMetric(string metricName, long totalBytes, long? exclusiveBytes = null);
     }
 }

src/Agent/NewRelic/Agent/Core/WireModels/MetricWireModel.cs

@@ -1016,6 +1016,12 @@ public MetricWireModel TryBuildCountMetric(string metricName, long count)
                 return BuildMetric(_metricNameService, metricName, null, data);
             }
 
+            public MetricWireModel TryBuildByteMetric(string metricName, long totalBytes, long? exclusiveBytes)
+            {
+                var data = MetricDataWireModel.BuildByteData(totalBytes, exclusiveBytes);
+                return BuildMetric(_metricNameService, metricName, null, data);
+            }
+
             #endregion
         }
     }

src/Agent/NewRelic/Agent/Extensions/NewRelic.Agent.Extensions/Api/Experimental/IAgentExperimental.cs

@@ -25,6 +25,13 @@ public interface IAgentExperimental
         /// <param name="metricName"></param>
         /// <param name="count"></param>
         void RecordCountMetric(string metricName, long count = 1);
+        /// <summary>
+        /// Records a byte count metric with the given name
+        /// </summary>
+        /// <param name="metricName"></param>
+        /// <param name="totalBytes"></param>
+        /// <param name="exclusiveBytes"></param>
+        void RecordByteMetric(string metricName, long totalBytes, long? exclusiveBytes = null);
 
         /// <summary>
         /// Records the log message in the transaction to later be forwarded if log forwarding is enabled.

src/Agent/NewRelic/Agent/Extensions/Providers/Wrapper/Kafka/KafkaConsumerWrapper.cs

@@ -122,7 +122,7 @@ private static void ReportSizeMetrics(IAgent agent, ITransaction transaction, st
             // Add metrics for bytes received and messages received
             var agentExp = agent.GetExperimentalApi();
             agentExp.RecordCountMetric($"Message/Kafka/Topic/Named/{topic}/Received/Messages", 1);
-            agentExp.RecordCountMetric($"Message/Kafka/Topic/Named/{topic}/Received/Bytes", totalSize);
+            agentExp.RecordByteMetric($"Message/Kafka/Topic/Named/{topic}/Received/Bytes", totalSize);
         }
 
         private static Func<object, string> GetTopicAccessorFunc(Type t) =>