feat: added loki streams processing at /loki/push

Signed-off-by: Lukas Schmidt <[email protected]>

Author: MrWong99

README.md

@@ -4,7 +4,7 @@
 [![Coverage Status](https://coveralls.io/repos/github/blind-oracle/cortex-tenant/badge.svg?branch=main)](https://coveralls.io/github/blind-oracle/cortex-tenant?branch=main)
 [![Artifact Hub](https://img.shields.io/endpoint?url=https://artifacthub.io/badge/repository/cortex-tenant)](https://artifacthub.io/packages/helm/cortex-tenant/cortex-tenant)
 
-Prometheus remote write proxy which marks timeseries with a Cortex/Mimir tenant ID based on labels.
+Prometheus/Loki remote write proxy which marks timeseries with a Cortex/Mimir/Loki tenant ID based on labels.
 
 ## Status
 
@@ -16,19 +16,20 @@ The project is stable and (hopefully) almost bug-free. I will accept PRs, but pr
 
 ## Overview
 
-Cortex/Mimir tenants (separate namespaces where metrics are stored to and queried from) are identified by `X-Scope-OrgID` HTTP header on both writes and queries.
+Cortex/Mimir/Loki tenants (separate namespaces where metrics are stored to and queried from) are identified by `X-Scope-OrgID` HTTP header on both writes and queries.
 
 ~~Problem is that Prometheus can't be configured to send this header~~ Actually in some recent version (year 2021 onwards) this functionality was added, but the tenant is the same for all jobs. This makes it impossible to use a single Prometheus (or an HA pair) to write to multiple tenants.
 
 This software solves the problem using the following logic:
 
-- Receive Prometheus remote write
-- Search each timeseries for a specific label name and extract a tenant ID from its value.
+- Receive Prometheus remote write or [Loki push](https://grafana.com/docs/loki/latest/reference/loki-http-api/#ingest-logs)
+- Search each timeseries/stream for a specific label name and extract a tenant ID from its value.
+  For Loki the stream labels are preferred over [structured metadata](https://grafana.com/docs/loki/latest/get-started/labels/structured-metadata/) labels.
   If the label wasn't found then it can fall back to a configurable default ID.
-  If none is configured then the write request will be rejected with HTTP code 400
-- Optionally removes this label from the timeseries
-- Groups timeseries by tenant
-- Issues a number of parallel per-tenant HTTP requests to Cortex/Mimir with the relevant tenant HTTP header (`X-Scope-OrgID` by default)
+  If none is configured then the write/push request will be rejected with HTTP code 400
+- Optionally removes this label from the timeseries/stream
+- Groups timeseries/streams by tenant
+- Issues a number of parallel per-tenant HTTP requests to Cortex/Mimir/Loki with the relevant tenant HTTP header (`X-Scope-OrgID` by default)
 
 ## Usage
 
@@ -38,6 +39,7 @@ This software solves the problem using the following logic:
 
 - GET `/alive` returns 200 by default and 503 if the service is shutting down (if `timeout_shutdown` setting is > 0)
 - POST `/push` receives metrics from Prometheus - configure remote write to send here
+- POST `/loki/push` receives logs from Loki - configure push to send here
 
 ### Configuration
 
@@ -59,6 +61,10 @@ listen_pprof: 0.0.0.0:7008
 # env: CT_TARGET
 target: http://127.0.0.1:9091/receive
 
+# Where to send the modified requests (Loki)
+# env: CT_TARGET_LOKI
+target_loki: http://127.0.0.1:3100/loki/api/v1/push
+
 # Whether to enable querying for IPv6 records
 # env: CT_ENABLE_IPV6
 enable_ipv6: false
@@ -137,10 +143,11 @@ tenant:
     - tenant
     - other_tenant
 
-  # Whether to remove the tenant label from the request
+  # Whether to remove the tenant label from the request.
+  # Has no effect for Loki stream messages, they are kept as is.
   # env: CT_TENANT_LABEL_REMOVE
   label_remove: true
-  
+
   # To which header to add the tenant ID
   # env: CT_TENANT_HEADER
   header: X-Scope-OrgID

config.go

@@ -3,6 +3,7 @@ package main
 import (
 	"fmt"
 	"os"
+	"slices"
 	"time"
 
 	"github.com/caarlos0/env/v8"
@@ -18,6 +19,7 @@ type config struct {
 	MetricsIncludeTenant bool   `yaml:"metrics_include_tenant" env:"CT_METRICS_INCLUDE_TENANT"`
 
 	Target     string `env:"CT_TARGET"`
+	TargetLoki string `yaml:"target_loki" env:"CT_TARGET_LOKI"`
 	EnableIPv6 bool   `yaml:"enable_ipv6" env:"CT_ENABLE_IPV6"`
 
 	LogLevel          string        `yaml:"log_level" env:"CT_LOG_LEVEL"`
@@ -86,6 +88,10 @@ func configLoad(file string) (*config, error) {
 		cfg.Target = "127.0.0.1:9090"
 	}
 
+	if cfg.TargetLoki == "" {
+		cfg.TargetLoki = "127.0.0.1:3100"
+	}
+
 	if cfg.Timeout == 0 {
 		cfg.Timeout = 10 * time.Second
 	}
@@ -109,6 +115,9 @@ func configLoad(file string) (*config, error) {
 	// Default to the Label if list is empty
 	if len(cfg.Tenant.LabelList) == 0 {
 		cfg.Tenant.LabelList = append(cfg.Tenant.LabelList, cfg.Tenant.Label)
+	} else {
+		// Reverse entries to always prefer last label in list when found
+		slices.Reverse(cfg.Tenant.LabelList)
 	}
 
 	if cfg.Auth.Egress.Username != "" {

go.mod

@@ -1,41 +1,137 @@
 module github.com/blind-oracle/cortex-tenant
 
-go 1.23
+go 1.25.0
 
 require (
 	github.com/blind-oracle/go-common v1.0.7
 	github.com/caarlos0/env/v8 v8.0.0
 	github.com/gogo/protobuf v1.3.2
-	github.com/golang/snappy v0.0.4
+	github.com/golang/snappy v1.0.0
 	github.com/google/uuid v1.6.0
+	github.com/grafana/loki/pkg/push v0.0.0-20240924133635-758364c7775f
+	github.com/grafana/loki/v3 v3.5.4
 	github.com/hashicorp/go-multierror v1.1.1
 	github.com/pkg/errors v0.9.1
-	github.com/prometheus/client_golang v1.20.5
-	github.com/prometheus/prometheus v0.300.1
+	github.com/prometheus/client_golang v1.21.1
+	github.com/prometheus/prometheus v0.302.1
 	github.com/sirupsen/logrus v1.9.3
 	github.com/stretchr/testify v1.10.0
 	github.com/valyala/fasthttp v1.58.0
 	gopkg.in/yaml.v2 v2.4.0
 )
 
 require (
+	dario.cat/mergo v1.0.1 // indirect
+	github.com/HdrHistogram/hdrhistogram-go v1.1.2 // indirect
+	github.com/Masterminds/goutils v1.1.1 // indirect
+	github.com/Masterminds/semver/v3 v3.3.1 // indirect
+	github.com/Masterminds/sprig/v3 v3.3.0 // indirect
+	github.com/alecthomas/units v0.0.0-20240927000941-0f3dac36c52b // indirect
 	github.com/andybalholm/brotli v1.1.1 // indirect
+	github.com/armon/go-metrics v0.4.1 // indirect
 	github.com/beorn7/perks v1.0.1 // indirect
+	github.com/c2h5oh/datasize v0.0.0-20231215233829-aa82cc1e6500 // indirect
 	github.com/cespare/xxhash/v2 v2.3.0 // indirect
+	github.com/coreos/go-semver v0.3.1 // indirect
+	github.com/coreos/go-systemd/v22 v22.5.0 // indirect
 	github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect
+	github.com/dennwc/varint v1.0.0 // indirect
+	github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect
+	github.com/dustin/go-humanize v1.0.1 // indirect
+	github.com/edsrzf/mmap-go v1.2.0 // indirect
+	github.com/facette/natsort v0.0.0-20181210072756-2cd4dd1e2dcb // indirect
+	github.com/fatih/color v1.18.0 // indirect
+	github.com/felixge/httpsnoop v1.0.4 // indirect
+	github.com/fsnotify/fsnotify v1.8.0 // indirect
+	github.com/go-kit/log v0.2.1 // indirect
+	github.com/go-logfmt/logfmt v0.6.0 // indirect
+	github.com/go-logr/logr v1.4.2 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/go-redsync/redsync/v4 v4.13.0 // indirect
+	github.com/gogo/googleapis v1.4.1 // indirect
+	github.com/gogo/status v1.1.1 // indirect
+	github.com/golang/protobuf v1.5.4 // indirect
+	github.com/google/btree v1.1.3 // indirect
+	github.com/gorilla/mux v1.8.1 // indirect
+	github.com/grafana/dskit v0.0.0-20250317084829-9cdd36a91f10 // indirect
+	github.com/grafana/gomemcache v0.0.0-20250228145437-da7b95fd2ac1 // indirect
+	github.com/grafana/jsonparser v0.0.0-20241004153430-023329977675 // indirect
+	github.com/grafana/pyroscope-go/godeltaprof v0.1.8 // indirect
 	github.com/grafana/regexp v0.0.0-20240518133315-a468a5bfb3bc // indirect
+	github.com/hashicorp/consul/api v1.31.2 // indirect
 	github.com/hashicorp/errwrap v1.1.0 // indirect
-	github.com/klauspost/compress v1.17.11 // indirect
-	github.com/kr/text v0.2.0 // indirect
-	github.com/matttproud/golang_protobuf_extensions/v2 v2.0.0 // indirect
+	github.com/hashicorp/go-cleanhttp v0.5.2 // indirect
+	github.com/hashicorp/go-hclog v1.6.3 // indirect
+	github.com/hashicorp/go-immutable-radix v1.3.1 // indirect
+	github.com/hashicorp/go-metrics v0.5.4 // indirect
+	github.com/hashicorp/go-msgpack/v2 v2.1.1 // indirect
+	github.com/hashicorp/go-rootcerts v1.0.2 // indirect
+	github.com/hashicorp/go-sockaddr v1.0.7 // indirect
+	github.com/hashicorp/golang-lru v1.0.2 // indirect
+	github.com/hashicorp/memberlist v0.5.3 // indirect
+	github.com/hashicorp/serf v0.10.1 // indirect
+	github.com/huandu/xstrings v1.5.0 // indirect
+	github.com/jpillora/backoff v1.0.0 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/klauspost/compress v1.18.0 // indirect
+	github.com/mattn/go-colorable v0.1.14 // indirect
+	github.com/mattn/go-isatty v0.0.20 // indirect
+	github.com/mdlayher/socket v0.5.1 // indirect
+	github.com/mdlayher/vsock v1.2.1 // indirect
+	github.com/miekg/dns v1.1.63 // indirect
+	github.com/mitchellh/copystructure v1.2.0 // indirect
+	github.com/mitchellh/go-homedir v1.1.0 // indirect
+	github.com/mitchellh/mapstructure v1.5.1-0.20220423185008-bf980b35cac4 // indirect
+	github.com/mitchellh/reflectwalk v1.0.2 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.2 // indirect
 	github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect
+	github.com/mwitkow/go-conntrack v0.0.0-20190716064945-2f068394615f // indirect
+	github.com/opentracing-contrib/go-grpc v0.1.1 // indirect
+	github.com/opentracing-contrib/go-stdlib v1.1.0 // indirect
+	github.com/opentracing/opentracing-go v1.2.1-0.20220228012449-10b1cf09e00b // indirect
+	github.com/pires/go-proxyproto v0.7.0 // indirect
 	github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 // indirect
 	github.com/prometheus/client_model v0.6.1 // indirect
-	github.com/prometheus/common v0.61.0 // indirect
+	github.com/prometheus/common v0.62.0 // indirect
+	github.com/prometheus/exporter-toolkit v0.13.2 // indirect
 	github.com/prometheus/procfs v0.15.1 // indirect
+	github.com/redis/go-redis/v9 v9.7.3 // indirect
+	github.com/sean-/seed v0.0.0-20170313163322-e2103e2c3529 // indirect
+	github.com/sercand/kuberesolver/v5 v5.1.1 // indirect
+	github.com/shopspring/decimal v1.4.0 // indirect
+	github.com/sony/gobreaker/v2 v2.1.0 // indirect
+	github.com/spf13/cast v1.7.1 // indirect
+	github.com/stretchr/objx v0.5.2 // indirect
+	github.com/tjhop/slog-gokit v0.1.4 // indirect
+	github.com/uber/jaeger-client-go v2.30.0+incompatible // indirect
+	github.com/uber/jaeger-lib v2.4.1+incompatible // indirect
 	github.com/valyala/bytebufferpool v1.0.0 // indirect
-	golang.org/x/sys v0.28.0 // indirect
-	golang.org/x/text v0.21.0 // indirect
-	google.golang.org/protobuf v1.36.0 // indirect
+	go.etcd.io/etcd/api/v3 v3.5.4 // indirect
+	go.etcd.io/etcd/client/pkg/v3 v3.5.4 // indirect
+	go.etcd.io/etcd/client/v3 v3.5.4 // indirect
+	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
+	go.opentelemetry.io/collector/pdata v1.28.1 // indirect
+	go.opentelemetry.io/otel v1.35.0 // indirect
+	go.opentelemetry.io/otel/metric v1.35.0 // indirect
+	go.opentelemetry.io/otel/trace v1.35.0 // indirect
+	go.uber.org/atomic v1.11.0 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	go.uber.org/zap v1.27.0 // indirect
+	go4.org/netipx v0.0.0-20230125063823-8449b0a6169f // indirect
+	golang.org/x/crypto v0.36.0 // indirect
+	golang.org/x/exp v0.0.0-20250106191152-7588d65b2ba8 // indirect
+	golang.org/x/mod v0.22.0 // indirect
+	golang.org/x/net v0.38.0 // indirect
+	golang.org/x/oauth2 v0.28.0 // indirect
+	golang.org/x/sync v0.12.0 // indirect
+	golang.org/x/sys v0.31.0 // indirect
+	golang.org/x/text v0.23.0 // indirect
+	golang.org/x/time v0.11.0 // indirect
+	golang.org/x/tools v0.29.0 // indirect
+	google.golang.org/genproto/googleapis/api v0.0.0-20250303144028-a0af3efb3deb // indirect
+	google.golang.org/genproto/googleapis/rpc v0.0.0-20250313205543-e70fdf4c4cb4 // indirect
+	google.golang.org/grpc v1.71.0 // indirect
+	google.golang.org/protobuf v1.36.6 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )