26219653858

Committed 21 May 2026 10:10AM UTC coverage: 69.007% (-1.0%) from 69.983%

Build # 26219653858

Build Type

Pull #100

github

Committed by

EmilyShepherd

Commit Message

Update customCA to use config.ca_bundle_file

The helm chart had a field called customCA which, when specified, would
cause a secret to be mounted in the containers /etc/ssl/certs directory,
implictly causing go to pick the CA bundle up and trust it for outbound
TLS connections.

With the added support to explictly define a CA to trust in the
application, this commit updates the helm field's logic to instead use
that. This has the advantage that _only_ this CA will be trusted, not it
in addition to all other standard CAs bundled with the container image.

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

Pull Request Pull Request #100: Improve trusted CA support

Coverage Stats

6 of 17 new or added lines in 2 files covered. (35.29%)

403 of 584 relevant lines covered (69.01%)

6.07 hits per line

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

82.99

/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/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,
        }

        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))
        }

        // 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
        }

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

blind-oracle / cortex-tenant / 26219653858

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