26573682638

Committed 26 May 2026 01:29PM UTC coverage: 71.736% (+0.7%) from 71.062%

Build # 26573682638

Build Type

Pull #103

github

Committed by

EmilyShepherd

Commit Message

Support setting ingress TLS fields in helm chart

This allows turning TLS  on and specifying where the certificate and key
files should be loaded from within the container, or optionally loading
these from an existing secret.

Signed-off-by: Emily Shepherd <emily@redcoat.dev>

Pull Request Pull Request #103: TLS server Support

Coverage Stats

20 of 22 new or added lines in 1 file covered. (90.91%)

434 of 605 relevant lines covered (71.74%)

6.69 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

91.07

/processor.go

package main

import (
        "bytes"
        "crypto/tls"
        "crypto/x509"
        "encoding/base64"
        "fmt"
        "io/ioutil"
        "net"
        "sync"
        "sync/atomic"
        "time"

        "github.com/blind-oracle/go-common/logger"
        "github.com/dyson/certman"
        "github.com/google/uuid"
        "github.com/pkg/errors"
        fh "github.com/valyala/fasthttp"
)

type result struct {
        code     int
        body     []byte
        duration float64
        tenant   string
        err      error
}

type processor struct {
        cfg config

        srv *fh.Server
        cli *fh.Client

        shuttingDown uint32

        logger.Logger

        auth struct {
                egressHeader []byte
        }
}

func newProcessor(c config) (*processor, error) {
        p := &processor{
                cfg:    c,
                Logger: logger.NewSimpleLogger("proc"),
        }

        p.srv = &fh.Server{
                Name:    "cortex-tenant",
                Handler: p.handle,

                MaxRequestBodySize: 8 * 1024 * 1024,

                ReadTimeout:  c.Timeout,
                WriteTimeout: c.Timeout,
                IdleTimeout:  c.IdleTimeout,

                Concurrency: c.Concurrency,

                TLSConfig: &tls.Config{},
        }

        p.cli = &fh.Client{
                Name:               "cortex-tenant",
                ReadTimeout:        c.Timeout,
                WriteTimeout:       c.Timeout,
                MaxConnWaitTimeout: 1 * time.Second,
                MaxConnsPerHost:    c.MaxConnsPerHost,
                DialDualStack:      c.EnableIPv6,
                MaxConnDuration:    c.MaxConnDuration,
                TLSConfig:          &tls.Config{},
        }

        if caFile := c.Auth.Egress.TlsConfig.CaBundleFile; caFile != "" {
                caCert, err := ioutil.ReadFile(caFile)
                if err != nil {
                        return nil, errors.Wrap(err, "Unable to load CA Bundle")
                }

                caCertPool := x509.NewCertPool()
                caCertPool.AppendCertsFromPEM(caCert)
                p.cli.TLSConfig.RootCAs = caCertPool
        }

        if c.Auth.Egress.Username != "" {
                authString := []byte(fmt.Sprintf("%s:%s", c.Auth.Egress.Username, c.Auth.Egress.Password))
                p.auth.egressHeader = []byte("Basic " + base64.StdEncoding.EncodeToString(authString))
        }

        if c.Auth.Ingress.TlsConfig.CertFile != "" && c.Auth.Ingress.TlsConfig.KeyFile != "" {
                cm, err := certman.New(
                        c.Auth.Ingress.TlsConfig.CertFile,
                        c.Auth.Ingress.TlsConfig.KeyFile,
                )
                if err != nil {
                        return nil, errors.Wrap(err, "Unable to configure server TLS")
                }
                err = cm.Watch()
                if err != nil {
                        return nil, errors.Wrap(err, "Unable to configure server TLS")
                }

                p.srv.TLSConfig.GetCertificate = cm.GetCertificate
        }

        // For testing
        if c.pipeOut != nil {
                p.cli.Dial = func(a string) (net.Conn, error) {
                        return c.pipeOut.Dial()
                }
        }

        return p, nil
}

func (p *processor) run() (err error) {
        var l net.Listener

        // For testing
        if p.cfg.pipeIn == nil {
                if l, err = net.Listen("tcp", p.cfg.Listen); err != nil {
                        return
                }
        } else {
                l = p.cfg.pipeIn
        }

        if p.srv.TLSConfig.GetCertificate == nil {
                go p.srv.Serve(l)
        } else {
                // Just pass empty certFile and keyFile to serveTLS because we have
                // overriden the static behaviour with CertMan in the processor setup.
                go p.srv.ServeTLS(l, "", "")
        }
        return
}

func (p *processor) handle(ctx *fh.RequestCtx) {
        if bytes.Equal(ctx.Path(), []byte("/alive")) {
                if atomic.LoadUint32(&p.shuttingDown) == 1 {
                        ctx.SetStatusCode(fh.StatusServiceUnavailable)
                }

                return
        }

        if !bytes.Equal(ctx.Request.Header.Method(), []byte("POST")) {
                ctx.Error("Expecting POST", fh.StatusBadRequest)
                return
        }

        if bytes.Equal(ctx.Path(), []byte("/push")) {
                p.handleMetrics(ctx)
                return
        }

        if bytes.Equal(ctx.Path(), []byte("/loki/push")) {
                p.handleLogs(ctx)
                return
        }

        ctx.SetStatusCode(fh.StatusNotFound)
}

func (p *processor) dispatch(target string, clientIP net.Addr, reqID uuid.UUID, tenantPrefix string, m map[string]func() ([]byte, error)) (res []result) {
        var wg sync.WaitGroup
        res = make([]result, len(m))

        i := 0
        for tenant, bodyFunc := range m {
                wg.Add(1)

                go func(idx int, tenant string, bodyFunc func() ([]byte, error)) {
                        defer wg.Done()

                        r := p.send(target, clientIP, reqID, tenant, bodyFunc)
                        res[idx] = r
                }(i, tenantPrefix+tenant, bodyFunc)

                i++
        }

        wg.Wait()
        return
}

func (p *processor) send(target string, clientIP net.Addr, reqID uuid.UUID, tenant string, bodyFunc func() ([]byte, error)) (r result) {
        start := time.Now()
        r.tenant = tenant

        req := fh.AcquireRequest()
        resp := fh.AcquireResponse()

        defer func() {
                fh.ReleaseRequest(req)
                fh.ReleaseResponse(resp)
        }()

        buf, err := bodyFunc()
        if err != nil {
                r.err = err
                return
        }

        p.fillRequestHeaders(clientIP, reqID, tenant, req)

        if p.auth.egressHeader != nil {
                req.Header.SetBytesV("Authorization", p.auth.egressHeader)
        }

        req.Header.SetMethod(fh.MethodPost)
        req.SetRequestURI(target)
        req.SetBody(buf)

        if err = p.cli.DoTimeout(req, resp, p.cfg.Timeout); err != nil {
                r.err = err
                return
        }

        r.code = resp.Header.StatusCode()
        r.body = make([]byte, len(resp.Body()))
        copy(r.body, resp.Body())
        r.duration = time.Since(start).Seconds() / 1000

        return
}

func (p *processor) fillRequestHeaders(
        clientIP net.Addr, reqID uuid.UUID, tenant string, req *fh.Request) {
        req.Header.Set("Content-Encoding", "snappy")
        req.Header.Set("Content-Type", "application/x-protobuf")
        req.Header.Set("X-Prometheus-Remote-Write-Version", "0.1.0")
        req.Header.Set("X-Cortex-Tenant-Client", clientIP.String())
        req.Header.Set("X-Cortex-Tenant-ReqID", reqID.String())
        req.Header.Set(p.cfg.Tenant.Header, tenant)
}

func (p *processor) close() (err error) {
        // Signal that we're shutting down
        atomic.StoreUint32(&p.shuttingDown, 1)
        // Let healthcheck detect that we're offline
        time.Sleep(p.cfg.TimeoutShutdown)
        // Shutdown
        return p.srv.Shutdown()
}

1	package main
2
3	import (
4	"bytes"
5	"crypto/tls"
6	"crypto/x509"
7	"encoding/base64"
8	"fmt"
9	"io/ioutil"
10	"net"
11	"sync"
12	"sync/atomic"
13	"time"
14
15	"github.com/blind-oracle/go-common/logger"
16	"github.com/dyson/certman"
17	"github.com/google/uuid"
18	"github.com/pkg/errors"
19	fh "github.com/valyala/fasthttp"
20	)
21
22	type result struct {
23	code int
24	body []byte
25	duration float64
26	tenant string
27	err error
28	}
29
30	type processor struct {
31	cfg config
32
33	srv *fh.Server
34	cli *fh.Client
35
36	shuttingDown uint32
37
38	logger.Logger
39
40	auth struct {
41	egressHeader []byte
42	}
43	}
44
45	func newProcessor(c config) (*processor, error) {	19✔
46	p := &processor{	19✔
47	cfg: c,	19✔
48	Logger: logger.NewSimpleLogger("proc"),	19✔
49	}	19✔
50		19✔
51	p.srv = &fh.Server{	19✔
52	Name: "cortex-tenant",	19✔
53	Handler: p.handle,	19✔
54		19✔
55	MaxRequestBodySize: 8 * 1024 * 1024,	19✔
56		19✔
57	ReadTimeout: c.Timeout,	19✔
58	WriteTimeout: c.Timeout,	19✔
59	IdleTimeout: c.IdleTimeout,	19✔
60		19✔
61	Concurrency: c.Concurrency,	19✔
62		19✔
63	TLSConfig: &tls.Config{},	19✔
64	}	19✔
65		19✔
66	p.cli = &fh.Client{	19✔
67	Name: "cortex-tenant",	19✔
68	ReadTimeout: c.Timeout,	19✔
69	WriteTimeout: c.Timeout,	19✔
70	MaxConnWaitTimeout: 1 * time.Second,	19✔
71	MaxConnsPerHost: c.MaxConnsPerHost,	19✔
72	DialDualStack: c.EnableIPv6,	19✔
73	MaxConnDuration: c.MaxConnDuration,	19✔
74	TLSConfig: &tls.Config{},	19✔
75	}	19✔
76		19✔
77	if caFile := c.Auth.Egress.TlsConfig.CaBundleFile; caFile != "" {	21✔
78	caCert, err := ioutil.ReadFile(caFile)	2✔
79	if err != nil {	3✔
80	return nil, errors.Wrap(err, "Unable to load CA Bundle")	1✔
81	}	1✔
82
83	caCertPool := x509.NewCertPool()	1✔
84	caCertPool.AppendCertsFromPEM(caCert)	1✔
85	p.cli.TLSConfig.RootCAs = caCertPool	1✔
86	}
87
88	if c.Auth.Egress.Username != "" {	19✔
89	authString := []byte(fmt.Sprintf("%s:%s", c.Auth.Egress.Username, c.Auth.Egress.Password))	1✔
90	p.auth.egressHeader = []byte("Basic " + base64.StdEncoding.EncodeToString(authString))	1✔
91	}	1✔
92
93	if c.Auth.Ingress.TlsConfig.CertFile != "" && c.Auth.Ingress.TlsConfig.KeyFile != "" {	20✔
94	cm, err := certman.New(	2✔
95	c.Auth.Ingress.TlsConfig.CertFile,	2✔
96	c.Auth.Ingress.TlsConfig.KeyFile,	2✔
97	)	2✔
98	if err != nil {	2✔
NEW 99	return nil, errors.Wrap(err, "Unable to configure server TLS")	×
NEW 100	}	×
101	err = cm.Watch()	2✔
102	if err != nil {	3✔
103	return nil, errors.Wrap(err, "Unable to configure server TLS")	1✔
104	}	1✔
105
106	p.srv.TLSConfig.GetCertificate = cm.GetCertificate	1✔
107	}
108
109	// For testing
110	if c.pipeOut != nil {	21✔
111	p.cli.Dial = func(a string) (net.Conn, error) {	10✔
112	return c.pipeOut.Dial()	6✔
113	}	6✔
114	}
115
116	return p, nil	17✔
117	}
118
119	func (p *processor) run() (err error) {	4✔
120	var l net.Listener	4✔
121		4✔
122	// For testing	4✔
123	if p.cfg.pipeIn == nil {	4✔
124	if l, err = net.Listen("tcp", p.cfg.Listen); err != nil {	×
125	return	×
126	}	×
127	} else {	4✔
128	l = p.cfg.pipeIn	4✔
129	}	4✔
130
131	if p.srv.TLSConfig.GetCertificate == nil {	7✔
132	go p.srv.Serve(l)	3✔
133	} else {	4✔
134	// Just pass empty certFile and keyFile to serveTLS because we have	1✔
135	// overriden the static behaviour with CertMan in the processor setup.	1✔
136	go p.srv.ServeTLS(l, "", "")	1✔
137	}	1✔
138	return	4✔
139	}
140
141	func (p processor) handle(ctx fh.RequestCtx) {	17✔
142	if bytes.Equal(ctx.Path(), []byte("/alive")) {	21✔
143	if atomic.LoadUint32(&p.shuttingDown) == 1 {	6✔
144	ctx.SetStatusCode(fh.StatusServiceUnavailable)	2✔
145	}	2✔
146
147	return	4✔
148	}
149
150	if !bytes.Equal(ctx.Request.Header.Method(), []byte("POST")) {	13✔
151	ctx.Error("Expecting POST", fh.StatusBadRequest)	×
152	return	×
153	}	×
154
155	if bytes.Equal(ctx.Path(), []byte("/push")) {	20✔
156	p.handleMetrics(ctx)	7✔
157	return	7✔
158	}	7✔
159
160	if bytes.Equal(ctx.Path(), []byte("/loki/push")) {	12✔
161	p.handleLogs(ctx)	6✔
162	return	6✔
163	}	6✔
164
165	ctx.SetStatusCode(fh.StatusNotFound)	×
166	}
167
168	func (p *processor) dispatch(target string, clientIP net.Addr, reqID uuid.UUID, tenantPrefix string, m map[string]func() ([]byte, error)) (res []result) {	6✔
169	var wg sync.WaitGroup	6✔
170	res = make([]result, len(m))	6✔
171		6✔
172	i := 0	6✔
173	for tenant, bodyFunc := range m {	18✔
174	wg.Add(1)	12✔
175		12✔
176	go func(idx int, tenant string, bodyFunc func() ([]byte, error)) {	24✔
177	defer wg.Done()	12✔
178		12✔
179	r := p.send(target, clientIP, reqID, tenant, bodyFunc)	12✔
180	res[idx] = r	12✔
181	}(i, tenantPrefix+tenant, bodyFunc)	12✔
182
183	i++	12✔
184	}
185
186	wg.Wait()	6✔
187	return	6✔
188	}
189
190	func (p *processor) send(target string, clientIP net.Addr, reqID uuid.UUID, tenant string, bodyFunc func() ([]byte, error)) (r result) {	14✔
191	start := time.Now()	14✔
192	r.tenant = tenant	14✔
193		14✔
194	req := fh.AcquireRequest()	14✔
195	resp := fh.AcquireResponse()	14✔
196		14✔
197	defer func() {	28✔
198	fh.ReleaseRequest(req)	14✔
199	fh.ReleaseResponse(resp)	14✔
200	}()	14✔
201
202	buf, err := bodyFunc()	14✔
203	if err != nil {	14✔
204	r.err = err	×
205	return	×
206	}	×
207
208	p.fillRequestHeaders(clientIP, reqID, tenant, req)	14✔
209		14✔
210	if p.auth.egressHeader != nil {	15✔
211	req.Header.SetBytesV("Authorization", p.auth.egressHeader)	1✔
212	}	1✔
213
214	req.Header.SetMethod(fh.MethodPost)	14✔
215	req.SetRequestURI(target)	14✔
216	req.SetBody(buf)	14✔
217		14✔
218	if err = p.cli.DoTimeout(req, resp, p.cfg.Timeout); err != nil {	14✔
219	r.err = err	×
220	return	×
221	}	×
222
223	r.code = resp.Header.StatusCode()	14✔
224	r.body = make([]byte, len(resp.Body()))	14✔
225	copy(r.body, resp.Body())	14✔
226	r.duration = time.Since(start).Seconds() / 1000	14✔
227		14✔
228	return	14✔
229	}
230
231	func (p *processor) fillRequestHeaders(
232	clientIP net.Addr, reqID uuid.UUID, tenant string, req *fh.Request) {	15✔
233	req.Header.Set("Content-Encoding", "snappy")	15✔
234	req.Header.Set("Content-Type", "application/x-protobuf")	15✔
235	req.Header.Set("X-Prometheus-Remote-Write-Version", "0.1.0")	15✔
236	req.Header.Set("X-Cortex-Tenant-Client", clientIP.String())	15✔
237	req.Header.Set("X-Cortex-Tenant-ReqID", reqID.String())	15✔
238	req.Header.Set(p.cfg.Tenant.Header, tenant)	15✔
239	}	15✔
240
241	func (p *processor) close() (err error) {	4✔
242	// Signal that we're shutting down	4✔
243	atomic.StoreUint32(&p.shuttingDown, 1)	4✔
244	// Let healthcheck detect that we're offline	4✔
245	time.Sleep(p.cfg.TimeoutShutdown)	4✔
246	// Shutdown	4✔
247	return p.srv.Shutdown()	4✔
248	}	4✔

blind-oracle / cortex-tenant / 26573682638

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous