[LiveMetrics] add LogProcessor, refactor DocumentHelper, update tests

sdk/monitor/Azure.Monitor.OpenTelemetry.LiveMetrics/src/Internals/DataCollection/DocumentHelper.cs

@@ -7,7 +7,9 @@
 using System.Runtime.CompilerServices;
 using System.Text;
 using Azure.Monitor.OpenTelemetry.Exporter.Internals;
+using Azure.Monitor.OpenTelemetry.LiveMetrics.Internals.Diagnostics;
 using Azure.Monitor.OpenTelemetry.LiveMetrics.Models;
+using OpenTelemetry.Logs;
 using ExceptionDocument = Azure.Monitor.OpenTelemetry.LiveMetrics.Models.Exception;
 
 namespace Azure.Monitor.OpenTelemetry.LiveMetrics.Internals.DataCollection
@@ -19,7 +21,87 @@ internal static class DocumentHelper
     {
         // TODO: NEED TO HANDLE UNIQUE MAXLENGTH VALUES FOR DOCUMENT TYPES. SEE SWAGGER FOR MAXLENGTH VALUES.
 
-        internal static RemoteDependency ConvertToRemoteDependency(Activity activity)
+        #region Document Buffer Extension Methods
+        public static void AddExceptionDocument(this DoubleBuffer buffer, ActivityEvent activityEvent)
+        {
+            try
+            {
+                var exceptionDocument = ConvertToExceptionDocument(activityEvent);
+                buffer.WriteDocument(exceptionDocument);
+            }
+            catch (System.Exception ex)
+            {
+                LiveMetricsExporterEventSource.Log.FailedToCreateTelemetryDocument("ExceptionDocument", ex);
+            }
+        }
+
+        public static void AddExceptionDocument(this DoubleBuffer buffer, System.Exception exception)
+        {
+            try
+            {
+                var exceptionDocument = ConvertToExceptionDocument(exception);
+                buffer.WriteDocument(exceptionDocument);
+            }
+            catch (System.Exception ex)
+            {
+                LiveMetricsExporterEventSource.Log.FailedToCreateTelemetryDocument("ExceptionDocument", ex);
+            }
+        }
+
+        public static void AddLogDocument(this DoubleBuffer buffer, ActivityEvent activityEvent)
+        {
+            try
+            {
+                var logDocument = ConvertToLogDocument(activityEvent);
+                buffer.WriteDocument(logDocument);
+            }
+            catch (System.Exception ex)
+            {
+                LiveMetricsExporterEventSource.Log.FailedToCreateTelemetryDocument("LogDocument", ex);
+            }
+        }
+
+        public static void AddLogDocument(this DoubleBuffer buffer, LogRecord logRecord)
+        {
+            try
+            {
+                var logDocument = ConvertToLogDocument(logRecord);
+                buffer.WriteDocument(logDocument);
+            }
+            catch (System.Exception ex)
+            {
+                LiveMetricsExporterEventSource.Log.FailedToCreateTelemetryDocument("LogDocument", ex);
+            }
+        }
+
+        public static void AddDependencyDocument(this DoubleBuffer buffer, Activity activity)
+        {
+            try
+            {
+                var dependencyDocument = ConvertToDependencyDocument(activity);
+                buffer.WriteDocument(dependencyDocument);
+            }
+            catch (System.Exception ex)
+            {
+                LiveMetricsExporterEventSource.Log.FailedToCreateTelemetryDocument("DependencyDocument", ex);
+            }
+        }
+
+        public static void AddRequestDocument(this DoubleBuffer buffer, Activity activity)
+        {
+            try
+            {
+                var requestDocument = ConvertToRequestDocument(activity);
+                buffer.WriteDocument(requestDocument);
+            }
+            catch (System.Exception ex)
+            {
+                LiveMetricsExporterEventSource.Log.FailedToCreateTelemetryDocument("RequestDocument", ex);
+            }
+        }
+        #endregion
+
+        internal static RemoteDependency ConvertToDependencyDocument(Activity activity)
         {
             // TODO: Investigate if we can have a minimal/optimized version of ActivityTagsProcessor for LiveMetric.
             var atp = new ActivityTagsProcessor();
@@ -86,7 +168,7 @@ internal static RemoteDependency ConvertToRemoteDependency(Activity activity)
             return remoteDependencyDocumentIngress;
         }
 
-        internal static Request ConvertToRequest(Activity activity)
+        internal static Request ConvertToRequestDocument(Activity activity)
         {
             string httpResponseStatusCode = string.Empty;
             string urlScheme = string.Empty;
@@ -163,8 +245,29 @@ internal static Request ConvertToRequest(Activity activity)
             return requestDocumentIngress;
         }
 
-        internal static ExceptionDocument CreateException(string exceptionType, string exceptionMessage)
+        internal static ExceptionDocument ConvertToExceptionDocument(ActivityEvent activityEvent)
         {
+            string exceptionType = string.Empty;
+            string exceptionMessage = string.Empty;
+
+            foreach (ref readonly var tag in activityEvent.EnumerateTagObjects())
+            {
+                if (tag.Value == null)
+                {
+                    continue;
+                }
+                else if (tag.Key == SemanticConventions.AttributeExceptionType)
+                {
+                    exceptionType = tag.Value.ToString()!;
+                    continue;
+                }
+                else if (tag.Key == SemanticConventions.AttributeExceptionMessage)
+                {
+                    exceptionMessage = tag.Value.ToString()!;
+                    continue;
+                }
+            }
+
             ExceptionDocument exceptionDocumentIngress = new()
             {
                 DocumentType = DocumentType.Exception,
@@ -176,6 +279,38 @@ internal static ExceptionDocument CreateException(string exceptionType, string e
             return exceptionDocumentIngress;
         }
 
+        internal static ExceptionDocument ConvertToExceptionDocument(System.Exception exception)
+        {
+            ExceptionDocument exceptionDocumentIngress = new()
+            {
+                DocumentType = DocumentType.Exception,
+                ExceptionType = exception.GetType().FullName,
+                ExceptionMessage = exception.Message,
+            };
+
+            return exceptionDocumentIngress;
+        }
+
+        internal static Models.Trace ConvertToLogDocument(LogRecord logRecord)
+        {
+            return new Models.Trace()
+            {
+                DocumentType = DocumentType.Trace,
+                Message = logRecord.FormattedMessage ?? logRecord.Body, // TODO: MAY NEED TO BUILD THE FORMATTED MESSAGE IF NOT AVAILABLE
+                // TODO: Properties = new Dictionary<string, string>(), - UX supports up to 10 custom properties
+            };
+        }
+
+        internal static Models.Trace ConvertToLogDocument(ActivityEvent activityEvent)
+        {
+            return new Models.Trace()
+            {
+                DocumentType = DocumentType.Trace,
+                Message = activityEvent.Name,
+                // TODO: Properties = new Dictionary<string, string>(), - UX supports up to 10 custom properties
+            };
+        }
+
         [MethodImpl(MethodImplOptions.AggressiveInlining)]
         internal static bool IsHttpSuccess(Activity activity, string? responseCode)
         {

sdk/monitor/Azure.Monitor.OpenTelemetry.LiveMetrics/src/Internals/Diagnostics/LiveMetricsExporterEventSource.cs

@@ -204,5 +204,17 @@ public void DroppedDocument(DocumentType documentType)
 
         [Event(15, Message = "Calculated Cpu Counter: Period: {0}. DiffValue: {1}. CalculatedValue: {2}. ProcessorCount: {3}. NormalizedValue: {4}", Level = EventLevel.Verbose)]
         public void ProcessCountersCpuCounter(long period, long diffValue, double calculatedValue, int processorCount, double normalizedValue) => WriteEvent(15, period, diffValue, calculatedValue, processorCount, normalizedValue);
+
+        [NonEvent]
+        public void FailedToCreateTelemetryDocument(string documentTypeName, System.Exception ex)
+        {
+            if (IsEnabled(EventLevel.Error))
+            {
+                FailedToCreateTelemetryDocument(documentTypeName, ex.ToInvariantString());
+            }
+        }
+
+        [Event(16, Message = "Failed to create telemetry document due to an exception. DocumentType: {0}. Exception: {1}", Level = EventLevel.Error)]
+        public void FailedToCreateTelemetryDocument(string documentTypeName, string exceptionMessage) => WriteEvent(16, documentTypeName, exceptionMessage);
     }
 }

sdk/monitor/Azure.Monitor.OpenTelemetry.LiveMetrics/src/LiveMetricsActivityProcessor.cs

@@ -37,59 +37,27 @@ public override void OnEnd(Activity activity)
 
             if (activity.Kind == ActivityKind.Server || activity.Kind == ActivityKind.Consumer)
             {
-                AddRequestDocument(activity);
+                _manager._documentBuffer.AddRequestDocument(activity);
             }
             else
             {
-                AddRemoteDependencyDocument(activity);
+                _manager._documentBuffer.AddDependencyDocument(activity);
             }
 
             if (activity.Events != null)
             {
                 foreach (ref readonly var @event in activity.EnumerateEvents())
                 {
-                    string exceptionType = string.Empty;
-                    string exceptionMessage = string.Empty;
-
-                    foreach (ref readonly var tag in @event.EnumerateTagObjects())
+                    if (@event.Name == SemanticConventions.AttributeExceptionEventName)
                     {
-                        if (tag.Value == null)
-                        {
-                            continue;
-                        }
-                        else if (tag.Key == SemanticConventions.AttributeExceptionType)
-                        {
-                            exceptionType = tag.Value.ToString()!;
-                            continue;
-                        }
-                        else if (tag.Key == SemanticConventions.AttributeExceptionMessage)
-                        {
-                            exceptionMessage = tag.Value.ToString()!;
-                            continue;
-                        }
+                        _manager._documentBuffer.AddExceptionDocument(@event);
+                    }
+                    else
+                    {
+                        _manager._documentBuffer.AddLogDocument(@event);
                     }
-
-                    AddExceptionDocument(exceptionType, exceptionMessage);
                 }
             }
         }
-
-        private void AddExceptionDocument(string exceptionType, string exceptionMessage)
-        {
-            var exceptionDocumentIngress = DocumentHelper.CreateException(exceptionType, exceptionMessage);
-            _manager._documentBuffer.WriteDocument(exceptionDocumentIngress);
-        }
-
-        private void AddRemoteDependencyDocument(Activity activity)
-        {
-            var remoteDependencyDocumentIngress = DocumentHelper.ConvertToRemoteDependency(activity);
-            _manager._documentBuffer.WriteDocument(remoteDependencyDocumentIngress);
-        }
-
-        private void AddRequestDocument(Activity activity)
-        {
-            var requestDocumentIngress = DocumentHelper.ConvertToRequest(activity);
-            _manager._documentBuffer.WriteDocument(requestDocumentIngress);
-        }
     }
 }

sdk/monitor/Azure.Monitor.OpenTelemetry.LiveMetrics/src/LiveMetricsLogProcessor.cs

@@ -0,0 +1,69 @@
+// Copyright (c) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License.
+
+using Azure.Monitor.OpenTelemetry.LiveMetrics.Internals;
+using Azure.Monitor.OpenTelemetry.LiveMetrics.Internals.DataCollection;
+using OpenTelemetry;
+using OpenTelemetry.Logs;
+
+namespace Azure.Monitor.OpenTelemetry.LiveMetrics
+{
+    internal class LiveMetricsLogProcessor : BaseProcessor<LogRecord>
+    {
+        private bool _disposed;
+        private LiveMetricsResource? _resource;
+        private readonly Manager _manager;
+
+        internal LiveMetricsResource? LiveMetricsResource => _resource ??= ParentProvider?.GetResource().CreateAzureMonitorResource();
+
+        public LiveMetricsLogProcessor(Manager manager)
+        {
+            _manager = manager;
+        }
+
+        public override void OnEnd(LogRecord data)
+        {
+            // Check if live metrics is enabled.
+            if (!_manager.ShouldCollect())
+            {
+                return;
+            }
+
+            // Resource is not available at initialization and must be set later.
+            if (_manager.LiveMetricsResource == null && LiveMetricsResource != null)
+            {
+                _manager.LiveMetricsResource = LiveMetricsResource;
+            }
+
+            if (data.Exception is null)
+            {
+                _manager._documentBuffer.AddLogDocument(data);
+            }
+            else
+            {
+                _manager._documentBuffer.AddExceptionDocument(data.Exception);
+            }
+        }
+
+        protected override void Dispose(bool disposing)
+        {
+            if (!_disposed)
+            {
+                if (disposing)
+                {
+                    try
+                    {
+                        _manager.Dispose();
+                    }
+                    catch (System.Exception)
+                    {
+                    }
+                }
+
+                _disposed = true;
+            }
+
+            base.Dispose(disposing);
+        }
+    }
+}