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/api/http/http_test.go

@@ -39,7 +39,6 @@ import (
 	"github.com/dapr/components-contrib/bindings"
 	"github.com/dapr/components-contrib/configuration"
 	"github.com/dapr/components-contrib/lock"
-	"github.com/dapr/components-contrib/middleware"
 	"github.com/dapr/components-contrib/pubsub"
 	"github.com/dapr/components-contrib/secretstores"
 	"github.com/dapr/components-contrib/state"
@@ -52,14 +51,14 @@ import (
 	httpEndpointsV1alpha1 "github.com/dapr/dapr/pkg/apis/httpEndpoint/v1alpha1"
 	"github.com/dapr/dapr/pkg/apis/resiliency/v1alpha1"
 	"github.com/dapr/dapr/pkg/channel/http"
-	httpMiddlewareLoader "github.com/dapr/dapr/pkg/components/middleware/http"
 	"github.com/dapr/dapr/pkg/config"
 	diag "github.com/dapr/dapr/pkg/diagnostics"
 	"github.com/dapr/dapr/pkg/encryption"
 	"github.com/dapr/dapr/pkg/expr"
 	"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"
+	middlewarehttp "github.com/dapr/dapr/pkg/middleware/http"
 	"github.com/dapr/dapr/pkg/resiliency"
 	"github.com/dapr/dapr/pkg/runtime/channels"
 	"github.com/dapr/dapr/pkg/runtime/compstore"
@@ -2123,7 +2122,7 @@ func TestEmptyPipelineWithTracer(t *testing.T) {
 
 	buffer := ""
 	spec := config.TracingSpec{SamplingRate: "1.0"}
-	pipe := httpMiddleware.Pipeline{}
+	pipe := func(next gohttp.Handler) gohttp.Handler { return next }
 
 	createExporters(&buffer)
 	compStore := compstore.New()
@@ -2138,7 +2137,7 @@ func TestEmptyPipelineWithTracer(t *testing.T) {
 	}
 	fakeServer.StartServer(testAPI.constructDirectMessagingEndpoints(), &fakeHTTPServerOptions{
 		spec:     &spec,
-		pipeline: &pipe,
+		pipeline: pipe,
 	})
 
 	t.Run("Invoke direct messaging without querystring - 200 OK", func(t *testing.T) {
@@ -2925,22 +2924,20 @@ func TestV1Beta1Workflow(t *testing.T) {
 	})
 }
 
-func buildHTTPPineline(spec config.PipelineSpec) httpMiddleware.Pipeline {
-	registry := httpMiddlewareLoader.NewRegistry()
-	registry.RegisterComponent(func(l logger.Logger) httpMiddlewareLoader.FactoryMethod {
-		return func(metadata middleware.Metadata) (httpMiddleware.Middleware, error) {
-			return utils.UppercaseRequestMiddleware, nil
-		}
-	}, "uppercase")
-	var handlers []httpMiddleware.Middleware
-	for i := 0; i < len(spec.Handlers); i++ {
-		handler, err := registry.Create(spec.Handlers[i].Type, spec.Handlers[i].Version, middleware.Metadata{}, "")
-		if err != nil {
-			return httpMiddleware.Pipeline{}
-		}
-		handlers = append(handlers, handler)
-	}
-	return httpMiddleware.Pipeline{Handlers: handlers}
+func buildHTTPPipeline(spec config.PipelineSpec) middleware.HTTP {
+	h := middlewarehttp.New()
+	h.Add(middlewarehttp.Spec{
+		Component: componentsV1alpha1.Component{
+			ObjectMeta: metaV1.ObjectMeta{Name: "middleware.http.uppercase"},
+			Spec: componentsV1alpha1.ComponentSpec{
+				Type:    "middleware.http.uppercase",
+				Version: "v1",
+			},
+		},
+		Implementation: utils.UppercaseRequestMiddleware,
+	})
+
+	return h.BuildPipelineFromSpec("test", &spec)
 }
 
 func TestSinglePipelineWithTracer(t *testing.T) {
@@ -2956,7 +2953,7 @@ func TestSinglePipelineWithTracer(t *testing.T) {
 	buffer := ""
 	spec := config.TracingSpec{SamplingRate: "1.0"}
 
-	pipeline := buildHTTPPineline(config.PipelineSpec{
+	pipeline := buildHTTPPipeline(config.PipelineSpec{
 		Handlers: []config.HandlerSpec{
 			{
 				Type: "middleware.http.uppercase",
@@ -2978,7 +2975,7 @@ func TestSinglePipelineWithTracer(t *testing.T) {
 	}
 	fakeServer.StartServer(testAPI.constructDirectMessagingEndpoints(), &fakeHTTPServerOptions{
 		spec:     &spec,
-		pipeline: &pipeline,
+		pipeline: pipeline,
 	})
 
 	t.Run("Invoke direct messaging without querystring - 200 OK", func(t *testing.T) {
@@ -3018,7 +3015,7 @@ func TestSinglePipelineWithNoTracing(t *testing.T) {
 	buffer := ""
 	spec := config.TracingSpec{SamplingRate: "0"}
 
-	pipeline := buildHTTPPineline(config.PipelineSpec{
+	pipeline := buildHTTPPipeline(config.PipelineSpec{
 		Handlers: []config.HandlerSpec{
 			{
 				Type: "middleware.http.uppercase",
@@ -3040,7 +3037,7 @@ func TestSinglePipelineWithNoTracing(t *testing.T) {
 	}
 	fakeServer.StartServer(testAPI.constructDirectMessagingEndpoints(), &fakeHTTPServerOptions{
 		spec:     &spec,
-		pipeline: &pipeline,
+		pipeline: pipeline,
 	})
 
 	t.Run("Invoke direct messaging without querystring - 200 OK", func(t *testing.T) {
@@ -3089,7 +3086,7 @@ type fakeHTTPResponse struct {
 
 type fakeHTTPServerOptions struct {
 	spec     *config.TracingSpec
-	pipeline *httpMiddleware.Pipeline
+	pipeline middleware.HTTP
 	apiAuth  bool
 }
 
@@ -3104,7 +3101,7 @@ func (f *fakeHTTPServer) StartServer(endpoints []endpoints.Endpoint, opts *fakeH
 	go func() {
 		var handler gohttp.Handler = r
 		if opts.pipeline != nil {
-			handler = opts.pipeline.Apply(handler)
+			handler = opts.pipeline(handler)
 		}
 		if opts.spec != nil {
 			handler = diag.HTTPTraceMiddleware(handler, "fakeAppID", *opts.spec)

pkg/api/http/server.go

@@ -43,7 +43,7 @@ import (
 	corsDapr "github.com/dapr/dapr/pkg/cors"
 	diag "github.com/dapr/dapr/pkg/diagnostics"
 	diagUtils "github.com/dapr/dapr/pkg/diagnostics/utils"
-	httpMiddleware "github.com/dapr/dapr/pkg/middleware/http"
+	"github.com/dapr/dapr/pkg/middleware"
 	"github.com/dapr/dapr/pkg/responsewriter"
 	"github.com/dapr/dapr/pkg/security"
 	"github.com/dapr/kit/logger"
@@ -65,7 +65,7 @@ type server struct {
 	config             ServerConfig
 	tracingSpec        config.TracingSpec
 	metricSpec         config.MetricSpec
-	pipeline           httpMiddleware.Pipeline
+	middleware         middleware.HTTP
 	api                API
 	apiSpec            config.APISpec
 	servers            []*http.Server
@@ -79,7 +79,7 @@ type NewServerOpts struct {
 	Config      ServerConfig
 	TracingSpec config.TracingSpec
 	MetricSpec  config.MetricSpec
-	Pipeline    httpMiddleware.Pipeline
+	Middleware  middleware.HTTP
 	APISpec     config.APISpec
 }
 
@@ -91,7 +91,7 @@ func NewServer(opts NewServerOpts) Server {
 		config:      opts.Config,
 		tracingSpec: opts.TracingSpec,
 		metricSpec:  opts.MetricSpec,
-		pipeline:    opts.Pipeline,
+		middleware:  opts.Middleware,
 		apiSpec:     opts.APISpec,
 	}
 }
@@ -358,11 +358,7 @@ func (s *server) useAPILogging(mux chi.Router) {
 }
 
 func (s *server) useComponents(r chi.Router) {
-	if len(s.pipeline.Handlers) == 0 {
-		return
-	}
-
-	r.Use(s.pipeline.Handlers...)
+	r.Use(s.middleware)
 }
 
 func (s *server) useCors(r chi.Router) {