App (OTLP) --> OTel Collector --> Elasticsearch / Loki / Azure Monitor
                  |
                  +--> Sampling / filtering / PII scrub

# otel-collector-config.yaml
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: "0.0.0.0:4317"
      http:
        endpoint: "0.0.0.0:4318"

processors:
  batch:
    timeout: 5s
    send_batch_size: 1024
  filter:
    logs:
      exclude:
        match_type: strict
        bodies:
          - "Health check endpoint hit"

exporters:
  elasticsearch:
    endpoints: ["https://es-cluster:9200"]
    logs_index: "app-logs"
  loki:
    endpoint: "http://loki:3100/loki/api/v1/push"

service:
  pipelines:
    logs:
      receivers: [otlp]
      processors: [batch, filter]
      exporters: [elasticsearch, loki]

App (Serilog) --> Seq / Elasticsearch sink

# Find errors for a specific order
level: "Error" AND OrderId: "abc-123"

# Find slow operations (custom Duration property)
Duration > 5000 AND ServiceName: "order-api"

# Wildcard on message template
message: "Failed to process*" AND NOT level: "Debug"

# Time-scoped with correlation
TraceId: "0af7651916cd43dd8448eb211c80319c" AND @timestamp >= "2025-01-15T10:00:00"

# Find errors for a specific order
@Level = 'Error' and OrderId = 'abc-123'

# Find slow operations
Duration > 5000 and Application = 'order-api'

# Free-text search combined with structured filter
@Message like '%timeout%' and @Level in ['Warning', 'Error']

# Correlation across services
TraceId = '0af7651916cd43dd8448eb211c80319c'

# Filter by labels then regex on log line
{service_name="order-api"} |= "Error" | json | OrderId="abc-123"

# Structured field extraction and filtering
{service_name="order-api"} | json | Duration > 5000

# Count errors per service over time (for dashboards)
sum(rate({service_name=~".+"} |= "Error" [5m])) by (service_name)

// Find errors for a specific order
traces
| where severityLevel >= 3
| where customDimensions.OrderId == "abc-123"
| order by timestamp desc

// Slow operations
traces
| where toint(customDimensions.Duration) > 5000
| where cloud_RoleName == "order-api"

// Cross-service correlation
union traces, exceptions
| where operation_Id == "0af7651916cd43dd8448eb211c80319c"
| order by timestamp asc

processors:
  filter:
    logs:
      exclude:
        match_type: regexp
        # Drop Debug and Trace logs at the collector level
        severity_texts: ["DEBUG", "TRACE"]
      exclude:
        match_type: strict
        # Exclude health check noise
        bodies:
          - "Health check endpoint hit"
  transform:
    log_statements:
      - context: log
        conditions:
          # Keep all Warning+ logs unconditionally
          - severity_number >= SEVERITY_NUMBER_WARN
        statements: []

// Serilog.Expressions package for conditional log filtering
builder.Host.UseSerilog((context, loggerConfiguration) =>
{
    loggerConfiguration
        .ReadFrom.Configuration(context.Configuration)
        // Drop health check logs entirely
        .Filter.ByExcluding("RequestPath = '/health/ready'")
        // Sample Debug logs at 10%
        .Filter.ByExcluding(
            "@Level = 'Debug' and Hash(@i) % 10 != 0");
});

<PackageReference Include="Serilog.Expressions" Version="5.*" />

// Enricher that masks known-sensitive properties on every log event
public sealed class PiiMaskingEnricher : ILogEventEnricher
{
    private static readonly HashSet<string> s_sensitiveKeys = new(
        StringComparer.OrdinalIgnoreCase)
    {
        "Email", "PhoneNumber", "IpAddress",
        "CreditCard", "SSN", "Password"
    };

    public void Enrich(LogEvent logEvent, ILogEventPropertyFactory factory)
    {
        var propertiesToMask = logEvent.Properties
            .Where(p => s_sensitiveKeys.Contains(p.Key))
            .Select(p => p.Key)
            .ToList();

        foreach (var key in propertiesToMask)
        {
            logEvent.AddOrUpdateProperty(
                factory.CreateProperty(key, "***REDACTED***"));
        }
    }
}

// Registration
loggerConfiguration.Enrich.With<PiiMaskingEnricher>();

processors:
  attributes:
    actions:
      # Mask email addresses using regex
      - key: user.email
        action: update
        value: "***@redacted.com"
      # Remove sensitive attributes entirely
      - key: http.request.header.authorization
        action: delete
      - key: user.password
        action: delete

// Query all logs for a distributed operation across services
// In Seq:
TraceId = '0af7651916cd43dd8448eb211c80319c'

// In Kibana:
TraceId: "0af7651916cd43dd8448eb211c80319c"

// In Azure Monitor:
traces | where operation_Id == "0af7651916cd43dd8448eb211c80319c"

// Propagate a business correlation ID through Serilog LogContext
public sealed class CorrelationIdMiddleware(RequestDelegate next)
{
    private const string CorrelationHeader = "X-Correlation-Id";

    public async Task InvokeAsync(HttpContext context)
    {
        var correlationId = context.Request.Headers[CorrelationHeader]
            .FirstOrDefault() ?? Guid.NewGuid().ToString("N");

        context.Response.Headers[CorrelationHeader] = correlationId;

        using (LogContext.PushProperty("CorrelationId", correlationId))
        {
            await next(context);
        }
    }
}

// Registration
app.UseMiddleware<CorrelationIdMiddleware>();

// Producer -- attach correlation to message
var message = new ServiceBusMessage(payload)
{
    CorrelationId = Activity.Current?.TraceId.ToString()
        ?? Guid.NewGuid().ToString("N"),
    ApplicationProperties =
    {
        ["BusinessCorrelationId"] = orderId.ToString()
    }
};

// Consumer -- restore correlation in log scope
processor.ProcessMessageAsync += async args =>
{
    using var scope = logger.BeginScope(new Dictionary<string, object>
    {
        ["CorrelationId"] = args.Message.CorrelationId,
        ["BusinessCorrelationId"] =
            args.Message.ApplicationProperties["BusinessCorrelationId"]
    });

    logger.LogInformation("Processing message {MessageId}", args.Message.MessageId);
    await ProcessAsync(args.Message, args.CancellationToken);
};