Merge remote-tracking branch 'origin/main'
This commit is contained in:
Sipke Schoorstra
commit
06b757bc95
|
|
@ -1,4 +1,9 @@
|
|||
using System.Collections;
|
||||
using System.Data;
|
||||
using System.Data.Common;
|
||||
using System.Diagnostics.CodeAnalysis;
|
||||
using Elsa.Diagnostics.StructuredLogs.Models;
|
||||
using Elsa.Diagnostics.StructuredLogs.Persistence.Relational.Models;
|
||||
using Elsa.Diagnostics.StructuredLogs.Persistence.Relational.Services;
|
||||
|
||||
namespace Elsa.Diagnostics.StructuredLogs.Persistence.Relational.UnitTests;
|
||||
|
|
@ -41,4 +46,210 @@ public class RelationalStructuredLogMapperTests
|
|||
Assert.Equal("2026-05-13T13:00:00.0000000+00:00", formatted);
|
||||
Assert.Equal(timestamp.ToUniversalTime(), RelationalStructuredLogMapper.ParseTimestamp(formatted));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void MapReader_DeserializesStoredJsonAndNullableFields()
|
||||
{
|
||||
var timestamp = new DateTimeOffset(2026, 5, 13, 15, 0, 0, TimeSpan.FromHours(2));
|
||||
var logEvent = new StructuredLogEvent
|
||||
{
|
||||
Id = "event-a",
|
||||
Sequence = 123,
|
||||
Timestamp = timestamp,
|
||||
ReceivedAt = timestamp.AddSeconds(1),
|
||||
Level = StructuredLogLevel.Warning,
|
||||
Category = "Elsa.Tests",
|
||||
EventId = 42,
|
||||
EventName = "TestEvent",
|
||||
Message = "Message",
|
||||
MessageTemplate = "Message {Value}",
|
||||
Exception = new("System.InvalidOperationException", "Boom", "Stack"),
|
||||
Scopes = new Dictionary<string, string?> { ["Scope"] = "Value" },
|
||||
Properties = new Dictionary<string, string?> { ["Property"] = "Value" },
|
||||
TraceId = "trace-a",
|
||||
SpanId = "span-a",
|
||||
CorrelationId = "correlation-a",
|
||||
TenantId = "tenant-a",
|
||||
WorkflowDefinitionId = "definition-a",
|
||||
WorkflowInstanceId = "instance-a",
|
||||
SourceId = "source-a"
|
||||
};
|
||||
|
||||
using var reader = CreateReader(_mapper.Map(logEvent));
|
||||
Assert.True(reader.Read());
|
||||
|
||||
var mapped = _mapper.Map(reader);
|
||||
|
||||
Assert.Equal(logEvent.Id, mapped.Id);
|
||||
Assert.Equal(logEvent.Sequence, mapped.Sequence);
|
||||
Assert.Equal(logEvent.Timestamp.ToUniversalTime(), mapped.Timestamp);
|
||||
Assert.Equal(logEvent.ReceivedAt.ToUniversalTime(), mapped.ReceivedAt);
|
||||
Assert.Equal(logEvent.Level, mapped.Level);
|
||||
Assert.Equal(logEvent.Category, mapped.Category);
|
||||
Assert.Equal(logEvent.EventId, mapped.EventId);
|
||||
Assert.Equal(logEvent.EventName, mapped.EventName);
|
||||
Assert.Equal(logEvent.Message, mapped.Message);
|
||||
Assert.Equal(logEvent.MessageTemplate, mapped.MessageTemplate);
|
||||
Assert.Equal(logEvent.Exception.Type, mapped.Exception!.Type);
|
||||
Assert.Equal(logEvent.Exception.Message, mapped.Exception.Message);
|
||||
Assert.Equal(logEvent.Exception.StackTrace, mapped.Exception.StackTrace);
|
||||
Assert.Equal(logEvent.Scopes, mapped.Scopes);
|
||||
Assert.Equal(logEvent.Properties, mapped.Properties);
|
||||
Assert.Equal(logEvent.TraceId, mapped.TraceId);
|
||||
Assert.Equal(logEvent.SpanId, mapped.SpanId);
|
||||
Assert.Equal(logEvent.CorrelationId, mapped.CorrelationId);
|
||||
Assert.Equal(logEvent.TenantId, mapped.TenantId);
|
||||
Assert.Equal(logEvent.WorkflowDefinitionId, mapped.WorkflowDefinitionId);
|
||||
Assert.Equal(logEvent.WorkflowInstanceId, mapped.WorkflowInstanceId);
|
||||
Assert.Equal(logEvent.SourceId, mapped.SourceId);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void MapReader_TreatsNullAndWhitespaceJsonAsEmptyValues()
|
||||
{
|
||||
var record = new RelationalStructuredLogRecord
|
||||
{
|
||||
Id = "event-a",
|
||||
Sequence = 123,
|
||||
Timestamp = RelationalStructuredLogMapper.FormatTimestamp(DateTimeOffset.UtcNow),
|
||||
ReceivedAt = RelationalStructuredLogMapper.FormatTimestamp(DateTimeOffset.UtcNow),
|
||||
Level = StructuredLogLevel.Information,
|
||||
Category = "Elsa.Tests",
|
||||
EventId = 42,
|
||||
EventName = null,
|
||||
Message = "Message",
|
||||
MessageTemplate = null,
|
||||
ExceptionJson = null,
|
||||
ScopesJson = " ",
|
||||
PropertiesJson = "",
|
||||
TraceId = null,
|
||||
SpanId = null,
|
||||
CorrelationId = null,
|
||||
TenantId = null,
|
||||
WorkflowDefinitionId = null,
|
||||
WorkflowInstanceId = null,
|
||||
SourceId = "source-a"
|
||||
};
|
||||
|
||||
using var reader = CreateReader(record);
|
||||
Assert.True(reader.Read());
|
||||
|
||||
var mapped = _mapper.Map(reader);
|
||||
|
||||
Assert.Null(mapped.EventName);
|
||||
Assert.Null(mapped.MessageTemplate);
|
||||
Assert.Null(mapped.Exception);
|
||||
Assert.Empty(mapped.Scopes);
|
||||
Assert.Empty(mapped.Properties);
|
||||
Assert.Null(mapped.TraceId);
|
||||
Assert.Null(mapped.SpanId);
|
||||
Assert.Null(mapped.CorrelationId);
|
||||
Assert.Null(mapped.TenantId);
|
||||
Assert.Null(mapped.WorkflowDefinitionId);
|
||||
Assert.Null(mapped.WorkflowInstanceId);
|
||||
}
|
||||
|
||||
private static DbDataReader CreateReader(RelationalStructuredLogRecord record)
|
||||
{
|
||||
var table = new DataTable();
|
||||
table.Columns.Add("Id", typeof(string));
|
||||
table.Columns.Add("Sequence", typeof(long));
|
||||
table.Columns.Add("Timestamp", typeof(string));
|
||||
table.Columns.Add("ReceivedAt", typeof(string));
|
||||
table.Columns.Add("Level", typeof(int));
|
||||
table.Columns.Add("Category", typeof(string));
|
||||
table.Columns.Add("EventId", typeof(int));
|
||||
table.Columns.Add("EventName", typeof(string));
|
||||
table.Columns.Add("Message", typeof(string));
|
||||
table.Columns.Add("MessageTemplate", typeof(string));
|
||||
table.Columns.Add("ExceptionJson", typeof(string));
|
||||
table.Columns.Add("ScopesJson", typeof(string));
|
||||
table.Columns.Add("PropertiesJson", typeof(string));
|
||||
table.Columns.Add("TraceId", typeof(string));
|
||||
table.Columns.Add("SpanId", typeof(string));
|
||||
table.Columns.Add("CorrelationId", typeof(string));
|
||||
table.Columns.Add("TenantId", typeof(string));
|
||||
table.Columns.Add("WorkflowDefinitionId", typeof(string));
|
||||
table.Columns.Add("WorkflowInstanceId", typeof(string));
|
||||
table.Columns.Add("SourceId", typeof(string));
|
||||
|
||||
table.Rows.Add(
|
||||
record.Id,
|
||||
record.Sequence,
|
||||
record.Timestamp,
|
||||
record.ReceivedAt,
|
||||
(int)record.Level,
|
||||
record.Category,
|
||||
record.EventId,
|
||||
record.EventName ?? (object)DBNull.Value,
|
||||
record.Message,
|
||||
record.MessageTemplate ?? (object)DBNull.Value,
|
||||
record.ExceptionJson ?? (object)DBNull.Value,
|
||||
record.ScopesJson ?? (object)DBNull.Value,
|
||||
record.PropertiesJson ?? (object)DBNull.Value,
|
||||
record.TraceId ?? (object)DBNull.Value,
|
||||
record.SpanId ?? (object)DBNull.Value,
|
||||
record.CorrelationId ?? (object)DBNull.Value,
|
||||
record.TenantId ?? (object)DBNull.Value,
|
||||
record.WorkflowDefinitionId ?? (object)DBNull.Value,
|
||||
record.WorkflowInstanceId ?? (object)DBNull.Value,
|
||||
record.SourceId);
|
||||
|
||||
return new DisposingDataReader(table, table.CreateDataReader());
|
||||
}
|
||||
|
||||
private sealed class DisposingDataReader(DataTable table, DataTableReader reader) : DbDataReader
|
||||
{
|
||||
public override object this[int ordinal] => reader[ordinal];
|
||||
public override object this[string name] => reader[name];
|
||||
public override int Depth => reader.Depth;
|
||||
public override int FieldCount => reader.FieldCount;
|
||||
public override bool HasRows => reader.HasRows;
|
||||
public override bool IsClosed => reader.IsClosed;
|
||||
public override int RecordsAffected => reader.RecordsAffected;
|
||||
public override bool GetBoolean(int ordinal) => reader.GetBoolean(ordinal);
|
||||
public override byte GetByte(int ordinal) => reader.GetByte(ordinal);
|
||||
public override long GetBytes(int ordinal, long dataOffset, byte[]? buffer, int bufferOffset, int length) => reader.GetBytes(ordinal, dataOffset, buffer, bufferOffset, length);
|
||||
public override char GetChar(int ordinal) => reader.GetChar(ordinal);
|
||||
public override long GetChars(int ordinal, long dataOffset, char[]? buffer, int bufferOffset, int length) => reader.GetChars(ordinal, dataOffset, buffer, bufferOffset, length);
|
||||
public override string GetDataTypeName(int ordinal) => reader.GetDataTypeName(ordinal);
|
||||
public override DateTime GetDateTime(int ordinal) => reader.GetDateTime(ordinal);
|
||||
public override decimal GetDecimal(int ordinal) => reader.GetDecimal(ordinal);
|
||||
public override double GetDouble(int ordinal) => reader.GetDouble(ordinal);
|
||||
public override IEnumerator GetEnumerator() => reader.GetEnumerator();
|
||||
|
||||
[return: DynamicallyAccessedMembers(DynamicallyAccessedMemberTypes.PublicFields | DynamicallyAccessedMemberTypes.PublicProperties)]
|
||||
public override Type GetFieldType(int ordinal) => ordinal switch
|
||||
{
|
||||
1 => typeof(long),
|
||||
4 or 6 => typeof(int),
|
||||
_ => typeof(string)
|
||||
};
|
||||
|
||||
public override float GetFloat(int ordinal) => reader.GetFloat(ordinal);
|
||||
public override Guid GetGuid(int ordinal) => reader.GetGuid(ordinal);
|
||||
public override short GetInt16(int ordinal) => reader.GetInt16(ordinal);
|
||||
public override int GetInt32(int ordinal) => reader.GetInt32(ordinal);
|
||||
public override long GetInt64(int ordinal) => reader.GetInt64(ordinal);
|
||||
public override string GetName(int ordinal) => reader.GetName(ordinal);
|
||||
public override int GetOrdinal(string name) => reader.GetOrdinal(name);
|
||||
public override DataTable? GetSchemaTable() => reader.GetSchemaTable();
|
||||
public override string GetString(int ordinal) => reader.GetString(ordinal);
|
||||
public override object GetValue(int ordinal) => reader.GetValue(ordinal);
|
||||
public override int GetValues(object[] values) => reader.GetValues(values);
|
||||
public override bool IsDBNull(int ordinal) => reader.IsDBNull(ordinal);
|
||||
public override bool NextResult() => reader.NextResult();
|
||||
public override bool Read() => reader.Read();
|
||||
|
||||
protected override void Dispose(bool disposing)
|
||||
{
|
||||
if (disposing)
|
||||
{
|
||||
reader.Dispose();
|
||||
table.Dispose();
|
||||
}
|
||||
|
||||
base.Dispose(disposing);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -52,6 +52,55 @@ public class RelationalStructuredLogSqlBuilderTests
|
|||
Assert.Contains("TimestampTo", query.Parameters.Keys);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void BuildQuery_AddsTextPredicateAcrossSearchableColumns()
|
||||
{
|
||||
var query = _builder.BuildQuery(new()
|
||||
{
|
||||
Text = "failure",
|
||||
TenantId = "tenant-a",
|
||||
SpanId = "span-a",
|
||||
Take = 25
|
||||
});
|
||||
|
||||
const string expectedTextPredicate = "([Message] LIKE @Text OR [MessageTemplate] LIKE @Text OR [Category] LIKE @Text OR [ExceptionJson] LIKE @Text OR [ScopesJson] LIKE @Text OR [PropertiesJson] LIKE @Text)";
|
||||
|
||||
Assert.Contains(expectedTextPredicate, query.Sql, StringComparison.Ordinal);
|
||||
Assert.Contains("[TenantId] = @TenantId", query.Sql, StringComparison.Ordinal);
|
||||
Assert.Contains("[SpanId] = @SpanId", query.Sql, StringComparison.Ordinal);
|
||||
Assert.Equal("%failure%", query.Parameters["Text"]);
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(-5, "FETCH 0")]
|
||||
[InlineData(5000, "FETCH 1000")]
|
||||
public void BuildQuery_ClampsTakeToSupportedRange(int take, string expectedLimit)
|
||||
{
|
||||
var query = _builder.BuildQuery(new()
|
||||
{
|
||||
Take = take
|
||||
});
|
||||
|
||||
Assert.Contains(expectedLimit, query.Sql, StringComparison.Ordinal);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void BuildListSources_GroupsBySourceAndOrdersBySource()
|
||||
{
|
||||
var sql = _builder.BuildListSources();
|
||||
|
||||
Assert.Equal("SELECT [SourceId], MAX([ReceivedAt]) AS [LastSeen] FROM [StructuredLogEvents] GROUP BY [SourceId] ORDER BY [SourceId]", sql);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void BuildDeleteOlderThan_UsesReceivedAtCutoffParameter()
|
||||
{
|
||||
var query = _builder.BuildDeleteOlderThan("2026-05-13T13:00:00.0000000+00:00");
|
||||
|
||||
Assert.Equal("DELETE FROM [StructuredLogEvents] WHERE [ReceivedAt] < @Cutoff", query.Sql);
|
||||
Assert.Equal("2026-05-13T13:00:00.0000000+00:00", query.Parameters["Cutoff"]);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void BuildDeleteRowsBeyondMax_DelegatesOffsetSyntaxToDialect()
|
||||
{
|
||||
|
|
|
|||
|
|
@ -0,0 +1,157 @@
|
|||
using System.Runtime.CompilerServices;
|
||||
using Elsa.Diagnostics.StructuredLogs.Contracts;
|
||||
using Elsa.Diagnostics.StructuredLogs.Models;
|
||||
using Elsa.Diagnostics.StructuredLogs.Services;
|
||||
|
||||
namespace Elsa.Diagnostics.StructuredLogs.UnitTests;
|
||||
|
||||
public class DefaultStructuredLogProviderTests
|
||||
{
|
||||
private readonly List<string> _calls = new();
|
||||
private readonly CapturingStructuredLogStore _store;
|
||||
private readonly CapturingStructuredLogLiveFeed _liveFeed;
|
||||
private readonly DefaultStructuredLogProvider _provider;
|
||||
|
||||
public DefaultStructuredLogProviderTests()
|
||||
{
|
||||
_store = new(_calls);
|
||||
_liveFeed = new(_calls);
|
||||
_provider = new(_store, _liveFeed);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task PublishAsync_WritesToStoreBeforePublishingToLiveFeed()
|
||||
{
|
||||
var logEvent = CreateLogEvent(1);
|
||||
|
||||
await _provider.PublishAsync(logEvent);
|
||||
|
||||
Assert.Equal(["store", "live-feed"], _calls);
|
||||
Assert.Same(logEvent, Assert.Single(_store.WrittenEvents));
|
||||
Assert.Same(logEvent, Assert.Single(_liveFeed.PublishedEvents));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task GetRecentAsync_DelegatesToStore()
|
||||
{
|
||||
var filter = new StructuredLogFilter { Take = 10 };
|
||||
var expected = new RecentStructuredLogsResult([CreateLogEvent(1)], 2);
|
||||
_store.RecentResult = expected;
|
||||
|
||||
var result = await _provider.GetRecentAsync(filter);
|
||||
|
||||
Assert.Same(filter, _store.LastQueryFilter);
|
||||
Assert.Same(expected, result);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SubscribeAsync_YieldsOnlyLogEventsFromLiveFeed()
|
||||
{
|
||||
var first = CreateLogEvent(1);
|
||||
var second = CreateLogEvent(2);
|
||||
_liveFeed.StreamItems.Add(StructuredLogStreamItem.FromLogEvent(first));
|
||||
_liveFeed.StreamItems.Add(StructuredLogStreamItem.FromDroppedEvents(new("source-a", 3, "SubscriberChannelFull")));
|
||||
_liveFeed.StreamItems.Add(StructuredLogStreamItem.FromLogEvent(second));
|
||||
|
||||
var result = new List<StructuredLogEvent>();
|
||||
await foreach (var logEvent in _provider.SubscribeAsync(new()))
|
||||
result.Add(logEvent);
|
||||
|
||||
Assert.Equal([first, second], result);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task SubscribeWithDroppedEventsAsync_DelegatesToLiveFeed()
|
||||
{
|
||||
var filter = new StructuredLogFilter { SourceId = "source-a" };
|
||||
var expected = StructuredLogStreamItem.FromDroppedEvents(new("source-a", 1, "SubscriberChannelFull"));
|
||||
_liveFeed.StreamItems.Add(expected);
|
||||
|
||||
var result = new List<StructuredLogStreamItem>();
|
||||
await foreach (var item in _provider.SubscribeWithDroppedEventsAsync(filter))
|
||||
result.Add(item);
|
||||
|
||||
Assert.Same(filter, _liveFeed.LastSubscribeFilter);
|
||||
Assert.Equal([expected], result);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public async Task ListSourcesAsync_DelegatesToStore()
|
||||
{
|
||||
var expected = new List<StructuredLogSource>
|
||||
{
|
||||
new() { Id = "source-a", DisplayName = "Source A" }
|
||||
};
|
||||
_store.Sources = expected;
|
||||
|
||||
var result = await _provider.ListSourcesAsync();
|
||||
|
||||
Assert.Same(expected, result);
|
||||
}
|
||||
|
||||
private static StructuredLogEvent CreateLogEvent(long sequence) =>
|
||||
new()
|
||||
{
|
||||
Sequence = sequence,
|
||||
Timestamp = DateTimeOffset.UtcNow,
|
||||
ReceivedAt = DateTimeOffset.UtcNow,
|
||||
Level = StructuredLogLevel.Information,
|
||||
Category = "Elsa.Workflows",
|
||||
Message = $"Message {sequence}",
|
||||
SourceId = "source-a"
|
||||
};
|
||||
|
||||
private sealed class CapturingStructuredLogStore(List<string> calls) : IStructuredLogStore
|
||||
{
|
||||
public List<StructuredLogEvent> WrittenEvents { get; } = new();
|
||||
public StructuredLogFilter? LastQueryFilter { get; private set; }
|
||||
public RecentStructuredLogsResult RecentResult { get; set; } = new([], 0);
|
||||
public IReadOnlyCollection<StructuredLogSource> Sources { get; set; } = [];
|
||||
|
||||
public ValueTask WriteAsync(StructuredLogEvent logEvent, CancellationToken cancellationToken = default)
|
||||
{
|
||||
calls.Add("store");
|
||||
WrittenEvents.Add(logEvent);
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
public ValueTask<RecentStructuredLogsResult> QueryAsync(StructuredLogFilter filter, CancellationToken cancellationToken = default)
|
||||
{
|
||||
LastQueryFilter = filter;
|
||||
return ValueTask.FromResult(RecentResult);
|
||||
}
|
||||
|
||||
public ValueTask<IReadOnlyCollection<StructuredLogSource>> ListSourcesAsync(CancellationToken cancellationToken = default)
|
||||
{
|
||||
return ValueTask.FromResult(Sources);
|
||||
}
|
||||
}
|
||||
|
||||
private sealed class CapturingStructuredLogLiveFeed(List<string> calls) : IStructuredLogLiveFeed
|
||||
{
|
||||
public List<StructuredLogEvent> PublishedEvents { get; } = new();
|
||||
public List<StructuredLogStreamItem> StreamItems { get; } = new();
|
||||
public StructuredLogFilter? LastSubscribeFilter { get; private set; }
|
||||
|
||||
public ValueTask PublishAsync(StructuredLogEvent logEvent, CancellationToken cancellationToken = default)
|
||||
{
|
||||
calls.Add("live-feed");
|
||||
PublishedEvents.Add(logEvent);
|
||||
return ValueTask.CompletedTask;
|
||||
}
|
||||
|
||||
public async IAsyncEnumerable<StructuredLogStreamItem> SubscribeAsync(
|
||||
StructuredLogFilter filter,
|
||||
[EnumeratorCancellation] CancellationToken cancellationToken = default)
|
||||
{
|
||||
LastSubscribeFilter = filter;
|
||||
|
||||
foreach (var item in StreamItems)
|
||||
{
|
||||
cancellationToken.ThrowIfCancellationRequested();
|
||||
yield return item;
|
||||
await Task.Yield();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
Loading…
Reference in a new issue