Fix cluster config environment interaction (#55)

* Fix errors

* Fix version

* Don't check SASL mechanism if running from cluster config

* Update README

* Update README

* Fix README

* Fix README

* Update go version to 1.17

* Fix README

* Bump version

README.md

@@ -49,7 +49,7 @@ that makes it easy to tail and summarize structured data in Kafka.
 
 Either:
 
-1. Run `GO111MODULE="on" go get github.com/segmentio/topicctl/cmd/topicctl`
+1. Run `go install github.com/segmentio/topicctl/cmd/topicctl@latest`
 2. Clone this repo and run `make install` in the repo root
 3. Use the Docker image: `docker pull segment/topicctl`
 
@@ -228,7 +228,12 @@ independently of an `apply` workflow.
 ### Version compatibility
 
 We've tested `topicctl` on Kafka clusters with versions between `0.10.1` and `2.7.1`, inclusive.
-If you run into any compatibility issues, please file a bug.
+
+Note, however, that clusters at versions prior to `2.4.0` cannot use broker APIs for applying and
+thus also require ZooKeeper API access for full functionality. See the
+[cluster access details](#cluster-access-details) section below for more details.
+
+If you run into any unexpected compatibility issues, please file a bug.
 
 ## Config formats
 
@@ -254,13 +259,15 @@ meta:
 spec:
  bootstrapAddrs: # One or more broker bootstrap addresses
  - my-cluster.example.com:9092
- clusterID: abc-123-xyz # Expected cluster ID for cluster (optional, used as safety check only)
+ clusterID: abc-123-xyz # Expected cluster ID for cluster (optional,
+ # used as safety check only)
 
  # ZooKeeper access settings (only required for pre-v2 clusters; leave off to force exclusive use
  # of broker APIs)
  zkAddrs: # One or more cluster zookeeper addresses; if these are
- - zk.example.com:2181 # omitted, then the cluster will only be accessed via broker APIs;
- # see the section below on cluster access for more details.
+ - zk.example.com:2181 # omitted, then the cluster will only be accessed via
+ # broker APIs; see the section below on cluster access for
+ # more details.
  zkPrefix: my-cluster # Prefix for zookeeper nodes if using zookeeper access
  zkLockPath: /topicctl/locks # Path used for apply locks (optional)
 
@@ -274,16 +281,22 @@ spec:
  # SASL settings (optional, not supported if using ZooKeeper)
  sasl:
  enabled: true # Whether SASL is enabled
- mechanism: SCRAM-SHA-512 # Mechanism to use; choices are PLAIN, SCRAM-SHA-256, and SCRAM-SHA-512
- username: my-username # Username; can also be set via TOPICCTL_SASL_USERNAME environment variable
- password: my-password # Password; can also be set via TOPICCTL_SASL_PASSWORD environment variable
+ mechanism: SCRAM-SHA-512 # Mechanism to use;
+ # choices are PLAIN, SCRAM-SHA-256, and SCRAM-SHA-512
+ username: my-username # SASL username
+ password: my-password # SASL password
 ```
 
 Note that the `name`, `environment`, `region`, and `description` fields are used
 for description/identification only, and don't appear in any API calls. They can
 be set arbitrarily, provided that they match up with the values set in the
 associated topic configs.
 
+If the tool is run with the `--expand-env` option, then the cluster config will be prepreocessed
+using [`os.ExpandEnv`](https://pkg.go.dev/os#ExpandEnv) at load time. The latter will replace
+references of the form `$ENV_VAR_NAME` or `${ENV_VAR_NAME}` with the associated values from the
+environment.
+
 ### Topics
 
 Each topic is configured in a YAML file. The following is an

cmd/topicctl/subcmd/apply.go

@@ -180,7 +180,7 @@ func applyTopic(
  return err
  }
 
- clusterConfig, err := config.LoadClusterFile(clusterConfigPath)
+ clusterConfig, err := config.LoadClusterFile(clusterConfigPath, applyConfig.shared.expandEnv)
  if err != nil {
  return err
  }

cmd/topicctl/subcmd/bootstrap.go

@@ -62,7 +62,10 @@ func bootstrapRun(cmd *cobra.Command, args []string) error {
  ctx, cancel := context.WithCancel(context.Background())
  defer cancel()
 
- clusterConfig, err := config.LoadClusterFile(bootstrapConfig.shared.clusterConfig)
+ clusterConfig, err := config.LoadClusterFile(
+ bootstrapConfig.shared.clusterConfig,
+ bootstrapConfig.shared.expandEnv,
+ )
  if err != nil {
  return err
  }

cmd/topicctl/subcmd/check.go

@@ -116,7 +116,7 @@ func checkTopicFile(
  return false, err
  }
 
- clusterConfig, err := config.LoadClusterFile(clusterConfigPath)
+ clusterConfig, err := config.LoadClusterFile(clusterConfigPath, checkConfig.shared.expandEnv)
  if err != nil {
  return false, err
  }

cmd/topicctl/subcmd/shared.go

@@ -16,6 +16,7 @@ import (
 type sharedOptions struct {
  brokerAddr string
  clusterConfig string
+ expandEnv bool
  saslMechanism string
  saslPassword string
  saslUsername string
@@ -47,10 +48,14 @@ func (s sharedOptions) validate() error {
  }
  if s.clusterConfig != "" &&
  (s.zkAddr != "" || s.zkPrefix != "" || s.brokerAddr != "" || s.tlsCACert != "" ||
- s.tlsCert != "" || s.tlsKey != "" || s.saslMechanism != "") {
+ s.tlsCert != "" || s.tlsKey != "" || s.tlsServerName != "" || s.saslMechanism != "") {
  log.Warn("Broker and zk flags are ignored when using cluster-config")
  }
 
+ if s.clusterConfig != "" {
+ return err
+ }
+
  useTLS := s.tlsEnabled || s.tlsCACert != "" || s.tlsCert != "" || s.tlsKey != ""
  useSASL := s.saslMechanism != "" || s.saslPassword != "" || s.saslUsername != ""
 
@@ -76,14 +81,14 @@ func (s sharedOptions) getAdminClient(
  readOnly bool,
 ) (admin.Client, error) {
  if s.clusterConfig != "" {
- clusterConfig, err := config.LoadClusterFile(s.clusterConfig)
+ clusterConfig, err := config.LoadClusterFile(s.clusterConfig, s.expandEnv)
  if err != nil {
  return nil, err
  }
  return clusterConfig.NewAdminClient(
  ctx,
  sess,
- true,
+ readOnly,
  s.saslUsername,
  s.saslPassword,
  )
@@ -125,8 +130,6 @@ func (s sharedOptions) getAdminClient(
  ZKAddrs: []string{s.zkAddr},
  ZKPrefix: s.zkPrefix,
  Sess: sess,
- // Run in read-only mode to ensure that tailing doesn't make any changes
- // in the cluster
  ReadOnly: readOnly,
  },
  )
@@ -141,6 +144,13 @@ func addSharedFlags(cmd *cobra.Command, options *sharedOptions) {
  "",
  "Broker address",
  )
+ cmd.Flags().BoolVarP(
+ &options.expandEnv,
+ "expand-env",
+ "",
+ false,
+ "Expand environment in cluster config",
+ )
  cmd.Flags().StringVar(
  &options.clusterConfig,
  "cluster-config",
@@ -223,6 +233,13 @@ func addSharedConfigOnlyFlags(cmd *cobra.Command, options *sharedOptions) {
  os.Getenv("TOPICCTL_CLUSTER_CONFIG"),
  "Cluster config",
  )
+ cmd.Flags().BoolVarP(
+ &options.expandEnv,
+ "expand-env",
+ "",
+ false,
+ "Expand environment in cluster config",
+ )
  cmd.Flags().StringVar(
  &options.saslPassword,
  "sasl-password",

go.mod

@@ -1,6 +1,6 @@
 module github.com/segmentio/topicctl
 
-go 1.16
+go 1.17
 
 // Use something like this to use a local kafka-go (useful for testing purposes).
 // replace github.com/segmentio/kafka-go => /Users/benjamin.yolken/dev/src/github.com/segmentio/kafka-go
@@ -12,20 +12,40 @@ require (
  github.com/fatih/color v1.9.0
  github.com/ghodss/yaml v1.0.0
  github.com/hashicorp/go-multierror v1.1.0
- github.com/mattn/go-colorable v0.1.6 // indirect
- github.com/mattn/go-tty v0.0.3 // indirect
- github.com/mgutz/ansi v0.0.0-20170206155736-9520e82c474b // indirect
  github.com/olekukonko/tablewriter v0.0.4
- github.com/onsi/gomega v1.5.0 // indirect
- github.com/pkg/term v0.0.0-20200520122047-c3ffed290a03 // indirect
  github.com/samuel/go-zookeeper v0.0.0-20190923202752-2cc03de413da
  github.com/segmentio/kafka-go v0.4.21-0.20211001205616-c03923d67699
  github.com/sirupsen/logrus v1.2.0
  github.com/spf13/cobra v1.0.0
  github.com/stretchr/testify v1.6.1
  github.com/x-cray/logrus-prefixed-formatter v0.5.2
  golang.org/x/crypto v0.0.0-20200204104054-c9f3fb736b72
+)
+
+require (
+ github.com/davecgh/go-spew v1.1.1 // indirect
+ github.com/golang/snappy v0.0.1 // indirect
+ github.com/hashicorp/errwrap v1.0.0 // indirect
+ github.com/inconshreveable/mousetrap v1.0.0 // indirect
+ github.com/jmespath/go-jmespath v0.0.0-20180206201540-c2b33e8439af // indirect
+ github.com/klauspost/compress v1.9.8 // indirect
+ github.com/konsorten/go-windows-terminal-sequences v1.0.1 // indirect
+ github.com/mattn/go-colorable v0.1.6 // indirect
+ github.com/mattn/go-isatty v0.0.12 // indirect
+ github.com/mattn/go-runewidth v0.0.7 // indirect
+ github.com/mattn/go-tty v0.0.3 // indirect
+ github.com/mgutz/ansi v0.0.0-20170206155736-9520e82c474b // indirect
+ github.com/onsi/ginkgo v1.6.0 // indirect
+ github.com/onsi/gomega v1.5.0 // indirect
+ github.com/pierrec/lz4 v2.6.0+incompatible // indirect
+ 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.3 // indirect
+ github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c // indirect
+ github.com/xdg/stringprep v1.0.0 // indirect
  golang.org/x/net v0.0.0-20200202094626-16171245cfb2 // indirect
+ golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae // indirect
+ golang.org/x/text v0.3.0 // indirect
  gopkg.in/check.v1 v1.0.0-20190902080502-41f04d3bba15 // indirect
  gopkg.in/yaml.v2 v2.2.8 // indirect
  gopkg.in/yaml.v3 v3.0.0-20200605160147-a5ece683394c // indirect

go.sum

@@ -131,8 +131,6 @@ github.com/rogpeppe/fastuuid v0.0.0-20150106093220-6724a57986af/go.mod h1:XWv6So
 github.com/russross/blackfriday/v2 v2.0.1/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM=
 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.21-0.20211001180856-4d75f822c8b8 h1:H45Dpqb99IyjZoq/+WCuVz0UWe2JOrajBVmg6QsYyaQ=
-github.com/segmentio/kafka-go v0.4.21-0.20211001180856-4d75f822c8b8/go.mod h1:XzMcoMjSzDGHcIwpWUI7GB43iKZ2fTVmryPSGLf/MPg=
 github.com/segmentio/kafka-go v0.4.21-0.20211001205616-c03923d67699 h1:DM1XDA47wY0myfsik7hUz+pkmI2uhkfKsa6ogOTNLxw=
 github.com/segmentio/kafka-go v0.4.21-0.20211001205616-c03923d67699/go.mod h1:XzMcoMjSzDGHcIwpWUI7GB43iKZ2fTVmryPSGLf/MPg=
 github.com/shurcooL/sanitized_anchor_name v1.0.0/go.mod h1:1NzhyTcUVG4SuEtjjoZeVRXNmyL/1OwPU0+IJeTBvfc=

pkg/admin/throttles_test.go

@@ -328,7 +328,7 @@ func TestParseBrokerThrottles(t *testing.T) {
  },
  }
  _, _, err = ParseBrokerThrottles(badBrokers)
- assert.NotNil(t, err)
+ assert.Error(t, err)
 }
 
 func TestParsePartitionThrottles(t *testing.T) {
@@ -385,5 +385,5 @@ func TestParsePartitionThrottles(t *testing.T) {
  },
  }
  _, _, err = ParsePartitionThrottles(badTopic)
- assert.NotNil(t, err)
+ assert.Error(t, err)
 }

pkg/admin/types_test.go

@@ -155,16 +155,16 @@ func TestTopicRackHelpers(t *testing.T) {
 
  brokerRacks := BrokerRacks(testBrokers)
  minRacks, maxRacks, err := testTopic.RackCounts(brokerRacks)
- assert.Nil(t, err)
+ assert.NoError(t, err)
  assert.Equal(t, minRacks, 1)
  assert.Equal(t, maxRacks, 3)
 
  numRacks, err := testTopic.Partitions[0].NumRacks(brokerRacks)
- assert.Nil(t, err)
+ assert.NoError(t, err)
  assert.Equal(t, 3, numRacks)
 
  racks, err := testTopic.Partitions[0].Racks(brokerRacks)
- assert.Nil(t, err)
+ assert.NoError(t, err)
  assert.Equal(t, []string{"rack1", "rack2", "rack3"}, racks)
 }
 
@@ -343,7 +343,7 @@ func TestPartitionAssignmentHelpers(t *testing.T) {
  result,
  )
  replicas, err := AssignmentsToReplicas(result)
- assert.Nil(t, err)
+ assert.NoError(t, err)
 
  assert.Equal(
  t,