CF-2067 : Add Flink artifact management commands for Confluent Platform

internal/flink/command.go

@@ -46,14 +46,14 @@ func New(cfg *config.Config, prerunner pcmd.PreRunner) *cobra.Command {
 	cmd.AddCommand(c.newSystemInfoCommand())
 
 	// On-Prem and Cloud Commands
+	cmd.AddCommand(c.newArtifactCommand(cfg))
 	cmd.AddCommand(c.newComputePoolCommand(cfg))
 	if !cfg.IsOnPremLogin() {
 		cmd.AddCommand(c.newShellCommand(prerunner, cfg))
 	}
 	cmd.AddCommand(c.newStatementCommand(cfg))
 
 	// Cloud Specific Commands
-	cmd.AddCommand(c.newArtifactCommand())
 	cmd.AddCommand(c.newComputePoolConfigCommand())
 	cmd.AddCommand(c.newConnectionCommand())
 	cmd.AddCommand(c.newConnectivityTypeCommand())

internal/flink/command_artifact.go

@@ -6,6 +6,7 @@ import (
 	flinkartifactv1 "github.com/confluentinc/ccloud-sdk-go-v2/flink-artifact/v1"
 
 	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+	"github.com/confluentinc/cli/v4/pkg/config"
 	"github.com/confluentinc/cli/v4/pkg/output"
 )
 
@@ -21,17 +22,26 @@ type flinkArtifactOut struct {
 	DocumentationLink string `human:"Documentation Link" serialized:"documentation_link"`
 }
 
-func (c *command) newArtifactCommand() *cobra.Command {
+func (c *command) newArtifactCommand(cfg *config.Config) *cobra.Command {
 	cmd := &cobra.Command{
-		Use:         "artifact",
-		Short:       "Manage Flink UDF artifacts.",
-		Annotations: map[string]string{pcmd.RunRequirement: pcmd.RequireNonAPIKeyCloudLogin},
+		Use:   "artifact",
+		Short: "Manage Flink UDF artifacts.",
 	}
 
-	cmd.AddCommand(c.newCreateCommand())
-	cmd.AddCommand(c.newDeleteCommand())
-	cmd.AddCommand(c.newDescribeCommand())
-	cmd.AddCommand(c.newListCommand())
+	if cfg.IsCloudLogin() {
+		cmd.Annotations = map[string]string{pcmd.RunRequirement: pcmd.RequireNonAPIKeyCloudLogin}
+		cmd.AddCommand(c.newCreateCommand())
+		cmd.AddCommand(c.newDeleteCommand())
+		cmd.AddCommand(c.newDescribeCommand())
+		cmd.AddCommand(c.newListCommand())
+	} else {
+		cmd.AddCommand(c.newArtifactCreateCommandOnPrem())
+		cmd.AddCommand(c.newArtifactDeleteCommandOnPrem())
+		cmd.AddCommand(c.newArtifactDescribeCommandOnPrem())
+		cmd.AddCommand(c.newArtifactListCommandOnPrem())
+		cmd.AddCommand(c.newArtifactUpdateCommandOnPrem())
+		cmd.AddCommand(c.newArtifactVersionCommandOnPrem())
+	}
 
 	return cmd
 }

internal/flink/command_artifact_create_onprem.go

@@ -0,0 +1,74 @@
+package flink
+
+import (
+	"github.com/spf13/cobra"
+
+	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+	"github.com/confluentinc/cli/v4/pkg/examples"
+)
+
+func (c *command) newArtifactCreateCommandOnPrem() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:         "create <name>",
+		Short:       "Create a Flink artifact in Confluent Platform.",
+		Long:        "Create a Flink artifact in Confluent Platform by uploading a JAR or ZIP file. This creates version 1 of the artifact.",
+		Args:        cobra.ExactArgs(1),
+		Annotations: map[string]string{pcmd.RunRequirement: pcmd.RequireCloudLogout},
+		RunE:        c.artifactCreateOnPrem,
+		Example: examples.BuildExampleString(
+			examples.Example{
+				Text: `Create Flink artifact "my-artifact" in the environment "my-environment".`,
+				Code: "confluent flink artifact create my-artifact --artifact-file artifact.jar --environment my-environment",
+			},
+		),
+	}
+
+	cmd.Flags().String("environment", "", "Name of the Flink environment.")
+	cmd.Flags().String("artifact-file", "", "Path to the Flink artifact JAR or ZIP file.")
+	cmd.Flags().StringSlice("label", nil, `A comma-separated list of "key=value" label pairs.`)
+	addCmfFlagSet(cmd)
+	pcmd.AddOutputFlag(cmd)
+
+	cobra.CheckErr(cmd.MarkFlagRequired("environment"))
+	cobra.CheckErr(cmd.MarkFlagRequired("artifact-file"))
+	cobra.CheckErr(cmd.MarkFlagFilename("artifact-file", "jar", "zip"))
+
+	return cmd
+}
+
+func (c *command) artifactCreateOnPrem(cmd *cobra.Command, args []string) error {
+	name := args[0]
+
+	environment, err := cmd.Flags().GetString("environment")
+	if err != nil {
+		return err
+	}
+
+	artifactFile, err := cmd.Flags().GetString("artifact-file")
+	if err != nil {
+		return err
+	}
+
+	labels, err := getLabelsFlag(cmd)
+	if err != nil {
+		return err
+	}
+
+	client, err := c.GetCmfClient(cmd)
+	if err != nil {
+		return err
+	}
+
+	file, err := openArtifactFile(artifactFile)
+	if err != nil {
+		return err
+	}
+	defer file.Close()
+
+	outputArtifact, err := client.CreateArtifact(c.createContext(), environment, newSdkArtifact(name, labels), file)
+	if err != nil {
+		return err
+	}
+
+	return printArtifactOnPrem(cmd, outputArtifact)
+}

internal/flink/command_artifact_delete_onprem.go

@@ -0,0 +1,60 @@
+package flink
+
+import (
+	"github.com/spf13/cobra"
+
+	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+	"github.com/confluentinc/cli/v4/pkg/deletion"
+	"github.com/confluentinc/cli/v4/pkg/errors"
+	"github.com/confluentinc/cli/v4/pkg/resource"
+)
+
+func (c *command) newArtifactDeleteCommandOnPrem() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:         "delete <name-1> [name-2] ... [name-n]",
+		Short:       "Delete one or more Flink artifacts in Confluent Platform.",
+		Long:        "Delete one or more Flink artifacts in Confluent Platform, including all of their versions.",
+		Args:        cobra.MinimumNArgs(1),
+		Annotations: map[string]string{pcmd.RunRequirement: pcmd.RequireCloudLogout},
+		RunE:        c.artifactDeleteOnPrem,
+	}
+
+	cmd.Flags().String("environment", "", "Name of the Flink environment.")
+	addCmfFlagSet(cmd)
+	pcmd.AddForceFlag(cmd)
+
+	cobra.CheckErr(cmd.MarkFlagRequired("environment"))
+
+	return cmd
+}
+
+func (c *command) artifactDeleteOnPrem(cmd *cobra.Command, args []string) error {
+	environment, err := cmd.Flags().GetString("environment")
+	if err != nil {
+		return err
+	}
+
+	client, err := c.GetCmfClient(cmd)
+	if err != nil {
+		return err
+	}
+
+	existenceFunc := func(name string) bool {
+		_, err := client.DescribeArtifact(c.createContext(), environment, name, "")
+		return err == nil
+	}
+
+	if err := deletion.ValidateAndConfirm(cmd, args, existenceFunc, resource.FlinkArtifact); err != nil {
+		suggestions := "List available Flink artifacts with `confluent flink artifact list`."
+		suggestions += "\nCheck that CMF is running and accessible."
+		return errors.NewErrorWithSuggestions(err.Error(), suggestions)
+	}
+
+	// An empty version deletes the artifact and all of its versions.
+	deleteFunc := func(name string) error {
+		return client.DeleteArtifact(c.createContext(), environment, name, "")
+	}
+
+	_, err = deletion.Delete(cmd, args, deleteFunc, resource.FlinkArtifact)
+	return err
+}

internal/flink/command_artifact_describe_onprem.go

@@ -0,0 +1,47 @@
+package flink
+
+import (
+	"github.com/spf13/cobra"
+
+	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+)
+
+func (c *command) newArtifactDescribeCommandOnPrem() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:         "describe <name>",
+		Short:       "Describe a Flink artifact in Confluent Platform.",
+		Long:        "Describe a Flink artifact in Confluent Platform. Details reflect the latest version of the artifact.",
+		Args:        cobra.ExactArgs(1),
+		Annotations: map[string]string{pcmd.RunRequirement: pcmd.RequireCloudLogout},
+		RunE:        c.artifactDescribeOnPrem,
+	}
+
+	cmd.Flags().String("environment", "", "Name of the Flink environment.")
+	addCmfFlagSet(cmd)
+	pcmd.AddOutputFlag(cmd)
+
+	cobra.CheckErr(cmd.MarkFlagRequired("environment"))
+
+	return cmd
+}
+
+func (c *command) artifactDescribeOnPrem(cmd *cobra.Command, args []string) error {
+	name := args[0]
+
+	environment, err := cmd.Flags().GetString("environment")
+	if err != nil {
+		return err
+	}
+
+	client, err := c.GetCmfClient(cmd)
+	if err != nil {
+		return err
+	}
+
+	sdkArtifact, err := client.DescribeArtifact(c.createContext(), environment, name, "")
+	if err != nil {
+		return err
+	}
+
+	return printArtifactOnPrem(cmd, sdkArtifact)
+}

internal/flink/command_artifact_list_onprem.go

@@ -0,0 +1,58 @@
+package flink
+
+import (
+	"github.com/spf13/cobra"
+
+	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+	"github.com/confluentinc/cli/v4/pkg/output"
+)
+
+func (c *command) newArtifactListCommandOnPrem() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:         "list",
+		Short:       "List Flink artifacts in Confluent Platform.",
+		Args:        cobra.NoArgs,
+		Annotations: map[string]string{pcmd.RunRequirement: pcmd.RequireCloudLogout},
+		RunE:        c.artifactListOnPrem,
+	}
+
+	cmd.Flags().String("environment", "", "Name of the Flink environment.")
+	addCmfFlagSet(cmd)
+	pcmd.AddOutputFlag(cmd)
+
+	cobra.CheckErr(cmd.MarkFlagRequired("environment"))
+
+	return cmd
+}
+
+func (c *command) artifactListOnPrem(cmd *cobra.Command, _ []string) error {
+	environment, err := cmd.Flags().GetString("environment")
+	if err != nil {
+		return err
+	}
+
+	client, err := c.GetCmfClient(cmd)
+	if err != nil {
+		return err
+	}
+
+	sdkArtifacts, err := client.ListArtifacts(c.createContext(), environment)
+	if err != nil {
+		return err
+	}
+
+	if output.GetFormat(cmd) == output.Human {
+		list := output.NewList(cmd)
+		for _, artifact := range sdkArtifacts {
+			list.Add(newArtifactOutOnPrem(artifact))
+		}
+		return list.Print()
+	}
+
+	localArtifacts := make([]LocalArtifact, 0, len(sdkArtifacts))
+	for _, sdkArtifact := range sdkArtifacts {
+		localArtifacts = append(localArtifacts, convertSdkArtifactToLocalArtifact(sdkArtifact))
+	}
+
+	return output.SerializedOutput(cmd, localArtifacts)
+}