init

Author: blind-oracle

.gitignore

@@ -0,0 +1,5 @@
+cortex-tenant
+sink
+config.yml
+.out
+*.rpm

LICENSE

@@ -0,0 +1,61 @@
+NAME := cortex-tenant
+MAINTAINER := Igor Novgorodov
+DESCRIPTION := Cortex tenant proxy
+URL := https://github.com/blind-oracle/cortex-tenant
+LICENSE := MPL
+
+VERSION := 1.0.0
+RELEASE := 1
+
+GO ?= go
+
+RPM := $(NAME)-$(VERSION)-$(RELEASE).x86_64.rpm
+DIR := $(NAME)-git
+OUT := .out
+
+REPO_HOST := tvovma-mgt035
+REPO := cortex
+
+all: rpm
+
+build:
+	GOARCH=amd64 \
+	GOOS=linux \
+	$(GO) build -ldflags "-s -w -extldflags \"-static\" -X main.version=$(VERSION)"
+
+prepare:
+	cd deploy && \
+	rm -rf $(OUT) && \
+	mkdir -p $(OUT)/usr/sbin $(OUT)/var/lib/$(NAME) $(OUT)/etc/sysconfig $(OUT)/usr/lib/systemd/system && \
+	cp $(NAME).env $(OUT)/etc/sysconfig/$(NAME) && \
+	cp $(NAME).service $(OUT)/usr/lib/systemd/system && \
+	cp $(NAME).yml $(OUT)/etc/$(NAME).yml && \
+	cp ../$(NAME) $(OUT)/usr/sbin
+
+rpm: build prepare build-rpm
+
+rpm-upload:
+	scp $(RPM) $(REPO_HOST):
+	ssh $(REPO_HOST) sudo pulp-admin rpm repo uploads rpm --repo-id $(REPO) -f $(RPM)
+	ssh $(REPO_HOST) sudo pulp-admin rpm repo publish run --repo-id $(REPO)
+
+build-rpm:
+	fpm \
+		-s dir \
+		--config-files etc/$(NAME).yml \
+		--config-files etc/sysconfig/$(NAME) \
+		-C deploy/$(OUT)/ \
+		-t rpm \
+		--after-install deploy/after_install.sh \
+		-n $(NAME) \
+		-v $(VERSION) \
+		--iteration $(RELEASE) \
+		--force \
+		--rpm-compression bzip2 \
+		--rpm-os linux \
+		--url $(URL) \
+		--description "$(DESCRIPTION)" \
+		-m "$(MAINTAINER)" \
+		--license "$(LICENSE)" \
+		-a amd64 \
+		.

Makefile

@@ -0,0 +1,3 @@
+# cortex-tenant
+
+Prometheus remote write proxy which groups metrics by Cortex tenants based on labels

README.md

@@ -0,0 +1,7 @@
+#!/bin/bash
+
+USER="cortex-tenant"
+HOME="/var/lib/$USER"
+
+useradd -d $HOME -s /bin/false -M $USER > /dev/null 2>&1 || true
+chown $USER:$USER $HOME

deploy/after_install.sh

@@ -0,0 +1 @@
+CONFIG_FILE="/etc/cortex-tenant.yml"

deploy/cortex-tenant.env

@@ -0,0 +1,18 @@
+[Unit]
+Description=Cortex-Tenant
+Documentation=https://github.com/blind-oracle/cortex-tenant
+Wants=network-online.target
+After=network-online.target
+
+[Service]
+Restart=always
+User=cortex-tenant
+EnvironmentFile=/etc/sysconfig/cortex-tenant
+ExecStart=/usr/sbin/cortex -config $CONFIG_FILE
+ExecReload=/bin/kill -HUP $MAINPID
+TimeoutStopSec=30
+SendSIGKILL=no
+WorkingDirectory=/var/lib/cortex-tenant
+
+[Install]
+WantedBy=multi-user.target

deploy/cortex-tenant.service

@@ -0,0 +1,17 @@
+listen: 0.0.0.0:8080
+target: http://127.0.0.1:9091/receive
+
+log_level: warn
+
+max_tenants: 100
+buffer_size: 10000
+batch_size: 100
+flush_interval: 1s
+timeout: 10s
+
+tenant:
+  label: tenant
+  label_remove: true
+  header: X-Scope-OrgID
+  default: default
+  recycle_age: 10m

deploy/cortex-tenant.yml

@@ -0,0 +1,24 @@
+module github.com/blind-oracle/cortex-tenant
+
+go 1.15
+
+require (
+	github.com/andybalholm/brotli v1.0.1 // indirect
+	github.com/blind-oracle/go-common v1.0.5
+	github.com/gogo/protobuf v1.3.1
+	github.com/golang/snappy v0.0.2
+	github.com/grpc-ecosystem/grpc-gateway v1.15.0 // indirect
+	github.com/hashicorp/errwrap v1.1.0 // indirect
+	github.com/hashicorp/go-multierror v1.1.0
+	github.com/klauspost/compress v1.11.1 // indirect
+	github.com/prometheus/common v0.14.0
+	github.com/prometheus/prometheus v2.5.0+incompatible
+	github.com/sirupsen/logrus v1.7.0
+	github.com/valyala/fasthttp v1.16.0
+	golang.org/x/net v0.0.0-20201002202402-0a1ea396d57c // indirect
+	golang.org/x/sys v0.0.0-20201005172224-997123666555 // indirect
+	golang.org/x/text v0.3.3 // indirect
+	google.golang.org/genproto v0.0.0-20201006033701-bcad7cf615f2 // indirect
+	google.golang.org/protobuf v1.25.0 // indirect
+	gopkg.in/yaml.v2 v2.3.0
+)

go.mod

@@ -0,0 +1,201 @@
+package main
+
+import (
+	"bytes"
+	"net"
+	"sync"
+	"time"
+
+	"github.com/blind-oracle/go-common/logger"
+	"github.com/gogo/protobuf/proto"
+	"github.com/golang/snappy"
+	me "github.com/hashicorp/go-multierror"
+	"github.com/prometheus/prometheus/prompb"
+	fh "github.com/valyala/fasthttp"
+)
+
+type httpServer struct {
+	cfg config
+	srv *fh.Server
+
+	tenants map[string]*tenant
+	chClose chan struct{}
+
+	wg sync.WaitGroup
+	sync.RWMutex
+	logger.Logger
+}
+
+func newHTTPServer(c config) (s *httpServer, err error) {
+	s = &httpServer{
+		cfg:     c,
+		tenants: make(map[string]*tenant, c.MaxTenants),
+		chClose: make(chan struct{}),
+		Logger:  logger.NewSimpleLogger("http"),
+	}
+
+	s.srv = &fh.Server{
+		Name:    "cortex-tenant",
+		Handler: s.handle,
+
+		MaxRequestBodySize: 8 * 1024 * 1024,
+
+		ReadTimeout:  c.Timeout,
+		WriteTimeout: c.Timeout,
+		IdleTimeout:  60 * time.Second,
+	}
+
+	l, err := net.Listen("tcp", c.Listen)
+	if err != nil {
+		return nil, err
+	}
+
+	go s.srv.Serve(l)
+	s.wg.Add(1)
+	go s.recycler()
+	return
+}
+
+func (s *httpServer) handle(ctx *fh.RequestCtx) {
+	var (
+		rq      prompb.WriteRequest
+		buf     *buffer
+		decoded []byte
+		err     error
+	)
+
+	if !bytes.Equal(ctx.Request.Header.Method(), []byte("POST")) {
+		goto bad
+	}
+
+	if !bytes.Equal(ctx.Path(), []byte("/push")) {
+		goto bad
+	}
+
+	buf = bufferPool.Get().(*buffer)
+	defer bufferPool.Put(buf)
+
+	buf.grow()
+	if decoded, err = snappy.Decode(buf.b, ctx.Request.Body()); err != nil {
+		goto bad
+	}
+
+	if err = proto.Unmarshal(decoded, &rq); err != nil {
+		goto bad
+	}
+
+	for _, ts := range rq.Timeseries {
+		s.processTimeseries(ts)
+	}
+
+	return
+
+bad:
+	msg := "Bad request"
+	if err != nil {
+		msg += ": " + err.Error()
+	}
+
+	ctx.Error(msg, fh.StatusBadRequest)
+}
+
+func (s *httpServer) processTimeseries(ts *prompb.TimeSeries) {
+	var (
+		tenant string
+		idx    int
+	)
+
+	for i, l := range ts.Labels {
+		if l.Name == s.cfg.Tenant.Label {
+			tenant = l.Value
+			idx = i
+			break
+		}
+	}
+
+	if tenant == "" {
+		tenant = s.cfg.Tenant.Default
+	} else if s.cfg.Tenant.LabelRemove {
+		cnt := len(ts.Labels)
+		ts.Labels[idx] = ts.Labels[cnt-1]
+		ts.Labels = ts.Labels[:cnt-1]
+	}
+
+	s.RLock()
+	t := s.tenants[tenant]
+	s.RUnlock()
+
+	if t == nil {
+		s.Lock()
+		if s.cfg.MaxTenants > 0 && len(s.tenants) >= s.cfg.MaxTenants {
+			s.Unlock()
+			s.Errorf("MaxTenants (%s) reached, new tenant (%s) dropped", s.cfg.MaxTenants, tenant)
+			return
+		}
+
+		s.Warnf("Creating tenant '%s'", tenant)
+		t = newTenant(tenant, s.cfg)
+
+		s.tenants[tenant] = t
+		s.Unlock()
+	}
+
+	t.push(ts)
+}
+
+func (s *httpServer) close() (err error) {
+	close(s.chClose)
+	s.wg.Wait()
+
+	var errs *me.Error
+	if err = s.srv.Shutdown(); err != nil {
+		me.Append(errs, err)
+	}
+
+	s.RLock()
+	for _, t := range s.tenants {
+		if err = t.close(); err != nil {
+			me.Append(errs, err)
+		}
+	}
+	s.RUnlock()
+
+	return errs.ErrorOrNil()
+}
+
+func (s *httpServer) recycler() {
+	ticker := time.NewTicker(10 * time.Second)
+
+	defer func() {
+		ticker.Stop()
+		s.wg.Done()
+	}()
+
+	for {
+		select {
+		case <-ticker.C:
+			now := time.Now()
+
+			toClose := []*tenant{}
+			s.Lock()
+			for tn, t := range s.tenants {
+				if now.Sub(t.lastFlush.Load().(time.Time)) >= s.cfg.Tenant.RecycleAge {
+					toClose = append(toClose, t)
+					delete(s.tenants, tn)
+				}
+			}
+			s.Unlock()
+
+			for _, t := range toClose {
+				s.Warnf("Recycling tenant '%s'", t.name)
+
+				if err := t.close(); err != nil {
+					s.Errorf("Errors while closing tenant '%s': %s", t.name, err)
+				}
+			}
+
+		case <-s.chClose:
+			return
+		}
+	}
+}