wip: tracelog support for streaming an eventpipe source

Author: vaind

src/TraceEvent/EventPipe/EventPipeEventSource.cs

@@ -125,6 +125,12 @@ public DateTime QPCTimeToTimeStamp(long QPCTime)
         /// </summary>
         public int MinimumVersionCanRead => 0;
 
+        /// <summary>
+        /// Called after headers are deserialized. This is especially useful in a streaming scenario
+        /// because the headers are only read after Process() is called.
+        /// </summary>
+        internal Action HeadersDeserialized;
+
         protected override void Dispose(bool disposing)
         {
             if (_deserializer != null)
@@ -144,6 +150,7 @@ public override bool Process()
                 Debug.Assert(_deserializer != null);
             }
 
+            HeadersDeserialized?.Invoke();
 
             if (FileFormatVersionNumber >= 3)
             {

src/TraceEvent/TraceEvent.Tests/Parsing/EventPipeParsing.cs

@@ -1,17 +1,22 @@
 using FastSerialization;
 using Microsoft.Diagnostics.Tracing;
+using Microsoft.Diagnostics.Tracing.AutomatedAnalysis;
 using Microsoft.Diagnostics.Tracing.Etlx;
 using Microsoft.Diagnostics.Tracing.EventPipe;
 using Microsoft.Diagnostics.Tracing.Parsers;
 using Microsoft.Diagnostics.Tracing.Parsers.Clr;
+using Microsoft.Diagnostics.Tracing.Parsers.Kernel;
 using System;
 using System.Collections.Generic;
 using System.Diagnostics;
+using System.Diagnostics.Tracing;
 using System.IO;
 using System.Linq;
+using System.Runtime.InteropServices;
 using System.Text;
 using Xunit;
 using Xunit.Abstractions;
+using static Microsoft.Diagnostics.Tracing.Etlx.TraceLog;
 
 namespace TraceEventTests
 {
@@ -94,8 +99,7 @@ public void Streaming(string eventPipeFileName)
                         // block to read the events
                         Assert.InRange(newStreamPosition, curStreamPosition, curStreamPosition + 103_000);
                         curStreamPosition = newStreamPosition;
-
-
+                        
                         string eventName = data.ProviderName + "/" + data.EventName;
 
                         // For whatever reason the parse filtering below produces a couple extra events
@@ -148,6 +152,52 @@ public void Streaming(string eventPipeFileName)
             ValidateEventStatistics(eventStatistics, eventPipeFileName);
         }
 
+        [Theory()]
+        [MemberData(nameof(StreamableTestEventPipeFiles))]
+        public void TraceLogStreaming(string eventPipeFileName)
+        {
+            // Initialize
+            PrepareTestData();
+
+            string eventPipeFilePath = Path.Combine(UnZippedDataDir, eventPipeFileName);
+            Output.WriteLine(string.Format("Processing the file {0}", Path.GetFullPath(eventPipeFilePath)));
+            var eventStatistics = new SortedDictionary<string, EventRecord>(StringComparer.Ordinal);
+
+            using (MockStreamingOnlyStream s = new MockStreamingOnlyStream(new FileStream(eventPipeFilePath, FileMode.Open, FileAccess.Read, FileShare.Read)))
+            {
+                using (var eventPipeSource = new EventPipeEventSource(s))
+                {
+                    using (var traceSource = TraceLog.CreateFromEventPipeEventSource(eventPipeSource))
+                    {
+                        Action<TraceEvent> handler = delegate (TraceEvent data)
+                        {
+                            string eventName = data.ProviderName + "/" + data.EventName;
+
+                            if (eventStatistics.ContainsKey(eventName))
+                            {
+                                eventStatistics[eventName].TotalCount++;
+                            }
+                            else
+                            {
+                                eventStatistics[eventName] = new EventRecord()
+                                {
+                                    TotalCount = 1,
+                                    FirstSeriazliedSample = new String(data.ToString().Replace("\n", "\\n").Replace("\r", "\\r").Take(1000).ToArray())
+                                };
+                            }
+                        };
+
+                        traceSource.AllEvents += handler;
+
+                        // Process
+                        traceSource.Process();
+                    }
+                }
+            }
+            // Validate
+            ValidateEventStatistics(eventStatistics, eventPipeFileName);
+        }
+
         [Fact]
         public void CanParseHeaderOfV1EventPipeFile()
         {

src/TraceEvent/TraceEvent.csproj

@@ -41,6 +41,7 @@
   </PropertyGroup>
 
   <ItemGroup>
+    <PackageReference Include="Microsoft.Diagnostics.NETCore.Client" Version="0.2.421201" />
     <PackageReference Include="Microsoft.NETCore.Portable.Compatibility" Version="$(MicrosoftNETCorePortableCompatibilityVersion)" />
     <PackageReference Include="Microsoft.Diagnostics.Tracing.TraceEvent.SupportFiles" Version="$(MicrosoftDiagnosticsTracingTraceEventSupportFilesVersion)" />
     <PackageReference Include="Microsoft.Win32.Registry" Version="$(MicrosoftWin32RegistryVersion)" />

src/TraceEvent/TraceLog.cs

@@ -5,6 +5,7 @@
 // #define DEBUG_SERIALIZE
 
 using FastSerialization;
+using Microsoft.Diagnostics.NETCore.Client;
 using Microsoft.Diagnostics.Symbols;
 using Microsoft.Diagnostics.Tracing.Compatibility;
 using Microsoft.Diagnostics.Tracing.EventPipe;
@@ -153,7 +154,38 @@ public static TraceLog OpenOrConvert(string etlOrEtlxFilePath, TraceLogOptions o
         /// </summary>
         public static TraceLogEventSource CreateFromTraceEventSession(TraceEventSession session)
         {
-            var traceLog = new TraceLog(session);
+            var traceLog = new TraceLog(session.Source);
+
+            // See if we are on Win7 and have a separate kernel session associated with 'session'
+            if (session.m_kernelSession != null)
+            {
+                // Make sure both sources only dispatch one at a time by taking a lock during dispatch.
+                session.m_kernelSession.Source.lockObj = traceLog.realTimeQueue;
+                session.m_associatedWithTraceLog = true;                         // Indicate that it is OK to have the m_kernelSession.
+                session.Source.lockObj = traceLog.realTimeQueue;
+
+                // Set up the callbacks to the kernel session.
+                traceLog.rawKernelEventSource = session.m_kernelSession.Source;
+                traceLog.SetupCallbacks(traceLog.rawKernelEventSource);
+                traceLog.rawKernelEventSource.unhandledEventTemplate.traceEventSource = traceLog;       // Make everything point to the log as its source.
+                // TODO fixme - onAllEvents is local to the constructor
+                // traceLog.rawKernelEventSource.AllEvents += traceLog.onAllEvents;
+            }
+
+            return traceLog.realTimeSource;
+        }
+
+        public static TraceLogEventSource CreateFromEventPipeSession(EventPipeSession session)
+        {
+            return CreateFromEventPipeEventSource(new EventPipeEventSource(session.EventStream));
+        }
+
+        public static TraceLogEventSource CreateFromEventPipeEventSource(EventPipeEventSource source)
+        {
+            var traceLog = new TraceLog(source);
+            var dynamicParser = source.Dynamic;
+            var clrParser = source.Clr;
+            var kernelParser = source.Kernel;
             return traceLog.realTimeSource;
         }
 
@@ -543,7 +575,7 @@ private TraceLog()
         /// Functionality of TraceLog that does not depend on either remembering past EVENTS or require future
         /// knowledge (e.g. stacks of kernel events), will 'just work'.
         /// </summary>
-        private unsafe TraceLog(TraceEventSession session)
+        private unsafe TraceLog(TraceEventDispatcher source)
             : this()
         {
             IsRealTime = true;
@@ -605,23 +637,8 @@ private unsafe TraceLog(TraceEventSession session)
                 }
             };
 
-            // See if we are on Win7 and have a separate kernel session associated with 'session'
-            if (session.m_kernelSession != null)
-            {
-                // Make sure both sources only dispatch one at a time by taking a lock during dispatch.
-                session.m_kernelSession.Source.lockObj = realTimeQueue;
-                session.m_associatedWithTraceLog = true;                         // Indicate that it is OK to have the m_kernelSession.
-                session.Source.lockObj = realTimeQueue;
-
-                // Set up the callbacks to the kernel session.
-                rawKernelEventSource = session.m_kernelSession.Source;
-                SetupCallbacks(rawKernelEventSource);
-                rawKernelEventSource.unhandledEventTemplate.traceEventSource = this;       // Make everything point to the log as its source.
-                rawKernelEventSource.AllEvents += onAllEvents;
-            }
-
             // We use the session's source for our input.
-            rawEventSourceToConvert = session.Source;
+            rawEventSourceToConvert = source;
             SetupCallbacks(rawEventSourceToConvert);
             rawEventSourceToConvert.unhandledEventTemplate.traceEventSource = this;       // Make everything point to the log as its source.
             rawEventSourceToConvert.AllEvents += onAllEvents;
@@ -1081,16 +1098,27 @@ private unsafe void SetupCallbacks(TraceEventDispatcher rawEvents)
             sourceFilesByID = new Dictionary<JavaScriptSourceKey, string>();
 
             // If this is a ETL file, we also need to compute all the normal TraceLog stuff the raw stream
-            pointerSize = rawEvents.PointerSize;
-            _syncTimeUTC = rawEvents._syncTimeUTC;
-            _syncTimeQPC = rawEvents._syncTimeQPC;
-            _QPCFreq = rawEvents._QPCFreq;
-            sessionStartTimeQPC = rawEvents.sessionStartTimeQPC;
-            sessionEndTimeQPC = rawEvents.sessionEndTimeQPC;
-            cpuSpeedMHz = rawEvents.CpuSpeedMHz;
-            numberOfProcessors = rawEvents.NumberOfProcessors;
-            eventsLost = rawEvents.EventsLost;
-            osVersion = rawEvents.OSVersion;
+            Action copyHeaders = delegate ()
+            {
+                pointerSize = rawEvents.PointerSize;
+                _syncTimeUTC = rawEvents._syncTimeUTC;
+                _syncTimeQPC = rawEvents._syncTimeQPC;
+                _QPCFreq = rawEvents._QPCFreq;
+                sessionStartTimeQPC = rawEvents.sessionStartTimeQPC;
+                sessionEndTimeQPC = rawEvents.sessionEndTimeQPC;
+                cpuSpeedMHz = rawEvents.CpuSpeedMHz;
+                numberOfProcessors = rawEvents.NumberOfProcessors;
+                eventsLost = rawEvents.EventsLost;
+                osVersion = rawEvents.OSVersion;
+            };
+
+            if (rawEvents is EventPipeEventSource eventPipeEventSource)
+            {
+                eventPipeEventSource.HeadersDeserialized += copyHeaders;
+            } else
+            {
+                copyHeaders();
+            }
 
             // These parsers create state and we want to collect that so we put it on our 'parsers' list that we serialize.
             var kernelParser = rawEvents.Kernel;