exp/api: Add experimental exp module; Add remote API with write client and handler. #1658

bwplotka · 2024-10-21T13:01:51Z

This proposes Remote Write API directly in client_golang main module. No dependency added (some new small packages are added under the internal vendored code).

Old attempt: #1656

Both API (client) and handler supports 1.0 and 2.0 Proto messages, but I intentionally only host writev2.Request generated Go code as we plan to deprecate 1.0 and users should use writev2 now. Both API and handler also will work against protobuf message types that use custom generators e.g. with gogoproto or anything else.

TODO:

Add examples, documentation
Open PR that uses it on Prometheus

Signed-off-by: bwplotka <bwplotka@gmail.com>

api/prometheus/v1/remote/remote_api.go

jmichalek132 · 2024-10-21T18:07:00Z

api/prometheus/v1/remote/genproto/v2/symbols.go

+}
+
+// DesymbolizeLabels decodes label references, with given symbols to labels.
+func DesymbolizeLabels(labelRefs []uint32, symbols, buf []string) []string {


yeah I just ran into DesymbolizeLabels not being public in prometheus codebase here open-telemetry/opentelemetry-collector-contrib#35751 (comment)

Do you use labels.Labels or some other struct (for labels in Otel)

In the translation layer we have createAttributes https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/pkg/translator/prometheusremotewrite/helper.go#L99 which returns []prompb.Label and here we store them https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/pkg/translator/prometheusremotewrite/metrics_to_prw_v2.go#L119

bwplotka · 2024-10-22T07:54:18Z

api/prometheus/v1/remote/remote_api.go

+
+// NewRemoteWriteHandler returns HTTP handler that receives Remote Write 2.0
+// protocol https://prometheus.io/docs/specs/remote_write_spec_2_0/.
+func NewRemoteWriteHandler(logger *slog.Logger, store writeStorage) http.Handler {


It's bit inconsistent with NewRemoteAPI in terms of options, let's unify one to one or another 🤔

bwplotka · 2024-10-22T10:23:41Z

api/prometheus/v1/remote/remote_headers.go

+// WriteProtoFullName represents the fully qualified name of the protobuf message
+// to use in Remote write 1.0 and 2.0 protocols.
+// See https://prometheus.io/docs/specs/remote_write_spec_2_0/#protocol.
+type WriteProtoFullName protoreflect.FullName


nit: Change to string, bit too hard to use

bwplotka · 2024-10-22T10:24:06Z

api/prometheus/v1/remote/remote_headers.go

+const (
+	// WriteProtoFullNameV1 represents the `prometheus.WriteRequest` protobuf
+	// message introduced in the https://prometheus.io/docs/specs/remote_write_spec/.
+	// DEPRECATED: Use WriteProtoFullNameV2 instead.


For receivers who wants to support it for longer, deprecation might cause some friction, but maybe worth it? 🤔

bwplotka · 2024-10-22T11:32:54Z

Example use of handler in sink project: https://github.com/bwplotka/sink/blob/main/go/sink/main.go#L51

cstyan

Both API (client) and handler supports 1.0 and 2.0 Proto messages, but I intentionally only host writev2.Request generated Go code as we plan to deprecate 1.0 and users should use writev2 now. Both API and handler also will work against protobuf message types that use custom generators e.g. with gogoproto or anything else.

Given that the spec is only just finalized recently and (afaik) still no receivers like Mimir support v2, we should include v1 request code imo.

cstyan · 2024-10-25T18:29:09Z

api/prometheus/v1/remote/remote_api.go

+}
+
+func (r *API) attemptWrite(ctx context.Context, compr Compression, proto WriteProtoFullName, payload []byte, attempt int) (WriteResponseStats, error) {
+	u := r.client.URL("api/v1/write", nil)


this should be configurable imo, not all remote write receivers use the same endpoint

cstyan · 2024-10-25T18:29:54Z

api/prometheus/v1/remote/remote_api.go

+	rs, err := parseWriteResponseStats(resp)
+	if err != nil {
+		r.opts.logger.Warn("parsing rw write statistics failed; partial or no stats", "err", err)
+	}


should we only try to get write response stats/log a warning if the request type was v2?

cstyan · 2024-10-25T18:32:03Z

api/prometheus/v1/remote/remote_api.go

+	return &handler{logger: logger, store: store}
+}
+
+func parseProtoMsg(contentType string) (WriteProtoFullName, error) {


a comment with an example of what the string might look like would be useful for people who haven't read or fully understood the v2 spec

cstyan · 2024-10-25T18:39:27Z

internal/github.com/efficientgo/core/backoff/backoff.go

+// Copyright (c) The EfficientGo Authors.
+// Licensed under the Apache License 2.0.
+
+// Initially copied from Cortex project.
+
+// Package backoff implements backoff timers which increases wait time on every retry, incredibly useful
+// in distributed system timeout functionalities.


should we be pulling this via go.mod as opposed to having copies of the files? or is this the way we include dependencies for the client library?

We try as hard as we can to avoid dependencies. I'm in favor of copying other libraries code (respecting copyright) if possible

yeya24 · 2024-11-17T19:55:46Z

api/prometheus/v1/remote/remote_api.go

+	}
+
+	// Read the request body.
+	body, err := io.ReadAll(r.Body)


It might be out of scope for Prometheus but it would be nice to limit the request body size to protect the handler. https://github.com/cortexproject/cortex/blob/master/pkg/util/http.go#L180

yeya24 · 2024-11-17T19:58:08Z

api/prometheus/v1/remote/remote_api.go

+		return
+	}
+
+	decompressed, err := snappy.Decode(nil, body)


Similar here. It would be nice if we can check snappy decoded length and see if it exceeds the max size https://github.com/cortexproject/cortex/blob/master/pkg/util/http.go#L247

yeya24 · 2024-11-17T20:05:00Z

api/prometheus/v1/remote/remote_api.go

+	return &handler{logger: logger, store: store}
+}
+
+func parseProtoMsg(contentType string) (WriteProtoFullName, error) {


Is this something we can expose?

+1 to exposing this

yeya24 · 2024-11-17T20:16:15Z

api/prometheus/v1/remote/remote_api.go

+		h.logger.Error("Error decompressing remote write request", "err", err.Error())
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}


I wonder if we can expose a hook for the logic of decoding and decompressing part so that downstream project can just use their existing code without many adjustment.

func decodeRequestBody(r *http.Request) (body []byte, err error)

The main logic that can benefit us is handling v1 and v2 proto and response headers. It would be nice to still reuse the other code.

Yup some requestValidatorDecoder interface that can be overridden downstream would be cool to have.

saswatamcode · 2024-12-16T09:36:13Z

api/prometheus/v1/remote/remote_api.go

+	}
+
+	req.Header.Add("Content-Encoding", string(compr))
+	req.Header.Set("Content-Type", contentTypeHeader(proto))


Maybe there should be an option to specify custom headers here too?

You can do that with tripperware though 🤔

saswatamcode · 2024-12-16T09:43:26Z

api/prometheus/v1/remote/remote_api.go

+
+	err = fmt.Errorf("server returned HTTP status %s: %s", resp.Status, body)
+	if resp.StatusCode/100 == 5 ||
+		(r.opts.retryOnRateLimit && resp.StatusCode == http.StatusTooManyRequests) {


Should retryOn429 be configurable here?

saswatamcode · 2025-01-04T12:30:25Z

api/prometheus/v1/remote/remote_api.go

+		h.logger.Error("Error decompressing remote write request", "err", err.Error())
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}


Yup some requestValidatorDecoder interface that can be overridden downstream would be cool to have.

saswatamcode · 2025-01-04T12:42:54Z

api/prometheus/v1/remote/genproto/v2/types.pb.go

Maybe more of a future comment, but we can probably have some common/model codec here too, since we already have the types, like https://github.com/prometheus/prometheus/blob/main/prompb/io/prometheus/write/v2/codec.go, that would translate to common/model types. Would be good for downstream clients, that can choose to not import prometheus at all.

saswatamcode · 2025-01-04T12:44:27Z

api/prometheus/v1/remote/remote_api.go

+	return &handler{logger: logger, store: store}
+}
+
+func parseProtoMsg(contentType string) (WriteProtoFullName, error) {


+1 to exposing this

saswatamcode · 2025-01-04T12:50:42Z

api/prometheus/v1/remote/remote_api.go

+			code = http.StatusInternalServerError
+		}
+		if code/5 == 100 { // 5xx
+			h.logger.Error("Error while remote writing the v2 request", "err", storeErr.Error())


store.Store can ingest both I think

Suggested change

h.logger.Error("Error while remote writing the v2 request", "err", storeErr.Error())

h.logger.Error("Error while storing the remote write request", "err", storeErr.Error())

api/prometheus/v1/remote/remote_api.go

* Address remaining feedback Signed-off-by: Saswata Mukherjee <saswataminsta@yahoo.com> * Make Write message type more flexible Signed-off-by: Saswata Mukherjee <saswataminsta@yahoo.com> --------- Signed-off-by: Saswata Mukherjee <saswataminsta@yahoo.com>

* Move remote write API to client_golang/exp Signed-off-by: Saswata Mukherjee <saswataminsta@yahoo.com> * Don't use api.Client structs, add options for middleware Signed-off-by: Saswata Mukherjee <saswataminsta@yahoo.com> * Fix reqBuf usage Signed-off-by: Saswata Mukherjee <saswataminsta@yahoo.com> * Fix url path Signed-off-by: Saswata Mukherjee <saswataminsta@yahoo.com> * Add separate mod file (and workspace file) Signed-off-by: Saswata Mukherjee <saswataminsta@yahoo.com> * Hook exp tests fmt; Test handler error case; Configure backoff Signed-off-by: Saswata Mukherjee <saswataminsta@yahoo.com> --------- Signed-off-by: Saswata Mukherjee <saswataminsta@yahoo.com>

Signed-off-by: Bartlomiej Plotka <bwplotka@gmail.com>

bwplotka · 2025-02-03T10:14:20Z

exp/api/remote/remote_api.go

+//
+// It is not safe to use the returned API from multiple goroutines, create a
+// separate *API for each goroutine.
+func NewAPI(client *http.Client, baseURL string, opts ...APIOption) (*API, error) {


healthy default for client?

bwplotka · 2025-02-03T10:19:29Z

exp/api/remote/remote_api.go

+//     will be used
+//   - If neither is supported, it will marshaled using generic google.golang.org/protobuf methods and
+//     error out on unknown scheme.
+func (r *API) Write(ctx context.Context, msg any) (_ WriteResponseStats, err error) {


It might be more future proof to allow setting content type here (with our constant).

bwplotka · 2025-02-03T10:20:38Z

exp/api/remote/remote_api.go

+//     error out on unknown scheme.
+func (r *API) Write(ctx context.Context, msg any) (_ WriteResponseStats, err error) {
+	// Reset the buffer.
+	r.reqBuf = r.reqBuf[:0]


This is good for single thread, but people might want to use the same http client etc for multiple concurrent streams, so we probably can't do this. I would remove and in future add (without breaking variadic options for write, we could do it now too it's a common with WithWriteBuffer)

bwplotka · 2025-02-03T10:26:51Z

exp/api/remote/remote_api.go

+
+// NewRemoteWriteHandler returns HTTP handler that receives Remote Write 2.0
+// protocol https://prometheus.io/docs/specs/remote_write_spec_2_0/.
+func NewRemoteWriteHandler(store writeStorage, opts ...HandlerOption) http.Handler {


Suggested change

func NewRemoteWriteHandler(store writeStorage, opts ...HandlerOption) http.Handler {

func NewHandler(store writeStorage, opts ...HandlerOption) http.Handler {

bwplotka · 2025-02-03T10:28:43Z

exp/api/remote/remote_api.go

+// writeStorage represents the storage for RemoteWriteHandler.
+// This interface is intentionally private due its experimental state.
+type writeStorage interface {
+	Store(ctx context.Context, proto WriteProtoFullName, serializedRequest []byte) (_ WriteResponseStats, code int, _ error)


Discussion, should it by any and people provide a type for NewHandler? This feels tricky as user would need to give us a map of content type -> type and the exacty method to use for parsing, kind of sily.

However, if we don't do this, should we just give http.Request?

Let's focus on value we bring vs write your own ServeHTTP:

We write write response stats for you (limiting error surface) with correct success response code

We validate headers and respond accordingly

We decompress for you

After that we can't do much.

Disadventages of http.Request

user don't know what we validated, we can write down we validated proto full name, but should we give it in a separate parameter or double write with both param and header, or only header

should we remove proto and compression (encoding) headers from request? What if user want to have a metric that instruments what compression was used (then, they could use middleware).

We kind of have to put decompressed content into io.Reader of request to avoid duplicate data.

We have to give options to limit known ctypes to X.

Suggested change

Store(ctx context.Context, proto WriteProtoFullName, serializedRequest []byte) (_ WriteResponseStats, code int, _ error)

// Store stores remote write metrics encoded in the given WriteFullName content type. Provided http.Request contains the encoded bytes in the req.Body with all the HTTP information, except "Content-Type" header which is provided in a separate, validated ctype. Other headers might be trimmed,

// depending on the configured middlewares e.g. a default SnappyMiddleware trims "Content-Encoding".

//

// Store blocks an HTTP request and waits for WriteResponseStats to provide to the client.

Store(ctx context.Context, ctype WriteFullName, req *http.Request) (_ WriteResponseStats, code int, _ error)

Questions:

How to attach extra response headers e.g. Retry-After.

For questions, we might want to create more complex WriteResponse type that gives a way to return stats (force user ideally) and header. However, should we HTTP style input param?

Suggested change

Store(ctx context.Context, proto WriteProtoFullName, serializedRequest []byte) (_ WriteResponseStats, code int, _ error)

// Store stores remote write metrics encoded in the given WriteFullName content type. Provided http.Request contains the encoded bytes in the req.Body with all the HTTP information, except "Content-Type" header which is provided in a separate, validated ctype. Other headers might be trimmed,

// depending on the configured middlewares e.g. a default SnappyMiddleware trims "Content-Encoding".

//

// Store blocks an HTTP request and waits for WriteResponseStats to provide to the client.

Store(ctx context.Context, ctype WriteFullName, req *http.Request) (_ *WriteResponse, _ error)

type Stats struct { // Samples represents X-Prometheus-Remote-Write-Written-Samples Samples int // Histograms represents X-Prometheus-Remote-Write-Written-Histograms Histograms int // Exemplars represents X-Prometheus-Remote-Write-Written-Exemplars Exemplars int // Confirmed means we can trust those statistics from the point of view // of the PRW 2.0 spec. When parsed from headers, it means we got at least one // response header from the Receiver to confirm those numbers, meaning it must // be at least 2.0 Receiver. See ParseWriteResponseStats for details. confirmed bool } func StatsFromV2() Stats func NewWriteResponse(s Stats) *WriteResponse { .... } func (r *WriteResponse) SetHeaders func (r *WriteResponse) SetCode

bwplotka · 2025-02-03T10:52:24Z

exp/api/remote/remote_headers.go

+}
+
+var contentTypeHeaders = map[WriteProtoFullName]string{
+	WriteProtoFullNameV1: appProtoContentType, // Also application/x-protobuf;proto=prometheus.WriteRequest but simplified for compatibility with 1.x spec.


Let's make it WriteV1ContentType

bwplotka

Thanks @saswatamcode for helping a lot. Let's get those last things to this branch + quick README and let's get it merged in the experimental module!

GiedriusS · 2025-02-04T10:21:39Z

exp/api/remote/remote_api.go

+			}
+
+			// Read the request body.
+			bodyBytes, err := io.ReadAll(r.Body)


I think there's an opportunity to use a sync.Pool here.

GiedriusS · 2025-02-04T10:24:43Z

exp/api/remote/remote_api.go

+	}
+
+	// Read the already decompressed body
+	body, err := io.ReadAll(r.Body)


By using a bytes.Buffer instead of a bytes.Reader we could access the original bytes directly through Bytes() and avoid a copy/allocation here. 99% of users if not all will use the middleware that does the decompression 😄

bwplotka requested review from cstyan and ArthurSens October 21, 2024 13:01

bwplotka mentioned this pull request Oct 21, 2024

api: Create a new Go module for API; with remote write client and handler. #1656

Closed

3 tasks

bwplotka force-pushed the prwclient-em branch 4 times, most recently from 17b12a6 to 1985d61 Compare October 21, 2024 13:14

api: Add remote API with write client; add remote handler.

bffa922

Signed-off-by: bwplotka <bwplotka@gmail.com>

bwplotka force-pushed the prwclient-em branch from 1985d61 to bffa922 Compare October 21, 2024 13:18

bwplotka commented Oct 21, 2024

View reviewed changes

api/prometheus/v1/remote/remote_api.go Outdated Show resolved Hide resolved

jmichalek132 reviewed Oct 21, 2024

View reviewed changes

bwplotka commented Oct 22, 2024

View reviewed changes

cstyan reviewed Oct 25, 2024

View reviewed changes

jmichalek132 mentioned this pull request Nov 13, 2024

Make DesymbolizeLabels func public prometheus/prometheus#15395

Closed

yeya24 mentioned this pull request Nov 14, 2024

Support remote write v2 by converting request cortexproject/cortex#6330

Open

3 tasks

SungJin1212 mentioned this pull request Nov 15, 2024

Prometheus Remote Write v2 Implementation cortexproject/cortex#6324

Open

6 tasks

yeya24 reviewed Nov 17, 2024

View reviewed changes

bwplotka mentioned this pull request Nov 27, 2024

Promsafe: Strongly-typed safe labels #1598

Open

saswatamcode reviewed Dec 16, 2024

View reviewed changes

saswatamcode mentioned this pull request Jan 3, 2025

Support remote write 2.0 on receive thanos-io/thanos#8033

Draft

2 tasks

saswatamcode reviewed Jan 4, 2025

View reviewed changes

saswatamcode reviewed Jan 16, 2025

View reviewed changes

api/prometheus/v1/remote/remote_api.go Outdated Show resolved Hide resolved

saswatamcode mentioned this pull request Jan 17, 2025

Make Write message type more flexble, address some feedback #1710

Merged

saswatamcode added 2 commits January 17, 2025 11:24

bwplotka changed the title ~~api: Add remote API with write client; add remote handler.~~ exp/api: Add experimental exp module; Add remote API with write client and handler. Jan 31, 2025

Merge branch 'main' into prwclient-em

49dc00b

Signed-off-by: Bartlomiej Plotka <bwplotka@gmail.com>

bwplotka commented Feb 3, 2025

View reviewed changes

bwplotka marked this pull request as ready for review February 3, 2025 10:50

bwplotka requested review from kakkoyun and vesari as code owners February 3, 2025 10:50

bwplotka commented Feb 3, 2025

View reviewed changes

GiedriusS reviewed Feb 4, 2025

View reviewed changes

	h.logger.Error("Error while remote writing the v2 request", "err", storeErr.Error())
	h.logger.Error("Error while storing the remote write request", "err", storeErr.Error())

	func NewRemoteWriteHandler(store writeStorage, opts ...HandlerOption) http.Handler {
	func NewHandler(store writeStorage, opts ...HandlerOption) http.Handler {

-	Store(ctx context.Context, proto WriteProtoFullName, serializedRequest []byte) (_ WriteResponseStats, code int, _ error)
+        // Store stores remote write metrics encoded in the given WriteFullName content type. Provided http.Request contains the encoded bytes in the req.Body with all the HTTP information, except "Content-Type" header which is provided in a separate, validated ctype. Other headers might be trimmed,
+        // depending on the configured middlewares e.g. a default SnappyMiddleware trims "Content-Encoding".
+        //
+        // Store blocks an HTTP request and waits for WriteResponseStats to provide to the client.
+	Store(ctx context.Context, ctype WriteFullName, req *http.Request) (_ WriteResponseStats, code int, _ error)

exp/api: Add experimental exp module; Add remote API with write client and handler. #1658

Are you sure you want to change the base?

exp/api: Add experimental exp module; Add remote API with write client and handler. #1658

Conversation

bwplotka commented Oct 21, 2024 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bwplotka commented Oct 22, 2024

cstyan left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

bwplotka left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

GiedriusS Feb 4, 2025 • edited Loading

Choose a reason for hiding this comment

bwplotka commented Oct 21, 2024 •

edited

Loading

GiedriusS Feb 4, 2025 •

edited

Loading