api: Add all in-flight requests /reverse_proxy/upstreams (#7277)

Author: u5surf

modules/caddyhttp/reverseproxy/admin.go

@@ -102,6 +102,33 @@ func (adminUpstreams) handleUpstreams(w http.ResponseWriter, r *http.Request) er
 		})
 		return true
 	})
+	// Iterate over the inflight hosts
+	inflightHosts.Range(func(key, val any) bool {
+		address, ok := key.(string)
+		if !ok {
+			rangeErr = caddy.APIError{
+				HTTPStatus: http.StatusInternalServerError,
+				Err:        fmt.Errorf("could not type assert upstream address"),
+			}
+			return false
+		}
+
+		upstream, ok := val.(*Host)
+		if !ok {
+			rangeErr = caddy.APIError{
+				HTTPStatus: http.StatusInternalServerError,
+				Err:        fmt.Errorf("could not type assert upstream struct"),
+			}
+			return false
+		}
+
+		results = append(results, upstreamStatus{
+			Address:     address,
+			NumRequests: upstream.NumRequests(),
+			Fails:       upstream.Fails(),
+		})
+		return true
+	})
 
 	// If an error happened during the range, return it
 	if rangeErr != nil {

modules/caddyhttp/reverseproxy/hosts.go

@@ -132,6 +132,15 @@ func (u *Upstream) fillHost() {
 	u.Host = host
 }
 
+func (u *Upstream) fillInfilghtHost() {
+	host := new(Host)
+	existingHost, loaded := inflightHosts.LoadOrStore(u.String(), host)
+	if loaded {
+		host = existingHost.(*Host)
+	}
+	u.Host = host
+}
+
 // Host is the basic, in-memory representation of the state of a remote host.
 // Its fields are accessed atomically and Host values must not be copied.
 type Host struct {
@@ -268,6 +277,10 @@ func GetDialInfo(ctx context.Context) (DialInfo, bool) {
 // through config reloads.
 var hosts = caddy.NewUsagePool()
 
+// inflightHosts is the global repository for hosts that are
+// currently in use by inflight upstream request.
+var inflightHosts = caddy.NewUsagePool()
+
 // dialInfoVarKey is the key used for the variable that holds
 // the dial info for the upstream connection.
 const dialInfoVarKey = "reverse_proxy.dial_info"

modules/caddyhttp/reverseproxy/reverseproxy.go

@@ -394,6 +394,9 @@ func (h *Handler) Cleanup() error {
 
 	// remove hosts from our config from the pool
 	for _, upstream := range h.Upstreams {
+		if upstream.NumRequests() > 0 {
+			upstream.fillInfilghtHost()
+		}
 		_, _ = hosts.Delete(upstream.String())
 	}
 
@@ -460,6 +463,26 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request, next caddyht
 		var done bool
 		done, proxyErr = h.proxyLoopIteration(clonedReq, r, w, proxyErr, start, retries, repl, reqHeader, reqHost, next)
 		if done {
+			// Iterate over the upstream pool (needs to be fast)
+			var rangeErr error
+			inflightHosts.Range(func(key, val any) bool {
+				host, ok := val.(*Host)
+				if !ok {
+					rangeErr = caddy.APIError{
+						HTTPStatus: http.StatusInternalServerError,
+						Err:        fmt.Errorf("could not type assert upstream struct"),
+					}
+					return false
+				}
+				if host.NumRequests() == 0 {
+					_, _ = inflightHosts.Delete(reqHost)
+				}
+				return true
+			})
+			// If an error happened during the range, return it
+			if rangeErr != nil {
+				return statusError(rangeErr)
+			}
 			break
 		}
 		if h.VerboseLogs {