Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

support series relabeling on Thanos receiver #5391

Merged
merged 10 commits into from May 31, 2022
Merged

support series relabeling on Thanos receiver #5391

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
b72edac
support series relabeling on Thanos receiver
May 28, 2022
7cbd2e7
add changelog
May 28, 2022
54abfed
fix lint
May 28, 2022
2526c16
update lint
May 28, 2022
7a33da1
fix e2e test
May 28, 2022
fb00a3b
fix relabel config pass
May 28, 2022
65c3d26
cleanup white space
May 29, 2022
64502de
address review comments
May 30, 2022
3dd7fb5
address comments
May 30, 2022
fd07785
update comment
May 31, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Jump to
Jump to file
Failed to load files.
Diff view
Diff view
16 changes: 15 additions & 1 deletion cmd/thanos/receive.go
View file
Open in desktop
Expand Up @@ -5,6 +5,8 @@ package main

import (
"context"
"github.com/prometheus/prometheus/model/relabel"
yeya24 marked this conversation as resolved.
Show resolved Hide resolved
"gopkg.in/yaml.v2"
"io/ioutil"
"os"
"path"
Expand Down Expand Up @@ -173,6 +175,15 @@ func runReceive(
return errors.Wrapf(err, "migrate legacy storage in %v to default tenant %v", conf.dataDir, conf.defaultTenantID)
}

relabelContentYaml, err := conf.relabelConfigPath.Content()
GiedriusS marked this conversation as resolved.
Show resolved Hide resolved
if err != nil {
return errors.Wrap(err, "get content of relabel configuration")
}
var relabelConfig []*relabel.Config
if err := yaml.Unmarshal(relabelContentYaml, &relabelConfig); err != nil {
return errors.Wrap(err, "parsing relabel configuration")
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Tiiiiny nit: I think that by convention we use the present tense rather than the gerund when wrapping errors:

Suggested change
return errors.Wrap(err, "parsing relabel configuration")
return errors.Wrap(err, "parse relabel configuration")

}

dbs := receive.NewMultiTSDB(
conf.dataDir,
logger,
Expand Down Expand Up @@ -729,7 +740,8 @@ type receiveConfig struct {
ignoreBlockSize bool
allowOutOfOrderUpload bool

reqLogConfig *extflag.PathOrContent
reqLogConfig *extflag.PathOrContent
relabelConfigPath *extflag.PathOrContent
}

func (rc *receiveConfig) registerFlag(cmd extkingpin.FlagClause) {
Expand Down Expand Up @@ -783,6 +795,8 @@ func (rc *receiveConfig) registerFlag(cmd extkingpin.FlagClause) {

rc.forwardTimeout = extkingpin.ModelDuration(cmd.Flag("receive-forward-timeout", "Timeout for each forward request.").Default("5s").Hidden())

rc.relabelConfigPath = extflag.RegisterPathOrContent(cmd, "receive.relabel-config", "YAML file that contains relabelling configuration.", extflag.WithEnvSubstitution())
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Spelling nit: the correct form is the one used below in handler.go

Suggested change
rc.relabelConfigPath = extflag.RegisterPathOrContent(cmd, "receive.relabel-config", "YAML file that contains relabelling configuration.", extflag.WithEnvSubstitution())
rc.relabelConfigPath = extflag.RegisterPathOrContent(cmd, "receive.relabel-config", "YAML file that contains relabeling configuration.", extflag.WithEnvSubstitution())


rc.tsdbMinBlockDuration = extkingpin.ModelDuration(cmd.Flag("tsdb.min-block-duration", "Min duration for local TSDB blocks").Default("2h").Hidden())

rc.tsdbMaxBlockDuration = extkingpin.ModelDuration(cmd.Flag("tsdb.max-block-duration", "Max duration for local TSDB blocks").Default("2h").Hidden())
Expand Down
26 changes: 26 additions & 0 deletions pkg/receive/handler.go
View file
Open in desktop
Expand Up @@ -28,6 +28,7 @@ import (
"github.com/prometheus/client_golang/prometheus"
"github.com/prometheus/client_golang/prometheus/promauto"
"github.com/prometheus/common/route"
"github.com/prometheus/prometheus/model/relabel"
"github.com/prometheus/prometheus/storage"
"github.com/prometheus/prometheus/tsdb"
"google.golang.org/grpc"
Expand All @@ -38,6 +39,7 @@ import (
extpromhttp "github.com/thanos-io/thanos/pkg/extprom/http"
"github.com/thanos-io/thanos/pkg/runutil"
"github.com/thanos-io/thanos/pkg/server/http/middleware"
"github.com/thanos-io/thanos/pkg/store/labelpb"
"github.com/thanos-io/thanos/pkg/store/storepb"
"github.com/thanos-io/thanos/pkg/store/storepb/prompb"
"github.com/thanos-io/thanos/pkg/tracing"
Expand Down Expand Up @@ -81,6 +83,7 @@ type Options struct {
TLSConfig *tls.Config
DialOpts []grpc.DialOption
ForwardTimeout time.Duration
RelabelConfigs []*relabel.Config
}

// Handler serves a Prometheus remote write receiving HTTP endpoint.
Expand Down Expand Up @@ -353,6 +356,13 @@ func (h *Handler) receiveHTTP(w http.ResponseWriter, r *http.Request) {
return
}

// Apply relabeling configs.
h.relabel(&wreq)
if len(wreq.Timeseries) == 0 {
level.Debug(tLogger).Log("msg", "remote write request dropped due to relabeling.")
return
}

err = h.handleRequest(ctx, rep, tenant, &wreq)
if err != nil {
level.Debug(tLogger).Log("msg", "failed to handle request", "err", err)
Expand Down Expand Up @@ -682,6 +692,22 @@ func (h *Handler) RemoteWrite(ctx context.Context, r *storepb.WriteRequest) (*st
}
}

// relabel relabels the time series labels in the remote write request.
func (h *Handler) relabel(wreq *prompb.WriteRequest) {
if len(h.options.RelabelConfigs) > 0 {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Let's invert this check to make the code go from top to bottom instead of left to right:

if len(h.options.RelabelConfigs) == 0 {
  return
}
...

timeSeries := make([]prompb.TimeSeries, 0, len(wreq.Timeseries))
for _, ts := range wreq.Timeseries {
lbls := relabel.Process(labelpb.ZLabelsToPromLabels(ts.Labels), h.options.RelabelConfigs...)
if lbls == nil {
continue
}
ts.Labels = labelpb.ZLabelsFromPromLabels(lbls)
timeSeries = append(timeSeries, ts)
}
wreq.Timeseries = timeSeries
}
}

// isConflict returns whether or not the given error represents a conflict.
func isConflict(err error) bool {
if err == nil {
Expand Down