// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
using System.Collections.Concurrent;
using System.Diagnostics;
using System.Diagnostics.CodeAnalysis;
using System.Globalization;
using System.Runtime.InteropServices;
using System.Text;
using Aspire.Dashboard.Configuration;
using Aspire.Dashboard.Model;
using Aspire.Dashboard.Model.Otlp;
using Aspire.Dashboard.Otlp.Model;
using Aspire.Dashboard.Otlp.Model.MetricValues;
using Aspire.Dashboard.Utils;
using Google.Protobuf.Collections;
using Microsoft.Extensions.Logging;
using Microsoft.Extensions.Options;
using Microsoft.FluentUI.AspNetCore.Components;
using OpenTelemetry.Proto.Common.V1;
using OpenTelemetry.Proto.Logs.V1;
using OpenTelemetry.Proto.Metrics.V1;
using OpenTelemetry.Proto.Resource.V1;
using OpenTelemetry.Proto.Trace.V1;
using static Aspire.Tests.Shared.Telemetry.TelemetryTestHelpers;
using static OpenTelemetry.Proto.Trace.V1.Span.Types;
namespace Aspire.Dashboard.Otlp.Storage;
public sealed partial class InMemoryTelemetryRepository : ITelemetryRepository, ITelemetryRepositoryWriter
{
private readonly PauseManager _pauseManager;
private readonly IOutgoingPeerResolver[] _outgoingPeerResolvers;
private readonly ILogger _logger;
private bool _isReadOnly;
private readonly object _lock = new();
internal TimeSpan _subscriptionMinExecuteInterval = CallbackThrottler.DefaultMinExecuteInterval;
/// <summary>
/// Gets or sets a hook invoked at the start of each async read, identified by member name.
/// </summary>
/// <remarks>
/// Every async member of this fake otherwise completes synchronously, so callers never observe the
/// interleaving or faults that <c>SqliteTelemetryRepository</c> produces by running reads on the thread
/// pool. Version guards and loading-flag resets therefore go untested. Set this to return an incomplete
/// or faulted task to exercise those paths. When null, reads complete synchronously as before.
/// </remarks>
public Func<string, Task>? OnReadAsync { get; set; }
private Task ReadGateAsync(string readName) => OnReadAsync?.Invoke(readName) ?? Task.CompletedTask;
private readonly List<Subscription> _resourceSubscriptions = new();
private readonly List<Subscription> _logSubscriptions = new();
private readonly List<Subscription> _metricsSubscriptions = new();
private readonly List<Subscription> _tracesSubscriptions = new();
// Push-based streaming watchers - lazily initialized
private readonly object _watchersLock = new();
private List<SpanWatcher>? _spanWatchers;
private List<LogWatcher>? _logWatchers;
private readonly ConcurrentDictionary<ResourceKey, ResourceEntry> _resources = new();
private readonly ReaderWriterLockSlim _logsLock = new();
// Bounded by TelemetryRepositoryLimits.MaxScopeCount. Cleared when all logs are cleared.
private readonly Dictionary<string, OtlpScope> _logScopes = new();
private readonly CircularBuffer<OtlpLogEntry> _logs;
// Bounded by _resources count * MaxAttributeCount. Cleared per-resource or when all logs are cleared.
private readonly HashSet<(OtlpResource Resource, string PropertyKey)> _logPropertyKeys = new();
// Bounded by _resources count * MaxAttributeCount. Cleared per-resource or when all traces are cleared.
private readonly HashSet<(OtlpResource Resource, string PropertyKey)> _tracePropertyKeys = new();
private readonly Dictionary<ResourceKey, int> _resourceUnviewedErrorLogs = new();
private readonly ReaderWriterLockSlim _tracesLock = new();
// Bounded by TelemetryRepositoryLimits.MaxScopeCount. Cleared when all traces are cleared.
private readonly Dictionary<string, OtlpScope> _traceScopes = new();
private readonly CircularBuffer<OtlpTrace> _traces;
// Not explicitly capped per add — bounded only by the sum of span links across in-buffer traces.
// Cleaned up on trace eviction and clear, so growth is limited by the circular buffer capacity.
private readonly List<OtlpSpanLink> _spanLinks = new();
private readonly List<IDisposable> _peerResolverSubscriptions = new();
internal readonly OtlpContext _otlpContext;
public bool IsReadOnly => _isReadOnly;
public bool HasDisplayedMaxLogLimitMessage { get; set; }
public Message? MaxLogLimitMessage { get; set; }
public bool HasDisplayedMaxTraceLimitMessage { get; set; }
public Message? MaxTraceLimitMessage { get; set; }
// For testing.
internal List<OtlpSpanLink> SpanLinks => _spanLinks;
internal List<Subscription> TracesSubscriptions => _tracesSubscriptions;
internal void MakeReadOnly() => _isReadOnly = true;
private void ThrowIfReadOnly()
{
if (_isReadOnly)
{
throw new InvalidOperationException("Historical telemetry is read-only.");
}
}
public InMemoryTelemetryRepository(ILoggerFactory loggerFactory, IOptions<DashboardOptions> dashboardOptions, PauseManager pauseManager, IEnumerable<IOutgoingPeerResolver> outgoingPeerResolvers)
{
_logger = loggerFactory.CreateLogger(typeof(InMemoryTelemetryRepository));
_otlpContext = new OtlpContext
{
Logger = _logger,
Options = dashboardOptions.Value.TelemetryLimits
};
_pauseManager = pauseManager;
_outgoingPeerResolvers = outgoingPeerResolvers.ToArray();
_logs = new(_otlpContext.Options.MaxLogCount);
_traces = new(_otlpContext.Options.MaxTraceCount);
_traces.ItemRemovedForCapacity += TracesItemRemovedForCapacity;
foreach (var outgoingPeerResolver in _outgoingPeerResolvers)
{
_peerResolverSubscriptions.Add(outgoingPeerResolver.OnPeerChanges(OnPeerChanged));
}
}
private void TracesItemRemovedForCapacity(OtlpTrace trace)
{
// Remove links from central collection when the span is removed.
foreach (var span in trace.Spans)
{
foreach (var link in span.Links)
{
_spanLinks.Remove(link);
}
}
}
public List<OtlpResource> GetResources(bool includeUninstrumentedPeers = false)
{
return GetResourcesCore(includeUninstrumentedPeers, name: null);
}
public List<OtlpResource> GetResourcesByName(string name, bool includeUninstrumentedPeers = false)
{
return GetResourcesCore(includeUninstrumentedPeers, name);
}
private List<OtlpResource> GetResourcesCore(bool includeUninstrumentedPeers, string? name)
{
IEnumerable<OtlpResource> results = _resources.Values.Select(entry => entry.Resource);
if (!includeUninstrumentedPeers)
{
results = results.Where(a => !a.UninstrumentedPeer);
}
if (name != null)
{
results = results.Where(a => string.Equals(a.ResourceKey.Name, name, StringComparisons.ResourceName));
}
var resources = results.OrderBy(a => a.ResourceKey).ToList();
return resources;
}
public OtlpResource? GetResourceByCompositeName(string compositeName)
{
foreach (var kvp in _resources)
{
if (kvp.Key.EqualsCompositeName(compositeName))
{
return kvp.Value.Resource;
}
}
return null;
}
public OtlpResource? GetResource(ResourceKey key)
{
if (key.InstanceId == null)
{
throw new InvalidOperationException($"{nameof(ResourceKey)} must have an instance ID.");
}
return _resources.TryGetValue(key, out var entry) ? entry.Resource : null;
}
public List<OtlpResource> GetResources(ResourceKey key, bool includeUninstrumentedPeers = false)
{
if (key.InstanceId == null)
{
return GetResourcesByName(key.Name, includeUninstrumentedPeers: includeUninstrumentedPeers);
}
var resource = GetResource(key);
if (resource == null || (resource.UninstrumentedPeer && !includeUninstrumentedPeers))
{
return [];
}
return [resource];
}
private List<ResourceEntry> GetResourceEntries(ResourceKey key, bool includeUninstrumentedPeers = false)
{
IEnumerable<ResourceEntry> entries = key.InstanceId is null
? _resources.Values.Where(entry => string.Equals(entry.Resource.ResourceName, key.Name, StringComparisons.ResourceName))
: _resources.TryGetValue(key, out var entry) ? [entry] : [];
if (!includeUninstrumentedPeers)
{
entries = entries.Where(entry => !entry.Resource.UninstrumentedPeer);
}
return entries.ToList();
}
public Dictionary<ResourceKey, int> GetResourceUnviewedErrorLogsCount()
{
_logsLock.EnterReadLock();
try
{
return _resourceUnviewedErrorLogs.ToDictionary();
}
finally
{
_logsLock.ExitReadLock();
}
}
public void MarkViewedErrorLogs(ResourceKey? key)
{
_logsLock.EnterWriteLock();
try
{
if (key == null)
{
// Mark all logs as viewed.
if (_resourceUnviewedErrorLogs.Count > 0)
{
_resourceUnviewedErrorLogs.Clear();
RaiseSubscriptionChanged(_logSubscriptions);
}
return;
}
var resources = GetResources(key.Value);
foreach (var resource in resources)
{
// Mark one resource logs as viewed.
if (_resourceUnviewedErrorLogs.Remove(resource.ResourceKey))
{
RaiseSubscriptionChanged(_logSubscriptions);
}
}
}
finally
{
_logsLock.ExitWriteLock();
}
}
private OtlpResourceView GetOrAddResourceView(Resource resource) => GetOrAddResourceView(resource, out _);
private OtlpResourceView GetOrAddResourceView(Resource resource, out ResourceEntry resourceEntry)
{
ArgumentNullException.ThrowIfNull(resource);
var key = resource.GetResourceKey();
(resourceEntry, var isNew) = GetOrAddResourceEntry(key, uninstrumentedPeer: false);
if (isNew)
{
RaiseSubscriptionChanged(_resourceSubscriptions);
}
return resourceEntry.Resource.GetView(resource.Attributes);
}
private (ResourceEntry Entry, bool IsNew) GetOrAddResourceEntry(ResourceKey key, bool uninstrumentedPeer)
{
// Fast path.
if (_resources.TryGetValue(key, out var entry))
{
entry.Resource.SetUninstrumentedPeer(uninstrumentedPeer);
return (Entry: entry, IsNew: false);
}
// Check resource limit before adding a new resource.
// Note: This is a soft cap. Concurrent callers may both pass this check and slightly exceed the limit
// because _resources is a ConcurrentDictionary and the count check + GetOrAdd are not atomic.
if (_resources.Count >= _otlpContext.Options.MaxResourceCount)
{
throw new InvalidOperationException($"Resource limit of {_otlpContext.Options.MaxResourceCount} reached. Resource '{key}' will not be added.");
}
// Slower get or add path.
// This GetOrAdd allocates a closure, so we avoid it if possible.
var newResource = false;
entry = _resources.GetOrAdd(key, _ =>
{
newResource = true;
return new ResourceEntry(new OtlpResource(key.Name, key.InstanceId, uninstrumentedPeer, _otlpContext));
});
if (!newResource)
{
entry.Resource.SetUninstrumentedPeer(uninstrumentedPeer);
}
else
{
_logger.LogTrace("New resource added: {ResourceKey}", key);
}
return (Entry: entry, IsNew: newResource);
}
public Subscription OnNewResources(Func<Task> callback)
{
return AddSubscription(nameof(OnNewResources), null, SubscriptionType.Read, callback, _resourceSubscriptions);
}
public Subscription OnNewLogs(ResourceKey? resourceKey, SubscriptionType subscriptionType, Func<Task> callback)
{
return AddSubscription(nameof(OnNewLogs), resourceKey, subscriptionType, callback, _logSubscriptions);
}
public Subscription OnNewMetrics(ResourceKey? resourceKey, SubscriptionType subscriptionType, Func<Task> callback)
{
return AddSubscription(nameof(OnNewMetrics), resourceKey, subscriptionType, callback, _metricsSubscriptions);
}
public Subscription OnNewTraces(ResourceKey? resourceKey, SubscriptionType subscriptionType, Func<Task> callback)
{
return AddSubscription(nameof(OnNewTraces), resourceKey, subscriptionType, callback, _tracesSubscriptions);
}
private Subscription AddSubscription(string name, ResourceKey? resourceKey, SubscriptionType subscriptionType, Func<Task> callback, List<Subscription> subscriptions)
{
Subscription? subscription = null;
subscription = new Subscription(name, resourceKey, subscriptionType, callback, () =>
{
lock (_lock)
{
subscriptions.Remove(subscription!);
}
}, ExecutionContext.Capture(), _logger, _subscriptionMinExecuteInterval);
lock (_lock)
{
subscriptions.Add(subscription);
}
return subscription;
}
private void RaiseSubscriptionChanged(List<Subscription> subscriptions)
{
lock (_lock)
{
foreach (var subscription in subscriptions)
{
subscription.Execute();
}
}
}
public Task AddLogsAsync(AddContext context, RepeatedField<ResourceLogs> resourceLogs)
{
ThrowIfReadOnly();
if (_pauseManager.AreStructuredLogsPaused(out _))
{
_logger.LogTrace("{Count} incoming structured log(s) ignored because of an active pause.", resourceLogs.Count);
return Task.CompletedTask;
}
foreach (var rl in resourceLogs)
{
OtlpResourceView resourceView;
try
{
resourceView = GetOrAddResourceView(rl.Resource);
}
catch (Exception ex)
{
context.FailureCount += rl.ScopeLogs.Sum(s => s.LogRecords.Count);
_otlpContext.Logger.LogInformation(ex, "Error adding resource.");
continue;
}
AddLogsCore(context, resourceView, rl.ScopeLogs);
SetResourceHasLogs(resourceView.Resource, true);
}
RaiseSubscriptionChanged(_logSubscriptions);
return Task.CompletedTask;
}
public void AddLogsCore(AddContext context, OtlpResourceView resourceView, RepeatedField<ScopeLogs> scopeLogs)
{
List<OtlpLogEntry>? addedLogs = null;
_logsLock.EnterWriteLock();
try
{
foreach (var sl in scopeLogs)
{
if (!OtlpHelpers.TryGetOrAddScope(_logScopes, sl.Scope, _otlpContext, TelemetryType.Logs, out var scope))
{
context.FailureCount += sl.LogRecords.Count;
continue;
}
foreach (var record in sl.LogRecords)
{
try
{
var logEntry = CreateOtlpLogEntry(record, resourceView, scope, _otlpContext);
// Insert log entry in the correct position based on timestamp.
// Logs can be added out of order by different services.
var added = false;
for (var i = _logs.Count - 1; i >= 0; i--)
{
if (logEntry.TimeStamp > _logs[i].TimeStamp)
{
_logs.Insert(i + 1, logEntry);
added = true;
break;
}
}
if (!added)
{
_logs.Insert(0, logEntry);
}
// For log entries error and above, increment the unviewed count if there are no read log subscriptions for the resource.
// We don't increment the count if there are active read subscriptions because the count will be quickly decremented when the subscription callback is run.
// Notifying the user there are errors and then immediately clearing the notification is confusing.
if (logEntry.IsError)
{
if (!_logSubscriptions.Any(s => s.SubscriptionType == SubscriptionType.Read && (s.ResourceKey == resourceView.ResourceKey || s.ResourceKey == null)))
{
ref var count = ref CollectionsMarshal.GetValueRefOrAddDefault(_resourceUnviewedErrorLogs, resourceView.ResourceKey, out _);
// Adds to dictionary if not present.
count++;
}
}
foreach (var kvp in logEntry.Attributes)
{
_logPropertyKeys.Add((resourceView.Resource, kvp.Key));
}
// Collect log for push-based streaming (lazy init to avoid allocation when no watchers)
addedLogs ??= new List<OtlpLogEntry>();
addedLogs.Add(logEntry);
context.SuccessCount++;
}
catch (Exception ex)
{
context.FailureCount++;
_otlpContext.Logger.LogInformation(ex, "Error adding log entry.");
}
}
}
}
finally
{
_logsLock.ExitWriteLock();
}
// Push logs to watchers outside the lock
if (addedLogs is not null)
{
PushLogsToWatchers(addedLogs, resourceView.ResourceKey);
}
}
public Task<PagedResult<OtlpLogEntry>> GetLogsAsync(GetLogsContext context, CancellationToken cancellationToken = default) => Task.FromResult(GetLogs(context));
private PagedResult<OtlpLogEntry> GetLogs(GetLogsContext context)
{
List<OtlpResource>? resources = null;
if (context.ResourceKeys is { Count: > 0 } keys)
{
resources = [];
foreach (var key in keys)
{
resources.AddRange(GetResources(key));
}
if (resources.Count == 0)
{
return PagedResult<OtlpLogEntry>.Empty;
}
}
_logsLock.EnterReadLock();
try
{
var results = _logs.AsEnumerable();
if (resources?.Count > 0)
{
results = results.Where(l => MatchResources(l.ResourceView.ResourceKey, resources));
}
foreach (var filter in context.Filters.GetEnabledFilters())
{
results = filter.Apply(results);
}
if (context.TextFragments is { Length: > 0 } textFragments)
{
results = results.Where(l => MatchesLogTextFragments(l, textFragments));
}
var startIndex = context.StartIndex;
if (context.LatestItemCount is { } latestItemCount)
{
startIndex += Math.Max(results.Count() - Math.Max(latestItemCount, 0), 0);
}
return OtlpHelpers.GetItems(results, startIndex, context.Count, _logs.IsFull);
}
finally
{
_logsLock.ExitReadLock();
}
}
public Task<PagedResult<LogSummary>> GetLogSummariesAsync(GetLogsContext context, CancellationToken cancellationToken)
{
cancellationToken.ThrowIfCancellationRequested();
return Task.FromResult(GetLogSummaries(context));
}
private PagedResult<LogSummary> GetLogSummaries(GetLogsContext context)
{
var result = GetLogs(context);
return new PagedResult<LogSummary>
{
Items = result.Items.Select(log => new LogSummary
{
InternalId = log.InternalId,
TimeStamp = log.TimeStamp,
Severity = log.Severity,
Message = log.Message,
SpanId = log.SpanId,
TraceId = log.TraceId,
ScopeName = log.Scope.Name,
EventName = OtlpHelpers.GetEventName(log),
Resource = log.ResourceView.Resource,
ExceptionText = OtlpLogEntry.GetExceptionText(log),
HasGenAI = global::Aspire.Dashboard.Model.GenAI.GenAIHelpers.HasGenAIAttribute(log.Attributes) ||
GetSpan(log.TraceId, log.SpanId) is { } span && global::Aspire.Dashboard.Model.GenAI.GenAIHelpers.HasGenAIAttribute(span.Attributes)
}).ToList(),
TotalItemCount = result.TotalItemCount,
IsFull = result.IsFull
};
}
public OtlpLogEntry? GetLog(long logId)
{
_logsLock.EnterReadLock();
try
{
foreach (var logEntry in _logs)
{
if (logEntry.InternalId == logId)
{
return logEntry;
}
}
return null;
}
finally
{
_logsLock.ExitReadLock();
}
}
/// <summary>
/// Gets logs associated with a specific span, filtered by trace ID and span ID.
/// </summary>
/// <param name="traceId">The trace ID.</param>
/// <param name="spanId">The span ID.</param>
/// <returns>A list of log entries associated with the span.</returns>
public async Task<List<OtlpLogEntry>> GetLogsForSpanAsync(string traceId, string spanId, CancellationToken cancellationToken)
{
var result = await GetLogsAsync(CreateLogsForSpanContext(traceId, spanId), cancellationToken).ConfigureAwait(false);
return result.Items;
}
private static GetLogsContext CreateLogsForSpanContext(string traceId, string spanId)
{
return new GetLogsContext
{
ResourceKeys = [],
Count = int.MaxValue,
StartIndex = 0,
Filters =
[
new FieldTelemetryFilter
{
Field = KnownStructuredLogFields.TraceIdField,
Condition = FilterCondition.Equals,
Value = traceId
},
new FieldTelemetryFilter
{
Field = KnownStructuredLogFields.SpanIdField,
Condition = FilterCondition.Equals,
Value = spanId
}
]
};
}
/// <summary>
/// Gets logs associated with a specific trace, filtered by trace ID.
/// </summary>
/// <param name="traceId">The trace ID.</param>
/// <returns>A list of log entries associated with the trace.</returns>
public async Task<List<OtlpLogEntry>> GetLogsForTraceAsync(string traceId, CancellationToken cancellationToken)
{
var result = await GetLogsAsync(CreateLogsForTraceContext(traceId), cancellationToken).ConfigureAwait(false);
return result.Items;
}
private static GetLogsContext CreateLogsForTraceContext(string traceId)
{
return new GetLogsContext
{
ResourceKeys = [],
Count = int.MaxValue,
StartIndex = 0,
Filters =
[
new FieldTelemetryFilter
{
Field = KnownStructuredLogFields.TraceIdField,
Condition = FilterCondition.Equals,
Value = traceId
}
]
};
}
public async Task<List<string>> GetLogPropertyKeysAsync(ResourceKey? resourceKey, CancellationToken cancellationToken)
{
await ReadGateAsync(nameof(GetLogPropertyKeysAsync)).ConfigureAwait(false);
cancellationToken.ThrowIfCancellationRequested();
List<OtlpResource>? resources = null;
if (resourceKey != null)
{
resources = GetResources(resourceKey.Value);
}
_logsLock.EnterReadLock();
try
{
var resourceKeys = _logPropertyKeys.AsEnumerable();
if (resources?.Count > 0)
{
resourceKeys = resourceKeys.Where(keys => MatchResources(keys.Resource.ResourceKey, resources));
}
var keys = resourceKeys.Select(keys => keys.PropertyKey).Distinct();
return keys.OrderBy(k => k).ToList();
}
finally
{
_logsLock.ExitReadLock();
}
}
public async Task<List<string>> GetTracePropertyKeysAsync(ResourceKey? resourceKey, CancellationToken cancellationToken)
{
await ReadGateAsync(nameof(GetTracePropertyKeysAsync)).ConfigureAwait(false);
cancellationToken.ThrowIfCancellationRequested();
List<OtlpResource>? resources = null;
if (resourceKey != null)
{
resources = GetResources(resourceKey.Value, includeUninstrumentedPeers: true);
}
_tracesLock.EnterReadLock();
try
{
var resourceKeys = _tracePropertyKeys.AsEnumerable();
if (resources?.Count > 0)
{
resourceKeys = resourceKeys.Where(keys => MatchResources(keys.Resource.ResourceKey, resources));
}
var keys = resourceKeys.Select(keys => keys.PropertyKey).Distinct();
return keys.OrderBy(k => k).ToList();
}
finally
{
_tracesLock.ExitReadLock();
}
}
public Task<GetTracesResponse> GetTracesAsync(GetTracesRequest context, CancellationToken cancellationToken = default) => Task.FromResult(GetTraces(context));
private GetTracesResponse GetTraces(GetTracesRequest context)
{
List<OtlpResource>? resources = null;
if (context.ResourceKeys is { Count: > 0 } keys)
{
resources = [];
foreach (var key in keys)
{
resources.AddRange(GetResources(key, includeUninstrumentedPeers: true));
}
if (resources.Count == 0)
{
return new GetTracesResponse
{
PagedResult = PagedResult<OtlpTrace>.Empty,
MaxDuration = TimeSpan.Zero
};
}
}
_tracesLock.EnterReadLock();
try
{
var filters = context.Filters.GetEnabledFilters().ToList();
var optimizedFilters = CreateOptimizedTraceFilters(filters);
var resourceFilter = resources is { Count: > 0 } ? resources : null;
var hasTelemetryFilters = filters.Count > 0;
var hasFilterText = !string.IsNullOrWhiteSpace(context.TraceNameFilterText);
var hasTextFragments = context.TextFragments is { Length: > 0 };
var startIndex = Math.Max(context.StartIndex, 0);
var count = Math.Max(context.Count, 0);
List<OtlpTrace>? items = null;
var latestItemCount = Math.Max(context.LatestItemCount ?? 0, 0);
Queue<OtlpTrace>? latestItems = context.LatestItemCount is not null
? new Queue<OtlpTrace>(latestItemCount)
: null;
var totalItemCount = 0;
var maxDuration = default(TimeSpan);
foreach (var trace in _traces)
{
if (resourceFilter is not null && !MatchResources(trace, resourceFilter))
{
continue;
}
if (hasFilterText && !trace.FullName.Contains(context.TraceNameFilterText!, StringComparison.OrdinalIgnoreCase))
{
continue;
}
if (hasTelemetryFilters && !MatchesFilters(trace, filters, optimizedFilters))
{
continue;
}
if (hasTextFragments && !MatchesTraceTextFragments(trace, context.TextFragments!))
{
continue;
}
totalItemCount++;
var duration = trace.Duration;
if (duration > maxDuration)
{
maxDuration = duration;
}
if (latestItems is not null)
{
latestItems.Enqueue(trace);
if (latestItems.Count > latestItemCount)
{
latestItems.Dequeue();
}
continue;
}
// Keep paging, total count, and MaxDuration in the same scan. The dashboard
// needs MaxDuration for the full filtered set, while only the requested page
// should pay the clone cost needed to isolate callers from live span updates.
if (totalItemCount > startIndex && (items?.Count ?? 0) < count)
{
items ??= new List<OtlpTrace>(Math.Min(count, _traces.Count));
items.Add(OtlpTrace.Clone(trace));
}
}
if (latestItems is not null)
{
items = latestItems.Skip(startIndex).Take(count).Select(OtlpTrace.Clone).ToList();
}
var pagedResults = new PagedResult<OtlpTrace>
{
Items = items ?? new List<OtlpTrace>(),
TotalItemCount = totalItemCount,
IsFull = _traces.IsFull
};
return new GetTracesResponse
{
PagedResult = pagedResults,
MaxDuration = maxDuration
};
}
finally
{
_tracesLock.ExitReadLock();
}
}
public Task<GetTraceSummariesResponse> GetTraceSummariesAsync(GetTracesRequest context, CancellationToken cancellationToken)
{
cancellationToken.ThrowIfCancellationRequested();
return Task.FromResult(GetTraceSummaries(context));
}
private GetTraceSummariesResponse GetTraceSummaries(GetTracesRequest context)
{
var result = GetTraces(context);
return new GetTraceSummariesResponse
{
PagedResult = new PagedResult<TraceSummary>
{
Items = result.PagedResult.Items.Select(trace => new TraceSummary
{
TraceId = trace.TraceId,
FullName = trace.FullName,
StartTime = trace.FirstSpan.StartTime,
Duration = trace.Duration,
RootResource = trace.RootOrFirstSpan.Source.Resource,
Resources = TraceHelpers.GetOrderedResources(trace).Select(resource => new TraceResourceSummary
{
Resource = resource.Resource,
TotalSpans = resource.TotalSpans,
ErroredSpans = resource.ErroredSpans
}).ToList(),
HasError = trace.Spans.Any(span => span.Status == OtlpSpanStatusCode.Error),
HasGenAI = trace.Spans.Any(span => global::Aspire.Dashboard.Model.GenAI.GenAIHelpers.HasGenAIAttribute(span.Attributes))
}).ToList(),
TotalItemCount = result.PagedResult.TotalItemCount,
IsFull = result.PagedResult.IsFull
},
MaxDuration = result.MaxDuration
};
}
public Task<GetSpansResponse> GetSpansAsync(GetSpansRequest context, CancellationToken cancellationToken = default) => Task.FromResult(GetSpans(context));
private GetSpansResponse GetSpans(GetSpansRequest context)
{
List<OtlpResource>? resources = null;
if (context.ResourceKeys is { Count: > 0 } keys)
{
resources = [];
foreach (var key in keys)
{
resources.AddRange(GetResources(key, includeUninstrumentedPeers: true));
}
if (resources.Count == 0)
{
return new GetSpansResponse
{
PagedResult = PagedResult<OtlpSpan>.Empty
};
}
}
_tracesLock.EnterReadLock();
try
{
var filters = context.Filters.GetEnabledFilters().ToList();
var resourceFilter = resources is { Count: > 0 } ? resources : null;
var hasTraceIdFilter = !string.IsNullOrEmpty(context.TraceId);
var startIndex = Math.Max(context.StartIndex, 0);
var count = Math.Max(context.Count, 0);
List<OtlpSpan>? items = null;
var totalItemCount = 0;
foreach (var trace in _traces)
{
if (resourceFilter is not null && !MatchResources(trace, resourceFilter))
{
continue;
}
if (hasTraceIdFilter && !OtlpHelpers.MatchTelemetryId(context.TraceId!, trace.TraceId))
{
continue;
}
foreach (var span in trace.Spans)
{
if (!MatchesSpanCriteria(span, context.TraceId, context.HasError, filters, context.TextFragments))
{
continue;
}
totalItemCount++;
if (totalItemCount > startIndex && (items?.Count ?? 0) < count)
{
items ??= new List<OtlpSpan>(Math.Min(count, 64));
items.Add(span);
}
}
}
var pagedResults = new PagedResult<OtlpSpan>
{
Items = items ?? new List<OtlpSpan>(),
TotalItemCount = totalItemCount,
IsFull = _traces.IsFull
};
return new GetSpansResponse
{
PagedResult = pagedResults
};
}
finally
{
_tracesLock.ExitReadLock();
}
}
/// <summary>
/// Applies traceId, hasError, telemetry filters, and text fragment matching to a span.
/// Shared between GetSpans (initial query) and PushSpansToWatchers (push path).
/// </summary>
private static bool MatchesSpanCriteria(OtlpSpan span, string? traceId, bool? hasError, List<TelemetryFilter> filters, string[]? textFragments)
{
if (!string.IsNullOrEmpty(traceId) && !OtlpHelpers.MatchTelemetryId(traceId, span.TraceId))
{
return false;
}
if (hasError.HasValue && (span.Status == OtlpSpanStatusCode.Error) != hasError.Value)
{
return false;
}
if (filters.Count > 0 && !MatchesSpanFilters(span, filters))
{
return false;
}
if (textFragments is { Length: > 0 } fragments && !MatchesSpanTextFragments(span, fragments))
{
return false;
}
return true;
}
/// <summary>
/// Returns true when the span matches all enabled filters applied directly to the span.
/// </summary>
private static bool MatchesSpanFilters(OtlpSpan span, List<TelemetryFilter> filters)
{
foreach (var filter in filters)
{
if (!filter.Enabled)
{
continue;
}
if (!filter.Apply(span))
{
return false;
}
}
return true;
}
/// <summary>
/// Returns true when the span's searchable fields match all text fragments.
/// </summary>
private static bool MatchesSpanTextFragments(OtlpSpan span, string[] fragments)
{
return SearchTextParser.MatchesAllFragments(fragments, span, static (span, fragment) =>
{
if (span.Name.Contains(fragment, StringComparisons.FullTextSearch) ||
span.SpanId.Contains(fragment, StringComparisons.FullTextSearch) ||
span.TraceId.Contains(fragment, StringComparisons.FullTextSearch) ||
span.Scope.Name.Contains(fragment, StringComparisons.FullTextSearch) ||
span.Source.Resource.ResourceName.Contains(fragment, StringComparisons.FullTextSearch) ||
span.Status.ToString().Contains(fragment, StringComparisons.FullTextSearch) ||
span.Kind.ToString().Contains(fragment, StringComparisons.FullTextSearch))
{
return true;
}
if (span.StatusMessage is not null && span.StatusMessage.Contains(fragment, StringComparisons.FullTextSearch))
{
return true;
}
foreach (var attribute in span.Attributes)
{
if (attribute.Key.Contains(fragment, StringComparisons.FullTextSearch) ||
attribute.Value.Contains(fragment, StringComparisons.FullTextSearch))
{
return true;
}
}
foreach (var evt in span.Events)
{
if (evt.Name.Contains(fragment, StringComparisons.FullTextSearch))
{
return true;
}
}
return false;
});
}
/// <summary>
/// Returns true when the trace matches all text fragments. A trace matches if its full name
/// matches all fragments or any of its spans matches all fragments.
/// </summary>
private static bool MatchesTraceTextFragments(OtlpTrace trace, string[] fragments)
{
if (SearchTextParser.MatchesAllFragments(fragments, trace.FullName, static (fullName, fragment) =>
fullName.Contains(fragment, StringComparisons.FullTextSearch)))
{
return true;
}
foreach (var span in trace.Spans)
{
if (MatchesSpanTextFragments(span, fragments))
{
return true;
}
}
return false;
}
/// <summary>
/// Returns true when the log entry's searchable fields match all text fragments.
/// </summary>
private static bool MatchesLogTextFragments(OtlpLogEntry log, string[] fragments)
{
return SearchTextParser.MatchesAllFragments(fragments, log, static (log, fragment) =>
{
if (log.Message.Contains(fragment, StringComparisons.FullTextSearch) ||
log.Scope.Name.Contains(fragment, StringComparisons.FullTextSearch) ||
log.TraceId.Contains(fragment, StringComparisons.FullTextSearch) ||
log.SpanId.Contains(fragment, StringComparisons.FullTextSearch) ||
log.Severity.ToString().Contains(fragment, StringComparisons.FullTextSearch) ||
log.ResourceView.Resource.ResourceName.Contains(fragment, StringComparisons.FullTextSearch))
{
return true;
}
if (log.EventName is not null && log.EventName.Contains(fragment, StringComparisons.FullTextSearch))
{
return true;
}
foreach (var attribute in log.Attributes)
{
if (attribute.Key.Contains(fragment, StringComparisons.FullTextSearch) ||
attribute.Value.Contains(fragment, StringComparisons.FullTextSearch))
{
return true;
}
}
return false;
});
}
private static List<TraceFilter>? CreateOptimizedTraceFilters(List<TelemetryFilter> filters)
{
List<TraceFilter>? result = null;
for (var i = 0; i < filters.Count; i++)
{
var filter = filters[i];
var traceFilter = TraceFilter.Create(filter);
if (traceFilter.IsOptimized)
{
result ??= new List<TraceFilter>(filters.Count);
for (var j = result.Count; j < i; j++)
{
result.Add(new TraceFilter(filters[j], null, null));
}
}
result?.Add(traceFilter);
}
return result;
}
private static bool MatchesFilters(OtlpTrace trace, List<TelemetryFilter> filters, List<TraceFilter>? optimizedFilters)
{
if (optimizedFilters is not null)
{
return MatchesFilters(trace, optimizedFilters);
}
// Duration filters apply to the trace's overall duration, not individual spans.
foreach (var filter in filters)
{
if (filter.IsTraceDurationFilter() && !filter.HasNumericMatch(trace.Duration.TotalMilliseconds))
{
return false;
}
}
// Single pass over spans handles both filter polarities:
// - Negative filters (not-equal, not-contains) use ALL-span semantics: the trace is
// excluded if ANY span violates the condition.
// - Positive filters use ANY-span semantics: the trace matches when at least one span
// satisfies all positive filters.
var hasPositiveMatch = false;
foreach (var span in trace.Spans)
{
// Once a positive match has been found on an earlier span, skip re-evaluating
// positive filters on subsequent spans (only negative filters still need checking).
var positiveMatch = !hasPositiveMatch;
foreach (var filter in filters)
{
if (filter.IsTraceDurationFilter())
{
continue;
}
if (filter.IsNegativeFilter)
{
if (!filter.Apply(span))
{
return false;
}
}
else if (positiveMatch)
{
if (!filter.Apply(span))
{
positiveMatch = false;
}
}
}
if (positiveMatch)
{
hasPositiveMatch = true;
}
}
return hasPositiveMatch;
}
private static bool MatchesFilters(OtlpTrace trace, List<TraceFilter> optimizedFilters)
{
// Duration filters apply to the trace's overall duration, not individual spans.
foreach (var filter in optimizedFilters)
{
if (filter.IsDurationFilter && !filter.ApplyDuration(trace.Duration.TotalMilliseconds))
{
return false;
}
}
// Single pass over spans handles both filter polarities:
// - Negative filters use ALL-span semantics (any violation excludes the trace).
// - Positive filters use ANY-span semantics (one span matching all suffices).
var hasPositiveMatch = false;
foreach (var span in trace.Spans)
{
// Once a positive match has been found on an earlier span, skip re-evaluating
// positive filters on subsequent spans (only negative filters still need checking).
var positiveMatch = !hasPositiveMatch;
foreach (var filter in optimizedFilters)
{
if (filter.IsDurationFilter)
{
continue;
}
if (filter.IsNegativeFilter)
{
if (!filter.Apply(span))
{
return false;
}
}
else if (positiveMatch)
{
if (!filter.Apply(span))
{
positiveMatch = false;
}
}
}
if (positiveMatch)
{
hasPositiveMatch = true;
}
}
return hasPositiveMatch;
}
private readonly record struct TraceFilter(TelemetryFilter Filter, DurationFilter? OptimizedDurationFilter, StringFilter? OptimizedStringFilter)
{
public bool IsOptimized => OptimizedDurationFilter is not null || OptimizedStringFilter is not null;
public bool IsNegativeFilter => Filter.IsNegativeFilter;
public bool IsDurationFilter => OptimizedDurationFilter is not null || Filter.IsTraceDurationFilter();
public static TraceFilter Create(TelemetryFilter filter)
{
if (DurationFilter.TryCreate(filter, out var durationFilter))
{
return new TraceFilter(filter, durationFilter, null);
}
if (StringFilter.TryCreate(filter, out var stringFilter))
{
return new TraceFilter(filter, null, stringFilter);
}
return new TraceFilter(filter, null, null);
}
public bool Apply(OtlpSpan span)
{
if (OptimizedStringFilter is { } stringFilter)
{
return stringFilter.Apply(span);
}
return Filter.Apply(span);
}
public bool ApplyDuration(double traceDurationMs)
{
if (OptimizedDurationFilter is { } durationFilter)
{
return durationFilter.Apply(traceDurationMs);
}
return Filter.HasNumericMatch(traceDurationMs);
}
}
private readonly record struct DurationFilter(FilterCondition Condition, double Value)
{
public static bool TryCreate(TelemetryFilter filter, out DurationFilter durationFilter)
{
if (filter is FieldTelemetryFilter { Field: KnownTraceFields.DurationField } fieldFilter &&
double.TryParse(fieldFilter.Value, NumberStyles.Float, CultureInfo.InvariantCulture, out var value) &&
double.IsFinite(value) &&
IsSupportedCondition(fieldFilter.Condition))
{
durationFilter = new DurationFilter(fieldFilter.Condition, value);
return true;
}
durationFilter = default;
return false;
}
public bool Apply(double durationMilliseconds)
{
if (!double.IsFinite(durationMilliseconds))
{
return false;
}
// Avoid formatting the span duration and reparsing the filter threshold for each
// span. Duration is a known numeric field, so this preserves FieldTelemetryFilter's
// numeric comparison semantics without the per-span string allocation.
return Condition switch
{
FilterCondition.Equals => durationMilliseconds == Value,
FilterCondition.GreaterThan => durationMilliseconds > Value,
FilterCondition.LessThan => durationMilliseconds < Value,
FilterCondition.GreaterThanOrEqual => durationMilliseconds >= Value,
FilterCondition.LessThanOrEqual => durationMilliseconds <= Value,
FilterCondition.NotEqual => durationMilliseconds != Value,
_ => false
};
}
private static bool IsSupportedCondition(FilterCondition condition)
{
return condition is FilterCondition.Equals
or FilterCondition.GreaterThan
or FilterCondition.LessThan
or FilterCondition.GreaterThanOrEqual
or FilterCondition.LessThanOrEqual
or FilterCondition.NotEqual;
}
}
private readonly record struct StringFilter(string Field, FilterCondition Condition, string Value)
{
public static bool TryCreate(TelemetryFilter filter, out StringFilter stringFilter)
{
if (filter is FieldTelemetryFilter fieldFilter &&
!FieldTelemetryFilter.IsNumericField(fieldFilter.Field) &&
!FieldTelemetryFilter.IsDateField(fieldFilter.Field) &&
IsSupportedCondition(fieldFilter.Condition))
{
stringFilter = new StringFilter(fieldFilter.Field, fieldFilter.Condition, fieldFilter.Value);
return true;
}
stringFilter = default;
return false;
}
public bool Apply(OtlpSpan span)
{
var fieldValue = OtlpSpan.GetFieldValue(span, Field);
var isNot = Condition is FilterCondition.NotEqual or FilterCondition.NotContains;
if (!isNot)
{
if (fieldValue.Value1 is not null && IsMatch(fieldValue.Value1))
{
return true;
}
if (fieldValue.Value2 is not null && IsMatch(fieldValue.Value2))
{
return true;
}
}
else
{
// And — both values must satisfy the not-equal/not-contains condition.
// When the field is absent (Value1 is null), the span trivially satisfies the
// negative condition — a span without the field cannot contain/equal the value.
if (fieldValue.Value1 is null)
{
return true;
}
if (IsMatch(fieldValue.Value1))
{
if (fieldValue.Value2 is null || IsMatch(fieldValue.Value2))
{
return true;
}
}
}
return false;
}
private bool IsMatch(string fieldValue)
{
return Condition switch
{
FilterCondition.Equals => string.Equals(fieldValue, Value, StringComparisons.OtlpFieldValue),
FilterCondition.Contains => fieldValue.Contains(Value, StringComparisons.OtlpFieldValue),
FilterCondition.NotEqual => !string.Equals(fieldValue, Value, StringComparisons.OtlpFieldValue),
FilterCondition.NotContains => !fieldValue.Contains(Value, StringComparisons.OtlpFieldValue),
_ => false
};
}
private static bool IsSupportedCondition(FilterCondition condition)
{
return condition is FilterCondition.Equals
or FilterCondition.Contains
or FilterCondition.NotEqual
or FilterCondition.NotContains;
}
}
private static bool MatchResources(ResourceKey resourceKey, List<OtlpResource> resources)
{
foreach (var resource in resources)
{
if (resourceKey == resource.ResourceKey)
{
return true;
}
}
return false;
}
private static bool MatchResources(OtlpTrace t, List<OtlpResource> resources)
{
for (var i = 0; i < resources.Count; i++)
{
var resourceKey = resources[i].ResourceKey;
// Spans collection type returns a struct enumerator so it's ok to foreach inside another loop.
foreach (var span in t.Spans)
{
if (span.Source.ResourceKey == resourceKey || span.UninstrumentedPeer?.ResourceKey == resourceKey)
{
return true;
}
}
}
return false;
}
private void SetResourceHasLogs(OtlpResource resource, bool value)
{
if (resource.HasLogs != value)
{
resource.HasLogs = value;
RaiseSubscriptionChanged(_resourceSubscriptions);
}
}
private void SetResourceHasTraces(OtlpResource resource, bool value)
{
if (resource.HasTraces != value)
{
resource.HasTraces = value;
RaiseSubscriptionChanged(_resourceSubscriptions);
}
}
private void SetResourceHasMetrics(OtlpResource resource, bool value)
{
if (resource.HasMetrics != value)
{
resource.HasMetrics = value;
RaiseSubscriptionChanged(_resourceSubscriptions);
}
}
/// <summary>
/// Clears selected telemetry signals for specified resources.
/// </summary>
/// <param name="selectedResources">Dictionary mapping resource names to the data types to clear.</param>
public Task ClearSelectedSignalsAsync(Dictionary<string, HashSet<AspireDataType>> selectedResources)
{
ThrowIfReadOnly();
var allOtlpResources = GetResources();
foreach (var otlpResource in allOtlpResources)
{
var resourceName = otlpResource.ResourceKey.GetCompositeName();
if (!selectedResources.TryGetValue(resourceName, out var dataTypes))
{
continue;
}
var clearStructuredLogs = IsDataTypeSelected(dataTypes, AspireDataType.StructuredLogs);
var clearTraces = IsDataTypeSelected(dataTypes, AspireDataType.Traces);
var clearMetrics = IsDataTypeSelected(dataTypes, AspireDataType.Metrics);
if (clearStructuredLogs)
{
ClearStructuredLogs(otlpResource.ResourceKey);
}
if (clearTraces)
{
ClearTraces(otlpResource.ResourceKey);
}
if (clearMetrics)
{
ClearMetrics(otlpResource.ResourceKey);
}
// If Resource flag is set, remove the resource itself
if (dataTypes.Contains(AspireDataType.Resource))
{
ClearResource(otlpResource.ResourceKey);
}
}
static bool IsDataTypeSelected(HashSet<AspireDataType> dataTypes, AspireDataType dataType)
{
// Always remove everything if the resource is being removed.
return dataTypes.Contains(dataType) || dataTypes.Contains(AspireDataType.Resource);
}
return Task.CompletedTask;
}
public Task ClearTracesAsync(ResourceKey? resourceKey = null)
{
ClearTraces(resourceKey);
return Task.CompletedTask;
}
private void ClearTraces(ResourceKey? resourceKey)
{
ThrowIfReadOnly();
List<OtlpResource>? resources = null;
if (resourceKey.HasValue)
{
resources = GetResources(resourceKey.Value, includeUninstrumentedPeers: true);
}
_tracesLock.EnterWriteLock();
try
{
if (resources is null || resources.Count == 0)
{
// Nothing selected, clear everything.
_traces.Clear();
_traceScopes.Clear();
_tracePropertyKeys.Clear();
_spanLinks.Clear();
foreach (var resource in _resources.Values)
{
SetResourceHasTraces(resource.Resource, false);
}
}
else
{
for (var i = _traces.Count - 1; i >= 0; i--)
{
var trace = _traces[i];
// Remove trace if any span matches one of the resources. This matches filter behavior.
if (MatchResources(trace, resources))
{
// Remove span links for the removed trace.
foreach (var span in trace.Spans)
{
foreach (var link in span.Links)
{
_spanLinks.Remove(link);
}
}
_traces.RemoveAt(i);
continue;
}
}
// Remove property keys for cleared resources.
foreach (var resource in resources)
{
_tracePropertyKeys.RemoveWhere(k => k.Resource.ResourceKey == resource.ResourceKey);
SetResourceHasTraces(resource, false);
}
}
}
finally
{
_tracesLock.ExitWriteLock();
}
RemoveOrphanedUninstrumentedPeers();
RaiseSubscriptionChanged(_tracesSubscriptions);
}
/// <summary>
/// Removes peer resources that no remaining span references.
/// </summary>
/// <remarks>
/// Uninstrumented peers are synthesised from span attributes rather than reported by a real resource, so
/// clearing traces is the only thing that can retire them. This mirrors the SQLite repository, which deletes
/// the same rows so peers do not accumulate across clears.
/// </remarks>
private void RemoveOrphanedUninstrumentedPeers()
{
List<ResourceKey>? orphanedKeys = null;
_tracesLock.EnterReadLock();
try
{
foreach (var (key, entry) in _resources)
{
if (!entry.Resource.UninstrumentedPeer)
{
continue;
}
var referenced = _traces.Any(trace => trace.Spans.Any(span =>
span.UninstrumentedPeer == entry.Resource || span.Source.Resource == entry.Resource));
if (!referenced)
{
(orphanedKeys ??= []).Add(key);
}
}
}
finally
{
_tracesLock.ExitReadLock();
}
if (orphanedKeys is null)
{
return;
}
foreach (var key in orphanedKeys)
{
_resources.TryRemove(key, out _);
}
RaiseSubscriptionChanged(_resourceSubscriptions);
}
public Task ClearStructuredLogsAsync(ResourceKey? resourceKey = null)
{
ClearStructuredLogs(resourceKey);
return Task.CompletedTask;
}
private void ClearStructuredLogs(ResourceKey? resourceKey)
{
ThrowIfReadOnly();
List<OtlpResource>? resources = null;
if (resourceKey.HasValue)
{
resources = GetResources(resourceKey.Value);
}
_logsLock.EnterWriteLock();
try
{
if (resources is null || resources.Count == 0)
{
// Nothing selected, clear everything.
_logs.Clear();
_logScopes.Clear();
_logPropertyKeys.Clear();
foreach (var resource in _resources.Values)
{
SetResourceHasLogs(resource.Resource, false);
}
_resourceUnviewedErrorLogs.Clear();
}
else
{
for (var i = _logs.Count - 1; i >= 0; i--)
{
if (MatchResources(_logs[i].ResourceView.ResourceKey, resources))
{
_logs.RemoveAt(i);
continue;
}
}
// Update HasLogs flag and remove property keys for cleared resources.
foreach (var resource in resources)
{
_logPropertyKeys.RemoveWhere(k => k.Resource.ResourceKey == resource.ResourceKey);
SetResourceHasLogs(resource, false);
_resourceUnviewedErrorLogs.Remove(resource.ResourceKey);
}
}
}
finally
{
_logsLock.ExitWriteLock();
}
RaiseSubscriptionChanged(_logSubscriptions);
}
private void ClearResource(ResourceKey resourceKey)
{
if (_resources.TryRemove(resourceKey, out _))
{
RaiseSubscriptionChanged(_resourceSubscriptions);
}
}
public Task ClearMetricsAsync(ResourceKey? resourceKey = null)
{
ClearMetrics(resourceKey);
return Task.CompletedTask;
}
private void ClearMetrics(ResourceKey? resourceKey)
{
ThrowIfReadOnly();
List<ResourceEntry> resources;
if (resourceKey.HasValue)
{
resources = GetResourceEntries(resourceKey.Value);
}
else
{
resources = _resources.Values.ToList();
}
foreach (var entry in resources)
{
entry.MetricsLock.EnterWriteLock();
try
{
entry.Instruments.Clear();
entry.Meters.Clear();
}
finally
{
entry.MetricsLock.ExitWriteLock();
}
SetResourceHasMetrics(entry.Resource, false);
}
RaiseSubscriptionChanged(_metricsSubscriptions);
}
public async Task<Dictionary<string, int>> GetTraceFieldValuesAsync(string attributeName, CancellationToken cancellationToken)
{
await ReadGateAsync(nameof(GetTraceFieldValuesAsync)).ConfigureAwait(false);
cancellationToken.ThrowIfCancellationRequested();
_tracesLock.EnterReadLock();
try
{
var fieldValues = OtlpSpan.GetFieldValuesFromTraces(_traces, attributeName);
return fieldValues;
}
finally
{
_tracesLock.ExitReadLock();
}
}
public async Task<Dictionary<string, int>> GetLogsFieldValuesAsync(string attributeName, CancellationToken cancellationToken)
{
await ReadGateAsync(nameof(GetLogsFieldValuesAsync)).ConfigureAwait(false);
cancellationToken.ThrowIfCancellationRequested();
var attributesValues = new Dictionary<string, int>(StringComparers.OtlpAttribute);
if (attributeName == KnownStructuredLogFields.TimestampField)
{
return attributesValues;
}
_logsLock.EnterReadLock();
try
{
foreach (var log in _logs)
{
var value = OtlpLogEntry.GetFieldValue(log, attributeName);
if (value != null)
{
ref var count = ref CollectionsMarshal.GetValueRefOrAddDefault(attributesValues, value, out _);
// Adds to dictionary if not present.
count++;
}
}
}
finally
{
_logsLock.ExitReadLock();
}
return attributesValues;
}
public bool HasUpdatedTrace(OtlpTrace trace)
{
_tracesLock.EnterReadLock();
try
{
var latestTrace = GetTraceUnsynchronized(trace.TraceId);
if (latestTrace == null)
{
// Trace must have been removed. Technically there is an update (nothing).
return true;
}
return latestTrace.LastUpdatedDate > trace.LastUpdatedDate;
}
finally
{
_tracesLock.ExitReadLock();
}
}
public OtlpTrace? GetTrace(string traceId)
{
_tracesLock.EnterReadLock();
try
{
return GetTraceAndCloneUnsynchronized(traceId);
}
finally
{
_tracesLock.ExitReadLock();
}
}
private OtlpTrace? GetTraceUnsynchronized(string traceId)
{
Debug.Assert(_tracesLock.IsReadLockHeld || _tracesLock.IsWriteLockHeld, $"Must get lock before calling {nameof(GetTraceUnsynchronized)}.");
foreach (var trace in _traces)
{
if (OtlpHelpers.MatchTelemetryId(traceId, trace.TraceId))
{
return trace;
}
}
return null;
}
private OtlpTrace? GetTraceAndCloneUnsynchronized(string traceId)
{
Debug.Assert(_tracesLock.IsReadLockHeld || _tracesLock.IsWriteLockHeld, $"Must get lock before calling {nameof(GetTraceAndCloneUnsynchronized)}.");
var trace = GetTraceUnsynchronized(traceId);
return trace != null ? OtlpTrace.Clone(trace) : null;
}
private OtlpSpan? GetSpanAndCloneUnsynchronized(string traceId, string spanId)
{
Debug.Assert(_tracesLock.IsReadLockHeld || _tracesLock.IsWriteLockHeld, $"Must get lock before calling {nameof(GetSpanAndCloneUnsynchronized)}.");
// Trace and its spans are cloned here.
var trace = GetTraceAndCloneUnsynchronized(traceId);
if (trace != null)
{
foreach (var span in trace.Spans)
{
if (span.SpanId == spanId)
{
return span;
}
}
}
return null;
}
public OtlpSpan? GetSpan(string traceId, string spanId)
{
_tracesLock.EnterReadLock();
try
{
return GetSpanAndCloneUnsynchronized(traceId, spanId);
}
finally
{
_tracesLock.ExitReadLock();
}
}
public Task AddMetricsAsync(AddContext context, RepeatedField<ResourceMetrics> resourceMetrics)
{
ThrowIfReadOnly();
if (_pauseManager.AreMetricsPaused(out _))
{
_logger.LogTrace("{Count} incoming metric(s) ignored because of an active pause.", resourceMetrics.Count);
return Task.CompletedTask;
}
foreach (var rm in resourceMetrics)
{
OtlpResourceView resourceView;
ResourceEntry resourceEntry;
try
{
resourceView = GetOrAddResourceView(rm.Resource, out resourceEntry);
}
catch (Exception ex)
{
context.FailureCount += rm.ScopeMetrics.Sum(sm => sm.Metrics.Sum(OtlpHelpers.GetMetricDataPointCount));
_otlpContext.Logger.LogInformation(ex, "Error adding resource.");
continue;
}
AddMetrics(resourceEntry, resourceView, context, rm.ScopeMetrics);
SetResourceHasMetrics(resourceView.Resource, true);
}
RaiseSubscriptionChanged(_metricsSubscriptions);
return Task.CompletedTask;
}
private void AddMetrics(ResourceEntry resourceEntry, OtlpResourceView resourceView, AddContext context, RepeatedField<ScopeMetrics> scopeMetrics)
{
resourceEntry.MetricsLock.EnterWriteLock();
try
{
foreach (var scopeMetric in scopeMetrics)
{
if (!OtlpHelpers.TryGetOrAddScope(resourceEntry.Meters, scopeMetric.Scope, _otlpContext, TelemetryType.Metrics, out var scope))
{
context.FailureCount += scopeMetric.Metrics.Sum(OtlpHelpers.GetMetricDataPointCount);
continue;
}
foreach (var metric in scopeMetric.Metrics)
{
InMemoryInstrument instrument;
try
{
if (string.IsNullOrEmpty(metric.Name))
{
throw new InvalidOperationException("Instrument name is required.");
}
var instrumentKey = new OtlpInstrumentKey(scope.Name, metric.Name);
if (resourceEntry.Instruments.TryGetValue(instrumentKey, out var existingInstrument))
{
instrument = existingInstrument;
}
else if (resourceEntry.Instruments.Count < TelemetryRepositoryLimits.MaxInstrumentCount)
{
instrument = new InMemoryInstrument
{
Summary = new OtlpInstrumentSummary
{
Name = metric.Name,
Description = metric.Description,
Unit = metric.Unit,
Type = MapMetricType(metric.DataCase),
AggregationTemporality = MapAggregationTemporality(metric),
Parent = scope,
ResourceView = resourceView
},
Context = _otlpContext
};
resourceEntry.Instruments.Add(instrumentKey, instrument);
_otlpContext.Logger.LogTrace("Added metric instrument '{InstrumentName}' for scope '{ScopeName}'.", instrument.Summary.Name, scope.Name);
}
else
{
throw new InvalidOperationException($"Instrument limit of {TelemetryRepositoryLimits.MaxInstrumentCount} reached. Instrument '{metric.Name}' will not be added.");
}
}
catch (Exception ex)
{
// If we can't create the instrument then all data points for it are failures.
context.FailureCount += OtlpHelpers.GetMetricDataPointCount(metric);
_otlpContext.Logger.LogInformation(ex, "Error adding metric instrument {MetricName}.", metric.Name);
continue;
}
AddMetrics(instrument, metric, context);
}
}
}
finally
{
resourceEntry.MetricsLock.ExitWriteLock();
}
}
private void AddMetrics(InMemoryInstrument instrument, Metric metric, AddContext context)
{
switch (metric.DataCase)
{
case Metric.DataOneofCase.Gauge:
foreach (var dataPoint in metric.Gauge.DataPoints)
{
try
{
OtlpHelpers.ValidateNumberDataPoint(dataPoint);
instrument.FindScope(dataPoint.Attributes).AddPointValue(dataPoint, _otlpContext);
context.SuccessCount++;
}
catch (Exception ex)
{
context.FailureCount++;
_otlpContext.Logger.LogInformation(ex, "Error adding metric.");
}
}
break;
case Metric.DataOneofCase.Sum:
foreach (var dataPoint in metric.Sum.DataPoints)
{
try
{
OtlpHelpers.ValidateNumberDataPoint(dataPoint);
instrument.FindScope(dataPoint.Attributes).AddPointValue(dataPoint, _otlpContext);
context.SuccessCount++;
}
catch (Exception ex)
{
context.FailureCount++;
_otlpContext.Logger.LogInformation(ex, "Error adding metric.");
}
}
break;
case Metric.DataOneofCase.Histogram:
foreach (var dataPoint in metric.Histogram.DataPoints)
{
try
{
instrument.FindScope(dataPoint.Attributes).AddHistogramValue(dataPoint, _otlpContext);
context.SuccessCount++;
}
catch (Exception ex)
{
context.FailureCount++;
_otlpContext.Logger.LogInformation(ex, "Error adding metric.");
}
}
break;
case Metric.DataOneofCase.Summary:
context.FailureCount += metric.Summary.DataPoints.Count;
_otlpContext.Logger.LogInformation("Error adding summary metrics. Summary is not supported.");
break;
case Metric.DataOneofCase.ExponentialHistogram:
context.FailureCount += metric.ExponentialHistogram.DataPoints.Count;
_otlpContext.Logger.LogInformation("Error adding exponential histogram metrics. Exponential histogram is not supported.");
break;
}
}
private static OtlpInstrumentType MapMetricType(Metric.DataOneofCase data)
{
return data switch
{
Metric.DataOneofCase.Gauge => OtlpInstrumentType.Gauge,
Metric.DataOneofCase.Sum => OtlpInstrumentType.Sum,
Metric.DataOneofCase.Histogram => OtlpInstrumentType.Histogram,
_ => OtlpInstrumentType.Unsupported
};
}
private static OtlpAggregationTemporality MapAggregationTemporality(Metric metric)
{
return metric.DataCase switch
{
Metric.DataOneofCase.Sum => (OtlpAggregationTemporality)metric.Sum.AggregationTemporality,
Metric.DataOneofCase.Histogram => (OtlpAggregationTemporality)metric.Histogram.AggregationTemporality,
Metric.DataOneofCase.ExponentialHistogram => (OtlpAggregationTemporality)metric.ExponentialHistogram.AggregationTemporality,
_ => OtlpAggregationTemporality.Unspecified
};
}
public Task AddTracesAsync(AddContext context, RepeatedField<ResourceSpans> resourceSpans)
{
ThrowIfReadOnly();
if (_pauseManager.AreTracesPaused(out _))
{
_logger.LogTrace("{Count} incoming trace(s) ignored because of an active pause.", resourceSpans.Count);
return Task.CompletedTask;
}
foreach (var rs in resourceSpans)
{
OtlpResourceView resourceView;
try
{
resourceView = GetOrAddResourceView(rs.Resource);
}
catch (Exception ex)
{
context.FailureCount += rs.ScopeSpans.Sum(s => s.Spans.Count);
_otlpContext.Logger.LogInformation(ex, "Error adding resource.");
continue;
}
AddTracesCore(context, resourceView, rs.ScopeSpans);
SetResourceHasTraces(resourceView.Resource, true);
}
RaiseSubscriptionChanged(_tracesSubscriptions);
return Task.CompletedTask;
}
private static OtlpSpanStatusCode ConvertStatus(Status? status)
{
return status?.Code switch
{
Status.Types.StatusCode.Ok => OtlpSpanStatusCode.Ok,
Status.Types.StatusCode.Error => OtlpSpanStatusCode.Error,
Status.Types.StatusCode.Unset => OtlpSpanStatusCode.Unset,
_ => OtlpSpanStatusCode.Unset
};
}
internal static OtlpSpanKind ConvertSpanKind(SpanKind? kind)
{
return OtlpHelpers.ConvertSpanKind(kind);
}
internal void AddTracesCore(AddContext context, OtlpResourceView resourceView, RepeatedField<ScopeSpans> scopeSpans)
{
List<OtlpSpan>? addedSpans = null;
_tracesLock.EnterWriteLock();
try
{
foreach (var scopeSpan in scopeSpans)
{
if (!OtlpHelpers.TryGetOrAddScope(_traceScopes, scopeSpan.Scope, _otlpContext, TelemetryType.Traces, out var scope))
{
context.FailureCount += scopeSpan.Spans.Count;
continue;
}
var updatedTraces = new Dictionary<ReadOnlyMemory<byte>, OtlpTrace>();
foreach (var span in scopeSpan.Spans)
{
try
{
OtlpTrace? trace;
var newTrace = false;
// Fast path to check if the span is in a trace that's been updated this add call.
if (!updatedTraces.TryGetValue(span.TraceId.Memory, out trace))
{
if (!TryGetTraceById(_traces, span.TraceId.Memory, out trace))
{
trace = new OtlpTrace(span.TraceId.Memory, DateTime.UtcNow);
newTrace = true;
}
}
var newSpan = CreateSpan(resourceView, span, trace, scope, _otlpContext);
trace.AddSpan(newSpan);
// The new span might be linked to by an existing span.
// Check current links to see if a backlink should be created.
foreach (var existingLink in _spanLinks)
{
if (existingLink.SpanId == newSpan.SpanId && existingLink.TraceId == newSpan.TraceId)
{
newSpan.BackLinks.Add(existingLink);
}
}
// Add links to central collection. Add backlinks to existing spans.
foreach (var link in newSpan.Links)
{
_spanLinks.Add(link);
var linkedSpan = GetSpanAndCloneUnsynchronized(link.TraceId, link.SpanId);
linkedSpan?.BackLinks.Add(link);
}
// Traces are sorted by the start time of the first span, then by trace ID.
// We need to ensure traces are in the correct order if we're:
// 1. Adding a new trace.
// 2. The first span of the trace has changed.
if (newTrace)
{
var added = false;
for (var i = _traces.Count - 1; i >= 0; i--)
{
var currentTrace = _traces[i];
if (CompareTraceOrder(trace, currentTrace) > 0)
{
_traces.Insert(i + 1, trace);
added = true;
break;
}
}
if (!added)
{
_traces.Insert(0, trace);
}
}
else
{
if (trace.FirstSpan == newSpan)
{
var moved = false;
var index = _traces.IndexOf(trace);
for (var i = index - 1; i >= 0; i--)
{
var currentTrace = _traces[i];
if (CompareTraceOrder(trace, currentTrace) > 0)
{
var insertPosition = i + 1;
if (index != insertPosition)
{
_traces.RemoveAt(index);
_traces.Insert(insertPosition, trace);
}
moved = true;
break;
}
}
if (!moved)
{
if (index != 0)
{
_traces.RemoveAt(index);
_traces.Insert(0, trace);
}
}
}
}
foreach (var kvp in newSpan.Attributes)
{
_tracePropertyKeys.Add((resourceView.Resource, kvp.Key));
}
// Newly added or updated trace should always been in the collection.
Debug.Assert(_traces.Contains(trace), "Trace not found in traces collection.");
updatedTraces[trace.Key] = trace;
// Collect span for push-based streaming (lazy init to avoid allocation when no watchers)
addedSpans ??= new List<OtlpSpan>();
addedSpans.Add(newSpan);
context.SuccessCount++;
}
catch (Exception ex)
{
context.FailureCount++;
_otlpContext.Logger.LogInformation(ex, "Error adding span.");
}
AssertTraceOrder();
AssertSpanLinks();
}
// After spans are updated, loop through traces and their spans and update uninstrumented peer values.
// These can change
foreach (var (_, updatedTrace) in updatedTraces)
{
CalculateTraceUninstrumentedPeers(updatedTrace);
}
}
}
finally
{
_tracesLock.ExitWriteLock();
}
// Push spans to watchers outside the lock
if (addedSpans is not null)
{
PushSpansToWatchers(addedSpans, resourceView.ResourceKey);
}
static bool TryGetTraceById(CircularBuffer<OtlpTrace> traces, ReadOnlyMemory<byte> traceId, [NotNullWhen(true)] out OtlpTrace? trace)
{
var s = traceId.Span;
for (var i = traces.Count - 1; i >= 0; i--)
{
if (traces[i].Key.Span.SequenceEqual(s))
{
trace = traces[i];
return true;
}
}
trace = null;
return false;
}
}
private static int CompareTraceOrder(OtlpTrace left, OtlpTrace right)
{
var timestampComparison = left.FirstSpan.StartTime.CompareTo(right.FirstSpan.StartTime);
return timestampComparison != 0 ? timestampComparison : string.CompareOrdinal(left.TraceId, right.TraceId);
}
public OtlpResource? GetPeerResource(OtlpSpan span)
{
return span.UninstrumentedPeer;
}
private void CalculateTraceUninstrumentedPeers(OtlpTrace trace)
{
foreach (var span in trace.Spans)
{
// A span may indicate a call to another service but the service isn't instrumented.
var hasPeerService = OtlpHelpers.GetPeerAddress(span.Attributes) != null;
var hasUninstrumentedPeer = hasPeerService && span.Kind is OtlpSpanKind.Client or OtlpSpanKind.Producer && !span.GetChildSpans().Any();
var uninstrumentedPeerKey = hasUninstrumentedPeer ? ResolveUninstrumentedPeerResourceKey(span, _outgoingPeerResolvers) : null;
if (uninstrumentedPeerKey is { } peerKey)
{
if (span.UninstrumentedPeer?.ResourceKey == peerKey)
{
// Already the correct value. No changes needed.
continue;
}
try
{
var (resource, _) = GetOrAddResourceEntry(peerKey, uninstrumentedPeer: true);
trace.SetSpanUninstrumentedPeer(span, resource.Resource);
}
catch (Exception ex)
{
_logger.LogInformation(ex, "Error adding uninstrumented peer resource.");
}
}
else
{
trace.SetSpanUninstrumentedPeer(span, null);
}
}
}
private static ResourceKey? ResolveUninstrumentedPeerResourceKey(OtlpSpan span, IEnumerable<IOutgoingPeerResolver> outgoingPeerResolvers)
{
foreach (var resolver in outgoingPeerResolvers)
{
if (!resolver.TryResolvePeer(span.Attributes, out var name, out var matchedResource))
{
continue;
}
if (matchedResource is not null)
{
return ResourceKey.Create(matchedResource.DisplayName, matchedResource.Name);
}
if (!string.IsNullOrEmpty(name))
{
return new ResourceKey(name, InstanceId: null);
}
}
return null;
}
[Conditional("DEBUG")]
private void AssertTraceOrder()
{
DateTime current = default;
for (var i = 0; i < _traces.Count; i++)
{
var trace = _traces[i];
if (trace.FirstSpan.StartTime < current)
{
throw new InvalidOperationException($"Traces not in order at index {i}.");
}
current = trace.FirstSpan.StartTime;
}
}
[Conditional("DEBUG")]
private void AssertSpanLinks()
{
// Create a local copy of span links.
var currentSpanLinks = _spanLinks.ToList();
// Remove span links that match span links on spans.
// Throw an error if an expected span link doesn't exist.
foreach (var trace in _traces)
{
foreach (var span in trace.Spans)
{
foreach (var link in span.Links)
{
if (!currentSpanLinks.Remove(link))
{
throw new InvalidOperationException($"Couldn't find expected link from span {span.SpanId} to span {link.SpanId}.");
}
}
}
}
// Throw error if there are orphaned span links.
if (currentSpanLinks.Count > 0)
{
var sb = new StringBuilder();
sb.AppendLine(CultureInfo.InvariantCulture, $"There are {currentSpanLinks.Count} orphaned span links.");
foreach (var link in currentSpanLinks)
{
sb.AppendLine(CultureInfo.InvariantCulture, $"\tSource span ID: {link.SourceSpanId}, Target span ID: {link.SpanId}");
}
throw new InvalidOperationException(sb.ToString());
}
}
private static OtlpSpan CreateSpan(OtlpResourceView resourceView, Span span, OtlpTrace trace, OtlpScope scope, OtlpContext context)
{
var id = span.SpanId?.ToHexString();
if (id is null)
{
throw new ArgumentException("Span has no SpanId");
}
var events = new List<OtlpSpanEvent>();
var links = new List<OtlpSpanLink>();
foreach (var e in span.Links)
{
links.Add(new OtlpSpanLink
{
SourceSpanId = id,
SourceTraceId = trace.TraceId,
TraceState = e.TraceState,
SpanId = e.SpanId.ToHexString(),
TraceId = e.TraceId.ToHexString(),
Attributes = e.Attributes.ToKeyValuePairs(context)
});
}
var newSpan = new OtlpSpan(resourceView, trace, scope)
{
SpanId = id,
ParentSpanId = span.ParentSpanId?.ToHexString(),
Name = span.Name,
Kind = ConvertSpanKind(span.Kind),
StartTime = OtlpHelpers.UnixNanoSecondsToDateTime(span.StartTimeUnixNano),
EndTime = OtlpHelpers.UnixNanoSecondsToDateTime(span.EndTimeUnixNano),
Status = ConvertStatus(span.Status),
StatusMessage = span.Status?.Message,
Attributes = span.Attributes.ToKeyValuePairs(context, filter: attribute => attribute.Key != OtlpHelpers.AspireDestinationNameAttribute),
State = !string.IsNullOrEmpty(span.TraceState) ? span.TraceState : null,
Events = events,
Links = links,
BackLinks = []
};
foreach (var e in span.Events.OrderBy(e => e.TimeUnixNano))
{
events.Add(new OtlpSpanEvent(newSpan)
{
InternalId = Guid.NewGuid(),
Name = e.Name,
Time = OtlpHelpers.UnixNanoSecondsToDateTime(e.TimeUnixNano),
Attributes = e.Attributes.ToKeyValuePairs(context)
});
if (events.Count >= context.Options.MaxSpanEventCount)
{
break;
}
}
return newSpan;
}
public List<OtlpInstrumentSummary> GetInstrumentSummaries(ResourceKey key)
{
var resources = GetResourceEntries(key);
var summaries = new List<OtlpInstrumentSummary>();
foreach (var resource in resources)
{
resource.MetricsLock.EnterReadLock();
try
{
summaries.AddRange(resource.Instruments.Values.Select(instrument => instrument.Summary));
}
finally
{
resource.MetricsLock.ExitReadLock();
}
}
return resources.Count > 1 ? summaries.DistinctBy(summary => summary.GetKey()).ToList() : summaries;
}
public OtlpInstrumentSummary? GetInstrumentSummary(ResourceKey resourceKey, string meterName, string instrumentName)
{
var instrumentKey = new OtlpInstrumentKey(meterName, instrumentName);
foreach (var resource in GetResourceEntries(resourceKey))
{
resource.MetricsLock.EnterReadLock();
try
{
if (resource.Instruments.TryGetValue(instrumentKey, out var instrument))
{
return instrument.Summary;
}
}
finally
{
resource.MetricsLock.ExitReadLock();
}
}
return null;
}
public async Task<OtlpInstrumentData?> GetInstrumentAsync(GetInstrumentRequest request, CancellationToken cancellationToken)
{
await ReadGateAsync(nameof(GetInstrumentAsync)).ConfigureAwait(false);
cancellationToken.ThrowIfCancellationRequested();
return GetInstrument(request);
}
private OtlpInstrumentData? GetInstrument(GetInstrumentRequest request)
{
var resources = GetResourceEntries(request.ResourceKey);
var instrumentKey = new OtlpInstrumentKey(request.MeterName, request.InstrumentName);
var instruments = resources
.Select(resource => CloneInstrument(resource, instrumentKey, request.StartTime, request.EndTime))
.OfType<InMemoryInstrument>()
.ToList();
if (instruments.Count == 0)
{
return null;
}
var allKnownAttributes = new Dictionary<string, List<string?>>();
var matchingDimensions = new List<DimensionScope>();
var hasOverflow = false;
foreach (var instrument in instruments)
{
foreach (var knownAttributeValues in instrument.KnownAttributeValues)
{
ref var values = ref CollectionsMarshal.GetValueRefOrAddDefault(allKnownAttributes, knownAttributeValues.Key, out _);
values = values is not null
? values.Union(knownAttributeValues.Value).ToList()
: knownAttributeValues.Value.ToList();
}
matchingDimensions.AddRange(instrument.Dimensions.Values.Where(dimension => MatchesDimensionFilters(dimension.Attributes, request.DimensionFilters)));
hasOverflow = hasOverflow || instrument.HasOverflow;
}
return new OtlpInstrumentData
{
Summary = instruments[0].Summary,
Dimensions = matchingDimensions,
KnownAttributeValues = allKnownAttributes,
HasOverflow = hasOverflow
};
}
private static InMemoryInstrument? CloneInstrument(ResourceEntry resource, OtlpInstrumentKey key, DateTime? valuesStart, DateTime? valuesEnd)
{
resource.MetricsLock.EnterReadLock();
try
{
return resource.Instruments.TryGetValue(key, out var instrument)
? InMemoryInstrument.Clone(instrument, valuesStart, valuesEnd)
: null;
}
finally
{
resource.MetricsLock.ExitReadLock();
}
}
private static bool MatchesDimensionFilters(
KeyValuePair<string, string>[] attributes,
IReadOnlyDictionary<string, IReadOnlyList<string?>> dimensionFilters)
{
foreach (var (key, values) in dimensionFilters)
{
if (!values.Contains(OtlpHelpers.GetValue(attributes, key)))
{
return false;
}
}
return true;
}
public DateTime? GetInstrumentLatestEndTime(ResourceKey resourceKey, string meterName, string instrumentName)
{
var instrument = GetInstrument(new GetInstrumentRequest
{
ResourceKey = resourceKey,
MeterName = meterName,
InstrumentName = instrumentName,
StartTime = DateTime.MinValue,
EndTime = DateTime.MaxValue
});
return instrument?.Dimensions
.SelectMany(dimension => dimension.Values)
.Select(value => (DateTime?)value.End)
.Max();
}
private Task OnPeerChanged()
{
_tracesLock.EnterWriteLock();
try
{
// When peers change then we need to recalculate the uninstrumented peers of spans.
foreach (var trace in _traces)
{
try
{
CalculateTraceUninstrumentedPeers(trace);
}
catch (Exception ex)
{
_logger.LogInformation(ex, "Error recalculating uninstrumented peers.");
}
}
}
finally
{
_tracesLock.ExitWriteLock();
}
return Task.CompletedTask;
}
public void Dispose()
{
foreach (var subscription in _peerResolverSubscriptions)
{
subscription.Dispose();
}
DisposeWatchers();
}
[DebuggerDisplay("Name = {Summary.Name}, Unit = {Summary.Unit}, Type = {Summary.Type}")]
private sealed class InMemoryInstrument
{
public required OtlpInstrumentSummary Summary { get; init; }
public required OtlpContext Context { get; init; }
public Dictionary<ReadOnlyMemory<KeyValuePair<string, string>>, DimensionScope> Dimensions { get; } = new(ScopeAttributesComparer.Instance);
private KnownAttributeValuesState IncomingKnownAttributeValues { get; } = new();
public Dictionary<string, List<string?>> KnownAttributeValues { get; } = [];
public bool HasOverflow { get; set; }
public DimensionScope FindScope(RepeatedField<KeyValue> attributes)
{
// See https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/sdk.md#overflow-attribute
// Inspect attributes before they're merged with parent attributes. "otel.metric.overflow" should be the only attribute.
if (!HasOverflow && attributes.Count == 1 && attributes[0].Key == "otel.metric.overflow" && attributes[0].Value.GetString() == "true")
{
HasOverflow = true;
}
var pointAttributes = attributes.ToKeyValuePairs(Context);
Array.Sort(pointAttributes, KeyValuePairComparer.Instance);
KeyValuePair<string, string>[] mergedAttributes = [.. pointAttributes, .. Summary.Parent.Attributes];
var comparableAttributes = mergedAttributes.AsMemory();
// Can't use CollectionsMarshal.GetValueRefOrAddDefault here because comparableAttributes is a view over mutable data.
// Need to add dimensions using durable attributes instance after scope is created.
if (!Dimensions.TryGetValue(comparableAttributes, out var dimension))
{
IncomingKnownAttributeValues.ValidateDimension(pointAttributes);
if (Dimensions.Count >= TelemetryRepositoryLimits.MaxDimensionCount)
{
throw new InvalidOperationException($"Dimension limit of {TelemetryRepositoryLimits.MaxDimensionCount} reached for instrument '{Summary.Name}'.");
}
IncomingKnownAttributeValues.AddDimension(pointAttributes);
dimension = CreateDimensionScope(comparableAttributes);
Dimensions.Add(dimension.Attributes, dimension);
}
return dimension;
}
private DimensionScope CreateDimensionScope(Memory<KeyValuePair<string, string>> comparableAttributes)
{
var isFirst = Dimensions.Count == 0;
var durableAttributes = comparableAttributes.ToArray();
var dimension = new DimensionScope(Context.Options.MaxMetricsCount, durableAttributes);
// Point and scope attributes were already accepted during ingestion, so intentionally do not limit their
// merged key or per-key value counts while building display metadata.
var keys = KnownAttributeValues.Keys.Union(durableAttributes.Select(attribute => attribute.Key)).Distinct();
foreach (var key in keys)
{
if (!KnownAttributeValues.TryGetValue(key, out var values))
{
values = [];
KnownAttributeValues.Add(key, values);
// If the key is new and there are already dimensions, add an empty value because there are dimensions without this key.
if (!isFirst)
{
TryAddValue(values, null);
}
}
var currentDimensionValue = OtlpHelpers.GetValue(durableAttributes, key);
TryAddValue(values, currentDimensionValue);
}
return dimension;
static void TryAddValue(List<string?> values, string? value)
{
if (!values.Contains(value))
{
values.Add(value);
}
}
}
public static InMemoryInstrument Clone(InMemoryInstrument instrument, DateTime? valuesStart, DateTime? valuesEnd)
{
var newInstrument = new InMemoryInstrument
{
Summary = instrument.Summary,
Context = instrument.Context,
HasOverflow = instrument.HasOverflow
};
foreach (var item in instrument.KnownAttributeValues)
{
newInstrument.KnownAttributeValues.Add(item.Key, item.Value.ToList());
}
foreach (var item in instrument.Dimensions)
{
newInstrument.Dimensions.Add(item.Key, DimensionScope.Clone(item.Value, valuesStart, valuesEnd));
}
return newInstrument;
}
private sealed class ScopeAttributesComparer : IEqualityComparer<ReadOnlyMemory<KeyValuePair<string, string>>>
{
public static readonly ScopeAttributesComparer Instance = new();
public bool Equals(ReadOnlyMemory<KeyValuePair<string, string>> x, ReadOnlyMemory<KeyValuePair<string, string>> y) =>
x.Span.SequenceEqual(y.Span);
public int GetHashCode([DisallowNull] ReadOnlyMemory<KeyValuePair<string, string>> obj)
{
var hashcode = new HashCode();
foreach (var pair in obj.Span)
{
hashcode.Add(pair.Key);
hashcode.Add(pair.Value);
}
return hashcode.ToHashCode();
}
}
private sealed class KeyValuePairComparer : IComparer<KeyValuePair<string, string>>
{
public static readonly KeyValuePairComparer Instance = new();
public int Compare(KeyValuePair<string, string> x, KeyValuePair<string, string> y) =>
string.Compare(x.Key, y.Key, StringComparison.Ordinal);
}
}
private sealed record ResourceEntry(OtlpResource Resource)
{
public ReaderWriterLockSlim MetricsLock { get; } = new();
// Bounded by TelemetryRepositoryLimits.MaxScopeCount. Cleared when metrics are cleared.
public Dictionary<string, OtlpScope> Meters { get; } = [];
// Bounded by TelemetryRepositoryLimits.MaxInstrumentCount. Cleared when metrics are cleared.
public Dictionary<OtlpInstrumentKey, InMemoryInstrument> Instruments { get; } = [];
}
}