// Licensed to the .NET Foundation under one or more agreements.
// The .NET Foundation licenses this file to you under the MIT license.
using System.Data;
using System.Globalization;
using System.Text;
using Aspire.Dashboard.Model;
using Aspire.Dashboard.Model.Otlp;
using Aspire.Dashboard.Otlp.Model;
using Dapper;
using Microsoft.Data.Sqlite;
namespace Aspire.Dashboard.Otlp.Storage;
public sealed partial class SqliteTelemetryRepository
{
private GetTracesResponse GetTracesFromDatabase(GetTracesRequest context, CancellationToken cancellationToken)
{
using var connection = _database.OpenConnection();
using var interrupt = connection.RegisterInterrupt(cancellationToken);
var query = BuildTraceQuery(context);
var aggregate = connection.QuerySingle<TraceAggregateRecord>($"""
SELECT COUNT(*) AS TotalItemCount, COALESCE(MAX(t.duration_ticks), 0) AS MaxDurationTicks
{query.FromAndWhere};
""", query.Parameters);
var hiddenItemCount = context.LatestItemCount is { } latestItemCount
? Math.Max(aggregate.TotalItemCount - Math.Max(latestItemCount, 0), 0)
: 0;
query.Parameters.Add("StartIndex", (long)Math.Max(context.StartIndex, 0) + hiddenItemCount);
query.Parameters.Add("Count", Math.Max(context.Count, 0));
var records = connection.Query<TraceSummaryRecord>($"""
SELECT
t.trace_id AS TraceId,
t.last_updated_timestamp_ticks AS LastUpdatedTimestampTicks
{query.FromAndWhere}
ORDER BY t.first_span_timestamp_ticks, t.trace_id
LIMIT @Count OFFSET @StartIndex;
""", query.Parameters).AsList();
var traces = new List<OtlpTrace>(records.Count);
foreach (var batch in records.Chunk(MaxTraceBatchSize))
{
var tracesById = MaterializeTraces(connection, batch.Select(record => record.TraceId).ToArray());
traces.AddRange(batch.Select(record => tracesById[record.TraceId]));
}
return new GetTracesResponse
{
PagedResult = new PagedResult<OtlpTrace>
{
Items = traces,
TotalItemCount = aggregate.TotalItemCount,
IsFull = connection.QuerySingle<int>("SELECT COUNT(*) FROM telemetry_traces;") >= _otlpContext.Options.MaxTraceCount
},
MaxDuration = TimeSpan.FromTicks(aggregate.MaxDurationTicks)
};
}
private GetTraceSummariesResponse GetTraceSummariesFromDatabase(GetTracesRequest context, CancellationToken cancellationToken)
{
using var connection = _database.OpenConnection();
using var interrupt = connection.RegisterInterrupt(cancellationToken);
var query = BuildTraceQuery(context);
query.Parameters.Add("StartIndex", Math.Max(context.StartIndex, 0));
query.Parameters.Add("Count", Math.Max(context.Count, 0));
query.Parameters.Add("LatestItemCount", Math.Max(context.LatestItemCount ?? int.MaxValue, 0));
query.Parameters.Add("MaxTraceCount", _otlpContext.Options.MaxTraceCount);
// Build the page and its pre-aggregated resource groups in one query.
var records = connection.Query<TracePageSummaryRecord>($"""
WITH
filtered_traces AS (
SELECT t.*
{query.FromAndWhere}
),
trace_aggregate AS (
SELECT COUNT(*) AS TotalItemCount, COALESCE(MAX(duration_ticks), 0) AS MaxDurationTicks
FROM filtered_traces
),
paged_traces AS (
SELECT *
FROM filtered_traces
ORDER BY first_span_timestamp_ticks, trace_id
LIMIT @Count OFFSET @StartIndex + MAX((SELECT TotalItemCount FROM trace_aggregate) - @LatestItemCount, 0)
),
resource_summaries AS (
SELECT
tr.trace_id,
r.resource_name,
r.instance_id,
r.uninstrumented_peer,
tr.resource_order_ticks,
tr.total_spans,
tr.errored_spans
FROM telemetry_trace_resources tr
JOIN paged_traces pt ON pt.trace_id = tr.trace_id
JOIN telemetry_resources r ON r.resource_id = tr.resource_id
WHERE tr.total_spans > 0
),
trace_summaries AS (
SELECT
pt.trace_id,
pt.full_name,
pt.first_span_timestamp_ticks,
pt.duration_ticks,
r.resource_name AS root_resource_name,
r.instance_id AS root_instance_id,
r.uninstrumented_peer AS root_uninstrumented_peer,
pt.has_error,
pt.has_gen_ai,
pt.first_span_timestamp_ticks AS trace_order_ticks
FROM paged_traces pt
JOIN telemetry_spans s ON s.trace_id = pt.trace_id AND s.span_id = pt.primary_span_id
JOIN telemetry_resources r ON r.resource_id = s.resource_id
)
SELECT
a.TotalItemCount,
a.MaxDurationTicks,
(SELECT COUNT(*) FROM telemetry_traces) >= @MaxTraceCount AS IsFull,
ts.trace_id AS TraceId,
ts.full_name AS FullName,
ts.first_span_timestamp_ticks AS StartTimeTicks,
ts.duration_ticks AS DurationTicks,
ts.root_resource_name AS RootResourceName,
ts.root_instance_id AS RootInstanceId,
ts.root_uninstrumented_peer AS RootUninstrumentedPeer,
ts.has_error AS HasError,
ts.has_gen_ai AS HasGenAI,
rs.resource_name AS ResourceName,
rs.instance_id AS InstanceId,
rs.uninstrumented_peer AS UninstrumentedPeer,
rs.total_spans AS TotalSpans,
rs.errored_spans AS ErroredSpans
FROM trace_aggregate a
LEFT JOIN trace_summaries ts ON 1 = 1
LEFT JOIN resource_summaries rs ON rs.trace_id = ts.trace_id
ORDER BY ts.trace_order_ticks, ts.trace_id, rs.resource_order_ticks DESC, rs.uninstrumented_peer, rs.resource_name, rs.instance_id;
""", query.Parameters).AsList();
var firstRecord = records[0];
var summaries = records
.Where(record => record.TraceId is not null)
.GroupBy(record => record.TraceId!, StringComparer.Ordinal)
.Select(group =>
{
var trace = group.First();
return new TraceSummary
{
TraceId = trace.TraceId!,
FullName = trace.FullName!,
StartTime = new DateTime(trace.StartTimeTicks!.Value, DateTimeKind.Utc),
Duration = TimeSpan.FromTicks(trace.DurationTicks!.Value),
RootResource = CreateSummaryResource(trace.RootResourceName!, trace.RootInstanceId, trace.RootUninstrumentedPeer!.Value),
Resources = group.Select(resource => new TraceResourceSummary
{
Resource = CreateSummaryResource(resource.ResourceName!, resource.InstanceId, resource.UninstrumentedPeer!.Value),
TotalSpans = resource.TotalSpans!.Value,
ErroredSpans = resource.ErroredSpans!.Value
}).ToList(),
HasError = trace.HasError!.Value,
HasGenAI = trace.HasGenAI!.Value
};
}).ToList();
return new GetTraceSummariesResponse
{
PagedResult = new PagedResult<TraceSummary>
{
Items = summaries,
TotalItemCount = firstRecord.TotalItemCount,
IsFull = firstRecord.IsFull
},
MaxDuration = TimeSpan.FromTicks(firstRecord.MaxDurationTicks)
};
OtlpResource CreateSummaryResource(string resourceName, string? instanceId, bool uninstrumentedPeer) =>
new(resourceName, instanceId, uninstrumentedPeer, _otlpContext);
}
private static TraceQuery BuildTraceQuery(GetTracesRequest context)
{
var sql = new StringBuilder("FROM telemetry_traces t WHERE 1 = 1");
var parameters = new DynamicParameters();
if (context.ResourceKeys.Count > 0)
{
var resourcePredicates = new List<string>(context.ResourceKeys.Count);
for (var index = 0; index < context.ResourceKeys.Count; index++)
{
var key = context.ResourceKeys[index];
parameters.Add($"ResourceName{index}", key.Name);
var resourcePredicate = $"r.resource_name = @ResourceName{index} COLLATE NOCASE";
if (key.InstanceId is not null)
{
parameters.Add($"InstanceId{index}", key.InstanceId);
resourcePredicate += $" AND r.instance_id = @InstanceId{index} COLLATE NOCASE";
}
resourcePredicates.Add($"({resourcePredicate})");
}
sql.Append(" AND EXISTS (SELECT 1 FROM telemetry_trace_resources tr JOIN telemetry_resources r ON r.resource_id = tr.resource_id WHERE tr.trace_id = t.trace_id AND tr.total_spans > 0 AND (");
sql.AppendJoin(" OR ", resourcePredicates);
sql.Append("))");
}
if (!string.IsNullOrWhiteSpace(context.TraceNameFilterText))
{
sql.Append(" AND t.full_name LIKE @TraceNameFilterText ESCAPE '!'");
parameters.Add("TraceNameFilterText", CreateContainsLikePattern(context.TraceNameFilterText));
}
var positivePredicates = new List<string>();
var filterIndex = 0;
foreach (var filter in context.Filters.Where(filter => filter.Enabled))
{
if (filter is SpanHasAttributeTelemetryFilter or SpanScopePrefixTelemetryFilter or SpanNoMatchTelemetryFilter)
{
positivePredicates.Add(BuildSpanTypePredicate(filter, parameters, ref filterIndex));
continue;
}
if (filter is not FieldTelemetryFilter fieldFilter)
{
continue;
}
if (fieldFilter.Field == KnownTraceFields.DurationField)
{
sql.Append(" AND ");
sql.Append(BuildTraceDurationPredicate(fieldFilter, parameters, filterIndex++));
continue;
}
if (fieldFilter.Condition is FilterCondition.NotEqual or FilterCondition.NotContains)
{
sql.Append(" AND NOT EXISTS (SELECT 1 FROM telemetry_spans s JOIN telemetry_resources r ON r.resource_id = s.resource_id JOIN telemetry_scopes sc ON sc.scope_id = s.scope_id JOIN telemetry_span_kinds sk ON sk.kind = s.kind JOIN telemetry_span_statuses ss ON ss.status = s.status LEFT JOIN telemetry_resources pr ON pr.resource_id = s.uninstrumented_peer_resource_id WHERE s.trace_id = t.trace_id AND ");
sql.Append(BuildSpanFieldPredicate(fieldFilter, parameters, filterIndex++, invertNegative: true));
sql.Append(')');
}
else
{
positivePredicates.Add(BuildSpanFieldPredicate(fieldFilter, parameters, filterIndex++, invertNegative: false));
}
}
if (positivePredicates.Count > 0)
{
sql.Append(" AND EXISTS (SELECT 1 FROM telemetry_spans s JOIN telemetry_resources r ON r.resource_id = s.resource_id JOIN telemetry_scopes sc ON sc.scope_id = s.scope_id JOIN telemetry_span_kinds sk ON sk.kind = s.kind JOIN telemetry_span_statuses ss ON ss.status = s.status LEFT JOIN telemetry_resources pr ON pr.resource_id = s.uninstrumented_peer_resource_id WHERE s.trace_id = t.trace_id AND ");
sql.AppendJoin(" AND ", positivePredicates);
sql.Append(')');
}
if (context.TextFragments is { Length: > 0 })
{
var fullNamePredicates = new List<string>(context.TextFragments.Length);
var spanPredicates = new List<string>(context.TextFragments.Length);
for (var index = 0; index < context.TextFragments.Length; index++)
{
var parameterName = $"TextFragment{index}";
parameters.Add(parameterName, CreateContainsLikePattern(context.TextFragments[index]));
fullNamePredicates.Add($"t.full_name LIKE @{parameterName} ESCAPE '!'");
spanPredicates.Add($"""
(
s.name LIKE @{parameterName} ESCAPE '!' OR
s.span_id LIKE @{parameterName} ESCAPE '!' OR
s.trace_id LIKE @{parameterName} ESCAPE '!' OR
sc.scope_name LIKE @{parameterName} ESCAPE '!' OR
r.resource_name LIKE @{parameterName} ESCAPE '!' OR
ss.status_name LIKE @{parameterName} ESCAPE '!' OR
sk.kind_name LIKE @{parameterName} ESCAPE '!' OR
COALESCE(s.status_message, '') LIKE @{parameterName} ESCAPE '!' OR
EXISTS (SELECT 1 FROM telemetry_span_attributes a WHERE a.trace_id = s.trace_id AND a.span_id = s.span_id AND (a.attribute_key LIKE @{parameterName} ESCAPE '!' OR a.attribute_value LIKE @{parameterName} ESCAPE '!')) OR
EXISTS (SELECT 1 FROM telemetry_span_events e WHERE e.trace_id = s.trace_id AND e.span_id = s.span_id AND e.event_name LIKE @{parameterName} ESCAPE '!')
)
""");
}
sql.Append(" AND ((");
sql.AppendJoin(" AND ", fullNamePredicates);
sql.Append(") OR EXISTS (SELECT 1 FROM telemetry_spans s JOIN telemetry_resources r ON r.resource_id = s.resource_id JOIN telemetry_scopes sc ON sc.scope_id = s.scope_id JOIN telemetry_span_kinds sk ON sk.kind = s.kind JOIN telemetry_span_statuses ss ON ss.status = s.status WHERE s.trace_id = t.trace_id AND ");
sql.AppendJoin(" AND ", spanPredicates);
sql.Append("))");
}
return new TraceQuery(sql.ToString(), parameters);
}
private static string BuildSpanTypePredicate(TelemetryFilter filter, DynamicParameters parameters, ref int filterIndex)
{
switch (filter)
{
case SpanHasAttributeTelemetryFilter attributeFilter:
var attributePredicates = new List<string>(attributeFilter.AttributeNames.Count);
foreach (var attributeName in attributeFilter.AttributeNames)
{
var parameterName = $"SpanTypeAttribute{filterIndex++}";
parameters.Add(parameterName, attributeName);
attributePredicates.Add($"a.attribute_key = @{parameterName}");
}
return $"EXISTS (SELECT 1 FROM telemetry_span_attributes a WHERE a.trace_id = s.trace_id AND a.span_id = s.span_id AND LENGTH(a.attribute_value) > 0 AND ({string.Join(" OR ", attributePredicates)}))";
case SpanScopePrefixTelemetryFilter scopeFilter:
var scopePredicates = new List<string>(scopeFilter.ScopePrefixes.Count);
foreach (var scopePrefix in scopeFilter.ScopePrefixes)
{
var parameterName = $"SpanTypeScope{filterIndex++}";
parameters.Add(parameterName, scopePrefix);
parameters.Add($"{parameterName}Prefix", $"{EscapeLikePattern(scopePrefix)}.%");
scopePredicates.Add($"(sc.scope_name = @{parameterName} COLLATE NOCASE OR sc.scope_name LIKE @{parameterName}Prefix ESCAPE '!')");
}
return $"({string.Join(" OR ", scopePredicates)})";
case SpanNoMatchTelemetryFilter noMatchFilter:
var matchPredicates = new List<string>(noMatchFilter.Filters.Count);
foreach (var nestedFilter in noMatchFilter.Filters)
{
matchPredicates.Add(BuildSpanTypePredicate(nestedFilter, parameters, ref filterIndex));
}
return $"NOT ({string.Join(" OR ", matchPredicates)})";
default:
throw new InvalidOperationException($"Unsupported span type filter: {filter.GetType().FullName}");
}
}
private static string BuildTraceDurationPredicate(FieldTelemetryFilter filter, DynamicParameters parameters, int filterIndex)
{
var parameterName = $"TraceDuration{filterIndex}";
if (!double.TryParse(filter.Value, NumberStyles.Float, CultureInfo.InvariantCulture, out var milliseconds) || !double.IsFinite(milliseconds))
{
return "0 = 1";
}
parameters.Add(parameterName, milliseconds);
return BuildNumericPredicate($"(CAST(t.duration_ticks AS REAL) / {TimeSpan.TicksPerMillisecond})", filter.Condition, parameterName);
}
private static string BuildSpanFieldPredicate(
FieldTelemetryFilter filter,
DynamicParameters parameters,
int filterIndex,
bool invertNegative)
{
var parameterName = $"TraceFilter{filterIndex}";
var condition = invertNegative
? filter.Condition switch
{
FilterCondition.NotEqual => FilterCondition.Equals,
FilterCondition.NotContains => FilterCondition.Contains,
_ => filter.Condition
}
: filter.Condition;
parameters.Add(
parameterName,
condition is FilterCondition.Contains or FilterCondition.NotContains
? CreateContainsLikePattern(filter.Value)
: filter.Value);
var expression = filter.Field switch
{
KnownResourceFields.ServiceNameField => null,
KnownTraceFields.TraceIdField => "s.trace_id",
KnownTraceFields.SpanIdField => "s.span_id",
KnownTraceFields.NameField => "s.name",
KnownTraceFields.KindField => "sk.kind_name",
KnownTraceFields.StatusField => "ss.status_name",
KnownSourceFields.NameField => "sc.scope_name",
KnownTraceFields.TimestampField => "s.start_time_ticks / 10000",
_ => null
};
if (filter.Field == KnownTraceFields.TimestampField)
{
if (!DateTime.TryParse(filter.Value, CultureInfo.InvariantCulture, DateTimeStyles.AdjustToUniversal | DateTimeStyles.AssumeLocal, out var date))
{
return "0 = 1";
}
parameters.Add(parameterName, date.ToUniversalTime().Ticks / TimeSpan.TicksPerMillisecond);
return BuildNumericPredicate(expression!, condition, parameterName);
}
if (expression is not null)
{
return BuildStringPredicate(expression, condition, parameterName);
}
if (filter.Field == KnownResourceFields.ServiceNameField)
{
var sourcePredicate = BuildStringPredicate("r.resource_name", condition, parameterName);
var peerPredicate = BuildStringPredicate("pr.resource_name", condition, parameterName);
return $"({sourcePredicate} OR (pr.resource_id IS NOT NULL AND {peerPredicate}))";
}
var attributePredicate = BuildStringPredicate("a.attribute_value", condition, parameterName);
parameters.Add($"TraceField{filterIndex}", filter.Field);
return $"EXISTS (SELECT 1 FROM telemetry_span_attributes a WHERE a.trace_id = s.trace_id AND a.span_id = s.span_id AND a.attribute_key = @TraceField{filterIndex} COLLATE NOCASE AND {attributePredicate})";
}
private GetSpansResponse GetSpansFromDatabase(GetSpansRequest context, CancellationToken cancellationToken)
{
using var connection = _database.OpenConnection();
using var interrupt = connection.RegisterInterrupt(cancellationToken);
var query = BuildSpanQuery(context);
var totalCount = connection.QuerySingle<int>($"SELECT COUNT(*) {query.FromAndWhere};", query.Parameters);
query.Parameters.Add("StartIndex", Math.Max(context.StartIndex, 0));
query.Parameters.Add("Count", Math.Max(context.Count, 0));
var identities = connection.Query<SpanIdentityRecord>($"""
SELECT s.trace_id AS TraceId, s.span_id AS SpanId
{query.FromAndWhere}
ORDER BY t.first_span_timestamp_ticks, t.trace_id, s.start_time_ticks, s.span_id
LIMIT @Count OFFSET @StartIndex;
""", query.Parameters).AsList();
var traces = identities.Select(identity => identity.TraceId).Distinct(StringComparer.Ordinal)
.ToDictionary(traceId => traceId, traceId => MaterializeTrace(connection, traceId)!, StringComparer.Ordinal);
return new GetSpansResponse
{
PagedResult = new PagedResult<OtlpSpan>
{
Items = identities.Select(identity => traces[identity.TraceId].Spans.Single(span => span.SpanId == identity.SpanId)).ToList(),
TotalItemCount = totalCount,
IsFull = connection.QuerySingle<int>("SELECT COUNT(*) FROM telemetry_traces;") >= _otlpContext.Options.MaxTraceCount
}
};
}
private static TraceQuery BuildSpanQuery(GetSpansRequest context)
{
var sql = new StringBuilder("""
FROM telemetry_spans s
JOIN telemetry_traces t ON t.trace_id = s.trace_id
JOIN telemetry_resources r ON r.resource_id = s.resource_id
JOIN telemetry_scopes sc ON sc.scope_id = s.scope_id
JOIN telemetry_span_kinds sk ON sk.kind = s.kind
JOIN telemetry_span_statuses ss ON ss.status = s.status
LEFT JOIN telemetry_resources pr ON pr.resource_id = s.uninstrumented_peer_resource_id
WHERE 1 = 1
""");
var parameters = new DynamicParameters();
if (context.ResourceKeys.Count > 0)
{
var predicates = new List<string>(context.ResourceKeys.Count);
for (var index = 0; index < context.ResourceKeys.Count; index++)
{
var key = context.ResourceKeys[index];
parameters.Add($"SpanResourceName{index}", key.Name);
var source = $"r.resource_name = @SpanResourceName{index} COLLATE NOCASE";
var peer = $"pr.resource_name = @SpanResourceName{index} COLLATE NOCASE";
if (key.InstanceId is not null)
{
parameters.Add($"SpanInstanceId{index}", key.InstanceId);
source += $" AND r.instance_id = @SpanInstanceId{index} COLLATE NOCASE";
peer += $" AND pr.instance_id = @SpanInstanceId{index} COLLATE NOCASE";
}
predicates.Add($"(({source}) OR ({peer}))");
}
sql.Append(" AND (");
sql.AppendJoin(" OR ", predicates);
sql.Append(')');
}
if (!string.IsNullOrEmpty(context.TraceId))
{
parameters.Add(
"SpanTraceId",
context.TraceId.Length >= OtlpHelpers.ShortenedIdLength
? CreateStartsWithLikePattern(context.TraceId)
: context.TraceId);
sql.Append(context.TraceId.Length >= OtlpHelpers.ShortenedIdLength
? " AND s.trace_id LIKE @SpanTraceId ESCAPE '!'"
: " AND s.trace_id = @SpanTraceId COLLATE NOCASE");
}
if (context.HasError is not null)
{
parameters.Add("SpanErrorStatus", (int)OtlpSpanStatusCode.Error);
sql.Append(context.HasError.Value ? " AND s.status = @SpanErrorStatus" : " AND s.status <> @SpanErrorStatus");
}
var filterIndex = 0;
foreach (var filter in context.Filters.Where(filter => filter.Enabled))
{
if (filter is not FieldTelemetryFilter fieldFilter)
{
continue;
}
if (fieldFilter.Field == KnownTraceFields.DurationField)
{
var parameterName = $"SpanDuration{filterIndex++}";
if (!double.TryParse(fieldFilter.Value, NumberStyles.Float, CultureInfo.InvariantCulture, out var milliseconds) || !double.IsFinite(milliseconds))
{
sql.Append(" AND 0 = 1");
continue;
}
parameters.Add(parameterName, milliseconds);
sql.Append(" AND ");
sql.Append(BuildNumericPredicate($"(CAST(s.end_time_ticks - s.start_time_ticks AS REAL) / {TimeSpan.TicksPerMillisecond})", fieldFilter.Condition, parameterName));
continue;
}
if (fieldFilter.Condition is FilterCondition.NotEqual or FilterCondition.NotContains &&
fieldFilter.Field is not (KnownResourceFields.ServiceNameField or KnownTraceFields.TraceIdField or KnownTraceFields.SpanIdField or KnownTraceFields.NameField or KnownTraceFields.KindField or KnownTraceFields.StatusField or KnownSourceFields.NameField or KnownTraceFields.TimestampField))
{
var violationFilter = new FieldTelemetryFilter
{
Field = fieldFilter.Field,
Condition = fieldFilter.Condition == FilterCondition.NotEqual ? FilterCondition.Equals : FilterCondition.Contains,
Value = fieldFilter.Value
};
sql.Append(" AND NOT ");
sql.Append(BuildSpanFieldPredicate(violationFilter, parameters, filterIndex++, invertNegative: false));
}
else
{
sql.Append(" AND ");
sql.Append(BuildSpanFieldPredicate(fieldFilter, parameters, filterIndex++, invertNegative: false));
}
}
if (context.TextFragments is { Length: > 0 })
{
for (var index = 0; index < context.TextFragments.Length; index++)
{
var parameterName = $"SpanTextFragment{index}";
parameters.Add(parameterName, CreateContainsLikePattern(context.TextFragments[index]));
sql.Append(CultureInfo.InvariantCulture, $"""
AND (
s.name LIKE @{parameterName} ESCAPE '!' OR
s.span_id LIKE @{parameterName} ESCAPE '!' OR
s.trace_id LIKE @{parameterName} ESCAPE '!' OR
sc.scope_name LIKE @{parameterName} ESCAPE '!' OR
r.resource_name LIKE @{parameterName} ESCAPE '!' OR
ss.status_name LIKE @{parameterName} ESCAPE '!' OR
sk.kind_name LIKE @{parameterName} ESCAPE '!' OR
COALESCE(s.status_message, '') LIKE @{parameterName} ESCAPE '!' OR
EXISTS (SELECT 1 FROM telemetry_span_attributes a WHERE a.trace_id = s.trace_id AND a.span_id = s.span_id AND (a.attribute_key LIKE @{parameterName} ESCAPE '!' OR a.attribute_value LIKE @{parameterName} ESCAPE '!')) OR
EXISTS (SELECT 1 FROM telemetry_span_events e WHERE e.trace_id = s.trace_id AND e.span_id = s.span_id AND e.event_name LIKE @{parameterName} ESCAPE '!')
)
""");
}
}
return new TraceQuery(sql.ToString(), parameters);
}
private List<string> GetTracePropertyKeysFromDatabase(ResourceKey? resourceKey, CancellationToken cancellationToken)
{
using var connection = _database.OpenConnection();
using var interrupt = connection.RegisterInterrupt(cancellationToken);
var parameters = new DynamicParameters();
var resourceWhere = string.Empty;
if (resourceKey is not null)
{
resourceWhere = " AND r.resource_name = @ResourceName COLLATE NOCASE";
parameters.Add("ResourceName", resourceKey.Value.Name);
if (resourceKey.Value.InstanceId is not null)
{
resourceWhere += " AND r.instance_id = @InstanceId COLLATE NOCASE";
parameters.Add("InstanceId", resourceKey.Value.InstanceId);
}
}
var propertyKeys = connection.Query<string>($"""
SELECT DISTINCT a.attribute_key
FROM telemetry_span_attributes a
JOIN telemetry_spans s ON s.trace_id = a.trace_id AND s.span_id = a.span_id
JOIN telemetry_resources r ON r.resource_id = s.resource_id
WHERE 1 = 1{resourceWhere}
ORDER BY a.attribute_key;
""", parameters);
return propertyKeys.AsList();
}
private Dictionary<string, int> GetTraceFieldValuesFromDatabase(string attributeName, CancellationToken cancellationToken)
{
if (attributeName is KnownTraceFields.DurationField or KnownTraceFields.TimestampField)
{
return new Dictionary<string, int>(StringComparers.OtlpAttribute);
}
using var connection = _database.OpenConnection();
using var interrupt = connection.RegisterInterrupt(cancellationToken);
var values = attributeName switch
{
KnownResourceFields.ServiceNameField => Query("""
SELECT resource_name AS FieldValue, COUNT(*) AS ValueCount
FROM (
SELECT r.resource_name
FROM telemetry_spans s
JOIN telemetry_resources r ON r.resource_id = s.resource_id
UNION ALL
SELECT r.resource_name
FROM telemetry_spans s
JOIN telemetry_resources r ON r.resource_id = s.uninstrumented_peer_resource_id
)
GROUP BY resource_name;
"""),
KnownTraceFields.TraceIdField => QueryFieldValues("trace_id", "telemetry_spans"),
KnownTraceFields.SpanIdField => QueryFieldValues("span_id", "telemetry_spans"),
KnownTraceFields.KindField => Query("""
SELECT sk.kind_name AS FieldValue, COUNT(*) AS ValueCount
FROM telemetry_spans s
JOIN telemetry_span_kinds sk ON sk.kind = s.kind
GROUP BY sk.kind, sk.kind_name;
"""),
KnownTraceFields.StatusField => Query("""
SELECT ss.status_name AS FieldValue, COUNT(*) AS ValueCount
FROM telemetry_spans s
JOIN telemetry_span_statuses ss ON ss.status = s.status
GROUP BY ss.status, ss.status_name;
"""),
KnownSourceFields.NameField => Query("""
SELECT sc.scope_name AS FieldValue, COUNT(*) AS ValueCount
FROM telemetry_spans s
JOIN telemetry_scopes sc ON sc.scope_id = s.scope_id
GROUP BY sc.scope_name;
"""),
KnownTraceFields.NameField => QueryFieldValues("name", "telemetry_spans"),
_ => Query("""
SELECT attribute_value AS FieldValue, COUNT(*) AS ValueCount
FROM telemetry_span_attributes
WHERE attribute_key = @AttributeName COLLATE NOCASE
GROUP BY attribute_value;
""", new { AttributeName = attributeName })
};
return values.ToDictionary(record => record.FieldValue!, record => record.ValueCount, StringComparers.OtlpAttribute);
IEnumerable<FieldValueRecord> Query(string sql, object? parameters = null)
{
return connection.Query<FieldValueRecord>(sql, parameters);
}
IEnumerable<FieldValueRecord> QueryFieldValues(string expression, string table)
{
return Query($"""
SELECT {expression} AS FieldValue, COUNT(*) AS ValueCount
FROM {table}
GROUP BY {expression};
""");
}
}
private bool HasUpdatedTraceInDatabase(OtlpTrace trace)
{
using var connection = _database.OpenConnection();
var lastUpdatedTicks = connection.QuerySingleOrDefault<long?>("""
SELECT last_updated_timestamp_ticks
FROM telemetry_traces
WHERE trace_id = @TraceId;
""", new { trace.TraceId });
return lastUpdatedTicks is null || lastUpdatedTicks.Value > trace.LastUpdatedDate.Ticks;
}
private OtlpTrace? GetTraceFromDatabase(string traceId)
{
using var connection = _database.OpenConnection();
var usePrefix = traceId.Length >= OtlpHelpers.ShortenedIdLength;
var storedTraceId = connection.QueryFirstOrDefault<string>(usePrefix
? "SELECT trace_id FROM telemetry_traces WHERE trace_id LIKE @TraceId ESCAPE '!' ORDER BY first_span_timestamp_ticks, trace_id LIMIT 1;"
: "SELECT trace_id FROM telemetry_traces WHERE trace_id = @TraceId COLLATE NOCASE ORDER BY first_span_timestamp_ticks, trace_id LIMIT 1;", new { TraceId = usePrefix ? CreateStartsWithLikePattern(traceId) : traceId });
return storedTraceId is null ? null : MaterializeTrace(connection, storedTraceId);
}
private OtlpSpan? GetSpanFromDatabase(string traceId, string spanId)
{
var trace = GetTraceFromDatabase(traceId);
return trace?.Spans.FirstOrDefault(span => span.SpanId == spanId);
}
private OtlpTrace? MaterializeTrace(SqliteConnection connection, string traceId, IDbTransaction? transaction = null)
{
return MaterializeTraces(connection, [traceId], transaction).GetValueOrDefault(traceId);
}
private Dictionary<string, OtlpTrace> MaterializeTraces(SqliteConnection connection, IReadOnlyList<string> traceIds, IDbTransaction? transaction = null)
{
var records = connection.Query<SpanRecord>("""
SELECT
s.trace_id AS TraceId,
s.span_id AS SpanId,
s.parent_span_id AS ParentSpanId,
s.resource_id AS ResourceId,
s.resource_view_id AS ResourceViewId,
s.scope_id AS ScopeId,
s.name AS Name,
s.kind AS Kind,
s.start_time_ticks AS StartTimeTicks,
s.end_time_ticks AS EndTimeTicks,
s.status AS Status,
s.status_message AS StatusMessage,
s.trace_state AS State,
s.uninstrumented_peer_resource_id AS PeerResourceId,
t.last_updated_timestamp_ticks AS LastUpdatedTimestampTicks,
r.resource_name AS ResourceName,
r.instance_id AS InstanceId,
r.uninstrumented_peer AS UninstrumentedPeer,
r.has_logs AS HasLogs,
r.has_traces AS HasTraces,
r.has_metrics AS HasMetrics,
sc.scope_name AS ScopeName,
sc.scope_version AS ScopeVersion,
pr.resource_name AS PeerResourceName,
pr.instance_id AS PeerInstanceId
FROM telemetry_spans s
JOIN telemetry_traces t ON t.trace_id = s.trace_id
JOIN telemetry_resources r ON r.resource_id = s.resource_id
JOIN telemetry_scopes sc ON sc.scope_id = s.scope_id
LEFT JOIN telemetry_resources pr ON pr.resource_id = s.uninstrumented_peer_resource_id
WHERE s.trace_id IN @TraceIds
ORDER BY s.trace_id, s.start_time_ticks, s.span_id;
""", new { TraceIds = traceIds }, transaction).AsList();
if (records.Count == 0)
{
return new Dictionary<string, OtlpTrace>(StringComparer.Ordinal);
}
var spanAttributes = connection.Query<TraceOwnedAttributeRecord>("""
SELECT trace_id AS TraceId, span_id AS OwnerId, attribute_key AS AttributeKey, attribute_value AS AttributeValue
FROM telemetry_span_attributes
WHERE trace_id IN @TraceIds
ORDER BY trace_id, span_id, ordinal;
""", new { TraceIds = traceIds }, transaction).ToLookup(record => (record.TraceId, record.OwnerId));
var eventRecords = connection.Query<SpanEventRecord>("""
SELECT trace_id AS TraceId, event_id AS EventId, span_id AS SpanId, event_name AS EventName, event_time_ticks AS EventTimeTicks
FROM telemetry_span_events
WHERE trace_id IN @TraceIds
ORDER BY trace_id, span_id, ordinal;
""", new { TraceIds = traceIds }, transaction).AsList();
var eventAttributeRecords = new List<TextOwnedAttributeRecord>();
foreach (var eventIdBatch in eventRecords.Select(record => record.EventId).Chunk(MaxSpanDetailBatchSize))
{
eventAttributeRecords.AddRange(connection.Query<TextOwnedAttributeRecord>("""
SELECT event_id AS OwnerId, attribute_key AS AttributeKey, attribute_value AS AttributeValue
FROM telemetry_span_event_attributes
WHERE event_id IN @Ids
ORDER BY event_id, ordinal;
""", new { Ids = eventIdBatch }, transaction));
}
var eventAttributes = eventAttributeRecords.ToLookup(record => record.OwnerId);
var events = eventRecords.ToLookup(record => (record.TraceId, record.SpanId));
var linkRecords = connection.Query<SpanLinkRecord>("""
SELECT
link_id AS LinkId,
source_trace_id AS SourceTraceId,
source_span_id AS SourceSpanId,
target_trace_id AS TraceId,
target_span_id AS SpanId,
trace_state AS TraceState
FROM telemetry_span_links
WHERE source_trace_id IN @TraceIds OR target_trace_id IN @TraceIds
ORDER BY link_id;
""", new { TraceIds = traceIds }, transaction).AsList();
var linkAttributeRecords = new List<LongOwnedAttributeRecord>();
foreach (var linkIdBatch in linkRecords.Select(record => record.LinkId).Chunk(MaxSpanDetailBatchSize))
{
linkAttributeRecords.AddRange(connection.Query<LongOwnedAttributeRecord>("""
SELECT link_id AS OwnerId, attribute_key AS AttributeKey, attribute_value AS AttributeValue
FROM telemetry_span_link_attributes
WHERE link_id IN @Ids
ORDER BY link_id, ordinal;
""", new { Ids = linkIdBatch }, transaction));
}
var linkAttributes = linkAttributeRecords.ToLookup(record => record.OwnerId);
var outgoingLinks = linkRecords.ToLookup(record => (record.SourceTraceId, record.SourceSpanId));
var incomingLinks = linkRecords.ToLookup(record => (record.TraceId, record.SpanId));
var traces = new Dictionary<string, OtlpTrace>(StringComparer.Ordinal);
foreach (var traceRecords in records.GroupBy(record => record.TraceId, StringComparer.Ordinal))
{
var traceId = traceRecords.Key;
var firstRecord = traceRecords.First();
var trace = new OtlpTrace(Convert.FromHexString(traceId), new DateTime(firstRecord.LastUpdatedTimestampTicks, DateTimeKind.Utc));
foreach (var record in traceRecords)
{
var (_, view, scope) = GetCachedTelemetryMetadata(record.ResourceId, record.ResourceViewId, record.ScopeId, CachedTelemetryType.Traces);
var modelSpan = new OtlpSpan(view, trace, scope)
{
SpanId = record.SpanId,
ParentSpanId = record.ParentSpanId,
Name = record.Name,
Kind = (OtlpSpanKind)record.Kind,
StartTime = new DateTime(record.StartTimeTicks, DateTimeKind.Utc),
EndTime = new DateTime(record.EndTimeTicks, DateTimeKind.Utc),
Status = (OtlpSpanStatusCode)record.Status,
StatusMessage = record.StatusMessage,
State = record.State,
Attributes = ToPairs(spanAttributes[(traceId, record.SpanId)]),
Events = [],
Links = outgoingLinks[(traceId, record.SpanId)].Select(CreateLink).ToList(),
BackLinks = incomingLinks[(traceId, record.SpanId)].Select(CreateLink).ToList()
};
if (record.PeerResourceId is not null)
{
modelSpan.SetUninstrumentedPeer(GetCachedResource(record.PeerResourceId.Value));
}
modelSpan.Events.AddRange(events[(traceId, record.SpanId)].Select(spanEvent => new OtlpSpanEvent(modelSpan)
{
InternalId = Guid.Parse(spanEvent.EventId),
Name = spanEvent.EventName,
Time = new DateTime(spanEvent.EventTimeTicks, DateTimeKind.Utc),
Attributes = ToPairs(eventAttributes[spanEvent.EventId])
}));
trace.AddSpan(modelSpan, skipLastUpdatedDate: true);
}
traces.Add(traceId, trace);
}
return traces;
OtlpSpanLink CreateLink(SpanLinkRecord link)
{
return new OtlpSpanLink
{
SourceTraceId = link.SourceTraceId,
SourceSpanId = link.SourceSpanId,
TraceId = link.TraceId,
SpanId = link.SpanId,
TraceState = link.TraceState,
Attributes = ToPairs(linkAttributes[link.LinkId])
};
}
static KeyValuePair<string, string>[] ToPairs(IEnumerable<AttributeRecord> attributes)
{
return attributes.Select(attribute => KeyValuePair.Create(attribute.AttributeKey, attribute.AttributeValue)).ToArray();
}
}
private sealed record TraceQuery(string FromAndWhere, DynamicParameters Parameters);
private sealed class TraceAggregateRecord
{
public required int TotalItemCount { get; init; }
public required long MaxDurationTicks { get; init; }
}
private sealed class TraceSummaryRecord
{
public required string TraceId { get; init; }
public required long LastUpdatedTimestampTicks { get; init; }
}
private sealed class TracePageSummaryRecord
{
public required int TotalItemCount { get; init; }
public required long MaxDurationTicks { get; init; }
public required bool IsFull { get; init; }
public string? TraceId { get; init; }
public string? FullName { get; init; }
public long? StartTimeTicks { get; init; }
public long? DurationTicks { get; init; }
public string? RootResourceName { get; init; }
public string? RootInstanceId { get; init; }
public bool? RootUninstrumentedPeer { get; init; }
public bool? HasError { get; init; }
public bool? HasGenAI { get; init; }
public string? ResourceName { get; init; }
public string? InstanceId { get; init; }
public bool? UninstrumentedPeer { get; init; }
public int? TotalSpans { get; init; }
public int? ErroredSpans { get; init; }
}
private sealed class SpanIdentityRecord
{
public required string TraceId { get; init; }
public required string SpanId { get; init; }
}
private sealed class TraceOwnedAttributeRecord : AttributeRecord
{
public required string TraceId { get; init; }
public required string OwnerId { get; init; }
}
private sealed class TextOwnedAttributeRecord : AttributeRecord
{
public required string OwnerId { get; init; }
}
private sealed class LongOwnedAttributeRecord : AttributeRecord
{
public required long OwnerId { get; init; }
}
private sealed class SpanEventRecord
{
public required string TraceId { get; init; }
public required string EventId { get; init; }
public required string SpanId { get; init; }
public required string EventName { get; init; }
public required long EventTimeTicks { get; init; }
}
private sealed class SpanLinkRecord
{
public required long LinkId { get; init; }
public required string SourceTraceId { get; init; }
public required string SourceSpanId { get; init; }
public required string TraceId { get; init; }
public required string SpanId { get; init; }
public required string TraceState { get; init; }
}
private sealed class SpanRecord
{
public required string TraceId { get; init; }
public required string SpanId { get; init; }
public string? ParentSpanId { get; init; }
public required long ResourceId { get; init; }
public required long ResourceViewId { get; init; }
public required long ScopeId { get; init; }
public required string Name { get; init; }
public required int Kind { get; init; }
public required long StartTimeTicks { get; init; }
public required long EndTimeTicks { get; init; }
public required int Status { get; init; }
public string? StatusMessage { get; init; }
public string? State { get; init; }
public long? PeerResourceId { get; init; }
public required long LastUpdatedTimestampTicks { get; init; }
public required string ResourceName { get; init; }
public string? InstanceId { get; init; }
public required bool UninstrumentedPeer { get; init; }
public required bool HasLogs { get; init; }
public required bool HasTraces { get; init; }
public required bool HasMetrics { get; init; }
public required string ScopeName { get; init; }
public required string ScopeVersion { get; init; }
public string? PeerResourceName { get; init; }
public string? PeerInstanceId { get; init; }
}
}