VAULT-1564 report in-flight requests

changelog/13024.txt

@@ -0,0 +1,3 @@
+```release-note:feature
+**Report in-flight requests**:Adding a trace capability to show in-flight requests, and a new gauge metric to show the total number of in-flight requests
+```

command/debug.go

@@ -17,6 +17,7 @@ import (
 	"github.com/hashicorp/go-secure-stdlib/gatedwriter"
 	"github.com/hashicorp/go-secure-stdlib/strutil"
 	"github.com/hashicorp/vault/api"
+	"github.com/hashicorp/vault/sdk/helper/jsonutil"
 	"github.com/hashicorp/vault/sdk/helper/logging"
 	"github.com/hashicorp/vault/sdk/version"
 	"github.com/mholt/archiver"
@@ -106,6 +107,7 @@ type DebugCommand struct {
 	metricsCollection           []map[string]interface{}
 	replicationStatusCollection []map[string]interface{}
 	serverStatusCollection      []map[string]interface{}
+	inFlightReqStatusCollection []map[string]interface{}
 
 	// cachedClient holds the client retrieved during preflight
 	cachedClient *api.Client
@@ -480,7 +482,7 @@ func (c *DebugCommand) preflight(rawArgs []string) (string, error) {
 }
 
 func (c *DebugCommand) defaultTargets() []string {
-	return []string{"config", "host", "metrics", "pprof", "replication-status", "server-status", "log"}
+	return []string{"config", "host", "requests", "metrics", "pprof", "replication-status", "server-status", "log"}
 }
 
 func (c *DebugCommand) captureStaticTargets() error {
@@ -492,6 +494,7 @@ func (c *DebugCommand) captureStaticTargets() error {
 		if err != nil {
 			c.captureError("config", err)
 			c.logger.Error("config: error capturing config state", "error", err)
+			return nil
 		}
 
 		if resp != nil && resp.Data != nil {
@@ -580,6 +583,16 @@ func (c *DebugCommand) capturePollingTargets() error {
 		})
 	}
 
+	// Collect in-flight request status if target is specified
+	if strutil.StrListContains(c.flagTargets, "requests") {
+		g.Add(func() error {
+			c.collectInFlightRequestStatus(ctx)
+			return nil
+		}, func(error) {
+			cancelFunc()
+		})
+	}
+
 	if strutil.StrListContains(c.flagTargets, "log") {
 		g.Add(func() error {
 			c.writeLogs(ctx)
@@ -611,7 +624,9 @@ func (c *DebugCommand) capturePollingTargets() error {
 	if err := c.persistCollection(c.hostInfoCollection, "host_info.json"); err != nil {
 		c.UI.Error(fmt.Sprintf("Error writing data to %s: %v", "host_info.json", err))
 	}
-
+	if err := c.persistCollection(c.inFlightReqStatusCollection, "requests.json"); err != nil {
+		c.UI.Error(fmt.Sprintf("Error writing data to %s: %v", "requests.json", err))
+	}
 	return nil
 }
 
@@ -635,13 +650,15 @@ func (c *DebugCommand) collectHostInfo(ctx context.Context) {
 		resp, err := c.cachedClient.RawRequestWithContext(ctx, r)
 		if err != nil {
 			c.captureError("host", err)
+			return
 		}
 		if resp != nil {
 			defer resp.Body.Close()
 
 			secret, err := api.ParseSecret(resp.Body)
 			if err != nil {
 				c.captureError("host", err)
+				return
 			}
 			if secret != nil && secret.Data != nil {
 				hostEntry := secret.Data
@@ -829,13 +846,15 @@ func (c *DebugCommand) collectReplicationStatus(ctx context.Context) {
 		resp, err := c.cachedClient.RawRequestWithContext(ctx, r)
 		if err != nil {
 			c.captureError("replication-status", err)
+			return
 		}
 		if resp != nil {
 			defer resp.Body.Close()
 
 			secret, err := api.ParseSecret(resp.Body)
 			if err != nil {
 				c.captureError("replication-status", err)
+				return
 			}
 			if secret != nil && secret.Data != nil {
 				replicationEntry := secret.Data
@@ -880,6 +899,48 @@ func (c *DebugCommand) collectServerStatus(ctx context.Context) {
 	}
 }
 
+func (c *DebugCommand) collectInFlightRequestStatus(ctx context.Context) {
+
+	idxCount := 0
+	intervalTicker := time.Tick(c.flagInterval)
+
+	for {
+		if idxCount > 0 {
+			select {
+			case <-ctx.Done():
+				return
+			case <-intervalTicker:
+			}
+		}
+
+		c.logger.Info("capturing in-flight request status", "count", idxCount)
+		idxCount++
+
+		req := c.cachedClient.NewRequest("GET", "/v1/sys/in-flight-req")
+		resp, err := c.cachedClient.RawRequestWithContext(ctx, req)
+		if err != nil {
+			c.captureError("requests", err)
+			return
+		}
+
+		var data map[string]interface{}
+		if resp != nil {
+			defer resp.Body.Close()
+			err = jsonutil.DecodeJSONFromReader(resp.Body, &data)
+			if err != nil {
+				c.captureError("requests", err)
+				return
+			}
+
+			statusEntry := map[string]interface{}{
+				"timestamp":          time.Now().UTC(),
+				"in_flight_requests": data,
+			}
+			c.inFlightReqStatusCollection = append(c.inFlightReqStatusCollection, statusEntry)
+		}
+	}
+}
+
 // persistCollection writes the collected data for a particular target onto the
 // specified file. If the collection is empty, it returns immediately.
 func (c *DebugCommand) persistCollection(collection []map[string]interface{}, outFile string) error {

command/debug_test.go

@@ -235,6 +235,11 @@ func TestDebugCommand_CaptureTargets(t *testing.T) {
 			[]string{"server-status"},
 			[]string{"server_status.json"},
 		},
+		{
+			"in-flight-req",
+			[]string{"requests"},
+			[]string{"requests.json"},
+		},
 		{
 			"all-minus-pprof",
 			[]string{"config", "host", "metrics", "replication-status", "server-status"},

command/server.go

@@ -1547,6 +1547,9 @@ func (c *ServerCommand) Run(args []string) int {
 				c.logger.Error(err.Error())
 			}
 
+			// Setting log request with the new value in the config after reload
+			core.ReloadLogRequestsLevel()
+
 			if config.LogLevel != "" {
 				configLogLevel := strings.ToLower(strings.TrimSpace(config.LogLevel))
 				switch configLogLevel {

command/server/config.go

@@ -77,6 +77,9 @@ type Config struct {
 	EnableResponseHeaderHostname    bool        `hcl:"-"`
 	EnableResponseHeaderHostnameRaw interface{} `hcl:"enable_response_header_hostname"`
 
+	LogRequestsLevel    string      `hcl:"-"`
+	LogRequestsLevelRaw interface{} `hcl:"log_requests_level"`
+
 	EnableResponseHeaderRaftNodeID    bool        `hcl:"-"`
 	EnableResponseHeaderRaftNodeIDRaw interface{} `hcl:"enable_response_header_raft_node_id"`
 
@@ -292,6 +295,11 @@ func (c *Config) Merge(c2 *Config) *Config {
 		result.EnableResponseHeaderHostname = c2.EnableResponseHeaderHostname
 	}
 
+	result.LogRequestsLevel = c.LogRequestsLevel
+	if c2.LogRequestsLevel != "" {
+		result.LogRequestsLevel = c2.LogRequestsLevel
+	}
+
 	result.EnableResponseHeaderRaftNodeID = c.EnableResponseHeaderRaftNodeID
 	if c2.EnableResponseHeaderRaftNodeID {
 		result.EnableResponseHeaderRaftNodeID = c2.EnableResponseHeaderRaftNodeID
@@ -480,6 +488,11 @@ func ParseConfig(d, source string) (*Config, error) {
 		}
 	}
 
+	if result.LogRequestsLevelRaw != nil {
+		result.LogRequestsLevel = strings.ToLower(strings.TrimSpace(result.LogRequestsLevelRaw.(string)))
+		result.LogRequestsLevelRaw = ""
+	}
+
 	if result.EnableResponseHeaderRaftNodeIDRaw != nil {
 		if result.EnableResponseHeaderRaftNodeID, err = parseutil.ParseBool(result.EnableResponseHeaderRaftNodeIDRaw); err != nil {
 			return nil, err
@@ -840,6 +853,8 @@ func (c *Config) Sanitized() map[string]interface{} {
 		"enable_response_header_hostname": c.EnableResponseHeaderHostname,
 
 		"enable_response_header_raft_node_id": c.EnableResponseHeaderRaftNodeID,
+
+		"log_requests_level": c.LogRequestsLevel,
 	}
 	for k, v := range sharedResult {
 		result[k] = v

command/server/config_test_helpers.go

@@ -701,6 +701,7 @@ func testConfig_Sanitized(t *testing.T) {
 		"enable_ui":                           true,
 		"enable_response_header_hostname":     false,
 		"enable_response_header_raft_node_id": false,
+		"log_requests_level":                   "basic",
 		"ha_storage": map[string]interface{}{
 			"cluster_addr":       "top_level_cluster_addr",
 			"disable_clustering": true,

command/server/listener_test.go

@@ -6,6 +6,7 @@ import (
 	"io"
 	"net"
 	"testing"
+
 )
 
 type testListenerConnFn func(net.Listener) (net.Conn, error)
@@ -66,3 +67,15 @@ func testListenerImpl(t *testing.T, ln net.Listener, connFn testListenerConnFn,
 		t.Fatalf("bad: %v", buf.String())
 	}
 }
+
+
+func TestProfilingUnauthenticatedInFlightAccess(t *testing.T) {
+
+	config, err := LoadConfigFile("./test-fixtures/unauth_in_flight_access.hcl")
+	if err != nil {
+		t.Fatalf("Error encountered when loading config %+v", err)
+	}
+	if !config.Listeners[0].InFlightRequestLogging.UnauthenticatedInFlightAccess {
+		t.Fatalf("failed to read UnauthenticatedInFlightAccess")
+	}
+}

command/server/test-fixtures/config3.hcl

@@ -1,5 +1,6 @@
 disable_cache = true
 disable_mlock = true
+log_requests_level = "Basic"
 
 ui = true
 

command/server/test-fixtures/unauth_in_flight_access.hcl

@@ -0,0 +1,9 @@
+storage "inmem" {}
+listener "tcp" {
+  address = "127.0.0.1:8200"
+  tls_disable = true
+  inflight_requests_logging {
+     unauthenticated_in_flight_requests_access = true
+  }
+}
+disable_mlock = true

http/handler.go

@@ -25,6 +25,7 @@ import (
 	"github.com/hashicorp/go-cleanhttp"
 	"github.com/hashicorp/go-secure-stdlib/parseutil"
 	"github.com/hashicorp/go-sockaddr"
+	"github.com/hashicorp/go-uuid"
 	"github.com/hashicorp/vault/helper/namespace"
 	"github.com/hashicorp/vault/internalshared/configutil"
 	"github.com/hashicorp/vault/sdk/helper/consts"
@@ -195,6 +196,11 @@ func Handler(props *vault.HandlerProperties) http.Handler {
 			mux.Handle("/v1/sys/pprof/", handleLogicalNoForward(core))
 		}
 
+		if props.ListenerConfig != nil && props.ListenerConfig.InFlightRequestLogging.UnauthenticatedInFlightAccess {
+			mux.Handle("/v1/sys/in-flight-req", handleUnAuthenticatedInFlightRequest(core))
+		} else {
+			mux.Handle("/v1/sys/in-flight-req", handleLogicalNoForward(core))
+		}
 		additionalRoutes(mux, core)
 	}
 
@@ -314,8 +320,11 @@ func wrapGenericHandler(core *vault.Core, h http.Handler, props *vault.HandlerPr
 				customHeaders = listenerCustomHeaders.StatusCodeHeaderMap
 			}
 		}
+		// saving start time for the in-flight requests
+		inFlightReqStartTime := time.Now()
 
 		nw := logical.NewStatusHeaderResponseWriter(w, customHeaders)
+
 		// Set the Cache-Control header for all the responses returned
 		// by Vault
 		nw.Header().Set("Cache-Control", "no-store")
@@ -367,6 +376,43 @@ func wrapGenericHandler(core *vault.Core, h http.Handler, props *vault.HandlerPr
 			return
 		}
 
+		// The uuid for the request is going to be generated when a logical
+		// request is generated. But, here we generate one to be able to track
+		// in-flight requests, and use that to update the req data with clientID
+		inFlightReqID, err := uuid.GenerateUUID()
+		if err != nil {
+			respondError(nw, http.StatusInternalServerError, fmt.Errorf("failed to generate an identifier for the in-flight request"))
+		}
+		// adding an entry to the context to enable updating in-flight
+		// data with ClientID in the logical layer
+		r = r.WithContext(context.WithValue(r.Context(), logical.CtxKeyInFlightRequestID{}, inFlightReqID))
+
+		// extracting the client address to be included in the in-flight request
+		var clientAddr string
+		headers := r.Header[textproto.CanonicalMIMEHeaderKey("X-Forwarded-For")]
+		if len(headers) == 0 {
+			clientAddr = r.RemoteAddr
+		} else {
+			clientAddr = headers[0]
+		}
+
+		// getting the request method
+		requestMethod := r.Method
+
+		// Storing the in-flight requests. Path should include namespace as well
+		core.StoreInFlightReqData(
+			inFlightReqID,
+			vault.InFlightReqData {
+				StartTime: inFlightReqStartTime,
+				ReqPath: r.URL.Path,
+				ClientRemoteAddr: clientAddr,
+				Method: requestMethod,
+			})
+		defer func() {
+			// Not expecting this fail, so skipping the assertion check
+			core.FinalizeInFlightReqData(inFlightReqID, nw.StatusCode)
+		}()
+
 		// Setting the namespace in the header to be included in the error message
 		ns := r.Header.Get(consts.NamespaceHeaderName)
 		if ns != "" {

http/handler_test.go

@@ -294,6 +294,45 @@ func TestHandler_CacheControlNoStore(t *testing.T) {
 	}
 }
 
+func TestHandler_InFlightRequest(t *testing.T) {
+	core, _, token := vault.TestCoreUnsealed(t)
+	ln, addr := TestServer(t, core)
+	defer ln.Close()
+	TestServerAuth(t, addr, token)
+
+	req, err := http.NewRequest("GET", addr+"/v1/sys/in-flight-req", nil)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+	req.Header.Set(consts.AuthHeaderName, token)
+
+	client := cleanhttp.DefaultClient()
+	resp, err := client.Do(req)
+	if err != nil {
+		t.Fatalf("err: %s", err)
+	}
+
+	if resp == nil {
+		t.Fatalf("nil response")
+	}
+
+	var actual map[string]interface{}
+	testResponseStatus(t, resp, 200)
+	testResponseBody(t, resp, &actual)
+	if actual == nil || len(actual) == 0 {
+		t.Fatal("expected to get at least one in-flight request, got nil or zero length map")
+	}
+	for _, v := range actual {
+		reqInfo, ok := v.(map[string]interface{})
+		if !ok {
+			t.Fatal("failed to read in-flight request")
+		}
+		if reqInfo["request_path"] != "/v1/sys/in-flight-req" {
+			t.Fatalf("expected /v1/sys/in-flight-req in-flight request path, got %s", actual["request_path"])
+		}
+	}
+}
+
 // TestHandler_MissingToken tests the response / error code if a request comes
 // in with a missing client token. See
 // https://github.com/hashicorp/vault/issues/8377

http/logical.go

@@ -183,7 +183,7 @@ func buildLogicalRequestNoAuth(perfStandby bool, w http.ResponseWriter, r *http.
 
 	requestId, err := uuid.GenerateUUID()
 	if err != nil {
-		return nil, nil, http.StatusBadRequest, fmt.Errorf("failed to generate identifier for the request: %w", err)
+		return nil, nil, http.StatusInternalServerError, fmt.Errorf("failed to generate identifier for the request: %w", err)
 	}
 
 	req := &logical.Request{

http/sys_config_state_test.go

@@ -48,6 +48,7 @@ func TestSysConfigState_Sanitized(t *testing.T) {
 		"plugin_directory":                    "",
 		"enable_response_header_hostname":     false,
 		"enable_response_header_raft_node_id": false,
+		"log_requests_level":                   "",
 	}
 
 	expected = map[string]interface{}{