Loading

TSDB metrics

TSDB (time-series database) mode optimizes data streams for metrics storage. It enables sorted indices, synthetic _source, automatic deduplication, and downsampling -- significantly reducing storage for high-cardinality metrics.

  • Ingesting metrics (CPU, memory, request latency, business KPIs)
  • You have natural dimension fields (host, service, metric name)
  • You want Elasticsearch's time-series optimizations (deduplication, downsampling)

TSDB requires at least one [Dimension] field and a [Timestamp]:

public class MetricEvent
{
    [Timestamp]
    [JsonPropertyName("@timestamp")]
    public DateTimeOffset Timestamp { get; set; }

    [Dimension]
    [Keyword]
    public string Host { get; set; }

    [Dimension]
    [Keyword]
    public string MetricName { get; set; }

    public double Value { get; set; }

    [Keyword]
    public string Unit { get; set; }
}

The combination of dimension fields and timestamp uniquely identifies a data point. Documents with the same dimensions and timestamp are deduplicated.

[ElasticsearchMappingContext]
[Entity<MetricEvent>(
    Target = EntityTarget.DataStream,
    DataStreamType = "metrics",
    DataStreamDataset = "myapp",
    DataStreamNamespace = "production",
    DataStreamMode = DataStreamMode.Tsdb
)]
public static partial class MetricsContext;
		
var strategy = IngestStrategies.DataStream<MetricEvent>(MetricsContext.MetricEvent.Context, "90d");
var options = new IngestChannelOptions<MetricEvent>(transport, strategy, MetricsContext.MetricEvent.Context);
using var channel = new IngestChannel<MetricEvent>(options);

await channel.BootstrapElasticsearchAsync(BootstrapMethod.Failure);

For continuous metrics collection, tune the buffer for throughput:

var strategy = IngestStrategies.DataStream<MetricEvent>(MetricsContext.MetricEvent.Context, "90d");
var options = new IngestChannelOptions<MetricEvent>(transport, strategy, MetricsContext.MetricEvent.Context)
{
    BufferOptions = new BufferOptions
    {
        InboundBufferMaxSize = 500_000,
        OutboundBufferMaxSize = 5_000,
        ExportMaxConcurrency = 8
    }
};
using var channel = new IngestChannel<MetricEvent>(options);

await channel.BootstrapElasticsearchAsync(BootstrapMethod.Failure);

// Collect metrics on a timer
timer.Elapsed += (_, _) =>
{
    channel.TryWrite(new MetricEvent
    {
        Timestamp = DateTimeOffset.UtcNow,
        Host = Environment.MachineName,
        MetricName = "cpu.usage",
        Value = GetCpuUsage(),
        Unit = "percent"
    });
};

// At shutdown
await channel.WaitForDrainAsync(TimeSpan.FromSeconds(10), ctx);