Add act wait command

cmd/cli/cmd/action/action.go

@@ -41,6 +41,13 @@ func NewCmd() *cobra.Command {
 		NewRun(),
 		NewGet(),
 		NewWatch(),
+		NewWait(),
 	)
 	return root
 }
+
+func panicOnError(err error) {
+	if err != nil {
+		panic(err)
+	}
+}

cmd/cli/cmd/action/wait.go

@@ -0,0 +1,42 @@
+package action
+
+import (
+	"os"
+	"time"
+
+	"capact.io/capact/internal/cli"
+	"capact.io/capact/internal/cli/action"
+	"capact.io/capact/internal/cli/client"
+	"capact.io/capact/internal/cli/heredoc"
+
+	"github.com/spf13/cobra"
+)
+
+// NewWait returns a new cobra.Command for waiting for a given Action's condition.
+func NewWait() *cobra.Command {
+	var opts action.WaitOptions
+
+	cmd := &cobra.Command{
+		Use:   "wait ACTION",
+		Short: "Wait for a specific condition of a given Action",
+		Args:  cobra.ExactArgs(1),
+		Example: heredoc.WithCLIName(`
+			# Wait for the Action "example" to contain the phase "READY_TO_RUN"
+			<cli> act wait --for=phase=READY_TO_RUN example
+		`, cli.Name),
+		RunE: func(cmd *cobra.Command, args []string) error {
+			opts.ActionName = args[0]
+			return action.Wait(cmd.Context(), opts, os.Stdout)
+		},
+	}
+
+	flags := cmd.Flags()
+	flags.StringVar(&opts.For, "for", "", "The field condition to wait on. Currently, only the 'phase' field is supported: 'phase={phase-name}'.")
+	flags.StringVarP(&opts.Namespace, "namespace", "n", "default", "Kubernetes namespace where the Action was created.")
+	flags.DurationVar(&opts.Timeout, "wait-timeout", 10*time.Minute, `Maximum time to wait before giving up. "0" means "infinite". Valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h".`)
+	client.RegisterFlags(flags)
+
+	panicOnError(cmd.MarkFlagRequired("for")) // this cannot happen
+
+	return cmd
+}

cmd/cli/docs/capact_action.md

@@ -26,5 +26,6 @@ This command consists of multiple subcommands to interact with target Actions
 * [capact action delete](capact_action_delete.md)	 - Deletes the Action
 * [capact action get](capact_action_get.md)	 - Displays one or multiple Actions
 * [capact action run](capact_action_run.md)	 - Queues up a specified Action for processing by the workflow engine
+* [capact action wait](capact_action_wait.md)	 - Wait for a specific condition of a given Action
 * [capact action watch](capact_action_watch.md)	 - Watch an Action until it has completed execution
 

cmd/cli/docs/capact_action_wait.md

@@ -0,0 +1,41 @@
+---
+title: capact action wait
+---
+
+## capact action wait
+
+Wait for a specific condition of a given Action
+
+```
+capact action wait ACTION [flags]
+```
+
+### Examples
+
+```
+# Wait for the Action "example" to contain the phase "READY_TO_RUN"
+capact act wait --for=phase=READY_TO_RUN example
+
+```
+
+### Options
+
+```
+      --for string              The field condition to wait on. Currently, only the 'phase' field is supported: 'phase={phase-name}'.
+  -h, --help                    help for wait
+  -n, --namespace string        Kubernetes namespace where the Action was created. (default "default")
+      --timeout duration        Timeout for HTTP request (default 30s)
+      --wait-timeout duration   Maximum time to wait before giving up. "0" means "infinite". Valid time units are "ns", "us" (or "µs"), "ms", "s", "m", "h". (default 10m0s)
+```
+
+### Options inherited from parent commands
+
+```
+  -c, --config string                 Path to the YAML config file
+  -v, --verbose int/string[=simple]   Prints more verbose output. Allowed values: 0 - disable, 1 - simple, 2 - trace (default 0 - disable)
+```
+
+### SEE ALSO
+
+* [capact action](capact_action.md)	 - This command consists of multiple subcommands to interact with target Actions
+

internal/cli/action/wait.go

@@ -0,0 +1,114 @@
+package action
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"strings"
+	"time"
+
+	"k8s.io/apimachinery/pkg/util/wait"
+
+	"capact.io/capact/internal/cli/client"
+	"capact.io/capact/internal/cli/config"
+	"capact.io/capact/internal/cli/printer"
+	"capact.io/capact/internal/k8s-engine/graphql/namespace"
+	gqlengine "capact.io/capact/pkg/engine/api/graphql"
+)
+
+const (
+	waitPollInterval = time.Second
+	forSeparator     = "="
+)
+
+type conditionFunc func(*gqlengine.Action) error
+
+// WaitOptions holds wait related configuration.
+type WaitOptions struct {
+	ActionName string
+	Namespace  string
+	Timeout    time.Duration
+	For        string
+}
+
+// Wait waits for a given Action's condition.
+func Wait(ctx context.Context, opts WaitOptions, w io.Writer) (err error) {
+	status := printer.NewStatus(w, "")
+	defer func() {
+		status.End(err == nil)
+	}()
+
+	server := config.GetDefaultContext()
+	actionCli, err := client.NewCluster(server)
+	if err != nil {
+		return err
+	}
+
+	conditionFn, kvPair, err := conditionFuncFor(opts.For)
+	if err != nil {
+		return err
+	}
+
+	status.Step("Waiting ≤ %s for %q to equal %q", opts.Timeout, kvPair[0], kvPair[1])
+	return waitUntilSatisfied(ctx, actionCli, conditionFn, opts)
+}
+
+func waitUntilSatisfied(ctx context.Context, actCli client.ClusterClient, conditionFn conditionFunc, opts WaitOptions) error {
+	var (
+		lastErr   error
+		ctxWithNs = namespace.NewContext(ctx, opts.Namespace)
+	)
+
+	err := wait.PollImmediate(waitPollInterval, opts.Timeout, func() (done bool, err error) {
+		if ctxWithNs.Err() != nil {
+			return false, ctxWithNs.Err()
+		}
+
+		act, err := actCli.GetAction(ctxWithNs, opts.ActionName)
+		if err != nil { // may be network issue, ignoring
+			lastErr = err
+			return false, nil
+		}
+
+		if act == nil {
+			return true, fmt.Errorf("Action %q not found in Namespace %q", opts.ActionName, opts.Namespace)
+		}
+
+		if err := conditionFn(act); err != nil {
+			lastErr = err // condition not met yet
+			return false, nil
+		}
+
+		return true, nil
+	})
+	if err != nil {
+		if err == wait.ErrWaitTimeout {
+			return lastErr
+		}
+		return err
+	}
+
+	return nil
+}
+
+func conditionFuncFor(condition string) (conditionFunc, []string, error) {
+	items := strings.SplitN(condition, forSeparator, 2)
+	if len(items) != 2 {
+		return nil, nil, fmt.Errorf("invalid format, require 'condition=condition-value'")
+	}
+
+	conditionName := items[0]
+	conditionValue := items[1]
+
+	switch conditionName {
+	case "phase":
+		return func(act *gqlengine.Action) error {
+			if act.Status.Phase != gqlengine.ActionStatusPhase(conditionValue) {
+				return fmt.Errorf("the Action still doesn't have phase=%s", conditionValue)
+			}
+			return nil
+		}, items, nil
+	default:
+		return nil, nil, fmt.Errorf("unrecognized condition: %q", condition)
+	}
+}