remotewrite

package module
v0.1.3 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Aug 18, 2026 License: MIT Imports: 31 Imported by: 0

README

remotewrite

A Go client for the Prometheus remote_write protocol with two transports:

  • Client for synchronous delivery (forward + push).
  • DurableClient for local disk spooling, restart survival, and background draining.

Install

go get github.com/baselinehq/remote-write

Forward Quickstart

Forward an incoming remote_write request to an upstream (typical proxy):

client, err := remotewrite.New(remotewrite.Config{
    UpstreamURL:  "http://localhost:9090/api/v1/write",
    TenantHeader: "X-Scope-OrgID",
    // Retry: nil => streaming mode (single attempt, no buffering)
})
if err != nil {
    log.Fatal(err)
}
defer client.Close()

func handleRemoteWrite(w http.ResponseWriter, r *http.Request) {
    resp, err := client.Forward(r.Context(), remotewrite.ForwardRequest{
        TenantID:           r.Header.Get("X-Scope-OrgID"),
        Body:               r.Body,
        ContentLength:      r.ContentLength,
        ContentType:        r.Header.Get("Content-Type"),
        ContentEncoding:    r.Header.Get("Content-Encoding"),
        RemoteWriteVersion: r.Header.Get("X-Prometheus-Remote-Write-Version"),
    })
    if err != nil {
        http.Error(w, err.Error(), http.StatusBadGateway)
        return
    }
    defer resp.Body.Close()

    w.WriteHeader(resp.StatusCode)
    io.Copy(w, resp.Body)
}

Use BodyBytes when the payload is already materialized:

resp, err := client.Forward(ctx, remotewrite.ForwardRequest{
    BodyBytes:   payload,
    ContentType: "application/x-protobuf",
})

Push Quickstart

Encode metrics from a prometheus.Gatherer and send them as a remote_write payload. The client handles protobuf encoding, snappy compression, and batching.

client, err := remotewrite.New(remotewrite.Config{
    UpstreamURL: "http://localhost:9090/api/v1/write",
})
if err != nil {
    log.Fatal(err)
}
defer client.Close()

err = client.Push(ctx, remotewrite.PushRequest{
    TenantID: "tenant-123",
    Gatherer: prometheus.DefaultGatherer,
    ExternalLabels: map[string]string{
        "job":      "my-daemon",
        "instance": hostname,
    },
})

The encoder supports classic Prometheus metric types: counter, gauge, untyped, classic histogram (_bucket/_count/_sum), and classic summary ({quantile=...}/_count/_sum). Native histograms, exemplars, metadata, and remote-write 2.0 are not implemented.

Durable Mode

DurableClient accepts fully materialized remote_write payloads, writes them to disk under QueueDir, and drains them in a background Run loop. Use it for long-running agents that must survive restarts or temporary upstream outages without dropping accepted payloads.

dc, err := remotewrite.NewDurable(remotewrite.DurableConfig{
    Client: remotewrite.Config{
        UpstreamURL:  "http://localhost:9090/api/v1/write",
        TenantHeader: "X-Scope-OrgID",
        Retry: &remotewrite.RetryConfig{
            MinWait: time.Second,
            MaxWait: 30 * time.Second,
        },
    },
    QueueDir:        "/var/lib/my-agent/remotewrite",
    QueueName:       "metrics",
    MaxPendingBytes: 10 * 1024 * 1024 * 1024,
    SendConcurrency: 1,
})
if err != nil {
    log.Fatal(err)
}
defer dc.Close()

go func() {
    if err := dc.Run(context.Background()); err != nil {
        log.Printf("durable drain stopped: %v", err)
    }
}()

err = dc.Push(ctx, remotewrite.PushRequest{
    TenantID: "tenant-a",
    Gatherer: prometheus.DefaultGatherer,
})

Durable delivery is at-least-once. If the upstream accepts a payload and the process exits before the spool record is removed, the payload may be replayed after restart. The default SendConcurrency: 1 preserves FIFO drain order. Setting it higher improves throughput but can produce out-of-order delivery.

The durable layer exports the following Prometheus metrics:

  • remotewrite_durable_enqueued_total
  • remotewrite_durable_sent_total
  • remotewrite_durable_send_failures_total
  • remotewrite_durable_retryable_failures_total
  • remotewrite_durable_permanent_failures_total
  • remotewrite_durable_dropped_total
  • remotewrite_durable_corrupt_records_total
  • remotewrite_durable_queue_pending_bytes
  • remotewrite_durable_queue_inmemory_blocks
  • remotewrite_durable_queue_blocked
  • remotewrite_durable_inflight

Configuration

UpstreamURL is required and must be an absolute URL with scheme and host. It may contain a single {tenant} placeholder in the path, which is replaced (path-escaped) with ForwardRequest.TenantID per call.

TenantHeader sets the header name for tenant injection; BearerToken adds an Authorization: Bearer ... header.

ExtraHeaders are forwarded but cannot override the reserved headers (Content-Type, Content-Encoding, User-Agent, X-Prometheus-Remote-Write-Version) or the configured auth/tenant headers.

Timeout, if non-zero, is applied to http.Client.Timeout and (for the default transport) ResponseHeaderTimeout. Prefer context deadlines and leave this at 0.

RateLimitBytesPerSec enforces a token-bucket budget on outgoing bytes. It respects the caller's context, including when a single payload is larger than the per-second budget.

External labels are merged with metric labels: metric labels take precedence when names collide.

Retries

Retries are disabled by default (streaming mode). Enable them by setting Retry:

client, _ := remotewrite.New(remotewrite.Config{
    UpstreamURL: "http://localhost:9090/api/v1/write",
    Retry: &remotewrite.RetryConfig{
        MaxRetries:  3,
        MinWait:     time.Second,
        MaxWait:     30 * time.Second,
        MaxBodySize: 10 * 1024 * 1024,
    },
})

Retry behavior:

  • Retries on 429 and 5xx, with exponential backoff plus jitter.
  • Honors Retry-After when present.
  • Uses GetBody when provided. Otherwise the request Body is buffered up to MaxBodySize so it can be replayed.

Benchmarks

The default benchmarks measure library and client overhead against a loopback httptest server. They do not prove receiver throughput — a real remote_write receiver is the dominant cost in production.

go test -run='^$' -bench=. -benchmem ./benchmarks/...

Integration Tests and Receiver Benchmarks

Integration correctness test (Docker required):

cd test/integration
go test -timeout 5m ./...

Real-VictoriaMetrics PushTimeSeries benchmarks (Docker required, opt-in):

cd test/integration
REMOTEWRITE_VM_BENCH=1 go test -run='^$' -bench=BenchmarkVictoriaMetrics -benchmem ./...

Set REMOTEWRITE_VM_URL=http://127.0.0.1:8428/api/v1/write to use an existing VictoriaMetrics instance instead of starting one.

Documentation

Overview

Package remotewrite provides clients for the Prometheus remote_write protocol.

It supports three common paths:

  • forwarding existing remote_write requests to another endpoint,
  • gathering Prometheus metrics, encoding them, and sending them upstream,
  • durably spooling encoded payloads on disk and draining them in the background.

The default Forward path streams request bodies once and does not buffer or retry unless retry behavior is explicitly configured. Push and PushTimeSeries encode remote_write protobuf payloads, snappy-compress them, and send the encoded bytes through the same transport.

Forwarding Requests

Forward an incoming remote_write request to an upstream endpoint:

client, err := remotewrite.New(remotewrite.Config{
	UpstreamURL:  "http://localhost:9090/api/v1/write",
	TenantHeader: "X-Scope-OrgID",
})
if err != nil {
	log.Fatal(err)
}
defer client.Close()

func handleRemoteWrite(w http.ResponseWriter, r *http.Request) {
	resp, err := client.Forward(r.Context(), remotewrite.ForwardRequest{
		TenantID:           r.Header.Get("X-Scope-OrgID"),
		Body:               r.Body,
		ContentLength:      r.ContentLength,
		ContentType:        r.Header.Get("Content-Type"),
		ContentEncoding:    r.Header.Get("Content-Encoding"),
		RemoteWriteVersion: r.Header.Get("X-Prometheus-Remote-Write-Version"),
	})
	if err != nil {
		http.Error(w, err.Error(), http.StatusBadGateway)
		return
	}
	defer resp.Body.Close()

	w.WriteHeader(resp.StatusCode)
	io.Copy(w, resp.Body)
}

If the request body is already materialized, pass BodyBytes so Forward can use net/http's in-memory body fast path:

resp, err := client.Forward(ctx, remotewrite.ForwardRequest{
	BodyBytes:   payload,
	ContentType: "application/x-protobuf",
})

Pushing Gathered Metrics

Push metrics from a prometheus.Gatherer to a remote_write endpoint:

client, err := remotewrite.New(remotewrite.Config{
	UpstreamURL: "http://localhost:9090/api/v1/write",
})
if err != nil {
	log.Fatal(err)
}
defer client.Close()

err = client.Push(ctx, remotewrite.PushRequest{
	Gatherer: prometheus.DefaultGatherer,
	ExternalLabels: map[string]string{
		"job":      "my-service",
		"instance": hostname,
	},
})

Use PushTimeSeries when callers already have prompb.TimeSeries values.

Durable Delivery

DurableClient accepts fully materialized remote_write payloads, stores them under QueueDir, and drains them in a background Run loop. It is useful for long-running agents that should survive restarts or temporary upstream outages without dropping accepted payloads.

dc, err := remotewrite.NewDurable(remotewrite.DurableConfig{
	Client: remotewrite.Config{
		UpstreamURL:  "http://localhost:9090/api/v1/write",
		TenantHeader: "X-Scope-OrgID",
	},
	QueueDir:        "/var/lib/my-agent/remotewrite",
	QueueName:       "metrics",
	MaxPendingBytes: 10 * 1024 * 1024 * 1024,
	SendConcurrency: 1,
})
if err != nil {
	log.Fatal(err)
}
defer dc.Close()

go func() {
	if err := dc.Run(context.Background()); err != nil {
		log.Printf("durable drain stopped: %v", err)
	}
}()

err = dc.Enqueue(ctx, remotewrite.DurableRequest{
	TenantID:  "tenant-a",
	BodyBytes: payload,
})

Durable delivery is at-least-once. If an upstream accepts a payload and the process exits before the local record is removed, the payload may be replayed after restart.

Retries

Retries are disabled by default (streaming mode). To enable retries:

client, err := remotewrite.New(remotewrite.Config{
	UpstreamURL: "http://localhost:9090/api/v1/write",
	Retry: &remotewrite.RetryConfig{
		MaxRetries: 3,
		MinWait:    time.Second,
		MaxWait:    30 * time.Second,
	},
})

Retries occur on 429 (Too Many Requests) and 5xx responses. The client honors Retry-After headers and uses exponential backoff with jitter.

Index

Constants

This section is empty.

Variables

View Source
var (
	ErrDurableClientClosed  = errors.New("durable client is closed")
	ErrDurableRunStarted    = errors.New("durable client run loop has already started")
	ErrDurableQueueBlocked  = errors.New("durable queue is blocked")
	ErrDurableSpoolLocked   = errors.New("durable spool is already locked")
	ErrDurableRecordCorrupt = errors.New("durable record is corrupt")
)

Functions

func IsBodyTooLarge

func IsBodyTooLarge(err error) bool

IsBodyTooLarge returns true if err (or any error in its chain) indicates the body exceeded RetryConfig.MaxBodySize.

Types

type Client

type Client struct {
	// contains filtered or unexported fields
}

func New

func New(cfg Config) (*Client, error)

New creates a new remote write client.

func (*Client) Close

func (c *Client) Close() error

Close shuts down the client.

func (*Client) Forward

func (c *Client) Forward(ctx context.Context, req ForwardRequest) (*http.Response, error)

Forward sends a remote write request to the upstream

func (*Client) Push

func (c *Client) Push(ctx context.Context, pr PushRequest) error

Push gathers metrics, batches them, and sends them to the remote write endpoint.

func (*Client) PushTimeSeries

func (c *Client) PushTimeSeries(ctx context.Context, pr PushTimeSeriesRequest) error

PushTimeSeries sends pre-built time series to the remote_write endpoint.

func (*Client) RetriesEnabled

func (c *Client) RetriesEnabled() bool

RetriesEnabled returns true if retries are configured.

type Config

type Config struct {
	// UpstreamURL is the base URL to send remote write requests to.
	// May contain {tenant} placeholder for tenant-based URL routing.
	UpstreamURL string

	// Timeout, if > 0, is applied to http.Client.Timeout and (for the default
	// transport) ResponseHeaderTimeout. Default: 0 (no client-level timeout).
	// Prefer context deadlines on each call.
	Timeout time.Duration

	// TenantHeader is the header name for tenant injection.
	TenantHeader string

	// BearerToken is an optional bearer token for authentication.
	BearerToken string

	// Transport is an optional custom http.Transport.
	Transport *http.Transport

	// MaxConnsPerHost limits connections to the upstream host. Default: 100.
	MaxConnsPerHost int

	// MaxIdleConnsPerHost limits idle connections per host. Default: 10.
	MaxIdleConnsPerHost int

	// Retry configures retry behavior. If nil, retries are DISABLED (streaming mode).
	Retry *RetryConfig

	// RateLimitBytesPerSec limits send rate. 0 = disabled.
	RateLimitBytesPerSec int64

	// UserAgent is the User-Agent header value. Default: "remotewrite-client".
	UserAgent string

	// DefaultContentType is the default Content-Type if not specified per request.
	// Common value: "application/x-protobuf"
	DefaultContentType string

	// DefaultContentEncoding is the default Content-Encoding if not specified per request.
	// Common value: "snappy"
	DefaultContentEncoding string

	// DefaultRemoteWriteVersion is the default X-Prometheus-Remote-Write-Version.
	// Common value: "0.1.0"
	DefaultRemoteWriteVersion string
}

Config configures the remote write client

type DurableClient

type DurableClient struct {
	// contains filtered or unexported fields
}

DurableClient persists encoded remote_write payloads locally and drains them in the background.

Accepted payloads are stored as record files under QueueDir, and the scheduler queue is used only to schedule record IDs for background delivery. This keeps accepted data durable even though the scheduler's read path is destructive.

Retry policy. The drain loop retries retryable failures indefinitely with exponential backoff (RetryConfig.MinWait → MaxWait, plus jitter). It honours Retry-After when present. RetryConfig.MaxRetries from the wrapped Client config is intentionally ignored — durable payloads have already been accepted on disk, so dropping them after N attempts would be surprising. Use Close (or cancel Run's context) to stop draining.

Platform support. Durable mode acquires an exclusive flock on the spool directory and is therefore Unix-only (Linux, macOS, *BSD). It is not supported on Windows.

func NewDurable

func NewDurable(cfg DurableConfig) (*DurableClient, error)

NewDurable creates a durable remote_write client.

func (*DurableClient) Close

func (dc *DurableClient) Close() error

Close stops the drain loop and closes the underlying queue and transport.

func (*DurableClient) Enqueue

func (dc *DurableClient) Enqueue(ctx context.Context, req DurableRequest) error

Enqueue persists a fully materialized remote_write payload.

func (*DurableClient) Push

func (dc *DurableClient) Push(ctx context.Context, req PushRequest) error

Push gathers metrics, encodes them into remote_write payloads, and enqueues them.

func (*DurableClient) PushTimeSeries

func (dc *DurableClient) PushTimeSeries(ctx context.Context, req PushTimeSeriesRequest) error

PushTimeSeries encodes pre-built series into remote_write payloads and enqueues them.

func (*DurableClient) Run

func (dc *DurableClient) Run(ctx context.Context) error

Run starts draining the durable queue until ctx is cancelled, Close is called, or a fatal error occurs. It is intended to be called once for the lifetime of the DurableClient.

type DurableConfig

type DurableConfig struct {
	// Client configures the underlying HTTP transport.
	Client Config

	// QueueDir is the base directory containing durable queue state.
	QueueDir string
	// QueueName scopes queue files within QueueDir.
	QueueName string
	// MaxInMemoryBlocks limits how many record references remain in memory before the
	// internal scheduler spills them to files.
	MaxInMemoryBlocks int
	// MaxPendingBytes bounds the total size of accepted spool records on disk.
	// When this budget is exceeded, Enqueue returns ErrDurableQueueBlocked.
	MaxPendingBytes int64
	// DisablePersistence disables normal file spillover for the internal scheduling queue.
	// Accepted durable spool records are still written to disk, and their scheduler
	// references may still be written during recovery or after acceptance.
	DisablePersistence bool

	// SendConcurrency controls how many queued payloads may be sent in parallel.
	SendConcurrency int

	// Logger receives durable queue lifecycle and failure logs.
	Logger *slog.Logger
	// Registerer receives durable Prometheus metrics. If nil, metrics are created but not registered.
	Registerer prometheus.Registerer
}

DurableConfig configures the persistent remote_write spooler.

type DurableRequest

type DurableRequest struct {
	// TenantID is used for tenant URL substitution or tenant header injection.
	TenantID string
	// BodyBytes contains the fully materialized request body to send or spool.
	BodyBytes []byte
	// ContentType is the Content-Type header for the upstream request.
	ContentType string
	// ContentEncoding is the Content-Encoding header for the upstream request.
	ContentEncoding string
	// RemoteWriteVersion is the X-Prometheus-Remote-Write-Version header value.
	RemoteWriteVersion string
	// ExtraHeaders are additional HTTP headers to forward upstream. It must not
	// contain protocol headers or headers managed by client configuration.
	ExtraHeaders http.Header
}

DurableRequest is a fully materialized remote_write request suitable for spooling.

type ForwardRequest

type ForwardRequest struct {
	// TenantID for URL substitution or header injection.
	TenantID string

	// Body is the request body. It will be closed by the transport.
	// Caller should not close it manually until the request is complete.
	Body io.ReadCloser

	// BodyBytes provides pre-buffered body bytes (best performance).
	BodyBytes []byte

	// GetBody returns a fresh body for retries.
	GetBody func() (io.ReadCloser, error)

	// ContentLength is the body size. Set to -1 (or leave 0 with a non-nil
	// Body) when the size is unknown — Forward treats 0 with a streaming Body
	// as "unknown" so net/http does not advertise an empty body.
	ContentLength int64

	// ContentType header value.
	ContentType string

	// ContentEncoding header value.
	ContentEncoding string

	// RemoteWriteVersion header value.
	RemoteWriteVersion string

	// ExtraHeaders are additional headers to forward. It must not contain
	// protocol headers or headers managed by client configuration.
	ExtraHeaders http.Header
}

ForwardRequest contains the data to forward to the upstream.

type PushRequest

type PushRequest struct {
	// TenantID is the tenant to push metrics for.
	TenantID string

	// Gatherer is the source of metrics.
	Gatherer prometheus.Gatherer

	// ExternalLabels are labels to add to every metric if not present.
	ExternalLabels map[string]string

	// WriteRelabelConfigs are relabeling rules applied before sending.
	// These can be used to filter, modify, or drop metrics.
	WriteRelabelConfigs []relabel.Config

	// Now returns the current time. If nil, time.Now() is used.
	Now func() time.Time

	// MaxBatchBytes is the target uncompressed batch size. Default: 3MB.
	// This is a soft target; a single metric family may push the batch over.
	MaxBatchBytes int

	// MaxSeriesPerBatch is a soft target for the number of series per batch.
	// Default: 10000. A batch may exceed this when the last appended metric
	// family pushes the count past the target — the series produced by a
	// single metric family are never split across batches.
	MaxSeriesPerBatch int

	// ExtraHeaders are additional headers to forward. It must not contain
	// protocol headers or headers managed by client configuration.
	ExtraHeaders http.Header
}

PushRequest contains the data for a push operation.

type PushTimeSeriesRequest

type PushTimeSeriesRequest struct {
	// TenantID is the tenant to push metrics for.
	TenantID string

	// TimeSeries are the series to send.
	TimeSeries []prompb.TimeSeries

	// MaxBatchBytes is the target uncompressed batch size. Default: 3MB.
	MaxBatchBytes int

	// MaxSeriesPerBatch is the maximum number of series per batch. Default: 10000.
	MaxSeriesPerBatch int

	// ExtraHeaders are additional headers to forward. It must not contain
	// protocol headers or headers managed by client configuration.
	ExtraHeaders http.Header
}

PushTimeSeriesRequest contains the data for pushing pre-built time series.

type RateLimiter

type RateLimiter struct {
	// contains filtered or unexported fields
}

RateLimiter implements a token bucket rate limiter for bytes per second.

The limiter refills perSecondLimit tokens every second. Register blocks when the bucket is empty.

func NewRateLimiter

func NewRateLimiter(perSecondLimit int64, stopCh <-chan struct{}) *RateLimiter

NewRateLimiter creates a new rate limiter with the given bytes per second limit. Pass stopCh to allow unblocking Register() when the limiter is no longer needed. If perSecondLimit <= 0, the limiter is disabled (Register is a no-op).

func (*RateLimiter) Enabled

func (rl *RateLimiter) Enabled() bool

Enabled returns true if the rate limiter is active.

func (*RateLimiter) Register

func (rl *RateLimiter) Register(ctx context.Context, n int64) error

Register blocks until n bytes can be sent under the rate limit or ctx is cancelled.

type RetryConfig

type RetryConfig struct {
	// MaxRetries is the maximum number of retries. Must be >= 1.
	MaxRetries int

	// MinWait is the minimum wait between retries. Default: 1s.
	MinWait time.Duration

	// MaxWait is the maximum wait between retries. Default: 30s.
	MaxWait time.Duration

	// MaxBodySize is the maximum body size to buffer for retries.
	// Bodies larger than this will fail with an error if retries are enabled
	// and content length is unknown. Default: 10MB.
	MaxBodySize int64
}

RetryConfig configures retry behavior.

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL