Configure alternate backend for Dapr workflow

cmd/daprd/components/wfbackend_actors.go

@@ -0,0 +1,27 @@
+//go:build allcomponents || stablecomponents
+
+/*
+Copyright 2023 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package components
+
+import (
+	wfbeLoader "github.com/dapr/dapr/pkg/components/wfbackend"
+	"github.com/dapr/dapr/pkg/runtime/wfengine/backends/actors"
+)
+
+func init() {
+	// Note that the actors backend is always enabled, even if this file is deleted
+	// This component is still registered here to allow users to explicitly instantiate the actors backend
+	wfbeLoader.DefaultRegistry.RegisterComponent(actors.NewActorBackend, "actor", "actors")
+}

cmd/daprd/components/wfbackend_sqlite.go

@@ -0,0 +1,25 @@
+//go:build allcomponents || stablecomponents
+
+/*
+Copyright 2023 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package components
+
+import (
+	wfbeLoader "github.com/dapr/dapr/pkg/components/wfbackend"
+	"github.com/dapr/dapr/pkg/runtime/wfengine/backends/sqlite"
+)
+
+func init() {
+	wfbeLoader.DefaultRegistry.RegisterComponent(sqlite.NewSQLiteBackend, "sqlite")
+}

cmd/daprd/main.go

@@ -34,6 +34,7 @@ import (
 	pubsubLoader "github.com/dapr/dapr/pkg/components/pubsub"
 	secretstoresLoader "github.com/dapr/dapr/pkg/components/secretstores"
 	stateLoader "github.com/dapr/dapr/pkg/components/state"
+	wfbeLoader "github.com/dapr/dapr/pkg/components/wfbackend"
 	workflowsLoader "github.com/dapr/dapr/pkg/components/workflows"
 	"github.com/dapr/dapr/pkg/modes"
 	"github.com/dapr/dapr/pkg/runtime"
@@ -91,6 +92,7 @@ func main() {
 	nrLoader.DefaultRegistry.Logger = logContrib
 	bindingsLoader.DefaultRegistry.Logger = logContrib
 	workflowsLoader.DefaultRegistry.Logger = logContrib
+	wfbeLoader.DefaultRegistry.Logger = logContrib
 	httpMiddlewareLoader.DefaultRegistry.Logger = log // Note this uses log on purpose
 
 	reg := registry.NewOptions().
@@ -103,7 +105,8 @@ func main() {
 		WithBindings(bindingsLoader.DefaultRegistry).
 		WithCryptoProviders(cryptoLoader.DefaultRegistry).
 		WithHTTPMiddlewares(httpMiddlewareLoader.DefaultRegistry).
-		WithWorkflows(workflowsLoader.DefaultRegistry)
+		WithWorkflows(workflowsLoader.DefaultRegistry).
+		WithWorkflowBackends(wfbeLoader.DefaultRegistry)
 
 	ctx := signals.Context()
 	secProvider, err := security.New(ctx, security.Options{

pkg/components/category.go

@@ -17,14 +17,15 @@ type Category string
 
 // supported components category
 const (
-	CategoryBindings       Category = "bindings"
-	CategoryPubSub         Category = "pubsub"
-	CategorySecretStore    Category = "secretstores"
-	CategoryStateStore     Category = "state"
-	CategoryWorkflow       Category = "workflow"
-	CategoryMiddleware     Category = "middleware"
-	CategoryConfiguration  Category = "configuration"
-	CategoryCryptoProvider Category = "crypto"
-	CategoryLock           Category = "lock"
-	CategoryNameResolution Category = "nameresolution"
+	CategoryBindings        Category = "bindings"
+	CategoryPubSub          Category = "pubsub"
+	CategorySecretStore     Category = "secretstores"
+	CategoryStateStore      Category = "state"
+	CategoryWorkflow        Category = "workflow"
+	CategoryWorkflowBackend Category = "workflowbackend"
+	CategoryMiddleware      Category = "middleware"
+	CategoryConfiguration   Category = "configuration"
+	CategoryCryptoProvider  Category = "crypto"
+	CategoryLock            Category = "lock"
+	CategoryNameResolution  Category = "nameresolution"
 )

pkg/components/state/pluggable.go

@@ -344,7 +344,7 @@
 	defer ss.lock.Unlock()
 
 	// Check the cached value again in case another goroutine set it
-    if multiMaxSize != nil {
+	if multiMaxSize != nil {
 		return *multiMaxSize
 	}
 

pkg/components/wfbackend/metadata.go

@@ -0,0 +1,26 @@
+/*
+Copyright 2023 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package wfbackend
+
+import (
+	"github.com/dapr/components-contrib/metadata"
+)
+
+// Metadata represents a set of properties specific for workflow backends.
+type Metadata struct {
+	metadata.Base `json:",inline"`
+
+	// Dapr app ID
+	AppID string
+}

pkg/components/wfbackend/registry.go

@@ -0,0 +1,86 @@
+/*
+Copyright 2023 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package wfbackend
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/microsoft/durabletask-go/backend"
+
+	"github.com/dapr/dapr/pkg/components"
+	"github.com/dapr/kit/logger"
+)
+
+// Registry is an interface for a component that returns registered workflow backend implementations.
+type Registry struct {
+	Logger                    logger.Logger
+	workflowBackendComponents map[string]WorkflowBackend
+}
+
+// DefaultRegistry is the singleton with the registry.
+var DefaultRegistry *Registry = NewRegistry()
+
+// NewRegistry is used to create workflow registry.
+func NewRegistry() *Registry {
+	return &Registry{
+		workflowBackendComponents: make(map[string]WorkflowBackend),
+	}
+}
+
+// RegisterComponent adds a new workflow to the registry.
+func (s *Registry) RegisterComponent(componentFactory WorkflowBackend, names ...string) {
+	for _, name := range names {
+		s.workflowBackendComponents[createFullName(name)] = componentFactory
+	}
+}
+
+func (s *Registry) Create(name, version, logName string) (func(Metadata) (backend.Backend, error), error) {
+	if method, ok := s.getWorkflowBackendComponent(name, version, logName); ok {
+		return method, nil
+	}
+	return nil, fmt.Errorf("couldn't find wokflow backend %s/%s", name, version)
+}
+
+func (s *Registry) getWorkflowBackendComponent(name, version, logName string) (func(Metadata) (backend.Backend, error), bool) {
+	nameLower := strings.ToLower(name)
+	versionLower := strings.ToLower(version)
+	workflowFn, ok := s.workflowBackendComponents[nameLower+"/"+versionLower]
+	if ok {
+		return s.wrapFn(workflowFn, logName), true
+	}
+	if components.IsInitialVersion(versionLower) {
+		workflowFn, ok = s.workflowBackendComponents[nameLower]
+		if ok {
+			return s.wrapFn(workflowFn, logName), true
+		}
+	}
+	return nil, false
+}
+
+func (s *Registry) wrapFn(componentFactory WorkflowBackend, logName string) func(Metadata) (backend.Backend, error) {
+	return func(m Metadata) (backend.Backend, error) {
+		l := s.Logger
+		if logName != "" && l != nil {
+			l = l.WithFields(map[string]any{
+				"component": logName,
+			})
+		}
+		return componentFactory(m, l)
+	}
+}
+
+func createFullName(name string) string {
+	return strings.ToLower("workflowbackend." + name)
+}

pkg/components/wfbackend/registry_test.go

@@ -0,0 +1,106 @@
+/*
+Copyright 2023 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package wfbackend_test
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/microsoft/durabletask-go/backend"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
+	wbe "github.com/dapr/dapr/pkg/components/wfbackend"
+	"github.com/dapr/kit/logger"
+)
+
+type mockWorkflowBackend struct {
+	backend.Backend
+}
+
+func TestRegistry(t *testing.T) {
+	testRegistry := wbe.NewRegistry()
+
+	t.Run("workflow backend is registered", func(t *testing.T) {
+		const (
+			backendName   = "testbackendname"
+			backendNameV2 = "testbackendname/v2"
+			componentType = "workflowbackend." + backendName
+		)
+
+		// Initiate mock object
+		wbeMock := &mockWorkflowBackend{}
+		wbeMockV2 := &mockWorkflowBackend{}
+		md := wbe.Metadata{}
+
+		// act
+		testRegistry.RegisterComponent(func(_ wbe.Metadata, _ logger.Logger) (backend.Backend, error) {
+			return wbeMock, nil
+		}, backendName)
+		testRegistry.RegisterComponent(func(_ wbe.Metadata, _ logger.Logger) (backend.Backend, error) {
+			return wbeMockV2, nil
+		}, backendNameV2)
+
+		// assert v0 and v1
+		wbeFn, err := testRegistry.Create(componentType, "v0", "")
+		require.NoError(t, err)
+		wbe, err := wbeFn(md)
+		require.NoError(t, err)
+		assert.Same(t, wbeMock, wbe)
+
+		wbeFn, err = testRegistry.Create(componentType, "v1", "")
+		require.NoError(t, err)
+		wbe, err = wbeFn(md)
+		require.NoError(t, err)
+		assert.Same(t, wbeMock, wbe)
+
+		// assert v2
+		wbeFn, err = testRegistry.Create(componentType, "v2", "")
+		require.NoError(t, err)
+		wbe, err = wbeFn(md)
+		require.NoError(t, err)
+		assert.Same(t, wbeMockV2, wbe)
+
+		// check case-insensitivity
+		wbeFn, err = testRegistry.Create(strings.ToUpper(componentType), "V2", "")
+		require.NoError(t, err)
+		wbe, err = wbeFn(md)
+		require.NoError(t, err)
+		assert.Same(t, wbeMockV2, wbe)
+
+		// check case-insensitivity
+		testRegistry.Logger = logger.NewLogger("wfengine.backend")
+		wbeFn, err = testRegistry.Create(strings.ToUpper(componentType), "V2", "workflowbackendlog")
+		require.NoError(t, err)
+		wbe, err = wbeFn(md)
+		require.NoError(t, err)
+		assert.Same(t, wbeMockV2, wbe)
+	})
+
+	t.Run("workflow backend is not registered", func(t *testing.T) {
+		const (
+			backendName   = "fakeBackend"
+			componentType = "workflowbackend." + backendName
+		)
+
+		// act
+		wbe, actualError := testRegistry.Create(componentType, "v1", "")
+		expectedError := fmt.Errorf("couldn't find wokflow backend %s/v1", componentType)
+
+		// assert
+		assert.Nil(t, wbe)
+		assert.Equal(t, expectedError.Error(), actualError.Error())
+	})
+}

pkg/components/wfbackend/wfbackend.go

@@ -0,0 +1,23 @@
+/*
+Copyright 2021 The Dapr Authors
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+    http://www.apache.org/licenses/LICENSE-2.0
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package wfbackend
+
+import (
+	"github.com/microsoft/durabletask-go/backend"
+
+	"github.com/dapr/kit/logger"
+)
+
+// WorkflowBackend is a function that returns a workflow backend
+type WorkflowBackend = func(Metadata, logger.Logger) (backend.Backend, error)