feat: create acls

.gitignore

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

README.md

@@ -155,6 +155,15 @@ The `check` command validates that each topic config has the correct fields set
 consistent with the associated cluster config. Unless `--validate-only` is set, it then
 checks the topic config against the state of the topic in the corresponding cluster.
 
+#### create
+```
+topicctl create [flags] [command]
+```
+
+The `create` command creates resources in the cluster from a configuration file. 
+Currently, only ACLs are supported. The create command is separate from the apply
+command as it is intended for usage with immutable resources managed by topicctl.
+
 #### get
 
 ```
@@ -419,6 +428,47 @@ This subcommand will not rebalance a topic if:
 1. a topic's `retention.ms` in the kafka cluster does not match the topic's `retentionMinutes` setting in the topic config
 1. a topic does not exist in the kafka cluster
 
+### ACLs
+
+Sets of ACLs can be configured in a YAML file. The following is an
+annotated example:
+
+```yaml
+meta:
+  name: acls-test                       # Name of the group of ACLs
+  cluster: my-cluster                   # Name of the cluster
+  environment: stage                    # Environment of the cluster
+  region: us-west-2                     # Region of the cluster
+  description: |                        # Free-text description of the topic (optional)
+    Test topic in my-cluster.
+  labels:                               # Custom key-value pairs purposed for ACL bookkeeping (optional)
+    key1: value1
+    key2: value2
+
+spec:
+  acls:
+    - resource:
+        type: topic                     # Type of resource (topic, group, cluster, etc.)
+        name: test-topic                # Name of the resource to apply an ACL to
+        patternType: literal            # Type of pattern (literal, prefixed, etc.)
+        principal: User:my-user         # Principal to apply the ACL to
+        host: *                         # Host to apply the ACL to
+        permission: allow			    # Permission to apply (allow, deny)
+      operations:                       # List of operations to use for the ACLs
+        - read
+        - describe
+```
+
+The `cluster`, `environment`, and `region` fields are used for matching
+against a cluster config and double-checking that the cluster we're applying
+in is correct; they don't appear in any API calls.
+
+See the [Kafka documentation](https://kafka.apache.org/documentation/#security_authz_primitives)
+for more details on the parameters that can be set in the `acls` field.
+
+Multiple groups of ACLs can be included in the same file, separated by `---` lines, provided
+that they reference the same cluster.
+
 ## Tool safety
 
 The `bootstrap`, `get`, `repl`, and `tail` subcommands are read-only and should never make
@@ -441,6 +491,9 @@ The `apply` subcommand can make changes, but under the following conditions:
 
 The `reset-offsets` command can also make changes in the cluster and should be used carefully.
 
+The `create` command can be used to create new resources in the cluster. It cannot be used with
+mutable resources.
+
 ### Idempotency
 
 Apply runs are designed to be idemponent- the effects should be the same no matter how many

cmd/topicctl/subcmd/create.go

@@ -0,0 +1,201 @@
+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.PersistentFlags().BoolVar(
+		&createConfig.dryRun,
+		"dry-run",
+		false,
+		"Do a dry-run",
+	)
+	createCmd.PersistentFlags().StringVar(
+		&createConfig.pathPrefix,
+		"path-prefix",
+		os.Getenv("TOPICCTL_ACL_PATH_PREFIX"),
+		"Prefix for ACL config paths",
+	)
+	createCmd.PersistentFlags().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,
+			createConfig.dryRun,
+			createConfig.shared.saslUsername,
+			createConfig.shared.saslPassword,
+		)
+		if err != nil {
+			return err
+		}
+		adminClients[clusterConfigPath] = adminClient
+	}
+
+	cliRunner := cli.NewCLIRunner(adminClient, log.Infof, false)
+
+	for _, aclConfig := range aclConfigs {
+		aclConfig.SetDefaults()
+		log.Infof(
+			"Processing ACL %s in config %s with cluster config %s",
+			aclConfig.Meta.Name,
+			aclConfigPath,
+			clusterConfigPath,
+		)
+
+		creatorConfig := create.ACLCreatorConfig{
+			DryRun:        createConfig.dryRun,
+			SkipConfirm:   createConfig.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
 			}

cmd/topicctl/subcmd/reset.go

@@ -6,9 +6,9 @@ import (
 	"fmt"
 	"strconv"
 
-	"github.com/segmentio/topicctl/pkg/apply"
 	"github.com/segmentio/topicctl/pkg/cli"
 	"github.com/segmentio/topicctl/pkg/groups"
+	"github.com/segmentio/topicctl/pkg/util"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 )
@@ -167,7 +167,7 @@ func resetOffsetsRun(cmd *cobra.Command, args []string) error {
 		"Please ensure that all other consumers are stopped, otherwise the reset might be overridden.",
 	)
 
-	ok, _ := apply.Confirm("OK to continue?", false)
+	ok, _ := util.Confirm("OK to continue?", false)
 	if !ok {
 		return errors.New("Stopping because of user response")
 	}

cmd/topicctl/subcmd/tester.go

@@ -10,7 +10,7 @@ import (
 	"time"
 
 	"github.com/segmentio/kafka-go"
-	"github.com/segmentio/topicctl/pkg/apply"
+	"github.com/segmentio/topicctl/pkg/util"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 )
@@ -104,7 +104,7 @@ func runTestReader(ctx context.Context) error {
 		testerConfig.readConsumer,
 	)
 
-	ok, _ := apply.Confirm("OK to continue?", false)
+	ok, _ := util.Confirm("OK to continue?", false)
 	if !ok {
 		return errors.New("Stopping because of user response")
 	}
@@ -153,7 +153,7 @@ func runTestWriter(ctx context.Context) error {
 		testerConfig.writeRate,
 	)
 
-	ok, _ := apply.Confirm("OK to continue?", false)
+	ok, _ := util.Confirm("OK to continue?", false)
 	if !ok {
 		return errors.New("Stopping because of user response")
 	}

examples/auth/acls/acl-default.yaml

@@ -0,0 +1,31 @@
+meta:
+  name: acl-default
+  cluster: local-cluster-auth
+  environment: local-env
+  region: local-region
+  description: |
+    This is a default ACL for the local cluster.
+    It grants read and describe access to the topic `my-topic` and read access to the group `my-group`
+    to the user `default`.
+
+spec:
+  acls:
+    - resource:
+        type: topic
+        name: my-topic
+        patternType: literal
+        principal: 'User:default'
+        host: '*'
+        permission: allow
+      operations:
+        - Read
+        - Describe
+    - resource:
+        type: group
+        name: my-group
+        patternType: prefixed
+        principal: 'User:default'
+        host: '*'
+        permission: allow
+      operations:
+        - Read

go.mod

@@ -11,7 +11,7 @@ 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
 	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