api: Add remote API with write client; add remote handler. #1658
Conversation
bwplotka
force-pushed
the
prwclient-em
branch
4 times, most recently
from
17b12a6
to
1985d61
Compare
Signed-off-by: bwplotka <[email protected]>
bwplotka
force-pushed
the
prwclient-em
branch
from
1985d61
to
bffa922
Compare
| // | ||
| // It is not safe to use the returned API from multiple goroutines, create a | ||
| // separate *API for each goroutine. | ||
| func NewAPI(c api.Client, opts ...APIOption) (*API, error) { |
There was a problem hiding this comment.
We could consider removing error from here and from APIOption as we don't have any option with validation or so. However I added it for future compatibility (we won't be able to easily extend it without breaking compatibility). Note: Currently api/ is experimenta... 🙈 I think I will remove error here and hope for the best, YAGNI.
| func NewAPI(c api.Client, opts ...APIOption) (*API, error) { | |
| func NewAPI(c api.Client, opts ...APIOption) *API { |
| } | ||
|
|
||
| // DesymbolizeLabels decodes label references, with given symbols to labels. | ||
| func DesymbolizeLabels(labelRefs []uint32, symbols, buf []string) []string { |
There was a problem hiding this comment.
yeah I just ran into DesymbolizeLabels not being public in prometheus codebase here open-telemetry/opentelemetry-collector-contrib#35751 (comment)
There was a problem hiding this comment.
Do you use labels.Labels or some other struct (for labels in Otel)
There was a problem hiding this comment.
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
|
|
||
| // 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 { |
There was a problem hiding this comment.
It's bit inconsistent with NewRemoteAPI in terms of options, let's unify one to one or another 🤔
| // 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 |
There was a problem hiding this comment.
nit: Change to string, bit too hard to use
| const ( | ||
| // WriteProtoFullNameV1 represents the `prometheus.WriteRequest` protobuf | ||
| // message introduced in the https://prometheus.io/docs/specs/remote_write_spec/. | ||
| // DEPRECATED: Use WriteProtoFullNameV2 instead. |
There was a problem hiding this comment.
For receivers who wants to support it for longer, deprecation might cause some friction, but maybe worth it? 🤔
|
Example use of handler in |
There was a problem hiding this comment.
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.
| } | ||
|
|
||
| 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) |
There was a problem hiding this comment.
this should be configurable imo, not all remote write receivers use the same endpoint
| rs, err := parseWriteResponseStats(resp) | ||
| if err != nil { | ||
| r.opts.logger.Warn("parsing rw write statistics failed; partial or no stats", "err", err) | ||
| } |
There was a problem hiding this comment.
should we only try to get write response stats/log a warning if the request type was v2?
| return &handler{logger: logger, store: store} | ||
| } | ||
|
|
||
| func parseProtoMsg(contentType string) (WriteProtoFullName, error) { |
There was a problem hiding this comment.
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
| // 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. |
There was a problem hiding this comment.
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?
There was a problem hiding this comment.
We try as hard as we can to avoid dependencies. I'm in favor of copying other libraries code (respecting copyright) if possible
| } | ||
|
|
||
| // Read the request body. | ||
| body, err := io.ReadAll(r.Body) |
There was a problem hiding this comment.
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
| return | ||
| } | ||
|
|
||
| decompressed, err := snappy.Decode(nil, body) |
There was a problem hiding this comment.
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
| return &handler{logger: logger, store: store} | ||
| } | ||
|
|
||
| func parseProtoMsg(contentType string) (WriteProtoFullName, error) { |
There was a problem hiding this comment.
Is this something we can expose?
| h.logger.Error("Error decompressing remote write request", "err", err.Error()) | ||
| http.Error(w, err.Error(), http.StatusBadRequest) | ||
| return | ||
| } |
There was a problem hiding this comment.
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.
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.Requestgenerated Go code as we plan to deprecate 1.0 and users should usewritev2now. Both API and handler also will work against protobuf message types that use custom generators e.g. with gogoproto or anything else.TODO: