Skip to content

Fix console logs content changing order and the line number increasing #5357

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
JamesNK merged 3 commits into from
Aug 20, 2024
Merged

Fix console logs content changing order and the line number increasing #5357

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
947a905
WIP
JamesNK Aug 20, 2024
1fc9913
Fix test
JamesNK Aug 20, 2024
5ddaf75
Apply suggestions from code review
JamesNK Aug 20, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
183 changes: 145 additions & 38 deletions src/Aspire.Hosting/ApplicationModel/ResourceLoggerService.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -58,6 +60,17 @@ public ILogger GetLogger(string resourceName)
return GetResourceLoggerState(resourceName).Logger;
}

/// <summary>
/// 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.
/// </summary>
internal Action<DateTime, string, bool> GetInternalLogger(string resourceName)
{
ArgumentNullException.ThrowIfNull(resourceName);

return GetResourceLoggerState(resourceName).AddLog;
}

/// <summary>
/// Watch for changes to the log stream for a resource.
/// </summary>
Expand Down Expand Up @@ -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();
Expand All @@ -168,15 +182,19 @@ private ResourceLoggerState GetResourceLoggerState(string resourceName) =>
},
this);

internal sealed record InternalLogLine(DateTime DateTimeUtc, string Message, bool IsError);

/// <summary>
/// A logger for the resource to write to.
/// </summary>
private sealed class ResourceLoggerState
internal sealed class ResourceLoggerState
{
private readonly ResourceLogger _logger;
private readonly CancellationTokenSource _logStreamCts = new();

private readonly CircularBuffer<LogLine> _backlog = new(10000);
private Task? _backlogReplayCompleteTask;
private long _lastLogReceivedTimestamp;
private readonly CircularBuffer<InternalLogLine> _backlog = new(10000);

/// <summary>
/// Creates a new <see cref="ResourceLoggerState"/>.
Expand Down Expand Up @@ -220,45 +238,46 @@ public event Action<bool> OnSubscribersChanged
/// <returns>The log stream for the resource.</returns>
public async IAsyncEnumerable<IReadOnlyList<LogLine>> WatchAsync([EnumeratorCancellation] CancellationToken cancellationToken = default)
{
var channel = Channel.CreateUnbounded<LogLine>();
// 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<InternalLogLine>();

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();
Comment thread
davidfowl marked this conversation as resolved.
}

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
Expand All @@ -267,11 +286,51 @@ void Log(LogLine log)

channel.Writer.TryComplete();
}

static LogLine[] CreateLogLines(ref int lineNumber, IReadOnlyList<InternalLogLine> 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);
Comment on lines +313 to +319

@eerhardt eerhardt Aug 20, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This seems fragile. Is there a way to use concurrency controls (reset events, semaphores, etc) to ensure the ordering happens correctly instead of delaying based on time?

@JamesNK JamesNK Aug 20, 2024

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, I agree that's not the best.

The problem is that ordering values in the backlog doesn't mean much once you start to stream log items as they're received. Sending items as they're received is fine after they logs have been replayed from DCP.

I think what is needed from DCP is an indication of how many lines are being replayed. Then we can tell the logger service that the backlog is full (and all the log items in it are sorted correctly), and that it can send the backlog and send new log lines in real time.

@eerhardt eerhardt Aug 20, 2024 •
edited
Loading

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't understand why the existing flushBacklogSync doesn't work. We delay writing to the channel until we've grabbed the backlog snapshot.

And then we check to ensure the same message wasn't already in the backlog.

@JamesNK JamesNK Aug 20, 2024

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That’s not the problem this change is fixing

see #4893 (comment)

@eerhardt eerhardt Aug 20, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why do we need to "Add a small delay to ensure the backlog is replayed from DCP and ordered correctly."? Why isn't the order of operations:

  1. Grab the backlog snapshot
  2. The following can happen concurrently
    a. After the snapshot has been made, as new logs get written, write them to the channel
    b. Write the snapshot to the listener
  3. After the snapshot has been written, write the messages from the channel

@JamesNK JamesNK Aug 20, 2024

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Grab the backlog snapshot

This doesn't work because it's empty. When you view the console logs page, and the request goes to the host to display logs, the snapshot has nothing in it. We can't immediately start streaming all the initial logs that DCP returns to us because we receive them out of order.

The delay is to give the opportunity for DCP to send us the initial logs, them to be put in order in the backlog, then the backlog returned to the UI, then new logs are streamed as they arrive.

lock (_backlog)
{
if (_lastLogReceivedTimestamp != 0 && Stopwatch.GetElapsedTime(_lastLogReceivedTimestamp) > delay)
{
break;
}
}
}
}
}

// This provides the fan out to multiple subscribers.
private Action<LogLine>? _onNewLog;
private event Action<LogLine> OnNewLog
private Action<InternalLogLine>? _onNewLog;
private event Action<InternalLogLine> OnNewLog
{
add
{
Expand Down Expand Up @@ -324,21 +383,48 @@ public void ClearBacklog()
lock (_backlog)
{
_backlog.Clear();
_backlogReplayCompleteTask = null;
}
}

private LogLine[] GetBacklogSnapshot()
internal InternalLogLine[] GetBacklogSnapshot()
{
lock (_backlog)
{
return [.. _backlog];
}
}

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>(TState state) => null;

bool ILogger.IsEnabled(LogLevel logLevel) => true;
Expand All @@ -351,21 +437,42 @@ public void Log<TState>(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;
}
}

Expand Down
2 changes: 1 addition & 1 deletion src/Aspire.Hosting/Dashboard/DashboardService.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
{
Expand Down
17 changes: 11 additions & 6 deletions src/Aspire.Hosting/Dcp/ApplicationExecutor.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -484,29 +484,34 @@ private void StartLogStream<T>(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);
Expand Down
10 changes: 4 additions & 6 deletions src/Aspire.Hosting/Dcp/ResourceLogSource.cs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,12 +34,10 @@ public async IAsyncEnumerator<LogEntryList> GetAsyncEnumerator(CancellationToken

var streamTasks = new List<Task>();

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);

@eerhardt eerhardt Aug 20, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It looks like we only ever call GetLogStreamAsync with timestamps: true. So we can remove that parameter to make the code easier to read.

@karolz-ms karolz-ms Aug 20, 2024

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'd prefer to keep the timestamps parameter on the KubernetesService class in case we find a use case for logs without timestamps in future.

@eerhardt eerhardt Aug 20, 2024

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We can always add it back, once we find a need. It is internal.

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);
Expand All @@ -48,8 +46,8 @@ public async IAsyncEnumerator<LogEntryList> 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);
Expand Down
Loading