[Test Optimization] Avoid crashing on Allure.Xunit usage (#7305)

## Summary of changes

This PR avoid crashing testhost when Allure.XUnit is being used.

## Reason for change

Currently using Allure in your XUnit tests will fail.

## Implementation details

Ensure to bypass some messages to the messagebus in order to trigger
some XUnit internal events in the MessageSink. Also when we disable a
test we stop sending messages in the messagebus regarding that test.

## Test coverage

Added the Allure.XUnit package in one of the XUnit sample test. Also it
tries to add the allure api inside a test. The goal is to don't crash
the samples app and have the same test passing without any exception
(errors in the logs).

## Other details
<!-- Fixes #{issue} -->

<!-- ⚠️ Note: where possible, please obtain 2 approvals prior to
merging. Unless CODEOWNERS specifies otherwise, for external teams it is
typically best to have one review from a team member, and one review
from apm-dotnet. Trivial changes do not require 2 reviews. -->

Author: tonyredondo

tracer/src/Datadog.Trace/Ci/Agent/CIVisibilityProtocolWriter.cs

@@ -16,6 +16,7 @@
 using Datadog.Trace.Telemetry;
 using Datadog.Trace.Util;
 using Datadog.Trace.Vendors.MessagePack;
+using Datadog.Trace.Vendors.Serilog.Events;
 
 namespace Datadog.Trace.Ci.Agent;
 
@@ -292,6 +293,11 @@ private static async Task InternalFlushEventsAsync(CIVisibilityProtocolWriter wr
 
         completionSource.TrySetResult(true);
         Log.Debug("CIVisibilityProtocolWriter: InternalFlushEventsAsync/ Finishing FlushEventsAsync loop");
+        if (Log.IsEnabled(LogEventLevel.Debug))
+        {
+            Log.Debug("CIVisibilityProtocolWriter: TestCycle stats: {Stats}", $"[Tests: {buffers.CiTestCycleBuffer.TestEventsCount}, TestSuites: {buffers.CiTestCycleBuffer.TestSuiteEventsCount}, TestModules: {buffers.CiTestCycleBuffer.TestModuleEventsCount}, TestSessions: {buffers.CiTestCycleBuffer.TestSessionEventsCount}, Spans: {buffers.CiTestCycleBuffer.SpanEventsCount}]");
+            Log.Debug<int>("CIVisibilityProtocolWriter: CodeCoverage stats: [Coverage: {Coverage}]", buffers.CiCodeCoverageBuffer.TestCoverageEventsCount);
+        }
     }
 
     internal class WatermarkEvent : IEvent

tracer/src/Datadog.Trace/Ci/Agent/Payloads/CICodeCoveragePayload.cs

@@ -45,6 +45,8 @@ public override MetricTags.CIVisibilityEndpointAndCompression TelemetryEndpointA
                ? MetricTags.CIVisibilityEndpointAndCompression.CodeCoverageRequestCompressed
                : MetricTags.CIVisibilityEndpointAndCompression.CodeCoverageUncompressed;
 
+    public int TestCoverageEventsCount { get; private set; }
+
     public override bool CanProcessEvent(IEvent @event)
     {
         return @event is TestCoverage;
@@ -62,6 +64,11 @@ public override bool TryProcessEvent(IEvent @event)
     {
         _serializationWatch.Restart();
         var success = base.TryProcessEvent(@event);
+        if (success && @event is TestCoverage)
+        {
+            TestCoverageEventsCount++;
+        }
+
         TelemetryFactory.Metrics.RecordDistributionCIVisibilityEndpointEventsSerializationMs(TelemetryEndpoint, _serializationWatch.Elapsed.TotalMilliseconds);
         return success;
     }

tracer/src/Datadog.Trace/Ci/Agent/Payloads/CIVisibilityProtocolPayload.cs

@@ -8,6 +8,7 @@
 using System.IO;
 using Datadog.Trace.Ci.Agent.MessagePack;
 using Datadog.Trace.Ci.Configuration;
+using Datadog.Trace.Ci.EventModel;
 using Datadog.Trace.Telemetry;
 using Datadog.Trace.Vendors.MessagePack;
 
@@ -35,10 +36,44 @@ public CIVisibilityProtocolPayload(TestOptimizationSettings settings, IFormatter
 
     internal EventsBuffer<IEvent> Events => _events;
 
+    public int TestEventsCount { get; private set; }
+
+    public int TestSuiteEventsCount { get; private set; }
+
+    public int TestModuleEventsCount { get; private set; }
+
+    public int TestSessionEventsCount { get; private set; }
+
+    public int SpanEventsCount { get; private set; }
+
     public override bool TryProcessEvent(IEvent @event)
     {
         _serializationWatch.Restart();
         var success = _events.TryWrite(@event);
+        if (success)
+        {
+            if (@event is TestEvent)
+            {
+                TestEventsCount++;
+            }
+            else if (@event is TestSuiteEvent)
+            {
+                TestSuiteEventsCount++;
+            }
+            else if (@event is TestModuleEvent)
+            {
+                TestModuleEventsCount++;
+            }
+            else if (@event is TestSessionEvent)
+            {
+                TestSessionEventsCount++;
+            }
+            else if (@event is EventModel.SpanEvent)
+            {
+                SpanEventsCount++;
+            }
+        }
+
         TelemetryFactory.Metrics.RecordDistributionCIVisibilityEndpointEventsSerializationMs(TelemetryEndpoint, _serializationWatch.Elapsed.TotalMilliseconds);
         return success;
     }

tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/Testing/XUnit/RetryMessageBus.cs

@@ -77,7 +77,7 @@ public bool QueueMessage(object? message)
         else
         {
             Common.Log.Debug("RetryMessageBus.QueueMessage: Message is not a supported message. Flushing: {Message}", message);
-            return _innerMessageBus.QueueMessage(message);
+            return InternalQueueMessage(message);
         }
 
         Common.Log.Debug("RetryMessageBus.QueueMessage: Message: {Message} | UniqueID: {UniqueID}", message, uniqueID);
@@ -88,7 +88,7 @@ public bool QueueMessage(object? message)
             if (metadata.Disposed)
             {
                 Common.Log.Debug("RetryMessageBus.QueueMessage: Metadata is disposed for: {UniqueID} direct flush of the message.", uniqueID);
-                return _innerMessageBus.QueueMessage(message);
+                return InternalQueueMessage(message);
             }
 
             var totalExecutions = metadata.TotalExecutions;
@@ -122,11 +122,25 @@ public bool QueueMessage(object? message)
             }
 
             lstRetryInstance.Add(message);
+
+            // Bypass some events to trigger MessageSink events. (Allure lib required it to create the test context)
+            // but just send the event once.
+            var messageTypeName = message.GetType().Name;
+            if (messageTypeName is "TestStarting" or "TestClassConstructionStarting" or "TestClassConstructionFinished")
+            {
+                if ((!metadata.Skipped && metadata.BypassedMessageTypes.Add(messageTypeName)) ||
+                    metadata.EarlyFlakeDetectionEnabled)
+                {
+                    Common.Log.Debug("RetryMessageBus.QueueMessage: Message bypass, flushing directly for: {UniqueID} | {MessageType}", uniqueID, messageTypeName);
+                    return InternalQueueMessage(message);
+                }
+            }
+
             return true;
         }
 
         Common.Log.Error("RetryMessageBus.QueueMessage: Message doesn't have an UniqueID. Flushing: {Message}", message);
-        return _innerMessageBus.QueueMessage(message);
+        return InternalQueueMessage(message);
     }
 
     public bool FlushMessages(string uniqueID)
@@ -135,9 +149,10 @@ public bool FlushMessages(string uniqueID)
 
         var metadata = (RetryTestCaseMetadata)GetMetadata(uniqueID);
         var listOfMessages = metadata.ListOfMessages;
-        if (listOfMessages is null || listOfMessages.Length == 0 || metadata.Disposed)
+        if (listOfMessages is null || listOfMessages.Length == 0 || metadata.Disposed || metadata.Skipped)
         {
             Common.Log.Debug("RetryMessageBus.FlushMessages: Nothing to flush for: {UniqueID}", uniqueID);
+            metadata.ListOfMessages = null;
             return true;
         }
 
@@ -169,9 +184,19 @@ public bool FlushMessages(string uniqueID)
         bool InternalFlushMessages(List<object> messages)
         {
             var retValue = true;
+            Common.Log.Debug("RetryMessageBus.InternalFlushMessages: Flushing messages for: {UniqueID}", uniqueID);
             foreach (var messageInList in messages)
             {
-                retValue = retValue && _innerMessageBus.QueueMessage(messageInList);
+                var messageTypeName = messageInList.GetType().Name;
+                if (messageTypeName is "TestStarting" or "TestClassConstructionStarting" or "TestClassConstructionFinished")
+                {
+                    Common.Log.Debug("RetryMessageBus.InternalFlushMessages: Skipping message: {Message} for: {UniqueID}", messageInList, uniqueID);
+                }
+                else
+                {
+                    Common.Log.Debug("RetryMessageBus.InternalFlushMessages: Flushing message: {Message} for: {UniqueID}", messageInList, uniqueID);
+                    retValue = InternalQueueMessage(messageInList) && retValue;
+                }
             }
 
             Common.Log.Debug<int, string>("RetryMessageBus.InternalFlushMessages: {Count} messages flushed for: {UniqueID}", messages.Count, uniqueID);
@@ -181,6 +206,19 @@ bool InternalFlushMessages(List<object> messages)
         }
     }
 
+    private bool InternalQueueMessage(object message)
+    {
+        try
+        {
+            return _innerMessageBus.QueueMessage(message);
+        }
+        catch (Exception ex)
+        {
+            Common.Log.Error(ex, "RetryMessageBus.InternalQueueMessage: Error while queueing message: {Message}", message);
+            return false;
+        }
+    }
+
 #pragma warning disable SA1201 // ElementsMustAppearInTheCorrectOrder
     internal interface ITestCaseMessage
     {
@@ -214,6 +252,14 @@ public List<object>?[]? ListOfMessages
 
         public bool Disposed { get; set; }
 
+        /// <summary>
+        /// Gets the messages types that were bypassed to trigger MessageSink events.
+        /// This is used to avoid sending the same message multiple times.
+        /// For example, TestStarting, TestClassConstructionStarting, TestClassConstructionFinished, etc.
+        /// "TestCaseStarting" or "TestMethodStarting" are not bypassed, as they are used to create the test context.
+        /// </summary>
+        public HashSet<string> BypassedMessageTypes { get; } = new();
+
         public void ResizeListOfMessages(int totalExecutions)
         {
             Array.Resize(ref _listOfMessages, totalExecutions);

tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/Testing/XUnit/TestCaseMetadata.cs

@@ -43,5 +43,7 @@ public TestCaseMetadata(string uniqueID, int totalExecution, int countDownExecut
 
     public bool AllRetriesFailed { get; set; } = true;
 
+    public bool Skipped { get; set; }
+
     public string UniqueID { get; }
 }

tracer/src/Datadog.Trace/ClrProfiler/AutoInstrumentation/Testing/XUnit/XUnitTestRunnerRunAsyncIntegration.cs

@@ -129,6 +129,7 @@ internal static CallTargetState OnMethodBegin<TTarget>(TTarget instance)
         {
             runnerInstance.SkipReason = "Flaky test is disabled by Datadog";
             testRunnerInstance.SkipReason = runnerInstance.SkipReason;
+            testCaseMetadata.Skipped = true;
             Common.Log.Debug("XUnitTestRunnerRunAsyncIntegration: Skipping test: {Class}.{Name} Reason: {Reason}", runnerInstance.TestClass?.ToString() ?? string.Empty, runnerInstance.TestMethod?.Name ?? string.Empty, runnerInstance.SkipReason);
             XUnitIntegration.CreateTest(ref runnerInstance, testCaseMetadata);
         }

tracer/test/Datadog.Trace.ClrProfiler.IntegrationTests/CI/TcpXUnitEvpTests.cs

@@ -4,6 +4,7 @@
 // </copyright>
 #if NETCOREAPP3_1_OR_GREATER
 using System.Threading.Tasks;
+using Datadog.Trace.Configuration;
 using Xunit;
 using Xunit.Abstractions;
 

tracer/test/test-applications/integrations/Samples.XUnitTests/Samples.XUnitTests.csproj

@@ -39,6 +39,10 @@
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
       <PrivateAssets>all</PrivateAssets>
     </PackageReference>
+    <PackageReference 
+      Condition="$([MSBuild]::VersionGreaterThanOrEquals($(ApiVersion), '2.9.0')) AND $([MSBuild]::VersionLessThan($(ApiVersion),  '3.0.0'))"
+      Include="Allure.Xunit"
+      Version="2.12.1" />
   </ItemGroup>
 
   <ItemGroup>

tracer/test/test-applications/integrations/Samples.XUnitTests/TestSuite.cs

@@ -1,6 +1,7 @@
 using System;
 using System.Collections;
 using System.Diagnostics;
+using System.Linq;
 using Xunit;
 using Xunit.Abstractions;
 
@@ -20,6 +21,16 @@ public TestSuite(ITestOutputHelper output)
         public void SimplePassTest()
         {
             _output.WriteLine("Test:SimplePassTest");
+            
+            // Let's call the Allure API to add tags if available
+            var allureApi = AppDomain.CurrentDomain.GetAssemblies()
+                                     .Select(asm => asm.GetType("Allure.Net.Commons.AllureApi", false))
+                                     .FirstOrDefault(type => type != null);
+            if (allureApi is not null)
+            {
+                _output.WriteLine("Allure API found, adding tags.");
+                allureApi.GetMethod("AddTags")?.Invoke(null, [new[] { "tag1", "tag2" }]);
+            }
         }
 
         [Fact(Skip = "Simple skip reason")]