diff --git a/src/Aspire.Hosting/ApplicationModel/ResourceLoggerService.cs b/src/Aspire.Hosting/ApplicationModel/ResourceLoggerService.cs
index 4a065cd1cbc..76778e42118 100644
--- a/src/Aspire.Hosting/ApplicationModel/ResourceLoggerService.cs
+++ b/src/Aspire.Hosting/ApplicationModel/ResourceLoggerService.cs
@@ -2,6 +2,8 @@
// The .NET Foundation licenses this file to you under the MIT license.
using System.Collections.Concurrent;
+using System.Diagnostics;
+using System.Globalization;
using System.Runtime.CompilerServices;
using System.Threading.Channels;
using Aspire.Dashboard.Otlp.Storage;
@@ -58,6 +60,17 @@ public ILogger GetLogger(string resourceName)
return GetResourceLoggerState(resourceName).Logger;
}
+ ///
+ /// The internal logger is used when adding logs from resource's stream logs.
+ /// It allows the parsed date from text to be used as the log line date.
+ ///
+ internal Action GetInternalLogger(string resourceName)
+ {
+ ArgumentNullException.ThrowIfNull(resourceName);
+
+ return GetResourceLoggerState(resourceName).AddLog;
+ }
+
///
/// Watch for changes to the log stream for a resource.
///
@@ -159,7 +172,8 @@ public void ClearBacklog(string resourceName)
}
}
- private ResourceLoggerState GetResourceLoggerState(string resourceName) =>
+ // Internal for testing.
+ internal ResourceLoggerState GetResourceLoggerState(string resourceName) =>
_loggers.GetOrAdd(resourceName, (name, context) =>
{
var state = new ResourceLoggerState();
@@ -168,15 +182,19 @@ private ResourceLoggerState GetResourceLoggerState(string resourceName) =>
},
this);
+ internal sealed record InternalLogLine(DateTime DateTimeUtc, string Message, bool IsError);
+
///
/// A logger for the resource to write to.
///
- private sealed class ResourceLoggerState
+ internal sealed class ResourceLoggerState
{
private readonly ResourceLogger _logger;
private readonly CancellationTokenSource _logStreamCts = new();
- private readonly CircularBuffer _backlog = new(10000);
+ private Task? _backlogReplayCompleteTask;
+ private long _lastLogReceivedTimestamp;
+ private readonly CircularBuffer _backlog = new(10000);
///
/// Creates a new .
@@ -220,45 +238,46 @@ public event Action OnSubscribersChanged
/// The log stream for the resource.
public async IAsyncEnumerable> WatchAsync([EnumeratorCancellation] CancellationToken cancellationToken = default)
{
- var channel = Channel.CreateUnbounded();
+ // Line number always restarts from 1 when watching logs.
+ // Note that this will need to be improved if the log source (DCP) is changed to return a maximum number of lines.
+ var lineNumber = 1;
+ var channel = Channel.CreateUnbounded();
using var _ = _logStreamCts.Token.Register(() => channel.Writer.TryComplete());
- LogLine[]? backlogSnapshot = default;
- var flushBacklogSync = new ManualResetEventSlim();
- void Log(LogLine log)
+ InternalLogLine[]? backlogSnapshot = null;
+ void Log(InternalLogLine log)
{
- if (flushBacklogSync.IsSet)
- {
- channel.Writer.TryWrite(log);
- return;
- }
-
- flushBacklogSync.Wait(cancellationToken);
- // We need to ensure we don't write this log to the channel if it was already in the backlog
- if (backlogSnapshot?.Contains(log) == false)
+ lock (_backlog)
{
- channel.Writer.TryWrite(log);
+ // Don't write to the channel until the backlog snapshot is accessed.
+ // This prevents duplicate logs in result.
+ if (backlogSnapshot != null)
+ {
+ channel.Writer.TryWrite(log);
+ }
}
}
-
- // From the moment we add this callback, logs will be written to the backlog & to our Log method above
- // so our Log method needs to ensure it de-dupes logs.
OnNewLog += Log;
- backlogSnapshot = GetBacklogSnapshot();
- flushBacklogSync.Set();
+ // Add a small delay to ensure the backlog is replayed from DCP and ordered correctly.
+ await EnsureBacklogReplayAsync(cancellationToken).ConfigureAwait(false);
+
+ lock (_backlog)
+ {
+ backlogSnapshot = GetBacklogSnapshot();
+ }
if (backlogSnapshot.Length > 0)
{
- yield return backlogSnapshot;
+ yield return CreateLogLines(ref lineNumber, backlogSnapshot);
}
try
{
await foreach (var entry in channel.GetBatchesAsync(cancellationToken: cancellationToken).ConfigureAwait(false))
{
- yield return entry;
+ yield return CreateLogLines(ref lineNumber, entry);
}
}
finally
@@ -267,11 +286,51 @@ void Log(LogLine log)
channel.Writer.TryComplete();
}
+
+ static LogLine[] CreateLogLines(ref int lineNumber, IReadOnlyList entry)
+ {
+ var logs = new LogLine[entry.Count];
+ for (var i = 0; i < entry.Count; i++)
+ {
+ logs[i] = new LogLine(lineNumber, entry[i].Message, entry[i].IsError);
+ lineNumber++;
+ }
+
+ return logs;
+ }
+ }
+
+ private Task EnsureBacklogReplayAsync(CancellationToken cancellationToken)
+ {
+ lock (_backlog)
+ {
+ _backlogReplayCompleteTask ??= StartBacklogReplayAsync(cancellationToken);
+ return _backlogReplayCompleteTask;
+ }
+
+ async Task StartBacklogReplayAsync(CancellationToken cancellationToken)
+ {
+ var delay = TimeSpan.FromMilliseconds(100);
+
+ // There could be an initial burst of logs as they're replayed. Give them the opportunity to be loaded
+ // into the backlog in the correct order and returned before streaming logs as they arrive.
+ for (var i = 0; i < 3; i++)
+ {
+ await Task.Delay(delay, cancellationToken).ConfigureAwait(false);
+ lock (_backlog)
+ {
+ if (_lastLogReceivedTimestamp != 0 && Stopwatch.GetElapsedTime(_lastLogReceivedTimestamp) > delay)
+ {
+ break;
+ }
+ }
+ }
+ }
}
// This provides the fan out to multiple subscribers.
- private Action? _onNewLog;
- private event Action OnNewLog
+ private Action? _onNewLog;
+ private event Action OnNewLog
{
add
{
@@ -324,10 +383,11 @@ public void ClearBacklog()
lock (_backlog)
{
_backlog.Clear();
+ _backlogReplayCompleteTask = null;
}
}
- private LogLine[] GetBacklogSnapshot()
+ internal InternalLogLine[] GetBacklogSnapshot()
{
lock (_backlog)
{
@@ -335,10 +395,36 @@ private LogLine[] GetBacklogSnapshot()
}
}
- private sealed class ResourceLogger(ResourceLoggerState loggerState) : ILogger
+ public void AddLog(DateTime dateTimeUtc, string logMessage, bool isErrorMessage)
{
- private int _lineNumber;
+ InternalLogLine logLine;
+ lock (_backlog)
+ {
+ logLine = new InternalLogLine(dateTimeUtc, logMessage, isErrorMessage);
+
+ var added = false;
+ for (var i = _backlog.Count - 1; i >= 0; i--)
+ {
+ if (dateTimeUtc >= _backlog[i].DateTimeUtc)
+ {
+ _backlog.Insert(i + 1, logLine);
+ added = true;
+ break;
+ }
+ }
+ if (!added)
+ {
+ _backlog.Insert(0, logLine);
+ }
+
+ _lastLogReceivedTimestamp = Stopwatch.GetTimestamp();
+ }
+
+ _onNewLog?.Invoke(logLine);
+ }
+ private sealed class ResourceLogger(ResourceLoggerState loggerState) : ILogger
+ {
IDisposable? ILogger.BeginScope(TState state) => null;
bool ILogger.IsEnabled(LogLevel logLevel) => true;
@@ -351,21 +437,42 @@ public void Log(LogLevel logLevel, EventId eventId, TState state, Except
return;
}
- var log = formatter(state, exception) + (exception is null ? "" : $"\n{exception}");
+ var logMessage = formatter(state, exception) + (exception is null ? "" : $"\n{exception}");
var isErrorMessage = logLevel >= LogLevel.Error;
- LogLine logLine;
- lock (loggerState._backlog)
- {
- _lineNumber++;
- logLine = new LogLine(_lineNumber, log, isErrorMessage);
+ loggerState.AddLog(DateTime.UtcNow, logMessage, isErrorMessage);
+ }
+ }
+ }
- loggerState._backlog.Add(logLine);
- }
+ internal static bool TryParseContentLineDate(string content, out DateTime value)
+ {
+ const int MinDateLength = 20; // Date + time without fractional seconds.
+ const int MaxDateLength = 30; // Date + time with fractional seconds.
- loggerState._onNewLog?.Invoke(logLine);
+ if (content.Length >= MinDateLength)
+ {
+ var firstSpaceIndex = content.IndexOf(' ', StringComparison.Ordinal);
+ if (firstSpaceIndex > 0)
+ {
+ if (DateTimeOffset.TryParse(content.AsSpan(0, firstSpaceIndex), CultureInfo.InvariantCulture, DateTimeStyles.AssumeUniversal, out var dateTime))
+ {
+ value = dateTime.UtcDateTime;
+ return true;
+ }
+ }
+ else if (content.Length <= MaxDateLength)
+ {
+ if (DateTimeOffset.TryParse(content, CultureInfo.InvariantCulture, DateTimeStyles.AssumeUniversal, out var dateTime))
+ {
+ value = dateTime.UtcDateTime;
+ return true;
+ }
}
}
+
+ value = default;
+ return false;
}
}
diff --git a/src/Aspire.Hosting/Dashboard/DashboardService.cs b/src/Aspire.Hosting/Dashboard/DashboardService.cs
index adf82dc5f62..393e4f1077e 100644
--- a/src/Aspire.Hosting/Dashboard/DashboardService.cs
+++ b/src/Aspire.Hosting/Dashboard/DashboardService.cs
@@ -133,7 +133,7 @@ async Task WatchResourceConsoleLogsInternal()
await foreach (var group in subscription.WithCancellation(cts.Token).ConfigureAwait(false))
{
- WatchResourceConsoleLogsUpdate update = new();
+ var update = new WatchResourceConsoleLogsUpdate();
foreach (var (lineNumber, content, isErrorMessage) in group)
{
diff --git a/src/Aspire.Hosting/Dcp/ApplicationExecutor.cs b/src/Aspire.Hosting/Dcp/ApplicationExecutor.cs
index e29d5ff2dea..e7fd400d028 100644
--- a/src/Aspire.Hosting/Dcp/ApplicationExecutor.cs
+++ b/src/Aspire.Hosting/Dcp/ApplicationExecutor.cs
@@ -484,29 +484,34 @@ private void StartLogStream(T resource) where T : CustomResource
{
if (_logger.IsEnabled(LogLevel.Debug))
{
- _logger.LogDebug("Starting log streaming for {ResourceName}", resource.Metadata.Name);
+ _logger.LogDebug("Starting log streaming for {ResourceName}.", resource.Metadata.Name);
}
// Pump the logs from the enumerable into the logger
- var logger = loggerService.GetLogger(resource.Metadata.Name);
+ var logger = loggerService.GetInternalLogger(resource.Metadata.Name);
await foreach (var batch in enumerable.WithCancellation(cancellation.Token).ConfigureAwait(false))
{
foreach (var (content, isError) in batch)
{
- var level = isError ? LogLevel.Error : LogLevel.Information;
- logger.Log(level, 0, content, null, (s, _) => s);
+ if (!ResourceLoggerService.TryParseContentLineDate(content, out var dateTimeUtc))
+ {
+ // If a date can't be read from the line content then use the current date.
+ dateTimeUtc = DateTime.UtcNow;
+ }
+
+ logger(dateTimeUtc, content, isError);
}
}
}
catch (OperationCanceledException)
{
// Ignore
- _logger.LogDebug("Log streaming for {ResourceName} was cancelled", resource.Metadata.Name);
+ _logger.LogDebug("Log streaming for {ResourceName} was cancelled.", resource.Metadata.Name);
}
catch (Exception ex)
{
- _logger.LogError(ex, "Error streaming logs for {ResourceName}", resource.Metadata.Name);
+ _logger.LogError(ex, "Error streaming logs for {ResourceName}.", resource.Metadata.Name);
}
},
cancellation.Token);
diff --git a/src/Aspire.Hosting/Dcp/ResourceLogSource.cs b/src/Aspire.Hosting/Dcp/ResourceLogSource.cs
index e32c2474ffd..24b62b23517 100644
--- a/src/Aspire.Hosting/Dcp/ResourceLogSource.cs
+++ b/src/Aspire.Hosting/Dcp/ResourceLogSource.cs
@@ -34,12 +34,10 @@ public async IAsyncEnumerator GetAsyncEnumerator(CancellationToken
var streamTasks = new List();
- var timestamps = resource is Container; // Timestamps are available only for Containers as of Aspire P5.
-
if (resource is Container && dcpVersion?.CompareTo(DcpVersion.MinimumVersionAspire_8_1) >= 0)
{
- var startupStderrStream = await kubernetesService.GetLogStreamAsync(resource, Logs.StreamTypeStartupStdErr, follow: true, timestamps: timestamps, cancellationToken).ConfigureAwait(false);
- var startupStdoutStream = await kubernetesService.GetLogStreamAsync(resource, Logs.StreamTypeStartupStdOut, follow: true, timestamps: timestamps, cancellationToken).ConfigureAwait(false);
+ var startupStderrStream = await kubernetesService.GetLogStreamAsync(resource, Logs.StreamTypeStartupStdErr, follow: true, timestamps: true, cancellationToken).ConfigureAwait(false);
+ var startupStdoutStream = await kubernetesService.GetLogStreamAsync(resource, Logs.StreamTypeStartupStdOut, follow: true, timestamps: true, cancellationToken).ConfigureAwait(false);
var startupStdoutStreamTask = Task.Run(() => StreamLogsAsync(startupStdoutStream, isError: false), cancellationToken);
streamTasks.Add(startupStdoutStreamTask);
@@ -48,8 +46,8 @@ public async IAsyncEnumerator GetAsyncEnumerator(CancellationToken
streamTasks.Add(startupStderrStreamTask);
}
- var stdoutStream = await kubernetesService.GetLogStreamAsync(resource, Logs.StreamTypeStdOut, follow: true, timestamps: timestamps, cancellationToken).ConfigureAwait(false);
- var stderrStream = await kubernetesService.GetLogStreamAsync(resource, Logs.StreamTypeStdErr, follow: true, timestamps: timestamps, cancellationToken).ConfigureAwait(false);
+ var stdoutStream = await kubernetesService.GetLogStreamAsync(resource, Logs.StreamTypeStdOut, follow: true, timestamps: true, cancellationToken).ConfigureAwait(false);
+ var stderrStream = await kubernetesService.GetLogStreamAsync(resource, Logs.StreamTypeStdErr, follow: true, timestamps: true, cancellationToken).ConfigureAwait(false);
var stdoutStreamTask = Task.Run(() => StreamLogsAsync(stdoutStream, isError: false), cancellationToken);
streamTasks.Add(stdoutStreamTask);
diff --git a/tests/Aspire.Hosting.Tests/Dcp/ApplicationExecutorTests.cs b/tests/Aspire.Hosting.Tests/Dcp/ApplicationExecutorTests.cs
index b5d2159dc5e..0a53c8748fd 100644
--- a/tests/Aspire.Hosting.Tests/Dcp/ApplicationExecutorTests.cs
+++ b/tests/Aspire.Hosting.Tests/Dcp/ApplicationExecutorTests.cs
@@ -1,19 +1,22 @@
// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
+using System.Globalization;
+using System.IO.Pipelines;
+using System.Text;
+using System.Threading.Channels;
using Aspire.Hosting.Dcp;
using Aspire.Hosting.Dcp.Model;
+using Aspire.Hosting.Eventing;
using Aspire.Hosting.Lifecycle;
+using Aspire.Hosting.Tests.Utils;
using k8s.Models;
using Microsoft.Extensions.Configuration;
using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging.Abstractions;
using Microsoft.Extensions.Options;
using Xunit;
-using System.Globalization;
-using Microsoft.Extensions.Hosting;
-using Aspire.Hosting.Tests.Utils;
-using Aspire.Hosting.Eventing;
namespace Aspire.Hosting.Tests.Dcp;
@@ -368,6 +371,218 @@ public async Task EndpointOtelServiceName(int replicaCount, string expectedName)
Assert.Equal(expectedName, ers.Spec?.Template.Annotations?[CustomResource.OtelServiceNameAnnotation]);
}
+ [Fact]
+ public async Task ResourceLogging_MultipleStreams_StreamedOverTime()
+ {
+ var builder = DistributedApplication.CreateBuilder(new DistributedApplicationOptions
+ {
+ AssemblyName = typeof(DistributedApplicationTests).Assembly.FullName
+ });
+
+ builder.AddContainer("database", "image");
+
+ var logStreamPipesChannel = Channel.CreateUnbounded<(string Type, Pipe Pipe)>();
+ var kubernetesService = new TestKubernetesService(startStream: (obj, logStreamType) =>
+ {
+ var s = new Pipe();
+ if (!logStreamPipesChannel.Writer.TryWrite((logStreamType, s)))
+ {
+ Assert.Fail("Pipe channel unexpectedly closed.");
+ }
+
+ return s.Reader.AsStream();
+ });
+ using var app = builder.Build();
+ var distributedAppModel = app.Services.GetRequiredService();
+ var dcpOptions = new DcpOptions { DashboardPath = "./dashboard" };
+ var resourceLoggerService = new ResourceLoggerService();
+ var appExecutor = CreateAppExecutor(distributedAppModel, app.Services, kubernetesService: kubernetesService, dcpOptions: dcpOptions, resourceLoggerService: resourceLoggerService);
+ await appExecutor.RunApplicationAsync();
+
+ var exeResource = Assert.Single(kubernetesService.CreatedResources.OfType());
+
+ // Start watching logs for container.
+ var watchCts = new CancellationTokenSource();
+ var watchSubscribers = resourceLoggerService.WatchAnySubscribersAsync();
+ var watchSubscribersEnumerator = watchSubscribers.GetAsyncEnumerator();
+ var watchLogs = resourceLoggerService.WatchAsync(exeResource.Metadata.Name);
+ var watchLogsEnumerator = watchLogs.GetAsyncEnumerator(watchCts.Token);
+
+ var moveNextTask = watchLogsEnumerator.MoveNextAsync().AsTask();
+ Assert.False(moveNextTask.IsCompletedSuccessfully, "No logs yet.");
+
+ await watchSubscribersEnumerator.MoveNextAsync();
+ Assert.Equal(exeResource.Metadata.Name, watchSubscribersEnumerator.Current.Name);
+ Assert.True(watchSubscribersEnumerator.Current.AnySubscribers);
+
+ exeResource.Status = new ContainerStatus { State = ContainerState.Running };
+ kubernetesService.PushResourceModified(exeResource);
+
+ var pipes = await GetStreamPipesAsync(logStreamPipesChannel);
+
+ // Write content to container output stream. This is read by logging and creates log lines.
+ await pipes.StandardOut.Writer.WriteAsync(Encoding.UTF8.GetBytes("2024-08-19T06:10:33.473275911Z Hello world" + Environment.NewLine));
+ Assert.True(await moveNextTask);
+ var logLine = watchLogsEnumerator.Current.Single();
+ Assert.Equal("2024-08-19T06:10:33.473275911Z Hello world", logLine.Content);
+ Assert.Equal(1, logLine.LineNumber);
+ Assert.False(logLine.IsErrorMessage);
+
+ moveNextTask = watchLogsEnumerator.MoveNextAsync().AsTask();
+ Assert.False(moveNextTask.IsCompletedSuccessfully, "No logs yet.");
+
+ // Note: This console log is earlier than the previous, but logs are displayed in real time as they're available.
+ await pipes.StandardErr.Writer.WriteAsync(Encoding.UTF8.GetBytes("2024-08-19T06:10:32.661Z Next" + Environment.NewLine));
+ Assert.True(await moveNextTask);
+ logLine = watchLogsEnumerator.Current.Single();
+ Assert.Equal("2024-08-19T06:10:32.661Z Next", logLine.Content);
+ Assert.Equal(2, logLine.LineNumber);
+ Assert.True(logLine.IsErrorMessage);
+
+ var loggerState = resourceLoggerService.GetResourceLoggerState(exeResource.Metadata.Name);
+ Assert.Collection(loggerState.GetBacklogSnapshot(),
+ l => Assert.Equal("2024-08-19T06:10:32.661Z Next", l.Message),
+ l => Assert.Equal("2024-08-19T06:10:33.473275911Z Hello world", l.Message));
+
+ // Stop watching.
+ moveNextTask = watchLogsEnumerator.MoveNextAsync().AsTask();
+ watchCts.Cancel();
+
+ await Assert.ThrowsAnyAsync(async () => await moveNextTask);
+
+ await watchSubscribersEnumerator.MoveNextAsync();
+ Assert.Equal(exeResource.Metadata.Name, watchSubscribersEnumerator.Current.Name);
+ Assert.False(watchSubscribersEnumerator.Current.AnySubscribers);
+
+ // State is clear when no longer watching.
+ await AsyncTestHelpers.AssertIsTrueRetryAsync(
+ () => loggerState.GetBacklogSnapshot().Length == 0,
+ "Backlog is asyncronously cleared after watch ends.");
+ }
+
+ [Fact]
+ public async Task ResourceLogging_ReplayBacklog_SentInBatch()
+ {
+ var builder = DistributedApplication.CreateBuilder(new DistributedApplicationOptions
+ {
+ AssemblyName = typeof(DistributedApplicationTests).Assembly.FullName
+ });
+
+ builder.AddContainer("database", "image");
+
+ var kubernetesService = new TestKubernetesService(startStream: (obj, logStreamType) =>
+ {
+ switch (logStreamType)
+ {
+ case Logs.StreamTypeStdOut:
+ return new MemoryStream(Encoding.UTF8.GetBytes("2024-08-19T06:10:01.000Z First" + Environment.NewLine));
+ case Logs.StreamTypeStdErr:
+ return new MemoryStream(Encoding.UTF8.GetBytes("2024-08-19T06:10:02.000Z Second" + Environment.NewLine));
+ case Logs.StreamTypeStartupStdOut:
+ return new MemoryStream(Encoding.UTF8.GetBytes("2024-08-19T06:10:03.000Z Third" + Environment.NewLine));
+ case Logs.StreamTypeStartupStdErr:
+ return new MemoryStream(Encoding.UTF8.GetBytes(
+ "2024-08-19T06:10:05.000Z Sixth" + Environment.NewLine +
+ "2024-08-19T06:10:05.000Z Seventh" + Environment.NewLine +
+ "2024-08-19T06:10:04.000Z Forth" + Environment.NewLine +
+ "2024-08-19T06:10:04.000Z Fifth" + Environment.NewLine));
+ default:
+ throw new InvalidOperationException("Unexpected type: " + logStreamType);
+ }
+ });
+ using var app = builder.Build();
+ var distributedAppModel = app.Services.GetRequiredService();
+ var dcpOptions = new DcpOptions { DashboardPath = "./dashboard" };
+ var resourceLoggerService = new ResourceLoggerService();
+ var appExecutor = CreateAppExecutor(distributedAppModel, app.Services, kubernetesService: kubernetesService, dcpOptions: dcpOptions, resourceLoggerService: resourceLoggerService);
+ await appExecutor.RunApplicationAsync();
+
+ var exeResource = Assert.Single(kubernetesService.CreatedResources.OfType());
+
+ // Start watching logs for container.
+ var watchCts = new CancellationTokenSource();
+ var watchSubscribers = resourceLoggerService.WatchAnySubscribersAsync();
+ var watchSubscribersEnumerator = watchSubscribers.GetAsyncEnumerator();
+ var watchLogs1 = resourceLoggerService.WatchAsync(exeResource.Metadata.Name);
+ var watchLogsEnumerator1 = watchLogs1.GetAsyncEnumerator(watchCts.Token);
+
+ var moveNextTask = watchLogsEnumerator1.MoveNextAsync().AsTask();
+ Assert.False(moveNextTask.IsCompletedSuccessfully, "No logs yet.");
+
+ await watchSubscribersEnumerator.MoveNextAsync();
+ Assert.Equal(exeResource.Metadata.Name, watchSubscribersEnumerator.Current.Name);
+ Assert.True(watchSubscribersEnumerator.Current.AnySubscribers);
+
+ exeResource.Status = new ContainerStatus { State = ContainerState.Running };
+ kubernetesService.PushResourceModified(exeResource);
+
+ Assert.True(await moveNextTask);
+ Assert.Collection(watchLogsEnumerator1.Current,
+ l => Assert.Equal("2024-08-19T06:10:01.000Z First", l.Content),
+ l => Assert.Equal("2024-08-19T06:10:02.000Z Second", l.Content),
+ l => Assert.Equal("2024-08-19T06:10:03.000Z Third", l.Content),
+ l => Assert.Equal("2024-08-19T06:10:04.000Z Forth", l.Content),
+ l => Assert.Equal("2024-08-19T06:10:04.000Z Fifth", l.Content),
+ l => Assert.Equal("2024-08-19T06:10:05.000Z Sixth", l.Content),
+ l => Assert.Equal("2024-08-19T06:10:05.000Z Seventh", l.Content));
+
+ var watchLogs2 = resourceLoggerService.WatchAsync(exeResource.Metadata.Name);
+ var watchLogsEnumerator2 = watchLogs2.GetAsyncEnumerator(watchCts.Token);
+
+ Assert.True(await watchLogsEnumerator2.MoveNextAsync());
+ Assert.Collection(watchLogsEnumerator2.Current,
+ l => Assert.Equal("2024-08-19T06:10:01.000Z First", l.Content),
+ l => Assert.Equal("2024-08-19T06:10:02.000Z Second", l.Content),
+ l => Assert.Equal("2024-08-19T06:10:03.000Z Third", l.Content),
+ l => Assert.Equal("2024-08-19T06:10:04.000Z Forth", l.Content),
+ l => Assert.Equal("2024-08-19T06:10:04.000Z Fifth", l.Content),
+ l => Assert.Equal("2024-08-19T06:10:05.000Z Sixth", l.Content),
+ l => Assert.Equal("2024-08-19T06:10:05.000Z Seventh", l.Content));
+ }
+
+ private sealed class LogStreamPipes
+ {
+ public Pipe StandardOut { get; set; } = default!;
+ public Pipe StandardErr { get; set; } = default!;
+ public Pipe StartupOut { get; set; } = default!;
+ public Pipe StartupErr { get; set; } = default!;
+ }
+
+ private static async Task GetStreamPipesAsync(Channel<(string Type, Pipe Pipe)> logStreamPipesChannel)
+ {
+ var pipeCount = 0;
+ var result = new LogStreamPipes();
+
+ await foreach (var item in logStreamPipesChannel.Reader.ReadAllAsync())
+ {
+ switch (item.Type)
+ {
+ case Logs.StreamTypeStdOut:
+ result.StandardOut = item.Pipe;
+ break;
+ case Logs.StreamTypeStdErr:
+ result.StandardErr = item.Pipe;
+ break;
+ case Logs.StreamTypeStartupStdOut:
+ result.StartupOut = item.Pipe;
+ break;
+ case Logs.StreamTypeStartupStdErr:
+ result.StartupErr = item.Pipe;
+ break;
+ default:
+ throw new InvalidOperationException("Unexpected type: " + item.Type);
+ }
+
+ pipeCount++;
+ if (pipeCount == 4)
+ {
+ logStreamPipesChannel.Writer.Complete();
+ }
+ }
+
+ return result;
+ }
+
[Fact]
public async Task EndpointPortsProjectNoPortNoTargetPort()
{
@@ -730,7 +945,8 @@ private static ApplicationExecutor CreateAppExecutor(
IServiceProvider serviceProvider,
IConfiguration? configuration = null,
IKubernetesService? kubernetesService = null,
- DcpOptions? dcpOptions = null)
+ DcpOptions? dcpOptions = null,
+ ResourceLoggerService? resourceLoggerService = null)
{
if (configuration == null)
{
@@ -759,7 +975,7 @@ private static ApplicationExecutor CreateAppExecutor(
ServiceProvider = TestServiceProvider.Instance
}),
new ResourceNotificationService(new NullLogger(), new TestHostApplicationLifetime()),
- new ResourceLoggerService(),
+ resourceLoggerService ?? new ResourceLoggerService(),
new TestDcpDependencyCheckService(),
new DistributedApplicationEventing(),
serviceProvider
diff --git a/tests/Aspire.Hosting.Tests/Dcp/TestKubernetesService.cs b/tests/Aspire.Hosting.Tests/Dcp/TestKubernetesService.cs
index ce18c5795f5..5c0e74721d3 100644
--- a/tests/Aspire.Hosting.Tests/Dcp/TestKubernetesService.cs
+++ b/tests/Aspire.Hosting.Tests/Dcp/TestKubernetesService.cs
@@ -21,7 +21,13 @@ internal sealed class TestKubernetesService : IKubernetesService
public ConcurrentQueue CreatedResources { get; } = [];
private readonly List> _watchChannels = [];
- private int _nextPort = StartOfAutoPortRange;
+ private readonly Func _startStream;
+ private int _nextPort = StartOfAutoPortRange;
+
+ public TestKubernetesService(Func? startStream = null)
+ {
+ _startStream = startStream ?? ((obj, logStreamType) => new MemoryStream(Encoding.UTF8.GetBytes($"Logs for {obj.Metadata.Name} ({logStreamType})")));
+ }
public Task GetAsync(string name, string? namespaceParameter = null, CancellationToken _ = default) where T : CustomResource
{
@@ -66,10 +72,21 @@ static T Clone(T r)
c.Writer.TryWrite((WatchEventType.Added, res));
}
}
-
+
return Task.FromResult(res);
}
+ public void PushResourceModified(CustomResource resource)
+ {
+ lock (CreatedResources)
+ {
+ foreach (var c in _watchChannels)
+ {
+ c.Writer.TryWrite((WatchEventType.Modified, resource));
+ }
+ }
+ }
+
public Task DeleteAsync(string name, string? namespaceParameter = null, CancellationToken cancellationToken = default) where T : CustomResource
{
throw new NotImplementedException();
@@ -118,7 +135,6 @@ public Task> ListAsync(string? namespaceParameter = null, Cancellatio
public Task GetLogStreamAsync(T obj, string logStreamType, bool? follow = true, bool? timestamps = false, CancellationToken cancellationToken = default) where T : CustomResource
{
- var ms = new MemoryStream(Encoding.UTF8.GetBytes($"Logs for {obj.Metadata.Name} ({logStreamType})"));
- return Task.FromResult((Stream) ms);
+ return Task.FromResult(_startStream(obj, logStreamType));
}
}
diff --git a/tests/Aspire.Hosting.Tests/ResourceLoggerServiceTests.cs b/tests/Aspire.Hosting.Tests/ResourceLoggerServiceTests.cs
index a72d9c7dbaf..056e169bf39 100644
--- a/tests/Aspire.Hosting.Tests/ResourceLoggerServiceTests.cs
+++ b/tests/Aspire.Hosting.Tests/ResourceLoggerServiceTests.cs
@@ -8,6 +8,22 @@ namespace Aspire.Hosting.Tests;
public class ResourceLoggerServiceTests
{
+ [Fact]
+ public void ParseStreamedLogLine()
+ {
+ DateTime dateTimeUtc;
+
+ Assert.False(ResourceLoggerService.TryParseContentLineDate("", out _));
+ Assert.False(ResourceLoggerService.TryParseContentLineDate(" ", out _));
+ Assert.False(ResourceLoggerService.TryParseContentLineDate("ABC-ABC-ABC-ABC-ABC-ABC-ABC-ABC-ABC-ABC-ABC-ABC", out _));
+
+ Assert.True(ResourceLoggerService.TryParseContentLineDate("2024-08-19T06:01:06.661Z", out dateTimeUtc));
+ Assert.Equal(new DateTime(2024, 8, 19, 6, 1, 6, 661, DateTimeKind.Utc), dateTimeUtc);
+
+ Assert.True(ResourceLoggerService.TryParseContentLineDate("2024-08-19T06:10:33.473275911Z", out dateTimeUtc));
+ Assert.Equal(new DateTime(2024, 8, 19, 6, 10, 33, 473, 275, DateTimeKind.Utc).Add(TimeSpan.FromTicks(9)), dateTimeUtc);
+ }
+
[Fact]
public async Task AddingResourceLoggerAnnotationAllowsLogging()
{
diff --git a/tests/Aspire.Hosting.Tests/Utils/AsyncTestHelpers.cs b/tests/Aspire.Hosting.Tests/Utils/AsyncTestHelpers.cs
new file mode 100644
index 00000000000..170074a3514
--- /dev/null
+++ b/tests/Aspire.Hosting.Tests/Utils/AsyncTestHelpers.cs
@@ -0,0 +1,37 @@
+// Licensed to the .NET Foundation under one or more agreements.
+// The .NET Foundation licenses this file to you under the MIT license.
+
+using Microsoft.Extensions.Logging;
+
+namespace Aspire.Hosting.Tests.Utils;
+
+internal static class AsyncTestHelpers
+{
+ public static Task AssertIsTrueRetryAsync(Func assert, string message, ILogger? logger = null)
+ {
+ return AssertIsTrueRetryAsync(() => Task.FromResult(assert()), message, logger);
+ }
+
+ public static async Task AssertIsTrueRetryAsync(Func> assert, string message, ILogger? logger = null)
+ {
+ const int Retries = 10;
+
+ logger?.LogInformation("Start: " + message);
+
+ for (var i = 0; i < Retries; i++)
+ {
+ if (i > 0)
+ {
+ await Task.Delay((i + 1) * (i + 1) * 10);
+ }
+
+ if (await assert())
+ {
+ logger?.LogInformation("End: " + message);
+ return;
+ }
+ }
+
+ throw new InvalidOperationException($"Assert failed after {Retries} retries: {message}");
+ }
+}