feat: create acls

.gitignore

@@ -17,3 +17,6 @@ vendor/
 build/
 
 .vscode
+
+# Emacs backups
+*~

cmd/topicctl/subcmd/create.go

@@ -0,0 +1,200 @@
+package subcmd
+
+import (
+	"context"
+	"fmt"
+	"os"
+	"os/signal"
+	"path/filepath"
+	"syscall"
+
+	"github.com/segmentio/topicctl/pkg/admin"
+	"github.com/segmentio/topicctl/pkg/cli"
+	"github.com/segmentio/topicctl/pkg/config"
+	"github.com/segmentio/topicctl/pkg/create"
+	log "github.com/sirupsen/logrus"
+	"github.com/spf13/cobra"
+)
+
+var createCmd = &cobra.Command{
+	Use:               "create [resource type]",
+	Short:             "creates one or more resources",
+	PersistentPreRunE: createPreRun,
+}
+
+type createCmdConfig struct {
+	dryRun      bool
+	pathPrefix  string
+	skipConfirm bool
+
+	shared sharedOptions
+}
+
+var createConfig createCmdConfig
+
+func init() {
+	createCmd.Flags().BoolVar(
+		&createConfig.dryRun,
+		"dry-run",
+		false,
+		"Do a dry-run",
+	)
+	createCmd.Flags().StringVar(
+		&createConfig.pathPrefix,
+		"path-prefix",
+		os.Getenv("TOPICCTL_ACL_PATH_PREFIX"),
+		"Prefix for ACL config paths",
+	)
+	createCmd.Flags().BoolVar(
+		&createConfig.skipConfirm,
+		"skip-confirm",
+		false,
+		"Skip confirmation prompts during creation process",
+	)
+
+	addSharedFlags(createCmd, &createConfig.shared)
+	createCmd.AddCommand(
+		createACLsCmd(),
+	)
+	RootCmd.AddCommand(createCmd)
+}
+
+func createPreRun(cmd *cobra.Command, args []string) error {
+	if err := RootCmd.PersistentPreRunE(cmd, args); err != nil {
+		return err
+	}
+	return createConfig.shared.validate()
+}
+
+func createACLsCmd() *cobra.Command {
+	cmd := &cobra.Command{
+		Use:     "acls [acl configs]",
+		Short:   "creates ACLs from configuration files",
+		Args:    cobra.MinimumNArgs(1),
+		RunE:    createACLRun,
+		PreRunE: createPreRun,
+	}
+
+	return cmd
+}
+
+func createACLRun(cmd *cobra.Command, args []string) error {
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	sigChan := make(chan os.Signal, 1)
+	signal.Notify(sigChan, os.Interrupt, syscall.SIGTERM)
+	go func() {
+		<-sigChan
+		cancel()
+	}()
+
+	// Keep a cache of the admin clients with the cluster config path as the key
+	adminClients := map[string]admin.Client{}
+
+	defer func() {
+		for _, adminClient := range adminClients {
+			adminClient.Close()
+		}
+	}()
+
+	matchCount := 0
+
+	for _, arg := range args {
+		if createConfig.pathPrefix != "" && !filepath.IsAbs(arg) {
+			arg = filepath.Join(createConfig.pathPrefix, arg)
+		}
+
+		matches, err := filepath.Glob(arg)
+		if err != nil {
+			return err
+		}
+
+		for _, match := range matches {
+			matchCount++
+			if err := createACL(ctx, match, adminClients); err != nil {
+				return err
+			}
+		}
+	}
+
+	if matchCount == 0 {
+		return fmt.Errorf("No ACL configs match the provided args (%+v)", args)
+	}
+
+	return nil
+}
+
+func createACL(
+	ctx context.Context,
+	aclConfigPath string,
+	adminClients map[string]admin.Client,
+) error {
+	clusterConfigPath, err := clusterConfigForACLCreate(aclConfigPath)
+	if err != nil {
+		return err
+	}
+
+	aclConfigs, err := config.LoadACLsFile(aclConfigPath)
+	if err != nil {
+		return err
+	}
+
+	clusterConfig, err := config.LoadClusterFile(clusterConfigPath, createConfig.shared.expandEnv)
+	if err != nil {
+		return err
+	}
+
+	adminClient, ok := adminClients[clusterConfigPath]
+	if !ok {
+		adminClient, err = clusterConfig.NewAdminClient(
+			ctx,
+			nil,
+			applyConfig.dryRun,
+			applyConfig.shared.saslUsername,
+			applyConfig.shared.saslPassword,
+		)
+		if err != nil {
+			return err
+		}
+		adminClients[clusterConfigPath] = adminClient
+	}
+
+	cliRunner := cli.NewCLIRunner(adminClient, log.Infof, false)
+
+	for _, aclConfig := range aclConfigs {
+		log.Infof(
+			"Processing ACL %s in config %s with cluster config %s",
+			aclConfig.Meta.Name,
+			aclConfigPath,
+			clusterConfigPath,
+		)
+
+		creatorConfig := create.ACLCreatorConfig{
+			DryRun:        applyConfig.dryRun,
+			SkipConfirm:   applyConfig.skipConfirm,
+			ACLConfig:     aclConfig,
+			ClusterConfig: clusterConfig,
+		}
+
+		if err := cliRunner.CreateACL(ctx, creatorConfig); err != nil {
+			return err
+		}
+	}
+
+	return nil
+}
+
+func clusterConfigForACLCreate(aclConfigPath string) (string, error) {
+	if createConfig.shared.clusterConfig != "" {
+		return createConfig.shared.clusterConfig, nil
+	}
+
+	return filepath.Abs(
+		filepath.Join(
+			filepath.Dir(aclConfigPath),
+			"..",
+			"cluster.yaml",
+		),
+	)
+}

cmd/topicctl/subcmd/rebalance.go

@@ -3,14 +3,15 @@ package subcmd
 import (
 	"context"
 	"fmt"
-	"github.com/spf13/cobra"
 	"os"
 	"os/signal"
 	"path/filepath"
 	"strconv"
 	"syscall"
 	"time"
 
+	"github.com/spf13/cobra"
+
 	"github.com/segmentio/topicctl/pkg/admin"
 	"github.com/segmentio/topicctl/pkg/apply"
 	"github.com/segmentio/topicctl/pkg/cli"
@@ -159,7 +160,7 @@ func rebalanceRun(cmd *cobra.Command, args []string) error {
 
 		for _, topicConfig := range topicConfigs {
 			// topic config should be consistent with the cluster config
-			if err := config.CheckConsistency(topicConfig, clusterConfig); err != nil {
+			if err := config.CheckConsistency(topicConfig.Meta, clusterConfig); err != nil {
 				log.Errorf("topic file: %s inconsistent with cluster: %s", topicFile, clusterConfigPath)
 				continue
 			}

go.mod

@@ -11,12 +11,13 @@ require (
 	github.com/hashicorp/go-multierror v1.1.1
 	github.com/olekukonko/tablewriter v0.0.5
 	github.com/samuel/go-zookeeper v0.0.0-20190923202752-2cc03de413da
-	github.com/segmentio/kafka-go v0.4.44
+	github.com/segmentio/kafka-go v0.4.45-0.20231030174323-c6378c391a97
 	github.com/segmentio/kafka-go/sasl/aws_msk_iam v0.0.0-20220211180808-78889264d070
 	github.com/sirupsen/logrus v1.9.0
 	github.com/spf13/cobra v1.5.0
 	github.com/stretchr/testify v1.8.0
 	github.com/x-cray/logrus-prefixed-formatter v0.5.2
+	github.com/xdg-go/pbkdf2 v1.0.0
 	golang.org/x/crypto v0.14.0
 )
 
@@ -37,7 +38,6 @@ require (
 	github.com/pkg/term v0.0.0-20200520122047-c3ffed290a03 // indirect
 	github.com/pmezard/go-difflib v1.0.0 // indirect
 	github.com/spf13/pflag v1.0.5 // indirect
-	github.com/xdg-go/pbkdf2 v1.0.0 // indirect
 	github.com/xdg-go/scram v1.1.2 // indirect
 	github.com/xdg-go/stringprep v1.0.4 // indirect
 	golang.org/x/sys v0.13.0 // indirect

go.sum

@@ -74,8 +74,8 @@ github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQD
 github.com/samuel/go-zookeeper v0.0.0-20190923202752-2cc03de413da h1:p3Vo3i64TCLY7gIfzeQaUJ+kppEO5WQG3cL8iE8tGHU=
 github.com/samuel/go-zookeeper v0.0.0-20190923202752-2cc03de413da/go.mod h1:gi+0XIa01GRL2eRQVjQkKGqKF3SF9vZR/HnPullcV2E=
 github.com/segmentio/kafka-go v0.4.28/go.mod h1:XzMcoMjSzDGHcIwpWUI7GB43iKZ2fTVmryPSGLf/MPg=
-github.com/segmentio/kafka-go v0.4.44 h1:Vjjksniy0WSTZ7CuVJrz1k04UoZeTc77UV6Yyk6tLY4=
-github.com/segmentio/kafka-go v0.4.44/go.mod h1:HjF6XbOKh0Pjlkr5GVZxt6CsjjwnmhVOfURM5KMd8qg=
+github.com/segmentio/kafka-go v0.4.45-0.20231030174323-c6378c391a97 h1:vKYoioQZ7SgGcES2pKoNq7zV8ncKNvblHp+0O+dOeI0=
+github.com/segmentio/kafka-go v0.4.45-0.20231030174323-c6378c391a97/go.mod h1:HjF6XbOKh0Pjlkr5GVZxt6CsjjwnmhVOfURM5KMd8qg=
 github.com/segmentio/kafka-go/sasl/aws_msk_iam v0.0.0-20220211180808-78889264d070 h1:ng1Z/x5LLOIrzgWUOtypsCkR+dHTux7slqOCVkuwQBo=
 github.com/segmentio/kafka-go/sasl/aws_msk_iam v0.0.0-20220211180808-78889264d070/go.mod h1:IjMUGcOJoATsnlqAProGN1ezXeEgU5GCWr1/EzmkEMA=
 github.com/sirupsen/logrus v1.9.0 h1:trlNQbNUG3OdDrDil03MCb1H2o9nJ1x4/5LYw7byDE0=

pkg/admin/brokerclient.go

@@ -110,6 +110,7 @@ func NewBrokerAdminClient(
 	if _, ok := maxVersions["DescribeUserScramCredentials"]; ok {
 		supportedFeatures.Users = true
 	}
+
 	log.Debugf("Supported features: %+v", supportedFeatures)
 
 	adminClient := &BrokerAdminClient{
@@ -407,13 +408,17 @@ func (c *BrokerAdminClient) GetUsers(
 		return nil, err
 	}
 
-	if err = util.DescribeUserScramCredentialsResponseResultsError(resp.Results); err != nil {
-		return nil, err
-	}
-
 	results := []UserInfo{}
 
 	for _, result := range resp.Results {
+		if result.Error != nil {
+			log.Debugf("got here")
+			if errors.Is(result.Error, kafka.ResourceNotFound) {
+				log.Debugf("Skipping over user %s because it does not exist", result.User)
+				continue
+			}
+			return nil, fmt.Errorf("Error getting description of user %s: %+v", result.User, result.Error)
+		}
 		var credentials []CredentialInfo
 		for _, credential := range result.CredentialInfos {
 			credentials = append(credentials, CredentialInfo{
@@ -808,7 +813,7 @@ func (c *BrokerAdminClient) GetACLs(
 	return aclinfos, nil
 }
 
-// CreateACLs creates an ACL in the cluster.
+// CreateACLs creates ACLs in the cluster.
 func (c *BrokerAdminClient) CreateACLs(
 	ctx context.Context,
 	acls []kafka.ACLEntry,

pkg/admin/brokerclient_test.go

@@ -770,5 +770,4 @@ func TestBrokerClientCreateACLReadOnly(t *testing.T) {
 
 	err = client.CreateACLs(ctx, []kafka.ACLEntry{})
 	assert.Equal(t, err, errors.New("Cannot create ACL in read-only mode"))
-
 }

pkg/admin/client.go

@@ -77,7 +77,7 @@ type Client interface {
 		config kafka.TopicConfig,
 	) error
 
-	// Create ACLs creates ACLs in the cluster.
+	// CreateACLs creates ACLs in the cluster.
 	CreateACLs(
 		ctx context.Context,
 		acls []kafka.ACLEntry,

pkg/admin/types.go

@@ -94,7 +94,7 @@ type ACLInfo struct {
 // as a Cobra flag.
 type ResourceType kafka.ResourceType
 
-var resourceTypeMap = map[string]kafka.ResourceType{
+var ResourceTypeMap = map[string]kafka.ResourceType{
 	"any":             kafka.ResourceTypeAny,
 	"topic":           kafka.ResourceTypeTopic,
 	"group":           kafka.ResourceTypeGroup,
@@ -125,7 +125,7 @@ func (r *ResourceType) String() string {
 
 // Set is used by Cobra to set the value of a variable from a Cobra flag.
 func (r *ResourceType) Set(v string) error {
-	rt, ok := resourceTypeMap[strings.ToLower(v)]
+	rt, ok := ResourceTypeMap[strings.ToLower(v)]
 	if !ok {
 		return errors.New(`must be one of "any", "topic", "group", "cluster", "transactionalid", or "delegationtoken"`)
 	}
@@ -144,7 +144,7 @@ func (r *ResourceType) Type() string {
 // as a Cobra flag.
 type PatternType kafka.PatternType
 
-var patternTypeMap = map[string]kafka.PatternType{
+var PatternTypeMap = map[string]kafka.PatternType{
 	"any":      kafka.PatternTypeAny,
 	"match":    kafka.PatternTypeMatch,
 	"literal":  kafka.PatternTypeLiteral,
@@ -169,7 +169,7 @@ func (p *PatternType) String() string {
 
 // Set is used by Cobra to set the value of a variable from a Cobra flag.
 func (p *PatternType) Set(v string) error {
-	pt, ok := patternTypeMap[strings.ToLower(v)]
+	pt, ok := PatternTypeMap[strings.ToLower(v)]
 	if !ok {
 		return errors.New(`must be one of "any", "match", "literal", or "prefixed"`)
 	}
@@ -188,7 +188,7 @@ func (r *PatternType) Type() string {
 // as a Cobra flag.
 type ACLOperationType kafka.ACLOperationType
 
-var aclOperationTypeMap = map[string]kafka.ACLOperationType{
+var AclOperationTypeMap = map[string]kafka.ACLOperationType{
 	"any":             kafka.ACLOperationTypeAny,
 	"all":             kafka.ACLOperationTypeAll,
 	"read":            kafka.ACLOperationTypeRead,
@@ -237,7 +237,7 @@ func (o *ACLOperationType) String() string {
 
 // Set is used by Cobra to set the value of a variable from a Cobra flag.
 func (o *ACLOperationType) Set(v string) error {
-	ot, ok := aclOperationTypeMap[strings.ToLower(v)]
+	ot, ok := AclOperationTypeMap[strings.ToLower(v)]
 	if !ok {
 		return errors.New(`must be one of "any", "all", "read", "write", "create", "delete", "alter", "describe", "clusteraction", "describeconfigs", "alterconfigs" or "idempotentwrite"`)
 	}
@@ -256,7 +256,7 @@ func (o *ACLOperationType) Type() string {
 // as a Cobra flag.
 type ACLPermissionType kafka.ACLPermissionType
 
-var aclPermissionTypeMap = map[string]kafka.ACLPermissionType{
+var AclPermissionTypeMap = map[string]kafka.ACLPermissionType{
 	"any":   kafka.ACLPermissionTypeAny,
 	"allow": kafka.ACLPermissionTypeAllow,
 	"deny":  kafka.ACLPermissionTypeDeny,
@@ -278,7 +278,7 @@ func (p *ACLPermissionType) String() string {
 
 // Set is used by Cobra to set the value of a variable from a Cobra flag.
 func (p *ACLPermissionType) Set(v string) error {
-	pt, ok := aclPermissionTypeMap[strings.ToLower(v)]
+	pt, ok := AclPermissionTypeMap[strings.ToLower(v)]
 	if !ok {
 		return errors.New(`must be one of "any", "allow", or "deny"`)
 	}