add influx push endpoint to mimir #10153
Conversation
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
|
|
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
pkg/api/api.go
Outdated
| // TODO(alexg): hidden behind a featureflag or experimental config option? | ||
| a.RegisterRoute(InfluxPushEndpoint, distributor.InfluxHandler( |
There was a problem hiding this comment.
I don't think a feature flag for this is needed. We can just state in the docs (about-versioning.md) that the endpoint is experimental.
pkg/distributor/influx.go
Outdated
| level.Debug(spanLogger).Log( | ||
| "msg", "decodeAndConvert complete", | ||
| "bytesRead", bytesRead, | ||
| ) |
There was a problem hiding this comment.
Opinionated style nit: I don't think we need 4 lines for this debug log.
pkg/distributor/influx.go
Outdated
| "bytesRead", bytesRead, | ||
| ) | ||
| if err != nil { | ||
| level.Error(logger).Log("err", err.Error()) |
There was a problem hiding this comment.
Does this logger have all the required context? Can you also add some context here about what was going on when this happened? I'm scared of finding this log:
ts=2024-12-17 err="unexpected EOF"
Also, nit, .Error() call is not needed, just pass err.
There was a problem hiding this comment.
Agreed, fixed in next push.
pkg/distributor/influx.go
Outdated
| "github.com/grafana/dskit/grpcutil" | ||
| "github.com/grafana/dskit/httpgrpc" | ||
| "github.com/grafana/dskit/middleware" | ||
| io2 "github.com/influxdata/influxdb/v2/kit/io" |
There was a problem hiding this comment.
| io2 "github.com/influxdata/influxdb/v2/kit/io" | |
| influxio "github.com/influxdata/influxdb/v2/kit/io" |
There was a problem hiding this comment.
Agreed, fixed in next push.
pkg/distributor/influx.go
Outdated
| "github.com/grafana/mimir/pkg/util/spanlogger" | ||
| ) | ||
|
|
||
| func parser(ctx context.Context, r *http.Request, maxSize int, _ *util.RequestBuffers, req *mimirpb.PreallocWriteRequest, logger log.Logger) error { |
There was a problem hiding this comment.
This function lives in pkg/distributor, I would say it's a little bit pretentious to take the name parser for this :D
How about influxRequestParser?
There was a problem hiding this comment.
Agreed, fixed in next push.
pkg/distributor/influx.go
Outdated
| // TODO(alexg): Do we even need httpgrpc here? | ||
| // Check for httpgrpc error, default to client error if parsing failed |
There was a problem hiding this comment.
I don't see any httpgrpc errors being returned by parser
There was a problem hiding this comment.
Oops, should have got rid of that one. Will resolve once I work out best way to wrap existing error in the StatusBadRequest.
There was a problem hiding this comment.
Ah, httpgprc is used to smuggle both the http status code and the error message out of the supplier() function. I've removed the misleading comment.
pkg/distributor/influxpush/parser.go
Outdated
| (charInt >= 48 && charInt <= 57) || // 0-9 | ||
| charInt == 95) { // _ | ||
|
|
||
| *in = (*in)[:charIndex] + "_" + (*in)[charIndex+1:] |
There was a problem hiding this comment.
This allocates a new string (hopefully just one, compiler should be clever) for each one of the invalid characters. How about you modify the bytes slice before transforming it to a string?
There was a problem hiding this comment.
Agreed. Done in latest commit.
pkg/distributor/influxpush/parser.go
Outdated
| // analog of invalidChars = regexp.MustCompile("[^a-zA-Z0-9_]") | ||
| func replaceInvalidChars(in *string) { | ||
| for charIndex, char := range *in { | ||
| charInt := int(char) |
There was a problem hiding this comment.
Why do you need it to be an int?
There was a problem hiding this comment.
It doesn't, removed unnecessary cast.
pkg/distributor/influxpush/parser.go
Outdated
| func replaceInvalidChars(in *string) { | ||
| for charIndex, char := range *in { | ||
| charInt := int(char) | ||
| if !((charInt >= 97 && charInt <= 122) || // a-z |
There was a problem hiding this comment.
If you use 'a' instead of 97 (is it correct?) you won't need the comment and the code will be less prone to have bugs.
There was a problem hiding this comment.
Moved to literal chars instead of ASCII codes.
pkg/distributor/influxpush/parser.go
Outdated
| } | ||
| } | ||
| // prefix with _ if first char is 0-9 | ||
| if int((*in)[0]) >= 48 && int((*in)[0]) <= 57 { |
There was a problem hiding this comment.
Should check first that *in isn't empty, otherwise it would panic.
There was a problem hiding this comment.
The values come from the influx library which won't allow an empty string to be created, but I've added an extra check just in case. Also added some extra tests to ensure the assumption about the influx library holds.
pkg/distributor/influxpush/parser.go
Outdated
| key := string(tag.Key) | ||
| if key == "__name__" || key == internalLabel { | ||
| continue | ||
| } | ||
| replaceInvalidChars(&key) | ||
| lbls = append(lbls, mimirpb.LabelAdapter{ | ||
| Name: key, | ||
| Value: string(tag.Value), |
There was a problem hiding this comment.
We can do yoloString here instead of string(), as the push code is crafted to avoid keeping references to the strings from LabelAdapter
There was a problem hiding this comment.
Interesting! Fixed as per suggestion. I take it no other strings are guaranteed not to be referenced and so it can't be used anywhere else?
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
|
OK, comments left to deal with
|
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
…trings from LabelAdapter Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
Signed-off-by: alexgreenbank <[email protected]>
WIP - early draft - do not bother looking at yet
Early commit(s) are just importing the original code from #1971 which will not compile within latest codebase.
--
What this PR does
Which issue(s) this PR fixes or relates to
Fixes #
Checklist
CHANGELOG.mdupdated - the order of entries should be[CHANGE],[FEATURE],[ENHANCEMENT],[BUGFIX].about-versioning.mdupdated with experimental features.