Hot Reloading: HTTP Middleware Reloading

Part of dapr#1172
Branched from dapr#7260

Adds support for HTTP middleware hot reloading.

PR adds support for HTTP middlewares to be dynamically updated. The HTTP
Middleware returns a middleware.HTTP pipleine built from a config spec.
This pipeline implements a single handler which it itself runs the spec
configured HTTP middleware chain. When a middleware Component is added
or removed, the HTTP middleware store is updated and both the HTTP
server and App HTTP channel's pipeline's chains are updated dynamically.

Like today, if a pipeline spec contains handlers whose names, version,
or type does not match that from the store, then the handler is skipped
for that chain. Pipeline handler order is preserved (i.e. reverse order).

The middleware store has been made generic so can be used in future for
planned gRPC middleware support.

Middleware init has been moved from the runtime channel manager to the
runtime processor init procedure to allow for dynamic loading. The
middleware HTTP manager is passed to the runtime processor to expose the
store, and relevant built pipelines passed to the HTTP app channel and
HTTP server.

Middleware integration tests has been added for daprd to check
functionality, ordering, and hotreloading,

Signed-off-by: joshvanl <me@joshvanl.dev>

cmd/daprd/components/middleware_http_bearer.go

@@ -18,16 +18,16 @@ package components
 import (
 	"context"
 
-	"github.com/dapr/components-contrib/middleware"
+	contribmiddleware "github.com/dapr/components-contrib/middleware"
 	"github.com/dapr/components-contrib/middleware/http/bearer"
 	httpMiddlewareLoader "github.com/dapr/dapr/pkg/components/middleware/http"
-	httpMiddleware "github.com/dapr/dapr/pkg/middleware/http"
+	"github.com/dapr/dapr/pkg/middleware"
 	"github.com/dapr/kit/logger"
 )
 
 func init() {
 	httpMiddlewareLoader.DefaultRegistry.RegisterComponent(func(log logger.Logger) httpMiddlewareLoader.FactoryMethod {
-		return func(metadata middleware.Metadata) (httpMiddleware.Middleware, error) {
+		return func(metadata contribmiddleware.Metadata) (middleware.HTTP, error) {
 			return bearer.NewBearerMiddleware(log).GetHandler(context.TODO(), metadata)
 		}
 	}, "bearer")

cmd/daprd/components/middleware_http_oauth2.go

@@ -18,16 +18,16 @@ package components
 import (
 	"context"
 
-	"github.com/dapr/components-contrib/middleware"
+	contribmiddleware "github.com/dapr/components-contrib/middleware"
 	"github.com/dapr/components-contrib/middleware/http/oauth2"
 	httpMiddlewareLoader "github.com/dapr/dapr/pkg/components/middleware/http"
-	httpMiddleware "github.com/dapr/dapr/pkg/middleware/http"
+	"github.com/dapr/dapr/pkg/middleware"
 	"github.com/dapr/kit/logger"
 )
 
 func init() {
 	httpMiddlewareLoader.DefaultRegistry.RegisterComponent(func(log logger.Logger) httpMiddlewareLoader.FactoryMethod {
-		return func(metadata middleware.Metadata) (httpMiddleware.Middleware, error) {
+		return func(metadata contribmiddleware.Metadata) (middleware.HTTP, error) {
 			return oauth2.NewOAuth2Middleware(log).GetHandler(context.TODO(), metadata)
 		}
 	}, "oauth2")

cmd/daprd/components/middleware_http_oauth2clientcredentials.go

@@ -18,16 +18,16 @@ package components
 import (
 	"context"
 
-	"github.com/dapr/components-contrib/middleware"
+	contribmiddleware "github.com/dapr/components-contrib/middleware"
 	"github.com/dapr/components-contrib/middleware/http/oauth2clientcredentials"
 	httpMiddlewareLoader "github.com/dapr/dapr/pkg/components/middleware/http"
-	httpMiddleware "github.com/dapr/dapr/pkg/middleware/http"
+	"github.com/dapr/dapr/pkg/middleware"
 	"github.com/dapr/kit/logger"
 )
 
 func init() {
 	httpMiddlewareLoader.DefaultRegistry.RegisterComponent(func(log logger.Logger) httpMiddlewareLoader.FactoryMethod {
-		return func(metadata middleware.Metadata) (httpMiddleware.Middleware, error) {
+		return func(metadata contribmiddleware.Metadata) (middleware.HTTP, error) {
 			return oauth2clientcredentials.NewOAuth2ClientCredentialsMiddleware(log).GetHandler(context.TODO(), metadata)
 		}
 	}, "oauth2clientcredentials")

cmd/daprd/components/middleware_http_opa.go

@@ -18,16 +18,16 @@ package components
 import (
 	"context"
 
-	"github.com/dapr/components-contrib/middleware"
+	contribmiddleware "github.com/dapr/components-contrib/middleware"
 	"github.com/dapr/components-contrib/middleware/http/opa"
 	httpMiddlewareLoader "github.com/dapr/dapr/pkg/components/middleware/http"
-	httpMiddleware "github.com/dapr/dapr/pkg/middleware/http"
+	"github.com/dapr/dapr/pkg/middleware"
 	"github.com/dapr/kit/logger"
 )
 
 func init() {
 	httpMiddlewareLoader.DefaultRegistry.RegisterComponent(func(log logger.Logger) httpMiddlewareLoader.FactoryMethod {
-		return func(metadata middleware.Metadata) (httpMiddleware.Middleware, error) {
+		return func(metadata contribmiddleware.Metadata) (middleware.HTTP, error) {
 			return opa.NewMiddleware(log).GetHandler(context.TODO(), metadata)
 		}
 	}, "opa")

cmd/daprd/components/middleware_http_ratelimit.go

@@ -18,16 +18,16 @@ package components
 import (
 	"context"
 
-	"github.com/dapr/components-contrib/middleware"
+	contribmiddleware "github.com/dapr/components-contrib/middleware"
 	"github.com/dapr/components-contrib/middleware/http/ratelimit"
 	httpMiddlewareLoader "github.com/dapr/dapr/pkg/components/middleware/http"
-	httpMiddleware "github.com/dapr/dapr/pkg/middleware/http"
+	"github.com/dapr/dapr/pkg/middleware"
 	"github.com/dapr/kit/logger"
 )
 
 func init() {
 	httpMiddlewareLoader.DefaultRegistry.RegisterComponent(func(log logger.Logger) httpMiddlewareLoader.FactoryMethod {
-		return func(metadata middleware.Metadata) (httpMiddleware.Middleware, error) {
+		return func(metadata contribmiddleware.Metadata) (middleware.HTTP, error) {
 			return ratelimit.NewRateLimitMiddleware(log).GetHandler(context.TODO(), metadata)
 		}
 	}, "ratelimit")

cmd/daprd/components/middleware_http_routeralias.go

@@ -18,16 +18,16 @@ package components
 import (
 	"context"
 
-	"github.com/dapr/components-contrib/middleware"
+	contribmiddleware "github.com/dapr/components-contrib/middleware"
 	"github.com/dapr/components-contrib/middleware/http/routeralias"
 	httpMiddlewareLoader "github.com/dapr/dapr/pkg/components/middleware/http"
-	httpMiddleware "github.com/dapr/dapr/pkg/middleware/http"
+	"github.com/dapr/dapr/pkg/middleware"
 	"github.com/dapr/kit/logger"
 )
 
 func init() {
 	httpMiddlewareLoader.DefaultRegistry.RegisterComponent(func(log logger.Logger) httpMiddlewareLoader.FactoryMethod {
-		return func(metadata middleware.Metadata) (httpMiddleware.Middleware, error) {
+		return func(metadata contribmiddleware.Metadata) (middleware.HTTP, error) {
 			return routeralias.NewMiddleware(log).GetHandler(context.TODO(), metadata)
 		}
 	}, "routeralias")

cmd/daprd/components/middleware_http_routerchecker.go

@@ -18,16 +18,16 @@ package components
 import (
 	"context"
 
-	"github.com/dapr/components-contrib/middleware"
+	contribmiddleware "github.com/dapr/components-contrib/middleware"
 	"github.com/dapr/components-contrib/middleware/http/routerchecker"
 	httpMiddlewareLoader "github.com/dapr/dapr/pkg/components/middleware/http"
-	httpMiddleware "github.com/dapr/dapr/pkg/middleware/http"
+	"github.com/dapr/dapr/pkg/middleware"
 	"github.com/dapr/kit/logger"
 )
 
 func init() {
 	httpMiddlewareLoader.DefaultRegistry.RegisterComponent(func(log logger.Logger) httpMiddlewareLoader.FactoryMethod {
-		return func(metadata middleware.Metadata) (httpMiddleware.Middleware, error) {
+		return func(metadata contribmiddleware.Metadata) (middleware.HTTP, error) {
 			return routerchecker.NewMiddleware(log).GetHandler(context.TODO(), metadata)
 		}
 	}, "routerchecker")

cmd/daprd/components/middleware_http_sentinel.go

@@ -18,16 +18,16 @@ package components
 import (
 	"context"
 
-	"github.com/dapr/components-contrib/middleware"
+	contribmiddleware "github.com/dapr/components-contrib/middleware"
 	"github.com/dapr/components-contrib/middleware/http/sentinel"
 	httpMiddlewareLoader "github.com/dapr/dapr/pkg/components/middleware/http"
-	httpMiddleware "github.com/dapr/dapr/pkg/middleware/http"
+	"github.com/dapr/dapr/pkg/middleware"
 	"github.com/dapr/kit/logger"
 )
 
 func init() {
 	httpMiddlewareLoader.DefaultRegistry.RegisterComponent(func(log logger.Logger) httpMiddlewareLoader.FactoryMethod {
-		return func(metadata middleware.Metadata) (httpMiddleware.Middleware, error) {
+		return func(metadata contribmiddleware.Metadata) (middleware.HTTP, error) {
 			return sentinel.NewMiddleware(log).GetHandler(context.TODO(), metadata)
 		}
 	}, "sentinel")

cmd/daprd/components/middleware_http_uppercase.go

@@ -19,16 +19,16 @@ import (
 	"net/http"
 	"strings"
 
-	"github.com/dapr/components-contrib/middleware"
+	contribmiddleware "github.com/dapr/components-contrib/middleware"
 	httpMiddlewareLoader "github.com/dapr/dapr/pkg/components/middleware/http"
-	httpMiddleware "github.com/dapr/dapr/pkg/middleware/http"
+	"github.com/dapr/dapr/pkg/middleware"
 	"github.com/dapr/dapr/utils"
 	"github.com/dapr/kit/logger"
 )
 
 func init() {
 	httpMiddlewareLoader.DefaultRegistry.RegisterComponent(func(log logger.Logger) httpMiddlewareLoader.FactoryMethod {
-		return func(metadata middleware.Metadata) (httpMiddleware.Middleware, error) {
+		return func(metadata contribmiddleware.Metadata) (middleware.HTTP, error) {
 			// Apply to request only by default
 			var request, response bool
 			switch strings.ToLower(metadata.Properties["direction"]) {

cmd/daprd/components/middleware_http_webassembly.go

@@ -18,17 +18,17 @@ package components
 import (
 	"context"
 
-	"github.com/dapr/components-contrib/middleware"
+	contribmiddleware "github.com/dapr/components-contrib/middleware"
 	"github.com/dapr/components-contrib/middleware/http/wasm"
 	"github.com/dapr/dapr/pkg/components"
 	httpMiddlewareLoader "github.com/dapr/dapr/pkg/components/middleware/http"
-	httpMiddleware "github.com/dapr/dapr/pkg/middleware/http"
+	"github.com/dapr/dapr/pkg/middleware"
 	"github.com/dapr/kit/logger"
 )
 
 func init() {
 	httpMiddlewareLoader.DefaultRegistry.RegisterComponent(func(log logger.Logger) httpMiddlewareLoader.FactoryMethod {
-		return func(metadata middleware.Metadata) (httpMiddleware.Middleware, error) {
+		return func(metadata contribmiddleware.Metadata) (middleware.HTTP, error) {
 			return wasm.NewMiddleware(log).GetHandler(context.TODO(), metadata)
 		}
 	}, "wasm")

pkg/channel/http/http_channel.go

@@ -35,7 +35,7 @@ import (
 	diagUtils "github.com/dapr/dapr/pkg/diagnostics/utils"
 	"github.com/dapr/dapr/pkg/messages"
 	invokev1 "github.com/dapr/dapr/pkg/messaging/v1"
-	httpMiddleware "github.com/dapr/dapr/pkg/middleware/http"
+	"github.com/dapr/dapr/pkg/middleware"
 	commonv1pb "github.com/dapr/dapr/pkg/proto/common/v1"
 	internalv1pb "github.com/dapr/dapr/pkg/proto/internals/v1"
 	"github.com/dapr/dapr/pkg/runtime/compstore"
@@ -64,7 +64,7 @@ type Channel struct {
 	maxResponseBodySizeMB int
 	appHealthCheckPath    string
 	appHealth             *apphealth.AppHealth
-	pipeline              httpMiddleware.Pipeline
+	middleware            middleware.HTTP
 }
 
 // ChannelConfiguration is the configuration used to create an HTTP AppChannel.
@@ -73,7 +73,7 @@ type ChannelConfiguration struct {
 	CompStore            *compstore.ComponentStore
 	Endpoint             string
 	MaxConcurrency       int
-	Pipeline             httpMiddleware.Pipeline
+	Middleware           middleware.HTTP
 	TracingSpec          *config.TracingSpec
 	MaxRequestBodySizeMB int
 	TLSClientCert        string
@@ -85,7 +85,7 @@ type ChannelConfiguration struct {
 // CreateHTTPChannel creates an HTTP AppChannel.
 func CreateHTTPChannel(config ChannelConfiguration) (channel.AppChannel, error) {
 	c := &Channel{
-		pipeline:              config.Pipeline,
+		middleware:            config.Middleware,
 		client:                config.Client,
 		compStore:             config.CompStore,
 		baseAddress:           config.Endpoint,
@@ -231,34 +231,26 @@ func (h *Channel) invokeMethodV1(ctx context.Context, req *invokev1.InvokeMethod
 	diag.DefaultHTTPMonitoring.ClientRequestStarted(ctx, int64(len(req.Message().GetData().GetValue())))
 	startRequest := time.Now()
 
-	var resp *http.Response
-	if len(h.pipeline.Handlers) > 0 {
-		// Exec pipeline only if at least one handler is specified
-		rw := &RWRecorder{
-			W: &bytes.Buffer{},
-		}
-		execPipeline := h.pipeline.Apply(http.HandlerFunc(func(wr http.ResponseWriter, r *http.Request) {
-			// Send request to user application
-			// (Body is closed below, but linter isn't detecting that)
-			//nolint:bodyclose
-			clientResp, clientErr := h.client.Do(r)
-			if clientResp != nil {
-				copyHeader(wr.Header(), clientResp.Header)
-				wr.WriteHeader(clientResp.StatusCode)
-				_, _ = io.Copy(wr, clientResp.Body)
-			}
-			if clientErr != nil {
-				err = clientErr
-			}
-		}))
-		execPipeline.ServeHTTP(rw, channelReq)
-		resp = rw.Result() //nolint:bodyclose
-	} else {
+	// Exec pipeline only if at least one handler is specified
+	rw := &RWRecorder{
+		W: &bytes.Buffer{},
+	}
+	execPipeline := h.middleware(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
 		// Send request to user application
 		// (Body is closed below, but linter isn't detecting that)
 		//nolint:bodyclose
-		resp, err = h.client.Do(channelReq)
-	}
+		clientResp, clientErr := h.client.Do(r)
+		if clientResp != nil {
+			copyHeader(w.Header(), clientResp.Header)
+			w.WriteHeader(clientResp.StatusCode)
+			_, _ = io.Copy(w, clientResp.Body)
+		}
+		if clientErr != nil {
+			err = clientErr
+		}
+	}))
+	execPipeline.ServeHTTP(rw, channelReq)
+	resp := rw.Result() //nolint:bodyclose
 
 	elapsedMs := float64(time.Since(startRequest) / time.Millisecond)