git.stella-ops.org/src/__Libraries/StellaOps.Ingestion.Telemetry/IngestionTelemetry.cs

using System.Diagnostics;
using System.Diagnostics.Metrics;

namespace StellaOps.Ingestion.Telemetry;

public static class IngestionTelemetry
{
    public const string ActivitySourceName = "StellaOps.Ingestion";
    public const string MeterName = "StellaOps.Ingestion";

    public const string PhaseFetch = "fetch";
    public const string PhaseTransform = "transform";
    public const string PhaseWrite = "write";

    public const string ResultOk = "ok";
    public const string ResultReject = "reject";
    public const string ResultNoop = "noop";

    private const string WriteMetricName = "ingestion_write_total";
    private const string ViolationMetricName = "aoc_violation_total";
    private const string LatencyMetricName = "ingestion_latency_seconds";

    private static readonly ActivitySource ActivitySource = new(ActivitySourceName);
    private static readonly Meter Meter = new(MeterName);

    private static readonly Counter<long> WriteCounter = Meter.CreateCounter<long>(
        WriteMetricName,
        unit: "count",
        description: "Counts raw advisory ingestion attempts grouped by tenant, source, and outcome.");

    private static readonly Counter<long> ViolationCounter = Meter.CreateCounter<long>(
        ViolationMetricName,
        unit: "count",
        description: "Counts Aggregation-Only Contract violations raised during ingestion.");

    private static readonly Histogram<double> LatencyHistogram = Meter.CreateHistogram<double>(
        LatencyMetricName,
        unit: "s",
        description: "Ingestion stage latency measured in seconds.");

    public static Activity? StartFetchActivity(
        string tenant,
        string source,
        string? upstreamId,
        string? contentHash,
        string? uri = null)
        => StartActivity("ingest.fetch", tenant, source, upstreamId, contentHash, builder: activity =>
        {
            if (!string.IsNullOrWhiteSpace(uri))
            {
                activity.SetTag("uri", uri);
            }
        });

    public static Activity? StartTransformActivity(
        string tenant,
        string source,
        string? upstreamId,
        string? contentHash,
        string? documentType = null,
        long? payloadBytes = null)
        => StartActivity("ingest.transform", tenant, source, upstreamId, contentHash, builder: activity =>
        {
            if (!string.IsNullOrWhiteSpace(documentType))
            {
                activity.SetTag("documentType", documentType);
            }

            if (payloadBytes.HasValue && payloadBytes.Value >= 0)
            {
                activity.SetTag("payloadBytes", payloadBytes.Value);
            }
        });

    public static Activity? StartWriteActivity(
        string tenant,
        string source,
        string? upstreamId,
        string? contentHash,
        string collection)
        => StartActivity("ingest.write", tenant, source, upstreamId, contentHash, builder: activity =>
        {
            activity.SetTag("collection", collection);
        });

    public static Activity? StartGuardActivity(
        string tenant,
        string source,
        string? upstreamId,
        string? contentHash,
        string? supersedes)
        => StartActivity("aoc.guard", tenant, source, upstreamId, contentHash, builder: activity =>
        {
            if (!string.IsNullOrWhiteSpace(supersedes))
            {
                activity.SetTag("supersedes", supersedes);
            }
        });

    public static void RecordWriteAttempt(string tenant, string source, string result)
    {
        if (string.IsNullOrWhiteSpace(tenant) || string.IsNullOrWhiteSpace(source) || string.IsNullOrWhiteSpace(result))
        {
            return;
        }

        var tags = new TagList
        {
            { "tenant", tenant },
            { "source", source },
            { "result", result }
        };

        WriteCounter.Add(1, tags);
    }

    public static void RecordViolation(string tenant, string source, string code)
    {
        if (string.IsNullOrWhiteSpace(tenant) || string.IsNullOrWhiteSpace(source) || string.IsNullOrWhiteSpace(code))
        {
            return;
        }

        var tags = new TagList
        {
            { "tenant", tenant },
            { "source", source },
            { "code", code }
        };

        ViolationCounter.Add(1, tags);
    }

    public static void RecordLatency(string tenant, string source, string phase, TimeSpan duration)
    {
        if (string.IsNullOrWhiteSpace(tenant) || string.IsNullOrWhiteSpace(source) || string.IsNullOrWhiteSpace(phase))
        {
            return;
        }

        var tags = new TagList
        {
            { "tenant", tenant },
            { "source", source },
            { "phase", phase }
        };

        var seconds = duration.TotalSeconds;
        if (double.IsNaN(seconds) || double.IsInfinity(seconds))
        {
            seconds = 0d;
        }

        if (seconds < 0)
        {
            seconds = 0d;
        }

        LatencyHistogram.Record(seconds, tags);
    }

    private static Activity? StartActivity(
        string name,
        string tenant,
        string source,
        string? upstreamId,
        string? contentHash,
        Action<Activity>? builder = null)
    {
        var activity = ActivitySource.StartActivity(name, ActivityKind.Internal);
        if (activity is null)
        {
            return null;
        }

        if (!string.IsNullOrWhiteSpace(tenant))
        {
            activity.SetTag("tenant", tenant);
        }

        if (!string.IsNullOrWhiteSpace(source))
        {
            activity.SetTag("source", source);
        }

        if (!string.IsNullOrWhiteSpace(upstreamId))
        {
            activity.SetTag("upstream.id", upstreamId);
        }

        if (!string.IsNullOrWhiteSpace(contentHash))
        {
            activity.SetTag("contentHash", contentHash);
        }

        builder?.Invoke(activity);
        return activity;
    }
}