diff --git a/src/Microsoft.Diagnostics.Monitoring.EventPipe/Configuration/ActivitySourceConfiguration.cs b/src/Microsoft.Diagnostics.Monitoring.EventPipe/Configuration/ActivitySourceConfiguration.cs
new file mode 100644
index 0000000000..7168691a45
--- /dev/null
+++ b/src/Microsoft.Diagnostics.Monitoring.EventPipe/Configuration/ActivitySourceConfiguration.cs
@@ -0,0 +1,69 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+#nullable enable
+
+using System;
+using System.Collections.Generic;
+using System.Diagnostics.Tracing;
+using System.Linq;
+using System.Text;
+using Microsoft.Diagnostics.NETCore.Client;
+
+namespace Microsoft.Diagnostics.Monitoring.EventPipe
+{
+    public sealed class ActivitySourceConfiguration : MonitoringSourceConfiguration
+    {
+        private readonly double _samplingRatio;
+        private readonly string[] _activitySourceNames;
+
+        public ActivitySourceConfiguration(
+            double samplingRatio,
+            IEnumerable<string>? activitySourceNames)
+        {
+            _samplingRatio = samplingRatio;
+            _activitySourceNames = activitySourceNames?.ToArray() ?? Array.Empty<string>();
+        }
+
+        public override IList<EventPipeProvider> GetProviders()
+        {
+            StringBuilder filterAndPayloadSpecs = new();
+            foreach (string activitySource in _activitySourceNames)
+            {
+                if (string.IsNullOrEmpty(activitySource))
+                {
+                    continue;
+                }
+
+                // Note: It isn't currently possible to get Events or Links off
+                // of Activity using this mechanism:
+                // Events=Events.*Enumerate;Links=Links.*Enumerate; See:
+                // https://github.com/dotnet/runtime/issues/102924
+
+                string sampler = string.Empty;
+
+                if (_samplingRatio < 1D)
+                {
+                    sampler = $"-ParentRatioSampler({_samplingRatio})";
+                }
+
+                filterAndPayloadSpecs.AppendLine($"[AS]{activitySource}/Stop{sampler}:-TraceId;SpanId;ParentSpanId;ActivityTraceFlags;TraceStateString;Kind;DisplayName;StartTimeTicks=StartTimeUtc.Ticks;DurationTicks=Duration.Ticks;Status;StatusDescription;Tags=TagObjects.*Enumerate;ActivitySourceVersion=Source.Version");
+            }
+
+            // Note: Microsoft-Diagnostics-DiagnosticSource only supports a
+            // single listener. There can only be one
+            // ActivitySourceConfiguration, AspNetTriggerSourceConfiguration, or
+            // HttpRequestSourceConfiguration in play.
+            return new[] {
+                new EventPipeProvider(
+                    DiagnosticSourceEventSource,
+                    keywords: DiagnosticSourceEventSourceEvents | DiagnosticSourceEventSourceMessages,
+                    eventLevel: EventLevel.Verbose,
+                    arguments: new Dictionary<string, string>()
+                    {
+                        { "FilterAndPayloadSpecs", filterAndPayloadSpecs.ToString() },
+                    })
+            };
+        }
+    }
+}
diff --git a/src/Microsoft.Diagnostics.Monitoring.EventPipe/Counters/TraceEventExtensions.cs b/src/Microsoft.Diagnostics.Monitoring.EventPipe/Counters/TraceEventExtensions.cs
index d6833da307..4b170a68dd 100644
--- a/src/Microsoft.Diagnostics.Monitoring.EventPipe/Counters/TraceEventExtensions.cs
+++ b/src/Microsoft.Diagnostics.Monitoring.EventPipe/Counters/TraceEventExtensions.cs
@@ -34,7 +34,7 @@ public CounterConfiguration(CounterFilter filter)
 
     internal record struct ProviderAndCounter(string ProviderName, string CounterName);
 
-    internal static class TraceEventExtensions
+    internal static partial class TraceEventExtensions
     {
         private static Dictionary<ProviderAndCounter, CounterMetadata> counterMetadataByName = new();
         private static Dictionary<int, CounterMetadata> counterMetadataById = new();
diff --git a/src/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTraces/ActivityData.cs b/src/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTraces/ActivityData.cs
new file mode 100644
index 0000000000..04ba907854
--- /dev/null
+++ b/src/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTraces/ActivityData.cs
@@ -0,0 +1,74 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+#nullable enable
+
+using System;
+using System.Diagnostics;
+
+namespace Microsoft.Diagnostics.Monitoring.EventPipe
+{
+    internal readonly struct ActivityData
+    {
+        public ActivityData(
+            ActivitySourceData source,
+            string operationName,
+            string? displayName,
+            ActivityKind kind,
+            ActivityTraceId traceId,
+            ActivitySpanId spanId,
+            ActivitySpanId parentSpanId,
+            ActivityTraceFlags traceFlags,
+            string? traceState,
+            DateTime startTimeUtc,
+            DateTime endTimeUtc,
+            ActivityStatusCode status,
+            string? statusDescription)
+        {
+            if (string.IsNullOrEmpty(operationName))
+            {
+                throw new ArgumentNullException(nameof(operationName));
+            }
+
+            Source = source;
+            OperationName = operationName;
+            DisplayName = displayName;
+            Kind = kind;
+            TraceId = traceId;
+            SpanId = spanId;
+            ParentSpanId = parentSpanId;
+            TraceFlags = traceFlags;
+            TraceState = traceState;
+            StartTimeUtc = startTimeUtc;
+            EndTimeUtc = endTimeUtc;
+            Status = status;
+            StatusDescription = statusDescription;
+        }
+
+        public readonly ActivitySourceData Source;
+
+        public readonly string OperationName;
+
+        public readonly string? DisplayName;
+
+        public readonly ActivityKind Kind;
+
+        public readonly ActivityTraceId TraceId;
+
+        public readonly ActivitySpanId SpanId;
+
+        public readonly ActivitySpanId ParentSpanId;
+
+        public readonly ActivityTraceFlags TraceFlags;
+
+        public readonly string? TraceState;
+
+        public readonly DateTime StartTimeUtc;
+
+        public readonly DateTime EndTimeUtc;
+
+        public readonly ActivityStatusCode Status;
+
+        public readonly string? StatusDescription;
+    }
+}
diff --git a/src/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTraces/ActivityPayload.cs b/src/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTraces/ActivityPayload.cs
new file mode 100644
index 0000000000..849055435a
--- /dev/null
+++ b/src/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTraces/ActivityPayload.cs
@@ -0,0 +1,17 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+#nullable enable
+
+using System;
+using System.Collections.Generic;
+
+namespace Microsoft.Diagnostics.Monitoring.EventPipe
+{
+    internal ref struct ActivityPayload
+    {
+        public ActivityData ActivityData;
+
+        public ReadOnlySpan<KeyValuePair<string, object?>> Tags;
+    }
+}
diff --git a/src/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTraces/ActivitySourceData.cs b/src/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTraces/ActivitySourceData.cs
new file mode 100644
index 0000000000..de09dbb457
--- /dev/null
+++ b/src/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTraces/ActivitySourceData.cs
@@ -0,0 +1,21 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+#nullable enable
+
+namespace Microsoft.Diagnostics.Monitoring.EventPipe
+{
+    internal sealed class ActivitySourceData
+    {
+        public ActivitySourceData(
+            string name,
+            string? version)
+        {
+            Name = name;
+            Version = version;
+        }
+
+        public string Name { get; }
+        public string? Version { get; }
+    }
+}
diff --git a/src/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTraces/DistributedTracesPipeline.cs b/src/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTraces/DistributedTracesPipeline.cs
new file mode 100644
index 0000000000..c03f92d28d
--- /dev/null
+++ b/src/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTraces/DistributedTracesPipeline.cs
@@ -0,0 +1,150 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+#nullable enable
+
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Diagnostics.NETCore.Client;
+using Microsoft.Diagnostics.Tracing;
+
+namespace Microsoft.Diagnostics.Monitoring.EventPipe
+{
+    internal class DistributedTracesPipeline : EventSourcePipeline<DistributedTracesPipelineSettings>
+    {
+        private readonly IActivityLogger[] _loggers;
+
+        public DistributedTracesPipeline(DiagnosticsClient client,
+            DistributedTracesPipelineSettings settings,
+            IEnumerable<IActivityLogger> loggers) : base(client, settings)
+        {
+            _loggers = loggers?.ToArray() ?? throw new ArgumentNullException(nameof(loggers));
+        }
+
+        protected override MonitoringSourceConfiguration CreateConfiguration()
+            => new ActivitySourceConfiguration(Settings.SamplingRatio, Settings.Sources);
+
+        protected override async Task OnRun(CancellationToken token)
+        {
+            double samplingRatio = Settings.SamplingRatio;
+            if (samplingRatio < 1D)
+            {
+                await ValidateEventSourceVersion().ConfigureAwait(false);
+            }
+
+            await base.OnRun(token).ConfigureAwait(false);
+        }
+
+        private async Task ValidateEventSourceVersion()
+        {
+            int majorVersion = 0;
+
+            using CancellationTokenSource cancellationTokenSource = new();
+
+            DiagnosticsEventPipeProcessor processor = new(
+                new ActivitySourceConfiguration(1D, activitySourceNames: null),
+                async (EventPipeEventSource eventSource, Func<Task> stopSessionAsync, CancellationToken token) => {
+                    eventSource.Dynamic.All += traceEvent => {
+                        try
+                        {
+                            if ("Version".Equals(traceEvent.EventName))
+                            {
+                                majorVersion = (int)traceEvent.PayloadValue(0);
+                            }
+
+                            if (!cancellationTokenSource.IsCancellationRequested)
+                            {
+                                // Note: Version should be the first message
+                                // written so cancel once we have received a
+                                // message.
+                                cancellationTokenSource.Cancel();
+                            }
+                        }
+                        catch (Exception)
+                        {
+                        }
+                    };
+
+                    using EventTaskSource<Action> sourceCompletedTaskSource = new(
+                        taskComplete => taskComplete,
+                        handler => eventSource.Completed += handler,
+                        handler => eventSource.Completed -= handler,
+                        token);
+
+                    await sourceCompletedTaskSource.Task.ConfigureAwait(false);
+                });
+
+            try
+            {
+                await processor.Process(Client, TimeSpan.FromSeconds(10), resumeRuntime: false, token: cancellationTokenSource.Token).ConfigureAwait(false);
+            }
+            catch (OperationCanceledException)
+            {
+            }
+
+            await processor.DisposeAsync().ConfigureAwait(false);
+
+            if (majorVersion < 9)
+            {
+                throw new PipelineException("Sampling ratio can only be set when listening to processes running System.Diagnostics.DiagnosticSource 9 or greater");
+            }
+        }
+
+        protected override async Task OnEventSourceAvailable(EventPipeEventSource eventSource, Func<Task> stopSessionAsync, CancellationToken token)
+        {
+            await ExecuteCounterLoggerActionAsync((logger) => logger.PipelineStarted(token)).ConfigureAwait(false);
+
+            eventSource.Dynamic.All += traceEvent => {
+                try
+                {
+                    if (traceEvent.TryGetActivityPayload(out ActivityPayload activity))
+                    {
+                        foreach (IActivityLogger logger in _loggers)
+                        {
+                            try
+                            {
+                                logger.Log(
+                                    in activity.ActivityData,
+                                    activity.Tags);
+                            }
+                            catch (ObjectDisposedException)
+                            {
+                            }
+                        }
+                    }
+                }
+                catch (Exception)
+                {
+                }
+            };
+
+            using EventTaskSource<Action> sourceCompletedTaskSource = new(
+                taskComplete => taskComplete,
+                handler => eventSource.Completed += handler,
+                handler => eventSource.Completed -= handler,
+                token);
+
+            await sourceCompletedTaskSource.Task.ConfigureAwait(false);
+
+            await ExecuteCounterLoggerActionAsync((logger) => logger.PipelineStopped(token)).ConfigureAwait(false);
+        }
+
+        private async Task ExecuteCounterLoggerActionAsync(Func<IActivityLogger, Task> action)
+        {
+            foreach (IActivityLogger logger in _loggers)
+            {
+                try
+                {
+                    await action(logger).ConfigureAwait(false);
+                }
+                catch (ObjectDisposedException)
+                {
+                }
+            }
+        }
+    }
+}
diff --git a/src/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTraces/DistributedTracesPipelineSettings.cs b/src/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTraces/DistributedTracesPipelineSettings.cs
new file mode 100644
index 0000000000..2f01db3376
--- /dev/null
+++ b/src/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTraces/DistributedTracesPipelineSettings.cs
@@ -0,0 +1,14 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+#nullable enable
+
+namespace Microsoft.Diagnostics.Monitoring.EventPipe
+{
+    internal class DistributedTracesPipelineSettings : EventSourcePipelineSettings
+    {
+        public double SamplingRatio { get; set; } = 1.0D;
+
+        public string[]? Sources { get; set; }
+    }
+}
diff --git a/src/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTraces/IActivityLogger.cs b/src/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTraces/IActivityLogger.cs
new file mode 100644
index 0000000000..89882085ca
--- /dev/null
+++ b/src/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTraces/IActivityLogger.cs
@@ -0,0 +1,22 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+#nullable enable
+
+using System;
+using System.Collections.Generic;
+using System.Threading;
+using System.Threading.Tasks;
+
+namespace Microsoft.Diagnostics.Monitoring.EventPipe
+{
+    internal interface IActivityLogger
+    {
+        void Log(
+            in ActivityData activity,
+            ReadOnlySpan<KeyValuePair<string, object?>> tags);
+
+        Task PipelineStarted(CancellationToken token);
+        Task PipelineStopped(CancellationToken token);
+    }
+}
diff --git a/src/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTraces/TraceEventExtensions.cs b/src/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTraces/TraceEventExtensions.cs
new file mode 100644
index 0000000000..f5da906821
--- /dev/null
+++ b/src/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTraces/TraceEventExtensions.cs
@@ -0,0 +1,223 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+#nullable enable
+
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using Microsoft.Diagnostics.Tracing;
+
+namespace Microsoft.Diagnostics.Monitoring.EventPipe
+{
+    internal static partial class TraceEventExtensions
+    {
+        private const string DefaultTraceId = "00000000000000000000000000000000";
+        private const string DefaultSpanId = "0000000000000000";
+
+        [ThreadStatic]
+        private static KeyValuePair<string, object?>[]? s_TagStorage;
+        private static readonly Dictionary<string, ActivitySourceData> s_Sources = new(StringComparer.OrdinalIgnoreCase);
+
+        public static bool TryGetActivityPayload(this TraceEvent traceEvent, out ActivityPayload payload)
+        {
+            if ("Activity/Stop".Equals(traceEvent.EventName))
+            {
+                string sourceName = (traceEvent.PayloadValue(0) as string) ?? string.Empty;
+                string? activityName = traceEvent.PayloadValue(1) as string;
+                Array? arguments = traceEvent.PayloadValue(2) as Array;
+
+                if (string.IsNullOrEmpty(activityName)
+                    || arguments == null)
+                {
+                    payload = default;
+                    return false;
+                }
+
+                ActivitySourceData source;
+                string? displayName = null;
+                ActivityTraceId traceId = default;
+                ActivitySpanId spanId = default;
+                ActivitySpanId parentSpanId = default;
+                ActivityTraceFlags traceFlags = default;
+                string? traceState = null;
+                ActivityKind kind = default;
+                ActivityStatusCode status = ActivityStatusCode.Unset;
+                string? statusDescription = null;
+                string? sourceVersion = null;
+                DateTime startTimeUtc = default;
+                long durationTicks = 0;
+                int tagCount = 0;
+
+                foreach (IDictionary<string, object> arg in arguments)
+                {
+                    string? key = arg["Key"] as string;
+                    object value = arg["Value"];
+
+                    switch (key)
+                    {
+                        case "TraceId":
+                            string? traceIdValue = value as string;
+                            if (!string.IsNullOrEmpty(traceIdValue)
+                                && traceIdValue != DefaultTraceId)
+                            {
+                                traceId = ActivityTraceId.CreateFromString(traceIdValue);
+                            }
+                            break;
+                        case "SpanId":
+                            string? spanIdValue = value as string;
+                            if (!string.IsNullOrEmpty(spanIdValue)
+                                && spanIdValue != DefaultSpanId)
+                            {
+                                spanId = ActivitySpanId.CreateFromString(spanIdValue);
+                            }
+                            break;
+                        case "ParentSpanId":
+                            string? parentSpanIdValue = value as string;
+                            if (!string.IsNullOrEmpty(parentSpanIdValue)
+                                && parentSpanIdValue != DefaultSpanId)
+                            {
+                                parentSpanId = ActivitySpanId.CreateFromString(parentSpanIdValue);
+                            }
+                            break;
+                        case "ActivityTraceFlags":
+                            if (value is string traceFlagsValue)
+                            {
+                                traceFlags = (ActivityTraceFlags)Enum.Parse(typeof(ActivityTraceFlags), traceFlagsValue);
+                            }
+                            break;
+                        case "TraceStateString":
+                            traceState = value as string;
+                            break;
+                        case "Kind":
+                            if (value is string kindValue)
+                            {
+                                kind = (ActivityKind)Enum.Parse(typeof(ActivityKind), kindValue);
+                            }
+                            break;
+                        case "DisplayName":
+                            string? displayNameValue = value as string;
+                            if (!string.IsNullOrEmpty(displayNameValue)
+                                && displayNameValue != activityName)
+                            {
+                                displayName = displayNameValue;
+                            }
+
+                            break;
+                        case "StartTimeTicks":
+                            if (value is string startTimeUtcValue)
+                            {
+                                startTimeUtc = DateTime.SpecifyKind(new DateTime(long.Parse(startTimeUtcValue)), DateTimeKind.Utc);
+                            }
+                            break;
+                        case "DurationTicks":
+                            if (value is string durationTicksValue)
+                            {
+                                durationTicks = long.Parse(durationTicksValue);
+                            }
+                            break;
+                        case "Status":
+                            if (value is string statusValue)
+                            {
+                                status = (ActivityStatusCode)Enum.Parse(typeof(ActivityStatusCode), statusValue);
+                            }
+                            break;
+                        case "StatusDescription":
+                            statusDescription = value as string;
+                            break;
+                        case "Tags":
+                            string? tagsValue = value as string;
+                            if (!string.IsNullOrEmpty(tagsValue))
+                            {
+                                tagCount = ParseTags(tagsValue);
+                            }
+                            break;
+                        case "ActivitySourceVersion":
+                            sourceVersion = value as string;
+                            break;
+                    }
+                }
+
+                if (!s_Sources.TryGetValue(sourceName, out source))
+                {
+                    source = new(sourceName, sourceVersion);
+                    s_Sources[sourceName] = source;
+                }
+
+                payload.ActivityData = new ActivityData(
+                    source,
+                    activityName,
+                    displayName,
+                    kind,
+                    traceId,
+                    spanId,
+                    parentSpanId,
+                    traceFlags,
+                    traceState,
+                    startTimeUtc,
+                    startTimeUtc + TimeSpan.FromTicks(durationTicks),
+                    status,
+                    statusDescription);
+
+                payload.Tags = new(s_TagStorage, 0, tagCount);
+
+                return true;
+            }
+
+            payload = default;
+            return false;
+        }
+
+        private static int ParseTags(string tagsValue)
+        {
+            ref KeyValuePair<string, object?>[]? tags = ref s_TagStorage;
+            tags ??= new KeyValuePair<string, object?>[16];
+
+            int tagCount = 0;
+
+            for (int i = 0; i < tagsValue.Length; i++)
+            {
+                if (tagsValue[i++] != '[')
+                {
+                    break;
+                }
+
+                int commaPosition = tagsValue.IndexOf(',', i);
+                if (commaPosition < 0)
+                {
+                    break;
+                }
+
+                string key = tagsValue.Substring(i, commaPosition - i);
+
+                i = commaPosition + 2;
+
+                int endPosition = tagsValue.IndexOf("]", i);
+                if (endPosition < 0)
+                {
+                    break;
+                }
+
+                string value = tagsValue.Substring(i, endPosition - i);
+
+                i = endPosition + 1;
+
+                AddToArrayGrowingAsNeeded(ref tags, new(key, value), ref tagCount);
+            }
+
+            return tagCount;
+        }
+
+        private static void AddToArrayGrowingAsNeeded<T>(ref T[] destination, T item, ref int index)
+        {
+            if (index >= destination.Length)
+            {
+                T[] newArray = new T[destination.Length * 2];
+                Array.Copy(destination, newArray, destination.Length);
+                destination = newArray;
+            }
+
+            destination[index++] = item;
+        }
+    }
+}
diff --git a/src/Microsoft.Diagnostics.Monitoring.EventPipe/EventSourcePipeline.cs b/src/Microsoft.Diagnostics.Monitoring.EventPipe/EventSourcePipeline.cs
index 1b36f04e52..6cd78b1546 100644
--- a/src/Microsoft.Diagnostics.Monitoring.EventPipe/EventSourcePipeline.cs
+++ b/src/Microsoft.Diagnostics.Monitoring.EventPipe/EventSourcePipeline.cs
@@ -86,7 +86,19 @@ public async Task<Task> StartAsync(CancellationToken token)
             // started task. Logically, the run task will not successfully complete before the session
             // started task. Thus, the combined task completes either when the session started task is
             // completed OR the run task has cancelled/failed.
-            await Task.WhenAny(_processor.Value.SessionStarted, runTask).Unwrap().ConfigureAwait(false);
+            try
+            {
+                await Task.WhenAny(_processor.Value.SessionStarted, runTask).Unwrap().ConfigureAwait(false);
+            }
+            catch (TaskCanceledException)
+            {
+                if (runTask.IsFaulted)
+                {
+                    throw runTask.Exception.InnerException;
+                }
+
+                throw;
+            }
 
             return runTask;
         }
diff --git a/src/tests/EventPipeTracee/Program.cs b/src/tests/EventPipeTracee/Program.cs
index 9c61e0db0f..bfac20e59d 100644
--- a/src/tests/EventPipeTracee/Program.cs
+++ b/src/tests/EventPipeTracee/Program.cs
@@ -33,8 +33,9 @@ public static async Task<int> Main(string[] args)
 
             bool diagMetrics = args.Any("DiagMetrics".Equals);
             bool duplicateNameMetrics = args.Any("DuplicateNameMetrics".Equals);
+            bool useActivitySource = args.Any("UseActivitySource".Equals);
 
-            Console.WriteLine($"{pid} EventPipeTracee: DiagMetrics {diagMetrics}");
+            Console.WriteLine($"{pid} EventPipeTracee: DiagMetrics {diagMetrics} UseActivitySource {useActivitySource}");
             Console.WriteLine($"{pid} EventPipeTracee: DuplicateNameMetrics {duplicateNameMetrics}");
 
             Console.WriteLine($"{pid} EventPipeTracee: start process");
@@ -59,6 +60,10 @@ public static async Task<int> Main(string[] args)
             ILogger customCategoryLogger = loggerFactory.CreateLogger(loggerCategory);
             ILogger appCategoryLogger = loggerFactory.CreateLogger(AppLoggerCategoryName);
 
+            using ActivitySource activitySource = useActivitySource
+                ? new ActivitySource("EventPipeTracee.ActivitySource", version: "1.0.0")
+                : null;
+
             Console.WriteLine($"{pid} EventPipeTracee: {DateTime.UtcNow} Awaiting start");
             Console.Out.Flush();
 
@@ -96,7 +101,7 @@ public static async Task<int> Main(string[] args)
                 }).ConfigureAwait(true);
             }
 
-            await TestBodyCore(customCategoryLogger, appCategoryLogger).ConfigureAwait(false);
+            await TestBodyCore(customCategoryLogger, appCategoryLogger, activitySource).ConfigureAwait(false);
 
             Console.WriteLine($"{pid} EventPipeTracee: signal end of test data");
             Console.Out.Flush();
@@ -130,8 +135,32 @@ public static async Task<int> Main(string[] args)
         }
 
         // TODO At some point we may want parameters to choose different test bodies.
-        private static async Task TestBodyCore(ILogger customCategoryLogger, ILogger appCategoryLogger)
+        private static async Task TestBodyCore(ILogger customCategoryLogger, ILogger appCategoryLogger, ActivitySource activitySource)
         {
+            using Activity activity = activitySource?.StartActivity(
+                name: "TestBodyCore",
+                kind: ActivityKind.Client,
+                links: new ActivityLink[] {
+                    new(
+                        ActivityContext.Parse(
+                            traceParent: "00-99d43cb30a4cdb4fbeee3a19c29201b0-e82825765f051b47-01",
+                            traceState: "k1=v1;k2=v2"))
+                });
+
+            if (activity != null)
+            {
+                activity.DisplayName = "Display name";
+                if (activity.IsAllDataRequested)
+                {
+                    activity.SetTag("custom.tag.string", "value1");
+                    activity.SetTag("custom.tag.int", 18);
+                }
+                activity.SetStatus(ActivityStatusCode.Error, "Error occurred");
+                activity.AddEvent(new ActivityEvent(
+                    name: "MyEvent",
+                    tags: new ActivityTagsCollection { ["tag1"] = "value1", ["tag2"] = 18 }));
+            }
+
             TaskCompletionSource secondSetScopes = new(TaskCreationOptions.RunContinuationsAsynchronously);
             TaskCompletionSource firstFinishedLogging = new(TaskCreationOptions.RunContinuationsAsynchronously);
             TaskCompletionSource secondFinishedLogging = new(TaskCreationOptions.RunContinuationsAsynchronously);
diff --git a/src/tests/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTracesPipelineUnitTests.cs b/src/tests/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTracesPipelineUnitTests.cs
new file mode 100644
index 0000000000..b9d6d00338
--- /dev/null
+++ b/src/tests/Microsoft.Diagnostics.Monitoring.EventPipe/DistributedTracesPipelineUnitTests.cs
@@ -0,0 +1,162 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+#nullable enable
+
+using System;
+using System.Collections.Generic;
+using System.Diagnostics;
+using System.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.Diagnostics.NETCore.Client;
+using Microsoft.Diagnostics.TestHelpers;
+using Xunit;
+using Xunit.Abstractions;
+using Xunit.Extensions;
+using TestRunner = Microsoft.Diagnostics.CommonTestRunner.TestRunner;
+
+namespace Microsoft.Diagnostics.Monitoring.EventPipe.UnitTests
+{
+    public class DistributedTracesPipelineUnitTests
+    {
+        private readonly ITestOutputHelper _output;
+
+        public static IEnumerable<object[]> Configurations => TestRunner.Configurations;
+
+        public DistributedTracesPipelineUnitTests(ITestOutputHelper output)
+        {
+            _output = output;
+        }
+
+        [SkippableTheory, MemberData(nameof(Configurations))]
+        public async Task TestTracesPipeline(TestConfiguration config)
+        {
+            TestActivityLogger logger = new();
+
+            await using (TestRunner testRunner = await PipelineTestUtilities.StartProcess(config, "TracesRemoteTest UseActivitySource", _output))
+            {
+                DiagnosticsClient client = new(testRunner.Pid);
+
+                await using DistributedTracesPipeline pipeline = new(client, new DistributedTracesPipelineSettings
+                {
+                    Sources = new[] { "*" },
+                    SamplingRatio = 1D,
+                    Duration = Timeout.InfiniteTimeSpan,
+                }, new[] { logger });
+
+                await PipelineTestUtilities.ExecutePipelineWithTracee(
+                    pipeline,
+                    testRunner);
+            }
+
+            Assert.Single(logger.LoggedActivities);
+
+            var activityData = logger.LoggedActivities[0];
+
+            var activity = activityData.Item1;
+
+            Assert.Equal("TestBodyCore", activity.OperationName);
+            Assert.Equal("Display name", activity.DisplayName);
+            Assert.NotEqual(default, activity.TraceId);
+            Assert.NotEqual(default, activity.SpanId);
+            Assert.Equal(default, activity.ParentSpanId);
+            Assert.Equal(ActivityTraceFlags.Recorded, activity.TraceFlags);
+            Assert.Equal(ActivityKind.Client, activity.Kind);
+            Assert.NotEqual(default, activity.StartTimeUtc);
+            Assert.NotEqual(default, activity.EndTimeUtc);
+            Assert.NotEqual(TimeSpan.Zero, activity.EndTimeUtc - activity.StartTimeUtc);
+            Assert.Equal(ActivityStatusCode.Error, activity.Status);
+            Assert.Equal("Error occurred", activity.StatusDescription);
+
+            Assert.NotNull(activity.Source);
+            Assert.Equal("EventPipeTracee.ActivitySource", activity.Source.Name);
+            Assert.Equal("1.0.0", activity.Source.Version);
+
+            Dictionary<string, object?> tags = activityData.Item2.ToDictionary(
+                i => i.Key,
+                i => i.Value);
+            Assert.Equal("value1", tags["custom.tag.string"]);
+            Assert.Equal("18", tags["custom.tag.int"]);
+        }
+
+        [SkippableTheory, MemberData(nameof(Configurations))]
+        public async Task TestTracesPipelineWithSamplingRatio(TestConfiguration config)
+        {
+            TestActivityLogger logger = new();
+
+            await using (TestRunner testRunner = await PipelineTestUtilities.StartProcess(config, "TracesRemoteTest UseActivitySource", _output))
+            {
+                DiagnosticsClient client = new(testRunner.Pid);
+
+                await using DistributedTracesPipeline pipeline = new(client, new DistributedTracesPipelineSettings
+                {
+                    Sources = new[] { "*" },
+                    SamplingRatio = 0.0D,
+                    Duration = Timeout.InfiniteTimeSpan,
+                }, new[] { logger });
+
+                if (config.RuntimeFrameworkVersionMajor < 9)
+                {
+                    // Note: Pipeline should throw PipelineException when
+                    // setting SamplingRatio on runtimes < .NET 9.
+                    await Assert.ThrowsAsync<PipelineException>(
+                        async () => {
+                            await PipelineTestUtilities.ExecutePipelineWithTracee(
+                                pipeline,
+                                testRunner);
+                        });
+
+                    return;
+                }
+                else
+                {
+                    await PipelineTestUtilities.ExecutePipelineWithTracee(
+                        pipeline,
+                        testRunner);
+                }
+            }
+
+            Assert.Single(logger.LoggedActivities);
+
+            var activityData = logger.LoggedActivities[0];
+
+            var activity = activityData.Item1;
+
+            Assert.Equal("TestBodyCore", activity.OperationName);
+            Assert.Equal("Display name", activity.DisplayName);
+            Assert.NotEqual(default, activity.TraceId);
+            Assert.NotEqual(default, activity.SpanId);
+            Assert.Equal(default, activity.ParentSpanId);
+            Assert.Equal(ActivityTraceFlags.None, activity.TraceFlags);
+            Assert.Equal(ActivityKind.Client, activity.Kind);
+            Assert.NotEqual(default, activity.StartTimeUtc);
+            Assert.NotEqual(default, activity.EndTimeUtc);
+            Assert.NotEqual(TimeSpan.Zero, activity.EndTimeUtc - activity.StartTimeUtc);
+            Assert.Equal(ActivityStatusCode.Error, activity.Status);
+            Assert.Equal("Error occurred", activity.StatusDescription);
+
+            Assert.NotNull(activity.Source);
+            Assert.Equal("EventPipeTracee.ActivitySource", activity.Source.Name);
+            Assert.Equal("1.0.0", activity.Source.Version);
+
+            Assert.Empty(activityData.Item2);
+        }
+
+        private sealed class TestActivityLogger : IActivityLogger
+        {
+            public List<(ActivityData, KeyValuePair<string, object?>[])> LoggedActivities { get; } = new();
+
+            public void Log(
+                in ActivityData activity,
+                ReadOnlySpan<KeyValuePair<string, object?>> tags)
+            {
+                LoggedActivities.Add((activity, tags.ToArray()));
+            }
+
+            public Task PipelineStarted(CancellationToken token) => Task.CompletedTask;
+
+            public Task PipelineStopped(CancellationToken token) => Task.CompletedTask;
+        }
+    }
+}
diff --git a/src/tests/Microsoft.Diagnostics.Monitoring.EventPipe/PipelineTestUtilities.cs b/src/tests/Microsoft.Diagnostics.Monitoring.EventPipe/PipelineTestUtilities.cs
index ffd52880e4..5630f2dd71 100644
--- a/src/tests/Microsoft.Diagnostics.Monitoring.EventPipe/PipelineTestUtilities.cs
+++ b/src/tests/Microsoft.Diagnostics.Monitoring.EventPipe/PipelineTestUtilities.cs
@@ -72,7 +72,16 @@ private static async Task ExecutePipelineWithTracee<TPipeline>(
             TaskCompletionSource<object> waitTaskSource = null)
             where TPipeline : Pipeline
         {
-            Task runTask = await startPipelineAsync(pipeline, token);
+            Task runTask;
+            try
+            {
+                runTask = await startPipelineAsync(pipeline, token);
+            }
+            catch
+            {
+                await TestRunnerUtilities.ExecuteCollection(Task.CompletedTask, testRunner, token);
+                throw;
+            }
 
             Func<CancellationToken, Task> waitForPipeline = async (cancellationToken) => {
                 // Optionally wait on caller before allowing the pipeline to stop.