pipelines: stop command (#3275)

## Changes
pipelines stop, new command allows for cancelling resources
Requires specifying KEY of resource to stop or autocompletes if there is
only one pipeline in the project and stops that pipeline.
No arguments.

## Tests
Tests in follow-up: #3276

---------

Co-authored-by: Jeffery Cheng <[email protected]>

Author: alyssa-db

acceptance/pipelines/install-pipelines-cli/output.txt

@@ -19,6 +19,7 @@ Available Commands:
   init        Initialize a new pipelines project
   open        Open a pipeline in the browser
   run         Run a pipeline
+  stop        Stop a pipeline
   version     Retrieve information about the current version of the Pipelines CLI
 
 Flags:
@@ -62,6 +63,7 @@ Available Commands:
   init        Initialize a new pipelines project
   open        Open a pipeline in the browser
   run         Run a pipeline
+  stop        Stop a pipeline
   version     Retrieve information about the current version of the Pipelines CLI
 
 Flags:

cmd/pipelines/open.go

@@ -44,21 +44,18 @@ func promptOpenArgument(ctx context.Context, b *bundle.Bundle) (string, error) {
 // When no arguments are specified, auto-selects a pipeline if there's exactly one,
 // otherwise prompts the user to select a pipeline to open.
 func resolveOpenArgument(ctx context.Context, b *bundle.Bundle, args []string) (string, error) {
-	if len(args) == 0 {
-		if key := autoSelectSinglePipeline(b); key != "" {
-			return key, nil
-		}
-
-		if cmdio.IsPromptSupported(ctx) {
-			return promptOpenArgument(ctx, b)
-		}
+	if len(args) == 1 {
+		return args[0], nil
 	}
 
-	if len(args) < 1 {
-		return "", errors.New("expected a KEY of the pipeline to open")
+	if key := autoSelectSinglePipeline(b); key != "" {
+		return key, nil
 	}
 
-	return args[0], nil
+	if cmdio.IsPromptSupported(ctx) {
+		return promptOpenArgument(ctx, b)
+	}
+	return "", errors.New("expected a KEY of the pipeline to open")
 }
 
 func openCommand() *cobra.Command {

cmd/pipelines/pipelines.go

@@ -17,6 +17,7 @@ func New(ctx context.Context) *cobra.Command {
 	cli.AddCommand(dryRunCommand())
 	cli.AddCommand(authCommand())
 	cli.AddCommand(destroyCommand())
+	cli.AddCommand(stopCommand())
 	cli.AddCommand(versionCommand())
 	return cli
 }

cmd/pipelines/stop.go

@@ -0,0 +1,130 @@
+package pipelines
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	"github.com/databricks/cli/bundle"
+	"github.com/databricks/cli/bundle/deploy/terraform"
+	"github.com/databricks/cli/bundle/phases"
+	"github.com/databricks/cli/bundle/resources"
+	"github.com/databricks/cli/bundle/run"
+	"github.com/databricks/cli/bundle/statemgmt"
+	"github.com/databricks/cli/cmd/bundle/utils"
+	"github.com/databricks/cli/cmd/root"
+	"github.com/databricks/cli/libs/cmdio"
+	"github.com/databricks/cli/libs/logdiag"
+	"github.com/spf13/cobra"
+	"golang.org/x/exp/maps"
+)
+
+// resolveStopArgument resolves the pipeline key to stop
+// If there is only one pipeline in the project, KEY is optional and the pipeline will be auto-selected.
+// Otherwise, the user will be prompted to select a pipeline.
+func resolveStopArgument(ctx context.Context, b *bundle.Bundle, args []string) (string, error) {
+	if len(args) == 1 {
+		return args[0], nil
+	}
+
+	if key := autoSelectSinglePipeline(b); key != "" {
+		return key, nil
+	}
+
+	if cmdio.IsPromptSupported(ctx) {
+		return promptRunnablePipeline(ctx, b)
+	}
+
+	return "", errors.New("expected a KEY of the pipeline to stop")
+}
+
+func stopCommand() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:   "stop [KEY]",
+		Short: "Stop a pipeline",
+		Long: `Stop the pipeline if it's running, identified by KEY.
+KEY is the unique name of the pipeline to stop, as defined in its YAML file.
+If there is only one pipeline in the project, KEY is optional and the pipeline will be auto-selected.`,
+		Args: root.MaximumNArgs(1),
+	}
+
+	cmd.RunE = func(cmd *cobra.Command, args []string) error {
+		ctx := logdiag.InitContext(cmd.Context())
+		cmd.SetContext(ctx)
+
+		b := utils.ConfigureBundleWithVariables(cmd)
+		if b == nil || logdiag.HasError(ctx) {
+			return root.ErrAlreadyPrinted
+		}
+
+		phases.Initialize(ctx, b)
+		if logdiag.HasError(ctx) {
+			return root.ErrAlreadyPrinted
+		}
+
+		key, err := resolveStopArgument(ctx, b, args)
+		if err != nil {
+			return err
+		}
+
+		if !b.DirectDeployment {
+			bundle.ApplySeqContext(ctx, b,
+				terraform.Interpolate(),
+				terraform.Write(),
+			)
+			if logdiag.HasError(ctx) {
+				return root.ErrAlreadyPrinted
+			}
+		}
+
+		bundle.ApplySeqContext(ctx, b,
+			statemgmt.StatePull(),
+			statemgmt.Load(statemgmt.ErrorOnEmptyState),
+		)
+		if logdiag.HasError(ctx) {
+			return root.ErrAlreadyPrinted
+		}
+
+		runner, err := keyToRunner(b, key)
+		if err != nil {
+			return err
+		}
+
+		cmdio.LogString(ctx, fmt.Sprintf("Stopping %s...", key))
+		err = runner.Cancel(ctx)
+		if err != nil {
+			return err
+		}
+
+		cmdio.LogString(ctx, key+" has been stopped.")
+		return nil
+	}
+
+	// TODO: This autocomplete functionality was copied from cmd/bundle/run.go and is not working properly.
+	cmd.ValidArgsFunction = func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
+		b := root.MustConfigureBundle(cmd)
+		if logdiag.HasError(cmd.Context()) {
+			return nil, cobra.ShellCompDirectiveError
+		}
+
+		// No completion in the context of a bundle.
+		// Source and destination paths are taken from bundle configuration.
+		if b == nil {
+			return nil, cobra.ShellCompDirectiveNoFileComp
+		}
+
+		if len(args) == 0 {
+			completions := resources.Completions(b, run.IsRunnable)
+			return maps.Keys(completions), cobra.ShellCompDirectiveNoFileComp
+		} else {
+			// If we know the resource to stop, we can complete additional positional arguments.
+			runner, err := keyToRunner(b, args[0])
+			if err != nil {
+				return nil, cobra.ShellCompDirectiveError
+			}
+			return runner.CompleteArgs(args[1:], toComplete)
+		}
+	}
+
+	return cmd
+}

cmd/pipelines/utils.go

@@ -13,16 +13,16 @@ import (
 )
 
 // Copied from cmd/bundle/run.go
-// promptRunArgument prompts the user to select a resource to run.
-func promptRunArgument(ctx context.Context, b *bundle.Bundle) (string, error) {
+// promptRunnablePipeline prompts the user to select a runnable pipeline.
+func promptRunnablePipeline(ctx context.Context, b *bundle.Bundle) (string, error) {
 	// Compute map of "Human readable name of resource" -> "resource key".
 	inv := make(map[string]string)
 	for k, ref := range resources.Completions(b, run.IsRunnable) {
 		title := fmt.Sprintf("%s: %s", ref.Description.SingularTitle, ref.Resource.GetName())
 		inv[title] = k
 	}
 
-	key, err := cmdio.Select(ctx, inv, "Pipeline to run")
+	key, err := cmdio.Select(ctx, inv, "Select a pipeline")
 	if err != nil {
 		return "", err
 	}
@@ -56,7 +56,7 @@ func resolveRunArgument(ctx context.Context, b *bundle.Bundle, args []string) (s
 		}
 
 		if cmdio.IsPromptSupported(ctx) {
-			key, err := promptRunArgument(ctx, b)
+			key, err := promptRunnablePipeline(ctx, b)
 			if err != nil {
 				return "", nil, err
 			}