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
@@ -865,10 +884,12 @@ func (c *DebugCommand) collectServerStatus(ctx context.Context) {
 		healthInfo, err := c.cachedClient.Sys().Health()
 		if err != nil {
 			c.captureError("server-status.health", err)
+			return
 		}
 		sealInfo, err := c.cachedClient.Sys().SealStatus()
 		if err != nil {
 			c.captureError("server-status.seal", err)
+			return
 		}
 
 		statusEntry := map[string]interface{}{
@@ -880,6 +901,50 @@ 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("inFlightReq-status", 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("inFlightReq-status", err)
+				return
+			}
+
+			if data != nil && len(data) > 0 {
+				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/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/profiling_unauth_in_flight_access.hcl")
+	if err != nil {
+		t.Fatalf("Error encountered when loading config %+v", err)
+	}
+	if !config.Listeners[0].Profiling.UnauthenticatedInFlightAccess {
+		t.Fatalf("failed to read UnauthenticatedInFlightAccess")
+	}
+}

command/server/test-fixtures/profiling_unauth_in_flight_access.hcl

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

http/handler.go

@@ -6,6 +6,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"github.com/hashicorp/go-uuid"
 	"io"
 	"io/fs"
 	"io/ioutil"
@@ -196,6 +197,11 @@ func Handler(props *vault.HandlerProperties) http.Handler {
 			mux.Handle("/v1/sys/pprof/", handleLogicalNoForward(core))
 		}
 
+		if props.ListenerConfig != nil && props.ListenerConfig.Profiling.UnauthenticatedInFlightAccess {
+			mux.Handle("/v1/sys/in-flight-req", handleUnAuthenticatedInFlightRequest(core))
+		} else {
+			mux.Handle("/v1/sys/in-flight-req", handleLogicalNoForward(core))
+		}
 		additionalRoutes(mux, core)
 	}
 
@@ -389,6 +395,27 @@ func wrapGenericHandler(core *vault.Core, h http.Handler, props *vault.HandlerPr
 	hostname, _ := os.Hostname()
 
 	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+
+		// 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
+		inFlightReqID, err := uuid.GenerateUUID()
+		if err != nil {
+			respondError(w, http.StatusInternalServerError, fmt.Errorf("failed to generate an identifier for the in-flight request"))
+		}
+		core.StoreInFlightReqData(
+			inFlightReqID,
+			&vault.InFlightReqData {
+				StartTime: time.Now(),
+				ClientRemoteAddr: r.RemoteAddr,
+				ReqPath: r.URL.Path,
+			})
+		if err != nil {
+			respondError(w, http.StatusBadRequest, err)
+		}
+		// deleting the in-flight request entry
+		defer core.DeleteInFlightReqData(inFlightReqID)
+
 		// This block needs to be here so that upon sending SIGHUP, custom response
 		// headers are also reloaded into the handlers.
 		if props.ListenerConfig != nil {

http/handler_test.go

@@ -11,6 +11,7 @@ import (
 	"net/textproto"
 	"net/url"
 	"reflect"
+	"strconv"
 	"strings"
 	"testing"
 	"time"
@@ -294,6 +295,103 @@ 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 {
+		t.Fatalf("")
+	}
+}
+
+func TestHandler_InFlightRequestWithLoad(t *testing.T) {
+	core, _, token := vault.TestCoreUnsealed(t)
+	ln, addr := TestServer(t, core)
+	defer ln.Close()
+	TestServerAuth(t, addr, token)
+
+	stop := make(chan string)
+
+	go func() {
+		i := 0
+		for {
+			select {
+			case <-stop:
+				return
+			default:
+				break
+			}
+			// WRITE
+			secResp := testHttpPut(t, token, addr+"/v1/secret/foo"+strconv.Itoa(i), map[string]interface{}{
+				"data": "bar",
+			})
+			testResponseStatus(t, secResp, 204)
+			i++
+		}
+	}()
+
+	timeout := time.After(10 * time.Second)
+
+	for {
+		select {
+		case <-timeout:
+			stop <- "done"
+			t.Fatalf("could not capture any in-flight-req")
+			return
+		default:
+		}
+		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 inFlightReqData map[string]interface{}
+		testResponseStatus(t, resp, 200)
+		testResponseBody(t, resp, &inFlightReqData)
+
+		if inFlightReqData == nil {
+			t.Fatalf("")
+		}
+
+		if inFlightReqData != nil || len(inFlightReqData) > 0 {
+			stop <- "done"
+			return
+		}
+	}
+}
+
 // 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{