Workflow

.github/workflows/self_hosted_e2e.yaml

@@ -131,7 +131,7 @@ jobs:
         shell: bash
       - name: Set the test timeout - MacOS
         if: matrix.os == 'macos-14-large'
-        run: echo "E2E_SH_TEST_TIMEOUT=30m" >> $GITHUB_ENV
+        run: echo "E2E_SH_TEST_TIMEOUT=40m" >> $GITHUB_ENV
       - name: Run E2E tests with GHCR
         # runs every 6hrs
         if: github.event.schedule == '0 */6 * * *'

Makefile

@@ -74,7 +74,7 @@ TEST_OUTPUT_FILE ?= test_output.json
 
 # Set the default timeout for tests to 10 minutes
 ifndef E2E_SH_TEST_TIMEOUT
-  override E2E_SH_TEST_TIMEOUT := 30m
+  override E2E_SH_TEST_TIMEOUT := 40m
 endif
 
 # Use the variable H to add a header (equivalent to =>) to informational output

cmd/dapr.go

@@ -22,6 +22,7 @@ import (
 	"github.com/spf13/viper"
 
 	"github.com/dapr/cli/cmd/scheduler"
+	"github.com/dapr/cli/cmd/workflow"
 	"github.com/dapr/cli/pkg/api"
 	"github.com/dapr/cli/pkg/print"
 	"github.com/dapr/cli/pkg/standalone"
@@ -111,4 +112,5 @@ func init() {
 	RootCmd.PersistentFlags().BoolVarP(&logAsJSON, "log-as-json", "", false, "Log output in JSON format")
 
 	RootCmd.AddCommand(scheduler.SchedulerCmd)
+	RootCmd.AddCommand(workflow.WorkflowCmd)
 }

cmd/workflow/history.go

@@ -0,0 +1,85 @@
+/*
+Copyright 2025 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 workflow
+
+import (
+	"os"
+
+	"github.com/gocarina/gocsv"
+	"github.com/spf13/cobra"
+
+	"github.com/dapr/cli/pkg/workflow"
+	"github.com/dapr/cli/utils"
+	"github.com/dapr/kit/signals"
+)
+
+var (
+	historyOutputFormat *string
+)
+
+var HistoryCmd = &cobra.Command{
+	Use:   "history",
+	Short: "Get the history of a workflow instance.",
+	Args:  cobra.ExactArgs(1),
+	RunE: func(cmd *cobra.Command, args []string) error {
+		ctx := signals.Context()
+
+		appID, err := getWorkflowAppID(cmd)
+		if err != nil {
+			return err
+		}
+
+		opts := workflow.HistoryOptions{
+			KubernetesMode: flagKubernetesMode,
+			Namespace:      flagDaprNamespace,
+			AppID:          appID,
+			InstanceID:     args[0],
+		}
+
+		var list any
+		if *historyOutputFormat == outputFormatShort {
+			list, err = workflow.HistoryShort(ctx, opts)
+		} else {
+			list, err = workflow.HistoryWide(ctx, opts)
+		}
+		if err != nil {
+			return err
+		}
+
+		switch *historyOutputFormat {
+		case outputFormatYAML:
+			err = utils.PrintDetail(os.Stdout, "yaml", list)
+		case outputFormatJSON:
+			err = utils.PrintDetail(os.Stdout, "json", list)
+		default:
+			var table string
+			table, err = gocsv.MarshalString(list)
+			if err != nil {
+				break
+			}
+
+			utils.PrintTable(table)
+		}
+		if err != nil {
+			return err
+		}
+
+		return nil
+	},
+}
+
+func init() {
+	historyOutputFormat = outputFunc(HistoryCmd)
+	WorkflowCmd.AddCommand(HistoryCmd)
+}

cmd/workflow/list.go

@@ -0,0 +1,113 @@
+/*
+Copyright 2025 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 workflow
+
+import (
+	"os"
+
+	"github.com/gocarina/gocsv"
+	"github.com/spf13/cobra"
+
+	"github.com/dapr/cli/pkg/print"
+	"github.com/dapr/cli/pkg/workflow"
+	"github.com/dapr/cli/utils"
+	"github.com/dapr/kit/signals"
+)
+
+var (
+	listFilter       *workflow.Filter
+	listOutputFormat *string
+
+	listConn *connFlag
+)
+
+var ListCmd = &cobra.Command{
+	Use:     "list",
+	Aliases: []string{"ls"},
+	Short:   "List workflows for the given app ID.",
+	Args:    cobra.NoArgs,
+	RunE: func(cmd *cobra.Command, args []string) error {
+		ctx := signals.Context()
+
+		appID, err := getWorkflowAppID(cmd)
+		if err != nil {
+			return err
+		}
+
+		opts := workflow.ListOptions{
+			KubernetesMode:   flagKubernetesMode,
+			Namespace:        flagDaprNamespace,
+			AppID:            appID,
+			ConnectionString: listConn.connectionString,
+			TableName:        listConn.tableName,
+			Filter:           *listFilter,
+		}
+
+		var list any
+		var empty bool
+
+		switch *listOutputFormat {
+		case outputFormatShort:
+			var ll []*workflow.ListOutputShort
+			ll, err = workflow.ListShort(ctx, opts)
+			if err != nil {
+				return err
+			}
+			empty = len(ll) == 0
+			list = ll
+
+		default:
+			var ll []*workflow.ListOutputWide
+			ll, err = workflow.ListWide(ctx, opts)
+			if err != nil {
+				return err
+			}
+			empty = len(ll) == 0
+			list = ll
+		}
+
+		if empty {
+			print.FailureStatusEvent(os.Stderr, "No workflow found in namespace %q for app ID %q", flagDaprNamespace, appID)
+			return nil
+		}
+
+		switch *listOutputFormat {
+		case outputFormatYAML:
+			err = utils.PrintDetail(os.Stdout, "yaml", list)
+		case outputFormatJSON:
+			err = utils.PrintDetail(os.Stdout, "json", list)
+		default:
+			var table string
+			table, err = gocsv.MarshalString(list)
+			if err != nil {
+				break
+			}
+
+			utils.PrintTable(table)
+		}
+
+		if err != nil {
+			return err
+		}
+
+		return nil
+	},
+}
+
+func init() {
+	listFilter = filterCmd(ListCmd)
+	listOutputFormat = outputFunc(ListCmd)
+	listConn = connectionCmd(ListCmd)
+	WorkflowCmd.AddCommand(ListCmd)
+}

cmd/workflow/purge.go

@@ -0,0 +1,89 @@
+/*
+Copyright 2025 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 workflow
+
+import (
+	"errors"
+
+	"github.com/dapr/cli/pkg/workflow"
+	"github.com/dapr/kit/signals"
+	"github.com/spf13/cobra"
+)
+
+var (
+	flagPurgeOlderThan string
+	flagPurgeAll       bool
+	flagPurgeConn      *connFlag
+	schedulerNamespace string
+)
+
+var PurgeCmd = &cobra.Command{
+	Use:   "purge",
+	Short: "Purge one or more workflow instances with a terminal state. Accepts a workflow instance ID argument or flags to purge multiple/all terminal instances. Also deletes all associated scheduler jobs.",
+	Args: func(cmd *cobra.Command, args []string) error {
+		switch {
+		case cmd.Flags().Changed("all-older-than"),
+			cmd.Flags().Changed("all"):
+			if len(args) > 0 {
+				return errors.New("no arguments are accepted when using purge all flags")
+			}
+		default:
+			if len(args) == 0 {
+				return errors.New("one or more workflow instance ID arguments are required when not using purge all flags")
+			}
+		}
+
+		return nil
+	},
+	RunE: func(cmd *cobra.Command, args []string) error {
+		ctx := signals.Context()
+
+		appID, err := getWorkflowAppID(cmd)
+		if err != nil {
+			return err
+		}
+
+		opts := workflow.PurgeOptions{
+			KubernetesMode:     flagKubernetesMode,
+			Namespace:          flagDaprNamespace,
+			SchedulerNamespace: schedulerNamespace,
+			AppID:              appID,
+			InstanceIDs:        args,
+			All:                flagPurgeAll,
+			ConnectionString:   flagPurgeConn.connectionString,
+			TableName:          flagPurgeConn.tableName,
+		}
+
+		if cmd.Flags().Changed("all-older-than") {
+			opts.AllOlderThan, err = parseWorkflowDurationTimestamp(flagPurgeOlderThan, true)
+			if err != nil {
+				return err
+			}
+		}
+
+		return workflow.Purge(ctx, opts)
+	},
+}
+
+func init() {
+	PurgeCmd.Flags().StringVar(&flagPurgeOlderThan, "all-older-than", "", "Purge workflow instances older than the specified Go duration or timestamp, e.g., '24h' or '2023-01-02T15:04:05Z'.")
+	PurgeCmd.Flags().BoolVar(&flagPurgeAll, "all", false, "Purge all workflow instances in a terminal state. Use with caution.")
+	PurgeCmd.MarkFlagsMutuallyExclusive("all-older-than", "all")
+
+	PurgeCmd.Flags().StringVar(&schedulerNamespace, "scheduler-namespace", "dapr-system", "Kubernetes namespace where the scheduler is deployed, only relevant if --kubernetes is set")
+
+	flagPurgeConn = connectionCmd(PurgeCmd)
+
+	WorkflowCmd.AddCommand(PurgeCmd)
+}

cmd/workflow/raiseevent.go

@@ -0,0 +1,74 @@
+/*
+Copyright 2025 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 workflow
+
+import (
+	"errors"
+	"os"
+	"strings"
+
+	"github.com/dapr/cli/pkg/print"
+	"github.com/dapr/cli/pkg/workflow"
+	"github.com/dapr/kit/signals"
+	"github.com/spf13/cobra"
+)
+
+var (
+	flagRaiseEventInput *inputFlag
+)
+
+var RaiseEventCmd = &cobra.Command{
+	Use:   "raise-event",
+	Short: "Raise an event for a workflow waiting for an external event. Expects a single argument '<instance-id>/<event-name>'.",
+	Args:  cobra.ExactArgs(1),
+	RunE: func(cmd *cobra.Command, args []string) error {
+		ctx := signals.Context()
+
+		split := strings.Split(args[0], "/")
+		if len(split) != 2 {
+			return errors.New("the argument must be in the format '<instance-id>/<event-name>'")
+		}
+		instanceID := split[0]
+		eventName := split[1]
+
+		appID, err := getWorkflowAppID(cmd)
+		if err != nil {
+			return err
+		}
+
+		opts := workflow.RaiseEventOptions{
+			KubernetesMode: flagKubernetesMode,
+			Namespace:      flagDaprNamespace,
+			AppID:          appID,
+			InstanceID:     instanceID,
+			Name:           eventName,
+			Input:          flagRaiseEventInput.input,
+		}
+
+		if err = workflow.RaiseEvent(ctx, opts); err != nil {
+			print.FailureStatusEvent(os.Stdout, err.Error())
+			os.Exit(1)
+		}
+
+		print.InfoStatusEvent(os.Stdout, "Workflow '%s' raised event '%s' successfully", instanceID, eventName)
+
+		return nil
+	},
+}
+
+func init() {
+	flagRaiseEventInput = inputCmd(RaiseEventCmd)
+
+	WorkflowCmd.AddCommand(RaiseEventCmd)
+}