[skip secret scan] CLI-1494 Add schema registry management commands with srsdk (#1185)

* add schema-registry schema commands

create, describe and delete

* add schema-registry subject commands onPrem

* add sr config commands for on-prem and cloud

* add check compatibility commands for on-prem and cloud

* update compatibility command example

* clean up code and structure

* clean up and add error catcher

* exclude schema-registry short length rule

* add examples and error catcher

* Update main.go

* address comments i

* rename response var to resp

* address comments ii

* fix bad merge and conflicts

* Update command_config_describe.go

* address comments iii

* add compatibility and config unit and integration tests

* Update command_compatibility_test.go

* Update command_compatibility_test.go

* address comments iv

* update test output files

* address comments v

* fix naming and descriptions

* update test fixture files

* address comments vi

* Update command_subject_update_onprem.go

* Update subject-update-help.golden

* address comments, revert breaking changes

* Update command_subject_update_onprem.go

Author: MuweiHe

internal/cmd/kafka/command_topic.go

@@ -1,26 +1,16 @@
 package kafka
 
 import (
-	"context"
-	"encoding/binary"
-	"encoding/json"
 	"fmt"
-	"io/ioutil"
-	"os"
-	"path/filepath"
-	"strconv"
 	"time"
 
 	ckafka "github.com/confluentinc/confluent-kafka-go/kafka"
-	srsdk "github.com/confluentinc/schema-registry-sdk-go"
 	"github.com/spf13/cobra"
 
 	pcmd "github.com/confluentinc/cli/internal/pkg/cmd"
 	v1 "github.com/confluentinc/cli/internal/pkg/config/v1"
 	"github.com/confluentinc/cli/internal/pkg/errors"
 	"github.com/confluentinc/cli/internal/pkg/log"
-	"github.com/confluentinc/cli/internal/pkg/output"
-	"github.com/confluentinc/cli/internal/pkg/utils"
 )
 
 const (
@@ -161,87 +151,3 @@ func (c *hasAPIKeyTopicCommand) validateTopic(client *ckafka.AdminClient, topic
 	log.CliLogger.Tracef("validateTopic succeeded")
 	return nil
 }
-
-func registerSchemaWithAuth(cmd *cobra.Command, subject, schemaType, schemaPath string, refs []srsdk.SchemaReference, srClient *srsdk.APIClient, ctx context.Context) ([]byte, error) {
-	schema, err := ioutil.ReadFile(schemaPath)
-	if err != nil {
-		return nil, err
-	}
-
-	response, _, err := srClient.DefaultApi.Register(ctx, subject, srsdk.RegisterSchemaRequest{Schema: string(schema), SchemaType: schemaType, References: refs})
-	if err != nil {
-		return nil, err
-	}
-
-	outputFormat, err := cmd.Flags().GetString(output.FlagName)
-	if err != nil {
-		return nil, err
-	}
-	if outputFormat == output.Human.String() {
-		utils.Printf(cmd, errors.RegisteredSchemaMsg, response.Id)
-	} else {
-		err = output.StructuredOutput(outputFormat, &struct {
-			Id int32 `json:"id" yaml:"id"`
-		}{response.Id})
-		if err != nil {
-			return nil, err
-		}
-	}
-
-	metaInfo := []byte{0x0}
-	schemaIdBuffer := make([]byte, 4)
-	binary.BigEndian.PutUint32(schemaIdBuffer, uint32(response.Id))
-	metaInfo = append(metaInfo, schemaIdBuffer...)
-	return metaInfo, nil
-}
-
-func readSchemaRefs(cmd *cobra.Command) ([]srsdk.SchemaReference, error) {
-	var refs []srsdk.SchemaReference
-	refPath, err := cmd.Flags().GetString("refs")
-	if err != nil {
-		return nil, err
-	}
-	if refPath != "" {
-		refBlob, err := ioutil.ReadFile(refPath)
-		if err != nil {
-			return nil, err
-		}
-		err = json.Unmarshal(refBlob, &refs)
-		if err != nil {
-			return nil, err
-		}
-	}
-	return refs, nil
-}
-
-func storeSchemaReferences(refs []srsdk.SchemaReference, srClient *srsdk.APIClient, ctx context.Context) (map[string]string, error) {
-	dir := filepath.Join(os.TempDir(), "ccloud-schema")
-	if _, err := os.Stat(dir); os.IsNotExist(err) {
-		err = os.Mkdir(dir, 0755)
-		if err != nil {
-			return nil, err
-		}
-	}
-
-	referencePathMap := map[string]string{}
-	for _, ref := range refs {
-		tempStorePath := filepath.Join(dir, ref.Name)
-		if !fileExists(tempStorePath) {
-			schema, _, err := srClient.DefaultApi.GetSchemaByVersion(ctx, ref.Subject, strconv.Itoa(int(ref.Version)), &srsdk.GetSchemaByVersionOpts{})
-			if err != nil {
-				return nil, err
-			}
-			err = os.MkdirAll(filepath.Dir(tempStorePath), 0755)
-			if err != nil {
-				return nil, err
-			}
-			err = ioutil.WriteFile(tempStorePath, []byte(schema.Schema), 0644)
-			if err != nil {
-				return nil, err
-			}
-		}
-		referencePathMap[ref.Name] = tempStorePath
-	}
-
-	return referencePathMap, nil
-}

internal/cmd/kafka/command_topic_consume_onprem.go

@@ -85,7 +85,7 @@ func (c *authenticatedTopicCommand) onPremConsume(cmd *cobra.Command, args []str
 		if c.State == nil { // require log-in to use oauthbearer token
 			return errors.NewErrorWithSuggestions(errors.NotLoggedInErrorMsg, errors.AuthTokenSuggestion)
 		}
-		srClient, ctx, err = sr.GetAPIClientWithToken(cmd, nil, c.Version, c.AuthToken())
+		srClient, ctx, err = sr.GetSrApiClientWithToken(cmd, nil, c.Version, c.AuthToken())
 		if err != nil {
 			return err
 		}

internal/cmd/kafka/command_topic_produce.go

@@ -79,7 +79,7 @@ func (c *hasAPIKeyTopicCommand) produce(cmd *cobra.Command, args []string) error
 	if err != nil {
 		return err
 	}
-	refs, err := readSchemaRefs(cmd)
+	refs, err := sr.ReadSchemaRefs(cmd)
 	if err != nil {
 		return err
 	}
@@ -243,12 +243,12 @@ func (c *hasAPIKeyTopicCommand) registerSchema(cmd *cobra.Command, valueFormat,
 			}
 		}
 
-		info, err := registerSchemaWithAuth(cmd, subject, schemaType, schemaPath, refs, srClient, ctx)
+		info, err := sr.RegisterSchemaWithAuth(cmd, subject, schemaType, schemaPath, refs, srClient, ctx)
 		if err != nil {
 			return nil, nil, err
 		}
 		metaInfo = info
-		referencePathMap, err = storeSchemaReferences(refs, srClient, ctx)
+		referencePathMap, err = sr.StoreSchemaReferences(refs, srClient, ctx)
 		if err != nil {
 			return metaInfo, nil, err
 		}

internal/cmd/kafka/command_topic_produce_onprem.go

@@ -93,7 +93,7 @@ func (c *authenticatedTopicCommand) onPremProduce(cmd *cobra.Command, args []str
 	if err != nil {
 		return err
 	}
-	refs, err := readSchemaRefs(cmd)
+	refs, err := sr.ReadSchemaRefs(cmd)
 	if err != nil {
 		return err
 	}
@@ -182,17 +182,16 @@ func (c *authenticatedTopicCommand) registerSchema(cmd *cobra.Command, valueForm
 		if c.State == nil { // require log-in to use oauthbearer token
 			return nil, nil, errors.NewErrorWithSuggestions(errors.NotLoggedInErrorMsg, errors.AuthTokenSuggestion)
 		}
-		srClient, ctx, err := sr.GetAPIClientWithToken(cmd, nil, c.Version, c.AuthToken())
+		srClient, ctx, err := sr.GetSrApiClientWithToken(cmd, nil, c.Version, c.AuthToken())
 		if err != nil {
 			return nil, nil, err
 		}
 
-		info, err := registerSchemaWithAuth(cmd, subject, schemaType, schemaPath, refs, srClient, ctx)
+		metaInfo, err = sr.RegisterSchemaWithAuth(cmd, subject, schemaType, schemaPath, refs, srClient, ctx)
 		if err != nil {
 			return nil, nil, err
 		}
-		metaInfo = info
-		referencePathMap, err = storeSchemaReferences(refs, srClient, ctx)
+		referencePathMap, err = sr.StoreSchemaReferences(refs, srClient, ctx)
 		if err != nil {
 			return metaInfo, nil, err
 		}

internal/cmd/kafka/confluent_kafka.go

@@ -20,6 +20,7 @@ import (
 	"github.com/google/uuid"
 	"github.com/spf13/cobra"
 
+	sr "github.com/confluentinc/cli/internal/cmd/schema-registry"
 	configv1 "github.com/confluentinc/cli/internal/pkg/config/v1"
 	"github.com/confluentinc/cli/internal/pkg/errors"
 	"github.com/confluentinc/cli/internal/pkg/form"
@@ -347,7 +348,7 @@ func (h *GroupHandler) RequestSchema(value []byte) (string, map[string]string, e
 	tempStorePath := filepath.Join(h.Properties.SchemaPath, fmt.Sprintf("%s-%d.txt", h.Subject, schemaID))
 	tempRefStorePath := filepath.Join(h.Properties.SchemaPath, fmt.Sprintf("%s-%d.ref", h.Subject, schemaID))
 	var references []srsdk.SchemaReference
-	if !fileExists(tempStorePath) || !fileExists(tempRefStorePath) {
+	if !utils.FileExists(tempStorePath) || !utils.FileExists(tempRefStorePath) {
 		// TODO: add handler for writing schema failure
 		getSchemaOpts := srsdk.GetSchemaOpts{
 			Subject: optional.NewString(h.Subject),
@@ -382,7 +383,7 @@ func (h *GroupHandler) RequestSchema(value []byte) (string, map[string]string, e
 	}
 
 	// Store the references in temporary files
-	referencePathMap, err := storeSchemaReferences(references, h.SrClient, h.Ctx)
+	referencePathMap, err := sr.StoreSchemaReferences(references, h.SrClient, h.Ctx)
 	if err != nil {
 		return "", nil, err
 	}

internal/cmd/kafka/utils.go

@@ -27,14 +27,6 @@ func copyMap(inputMap map[string]string) map[string]string {
 	return newMap
 }
 
-func fileExists(filename string) bool {
-	info, err := os.Stat(filename)
-	if os.IsNotExist(err) {
-		return false
-	}
-	return !info.IsDir()
-}
-
 func toCreateTopicConfigs(topicConfigsMap map[string]string) []kafkarestv3.ConfigData {
 	topicConfigs := make([]kafkarestv3.ConfigData, len(topicConfigsMap))
 	i := 0

internal/cmd/schema-registry/command.go

@@ -19,10 +19,11 @@ func New(cfg *v1.Config, prerunner pcmd.PreRunner, srClient *srsdk.APIClient) *c
 	c := pcmd.NewAuthenticatedCLICommand(cmd, prerunner)
 
 	c.AddCommand(newClusterCommand(cfg, prerunner, srClient))
+	c.AddCommand(newCompatibilityCommand(cfg, prerunner, srClient))
+	c.AddCommand(newConfigCommand(cfg, prerunner, srClient))
 	c.AddCommand(newExporterCommand(prerunner, srClient))
-	c.AddCommand(newSchemaCommand(prerunner, srClient))
-	c.AddCommand(newSubjectCommand(prerunner, srClient))
-
+	c.AddCommand(newSchemaCommand(cfg, prerunner, srClient))
+	c.AddCommand(newSubjectCommand(cfg, prerunner, srClient))
 	return c.Command
 }
 

internal/cmd/schema-registry/command_cluster.go

@@ -36,7 +36,7 @@ func newClusterCommand(cfg *v1.Config, prerunner pcmd.PreRunner, srClient *srsdk
 
 	c.AddCommand(c.newDescribeCommand(cfg))
 	c.AddCommand(c.newEnableCommand(cfg))
-	c.AddCommand(c.newListCommand())
+	c.AddCommand(c.newListCommandOnPrem())
 	c.AddCommand(c.newUpdateCommand(cfg))
 
 	return c.Command

…/schema-registry/command_cluster_list.go …-registry/command_cluster_list_onprem.gointernal/cmd/schema-registry/command_cluster_list.go renamed to internal/cmd/schema-registry/command_cluster_list_onprem.go internal/cmd/schema-registry/command_cluster_list.go renamed to internal/cmd/schema-registry/command_cluster_list_onprem.go

@@ -14,13 +14,13 @@ import (
 
 const clusterType = "schema-registry-cluster"
 
-func (c *clusterCommand) newListCommand() *cobra.Command {
+func (c *clusterCommand) newListCommandOnPrem() *cobra.Command {
 	cmd := &cobra.Command{
 		Use:         "list",
 		Short:       "List registered Schema Registry clusters.",
 		Long:        "List Schema Registry clusters that are registered with the MDS cluster registry.",
 		Args:        cobra.NoArgs,
-		RunE:        pcmd.NewCLIRunE(c.list),
+		RunE:        pcmd.NewCLIRunE(c.onPremList),
 		Annotations: map[string]string{pcmd.RunRequirement: pcmd.RequireOnPremLogin},
 	}
 
@@ -30,7 +30,7 @@ func (c *clusterCommand) newListCommand() *cobra.Command {
 	return cmd
 }
 
-func (c *clusterCommand) list(cmd *cobra.Command, _ []string) error {
+func (c *clusterCommand) onPremList(cmd *cobra.Command, _ []string) error {
 	ctx := context.WithValue(context.Background(), mds.ContextAccessToken, c.State.AuthToken)
 	opts := &mds.ClusterRegistryListOpts{ClusterType: optional.NewString(clusterType)}
 

internal/cmd/schema-registry/command_compatibility.go

@@ -0,0 +1,34 @@
+package schemaregistry
+
+import (
+	srsdk "github.com/confluentinc/schema-registry-sdk-go"
+	"github.com/spf13/cobra"
+
+	pcmd "github.com/confluentinc/cli/internal/pkg/cmd"
+	v1 "github.com/confluentinc/cli/internal/pkg/config/v1"
+)
+
+type compatibilityCommand struct {
+	*pcmd.AuthenticatedStateFlagCommand
+	srClient *srsdk.APIClient
+}
+
+func newCompatibilityCommand(cfg *v1.Config, prerunner pcmd.PreRunner, srClient *srsdk.APIClient) *cobra.Command {
+	cmd := &cobra.Command{
+		Use:         "compatibility",
+		Short:       "Validate schema compatibility.",
+		Annotations: map[string]string{pcmd.RunRequirement: pcmd.RequireCloudLoginOrOnPremLogin},
+	}
+
+	c := &compatibilityCommand{srClient: srClient}
+
+	if cfg.IsCloudLogin() {
+		c.AuthenticatedStateFlagCommand = pcmd.NewAuthenticatedStateFlagCommand(cmd, prerunner)
+		c.AddCommand(c.newValidateCommand())
+	} else {
+		c.AuthenticatedStateFlagCommand = pcmd.NewAuthenticatedWithMDSStateFlagCommand(cmd, prerunner)
+		c.AddCommand(c.newValidateCommandOnPrem())
+	}
+
+	return cmd
+}