feat: Added support for logging of HTTP Headers (#4803)

* feat: Added support for logging of HTTP Headers

Signed-off-by: Zach Speaks <[email protected]>

* Fixed test failing due to namespace overlap

Signed-off-by: Zach Speaks <[email protected]>

* Updates to reflect PR feedback

Signed-off-by: Zach Speaks <[email protected]>

* Rough draft propogation to ingester + logging in ingester

Signed-off-by: Zach Speaks <[email protected]>

* Added Header Propogation to Ingester and Querier

Added Header Propagation to Ingester and Querier
Signed-off-by: Zach Speaks <[email protected]>
Signed-off-by: Zachary Speaks <[email protected]>

* Addressed whitespace and linting issues

Signed-off-by: Zach Speaks <[email protected]>
Signed-off-by: Zachary Speaks <[email protected]>

* Additional linting fixes

Signed-off-by: Zach Speaks <[email protected]>
Signed-off-by: Zachary Speaks <[email protected]>

* Fixed failing DCO check

Signed-off-by: Zachary Speaks <[email protected]>

* Removed outdated line from prior test design

Signed-off-by: Zachary Speaks <[email protected]>

* Updates to reflect PR feedback

Signed-off-by: Zachary Speaks <[email protected]>

* Additional Updates to reflect feedback

Signed-off-by: Zachary Speaks <[email protected]>

* Improved string comparison when decoding

Signed-off-by: Zachary Speaks <[email protected]>

* Added header decoding to frontend_processor.go for when scheduler service is not being used

Signed-off-by: Zachary Speaks <[email protected]>

* Added encoding/decoding interceptors for streams and updated changelog

Signed-off-by: Zachary Speaks <[email protected]>

* Undo accidental updates to single-process-config-block.yaml file

Signed-off-by: Zachary Speaks <[email protected]>

* Whitespace updates

Signed-off-by: Zachary Speaks <[email protected]>

* Refactored encode/decode headers to inject/extract to better align with existing code

Signed-off-by: Zachary Speaks <[email protected]>

* Fixed failing linter

Signed-off-by: Zachary Speaks <[email protected]>

* Switched extraction to being done by a middleware on the querier API

Signed-off-by: Zachary Speaks <[email protected]>

* Linting

Signed-off-by: Zachary Speaks <[email protected]>

* Changes to reflect PR feedback

Signed-off-by: Zachary Speaks <[email protected]>

* Updates to tests to match go conventions

Signed-off-by: Zachary Speaks <[email protected]>

Signed-off-by: Zach Speaks <[email protected]>
Signed-off-by: Zachary Speaks <[email protected]>

Author: zspeaks

CHANGELOG.md

@@ -49,6 +49,7 @@
 * [FEATURE] Compactor: Added configurations for Azure MSI in blocks-storage, ruler-storage and alertmanager-storage. #4818
 * [FEATURE] Ruler: Add support to pass custom implementations of queryable and pusher. #4782
 * [FEATURE] Create OpenTelemetry Bridge for Tracing. Now cortex can send traces to multiple destinations using OTEL Collectors. #4834
+* [FEATURE] Added `-api.http-request-headers-to-log` allowing for the addition of HTTP Headers to logs #4803
 * [BUGFIX] Memberlist: Add join with no retrying when starting service. #4804
 * [BUGFIX] Ruler: Fix /ruler/rule_groups returns YAML with extra fields. #4767
 

docs/configuration/config-file-reference.md

@@ -82,6 +82,10 @@ api:
   # CLI flag: -http.prometheus-http-prefix
   [prometheus_http_prefix: <string> | default = "/prometheus"]
 
+  # Which HTTP Request headers to add to logs
+  # CLI flag: -api.http-request-headers-to-log
+  [http_request_headers_to_log: <list of string> | default = []]
+
 # The server_config configures the HTTP and gRPC server of the launched
 # service(s).
 [server: <server_config>]

pkg/api/api.go

@@ -34,6 +34,7 @@ import (
 	"github.com/cortexproject/cortex/pkg/scheduler/schedulerpb"
 	"github.com/cortexproject/cortex/pkg/storegateway"
 	"github.com/cortexproject/cortex/pkg/storegateway/storegatewaypb"
+	"github.com/cortexproject/cortex/pkg/util/flagext"
 	"github.com/cortexproject/cortex/pkg/util/push"
 )
 
@@ -61,11 +62,15 @@ type Config struct {
 	// initialized, the custom config handler will be used instead of
 	// DefaultConfigHandler.
 	CustomConfigHandler ConfigHandler `yaml:"-"`
+
+	// Allows and is used to configure the addition of HTTP Header fields to logs
+	HTTPRequestHeadersToLog flagext.StringSlice `yaml:"http_request_headers_to_log"`
 }
 
 // RegisterFlags adds the flags required to config this to the given FlagSet.
 func (cfg *Config) RegisterFlags(f *flag.FlagSet) {
 	f.BoolVar(&cfg.ResponseCompression, "api.response-compression-enabled", false, "Use GZIP compression for API responses. Some endpoints serve large YAML or JSON blobs which can benefit from compression.")
+	f.Var(&cfg.HTTPRequestHeadersToLog, "api.http-request-headers-to-log", "Which HTTP Request headers to add to logs")
 	cfg.RegisterFlagsWithPrefix("", f)
 }
 
@@ -85,13 +90,13 @@ func (cfg *Config) wrapDistributorPush(d *distributor.Distributor) push.Func {
 }
 
 type API struct {
-	AuthMiddleware middleware.Interface
-
-	cfg       Config
-	server    *server.Server
-	logger    log.Logger
-	sourceIPs *middleware.SourceIPExtractor
-	indexPage *IndexPageContent
+	AuthMiddleware       middleware.Interface
+	cfg                  Config
+	server               *server.Server
+	logger               log.Logger
+	sourceIPs            *middleware.SourceIPExtractor
+	indexPage            *IndexPageContent
+	HTTPHeaderMiddleware *HTTPHeaderMiddleware
 }
 
 func New(cfg Config, serverCfg server.Config, s *server.Server, logger log.Logger) (*API, error) {
@@ -121,6 +126,9 @@ func New(cfg Config, serverCfg server.Config, s *server.Server, logger log.Logge
 	if cfg.HTTPAuthMiddleware == nil {
 		api.AuthMiddleware = middleware.AuthenticateUser
 	}
+	if len(cfg.HTTPRequestHeadersToLog) > 0 {
+		api.HTTPHeaderMiddleware = &HTTPHeaderMiddleware{TargetHeaders: cfg.HTTPRequestHeadersToLog}
+	}
 
 	return api, nil
 }
@@ -139,6 +147,9 @@ func (a *API) RegisterRoute(path string, handler http.Handler, auth bool, method
 	if a.cfg.ResponseCompression {
 		handler = gziphandler.GzipHandler(handler)
 	}
+	if a.HTTPHeaderMiddleware != nil {
+		handler = a.HTTPHeaderMiddleware.Wrap(handler)
+	}
 
 	if len(methods) == 0 {
 		a.server.HTTP.Path(path).Handler(handler)
@@ -156,6 +167,9 @@ func (a *API) RegisterRoutesWithPrefix(prefix string, handler http.Handler, auth
 	if a.cfg.ResponseCompression {
 		handler = gziphandler.GzipHandler(handler)
 	}
+	if a.HTTPHeaderMiddleware != nil {
+		handler = a.HTTPHeaderMiddleware.Wrap(handler)
+	}
 
 	if len(methods) == 0 {
 		a.server.HTTP.PathPrefix(prefix).Handler(handler)

pkg/api/api_test.go

@@ -60,3 +60,37 @@ func TestNewApiWithInvalidSourceIPExtractor(t *testing.T) {
 	require.Error(t, err)
 	require.Nil(t, api)
 }
+
+func TestNewApiWithHeaderLogging(t *testing.T) {
+	cfg := Config{
+		HTTPRequestHeadersToLog: []string{"ForTesting"},
+	}
+	serverCfg := server.Config{
+		HTTPListenNetwork: server.DefaultNetwork,
+		MetricsNamespace:  "with_header_logging",
+	}
+	server, err := server.New(serverCfg)
+	require.NoError(t, err)
+
+	api, err := New(cfg, serverCfg, server, &FakeLogger{})
+	require.NoError(t, err)
+	require.NotNil(t, api.HTTPHeaderMiddleware)
+
+}
+
+func TestNewApiWithoutHeaderLogging(t *testing.T) {
+	cfg := Config{
+		HTTPRequestHeadersToLog: []string{},
+	}
+	serverCfg := server.Config{
+		HTTPListenNetwork: server.DefaultNetwork,
+		MetricsNamespace:  "without_header_logging",
+	}
+	server, err := server.New(serverCfg)
+	require.NoError(t, err)
+
+	api, err := New(cfg, serverCfg, server, &FakeLogger{})
+	require.NoError(t, err)
+	require.Nil(t, api.HTTPHeaderMiddleware)
+
+}

pkg/api/middlewares.go

@@ -1,13 +1,15 @@
 package api
 
 import (
+	"context"
 	"net/http"
 
 	"github.com/weaveworks/common/middleware"
 
 	"github.com/cortexproject/cortex/pkg/chunk/purger"
 	"github.com/cortexproject/cortex/pkg/querier/queryrange"
 	"github.com/cortexproject/cortex/pkg/tenant"
+	util_log "github.com/cortexproject/cortex/pkg/util/log"
 )
 
 // middleware for setting cache gen header to let consumer of response know all previous responses could be invalid due to delete operation
@@ -27,3 +29,35 @@ func getHTTPCacheGenNumberHeaderSetterMiddleware(cacheGenNumbersLoader purger.To
 		})
 	})
 }
+
+// HTTPHeaderMiddleware adds specified HTTPHeaders to the request context
+type HTTPHeaderMiddleware struct {
+	TargetHeaders []string
+}
+
+// InjectTargetHeadersIntoHTTPRequest injects specified HTTPHeaders into the request context
+func (h HTTPHeaderMiddleware) InjectTargetHeadersIntoHTTPRequest(r *http.Request) context.Context {
+	headerMap := make(map[string]string)
+
+	// Check to make sure that Headers have not already been injected
+	checkMapInContext := util_log.HeaderMapFromContext(r.Context())
+	if checkMapInContext != nil {
+		return r.Context()
+	}
+
+	for _, target := range h.TargetHeaders {
+		contents := r.Header.Get(target)
+		if contents != "" {
+			headerMap[target] = contents
+		}
+	}
+	return util_log.ContextWithHeaderMap(r.Context(), headerMap)
+}
+
+// Wrap implements Middleware
+func (h HTTPHeaderMiddleware) Wrap(next http.Handler) http.Handler {
+	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		ctx := h.InjectTargetHeadersIntoHTTPRequest(r)
+		next.ServeHTTP(w, r.WithContext(ctx))
+	})
+}

pkg/api/middlewares_test.go

@@ -0,0 +1,74 @@
+package api
+
+import (
+	"context"
+	"net/http"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+
+	util_log "github.com/cortexproject/cortex/pkg/util/log"
+)
+
+var HTTPTestMiddleware = HTTPHeaderMiddleware{TargetHeaders: []string{"TestHeader1", "TestHeader2", "Test3"}}
+
+func TestHeaderInjection(t *testing.T) {
+	ctx := context.Background()
+	h := http.Header{}
+	contentsMap := make(map[string]string)
+	contentsMap["TestHeader1"] = "RequestID"
+	contentsMap["TestHeader2"] = "ContentsOfTestHeader2"
+	contentsMap["Test3"] = "SomeInformation"
+
+	h.Add("TestHeader1", contentsMap["TestHeader1"])
+	h.Add("TestHeader2", contentsMap["TestHeader2"])
+	h.Add("Test3", contentsMap["Test3"])
+
+	req := &http.Request{
+		Method:     "GET",
+		RequestURI: "/HTTPHeaderTest",
+		Body:       http.NoBody,
+		Header:     h,
+	}
+
+	req = req.WithContext(ctx)
+	ctx = HTTPTestMiddleware.InjectTargetHeadersIntoHTTPRequest(req)
+
+	headerMap := util_log.HeaderMapFromContext(ctx)
+	require.NotNil(t, headerMap)
+
+	for _, header := range HTTPTestMiddleware.TargetHeaders {
+		require.Equal(t, contentsMap[header], headerMap[header])
+	}
+	for header, contents := range contentsMap {
+		require.Equal(t, contents, headerMap[header])
+	}
+}
+
+func TestExistingHeaderInContextIsNotOverridden(t *testing.T) {
+	ctx := context.Background()
+
+	h := http.Header{}
+	contentsMap := make(map[string]string)
+	contentsMap["TestHeader1"] = "RequestID"
+	contentsMap["TestHeader2"] = "ContentsOfTestHeader2"
+	contentsMap["Test3"] = "SomeInformation"
+
+	h.Add("TestHeader1", "Fail1")
+	h.Add("TestHeader2", "Fail2")
+	h.Add("Test3", "Fail3")
+
+	ctx = util_log.ContextWithHeaderMap(ctx, contentsMap)
+	req := &http.Request{
+		Method:     "GET",
+		RequestURI: "/HTTPHeaderTest",
+		Body:       http.NoBody,
+		Header:     h,
+	}
+
+	req = req.WithContext(ctx)
+	ctx = HTTPTestMiddleware.InjectTargetHeadersIntoHTTPRequest(req)
+
+	require.Equal(t, contentsMap, util_log.HeaderMapFromContext(ctx))
+
+}

pkg/cortex/cortex.go

@@ -353,6 +353,7 @@ func New(cfg Config) (*Cortex, error) {
 	}
 
 	cortex.setupThanosTracing()
+	cortex.setupGRPCHeaderForwarding()
 
 	if err := cortex.setupModuleManager(); err != nil {
 		return nil, err
@@ -368,6 +369,15 @@ func (t *Cortex) setupThanosTracing() {
 	t.Cfg.Server.GRPCStreamMiddleware = append(t.Cfg.Server.GRPCStreamMiddleware, ThanosTracerStreamInterceptor)
 }
 
+// setupGRPCHeaderForwarding appends a gRPC middleware used to enable the propagation of
+// HTTP Headers through child gRPC calls
+func (t *Cortex) setupGRPCHeaderForwarding() {
+	if len(t.Cfg.API.HTTPRequestHeadersToLog) > 0 {
+		t.Cfg.Server.GRPCMiddleware = append(t.Cfg.Server.GRPCMiddleware, grpcutil.HTTPHeaderPropagationServerInterceptor)
+		t.Cfg.Server.GRPCStreamMiddleware = append(t.Cfg.Server.GRPCStreamMiddleware, grpcutil.HTTPHeaderPropagationStreamServerInterceptor)
+	}
+}
+
 // Run starts Cortex running, and blocks until a Cortex stops.
 func (t *Cortex) Run() error {
 	// Register custom process metrics.

pkg/cortex/modules.go

@@ -328,6 +328,10 @@ func (t *Cortex) initQuerier() (serv services.Service, err error) {
 		// HTTP router with middleware to parse the tenant ID from the HTTP header and inject it into the
 		// request context.
 		internalQuerierRouter = t.API.AuthMiddleware.Wrap(internalQuerierRouter)
+
+		if len(t.Cfg.API.HTTPRequestHeadersToLog) > 0 {
+			internalQuerierRouter = t.API.HTTPHeaderMiddleware.Wrap(internalQuerierRouter)
+		}
 	}
 
 	// If neither frontend address or scheduler address is configured, no worker is needed.

pkg/distributor/distributor.go

@@ -786,6 +786,10 @@ func (d *Distributor) Push(ctx context.Context, req *cortexpb.WriteRequest) (*co
 		if sp := opentracing.SpanFromContext(ctx); sp != nil {
 			localCtx = opentracing.ContextWithSpan(localCtx, sp)
 		}
+		// Get any HTTP headers that are supposed to be added to logs and add to localCtx for later use
+		if headerMap := util_log.HeaderMapFromContext(ctx); headerMap != nil {
+			localCtx = util_log.ContextWithHeaderMap(localCtx, headerMap)
+		}
 
 		// Get clientIP(s) from Context and add it to localCtx
 		localCtx = util.AddSourceIPsToOutgoingContext(localCtx, source)