Skip to content

Commit 78b4049

Browse files
agapoffagapoff
committed
Add help
1 parent fefe9cc commit 78b4049

File tree

1 file changed

+118
-48
lines changed

1 file changed

+118
-48
lines changed

‎kafka-ops.go‎

Lines changed: 118 additions & 48 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ import (
1111
"io/ioutil"
1212
"gopkg.in/yaml.v2"
1313
"encoding/json"
14+
"errors"
1415
)
1516

1617
var (
@@ -25,6 +26,7 @@ var (
2526
isJSON bool
2627
actionApply bool
2728
actionDump bool
29+
actionHelp bool
2830
errorStop bool
2931
)
3032

@@ -59,6 +61,8 @@ type Resource struct {
5961

6062
type OperationHost string
6163

64+
type Exit struct { Code int }
65+
6266
const (
6367
Ok = "\033[0;32m"
6468
Changed = "\033[0;33m"
@@ -75,20 +79,27 @@ func init() {
7579
flag.StringVar(&password, "password", "", "Password for authentication (can be also set by Env variable KAFKA_PASSWORD")
7680
flag.BoolVar(&actionApply, "apply", false, "Apply spec-file to the broker, create all entities that do not exist there; this is the default action")
7781
flag.BoolVar(&actionDump, "dump", false, "Dump broker entities in YAML (default) or JSON format to stdout or to a file if --spec option is defined")
82+
flag.BoolVar(&actionHelp, "help", false, "Print usage")
7883
flag.BoolVar(&isYAML, "yaml", false, "Spec-file is in YAML format (will try to detect format if none of --yaml or --json is set)")
7984
flag.BoolVar(&isJSON, "json", false, "Spec-file is in JSON format (will try to detect format if none of --yaml or --json is set)")
8085
flag.BoolVar(&errorStop, "stop-on-error", false, "Exit on first occurred error")
8186
flag.BoolVar(&verbose, "verbose", false, "Verbose output")
87+
flag.Usage = func() {
88+
usage()
89+
}
8290
flag.Parse()
8391

84-
if !actionApply && !actionDump {
85-
actionApply = true
92+
if !actionApply && !actionDump && !actionHelp {
93+
fmt.Println("Please define one of the actions: --dump, --apply, --help")
94+
os.Exit(1)
8695
}
8796
if actionApply && actionDump {
88-
panic("Please define one of the actions: --dump, --apply")
97+
fmt.Println("Please define one of the actions: --dump, --apply. Refer to kafka-ops --help for details")
98+
os.Exit(1)
8999
}
90100
if isJSON && isYAML {
91-
panic("Please define one of the formats: --json, --yaml")
101+
fmt.Println("Please define one of the formats: --json, --yaml")
102+
os.Exit(1)
92103
}
93104
if broker == "" {
94105
broker = loadEnvVar("KAFKA_BROKER")
@@ -99,7 +110,8 @@ func init() {
99110
if specfile == "" {
100111
specfile = loadEnvVar("KAFKA_SPEC_FILE")
101112
if specfile == "" && actionApply {
102-
panic("Please define spec file with --spec option or with KAFKA_SPEC_FILE env variable")
113+
fmt.Println("Please define spec file with --spec option or with KAFKA_SPEC_FILE env variable")
114+
os.Exit(1)
103115
}
104116
}
105117
protocol = strings.ToLower(protocol)
@@ -115,6 +127,7 @@ func init() {
115127
}
116128

117129
func main() {
130+
defer handleExit()
118131
// Connect to Kafka broker
119132
brokerAddrs := strings.Split(broker, ",")
120133
config := sarama.NewConfig()
@@ -131,34 +144,60 @@ func main() {
131144
config.Net.SASL.Mechanism = sarama.SASLTypeSCRAMSHA512
132145
config.Net.SASL.SCRAMClientGeneratorFunc = func() sarama.SCRAMClient { return &client.XDGSCRAMClient{HashGeneratorFcn: client.SHA512} }
133146
} else {
134-
panic("The only supported SASL mechanisms: scram-sha-256, scram-sha-512")
147+
fmt.Println("The only supported SASL mechanisms: scram-sha-256, scram-sha-512")
148+
os.Exit(1)
135149
}
136150
}
137151

138152
admin, err := sarama.NewClusterAdmin(brokerAddrs, config)
139153
if err != nil {
140-
panic("Error while creating cluster admin: " + err.Error())
154+
fmt.Println("Error while creating cluster admin: " + err.Error())
155+
os.Exit(2)
141156
}
142157
defer func() { _ = admin.Close() }()
143-
158+
//defer fmt.Println("closed")
144159
if actionApply {
145-
applySpecFile(&admin)
160+
err = applySpecFile(&admin)
161+
if err != nil {
162+
if err.Error() != "" {
163+
fmt.Println(err.Error())
164+
}
165+
panic(Exit{2})
166+
}
146167
} else if actionDump {
147-
dumpSpec(&admin)
168+
err = dumpSpec(&admin)
169+
if err != nil {
170+
if err.Error() != "" {
171+
fmt.Println(err.Error())
172+
}
173+
panic(Exit{2})
174+
}
175+
} else if actionHelp {
176+
usage()
148177
}
149178
}
150179

151-
func dumpSpec(adminref *sarama.ClusterAdmin) {
152-
admin := *adminref
180+
func handleExit() {
181+
if e := recover(); e != nil {
182+
if exit, ok := e.(Exit); ok == true {
183+
os.Exit(exit.Code)
184+
}
185+
panic(e)
186+
}
187+
}
153188

189+
func dumpSpec(admin *sarama.ClusterAdmin) error {
154190
// Get current topics from broker
155-
currentTopics, err := admin.ListTopics()
191+
currentTopics, err := (*admin).ListTopics()
156192
if err != nil {
157-
panic(err)
193+
return err
158194
}
159195

160196
// Get current ACLs from broker
161-
currentAcls := listAllAcls(&admin)
197+
currentAcls,err := listAllAcls(admin)
198+
if err != nil {
199+
return err
200+
}
162201

163202
var spec Spec
164203

@@ -209,6 +248,7 @@ func dumpSpec(adminref *sarama.ClusterAdmin) {
209248
yamlTopic,_ := yaml.Marshal(spec)
210249
fmt.Printf(string(yamlTopic))
211250
}
251+
return nil
212252
}
213253

214254
func (s *Spec) AddAcl(acl Acl) {
@@ -244,7 +284,7 @@ func (o OperationHost) Host() string {
244284
return strings.Split(string(o), ":")[1]
245285
}
246286

247-
func listAllAcls(adminref *sarama.ClusterAdmin) []sarama.ResourceAcls {
287+
func listAllAcls(admin *sarama.ClusterAdmin) ([]sarama.ResourceAcls,error) {
248288
filter := sarama.AclFilter{
249289
Version: 1,
250290
ResourceType: sarama.AclResourceAny,
@@ -253,24 +293,25 @@ func listAllAcls(adminref *sarama.ClusterAdmin) []sarama.ResourceAcls {
253293
PermissionType: sarama.AclPermissionAny,
254294
}
255295

256-
currentAcls, err := (*adminref).ListAcls(filter)
296+
currentAcls, err := (*admin).ListAcls(filter)
257297
if err != nil {
258-
panic(err)
298+
return nil, err
259299
}
260-
return currentAcls
300+
return currentAcls, nil
261301
}
262302

263-
func applySpecFile(adminref *sarama.ClusterAdmin) {
264-
admin := *adminref
265-
303+
func applySpecFile(admin *sarama.ClusterAdmin) error {
266304
var numOk, numChanged, numError int
267305

268-
spec := parseSpecFile()
306+
spec, err := parseSpecFile()
307+
if err != nil {
308+
return errors.New("Can't parse spec manifest: " + err.Error())
309+
}
269310

270311
// Get number of brokers
271-
brokers, _, err := admin.DescribeCluster()
312+
brokers, _, err := (*admin).DescribeCluster()
272313
if err != nil {
273-
panic(err)
314+
return errors.New("Can't get number of brokers: " + err.Error())
274315
}
275316
var autoReplicationFactor int
276317
if len(brokers) > 1 {
@@ -280,9 +321,9 @@ func applySpecFile(adminref *sarama.ClusterAdmin) {
280321
}
281322

282323
// Get current topics from broker
283-
currentTopics, err := admin.ListTopics()
324+
currentTopics, err := (*admin).ListTopics()
284325
if err != nil {
285-
panic(err)
326+
return errors.New("Can't list topics: " + err.Error())
286327
}
287328

288329
// Iterate over topics
@@ -295,7 +336,7 @@ func applySpecFile(adminref *sarama.ClusterAdmin) {
295336

296337
if !found {
297338
// Topic doesn't exist - need to create one
298-
err := createTopic(topic, &admin)
339+
err := createTopic(topic, admin)
299340
if err != nil {
300341
printResult(Error, broker, err.Error(), topic)
301342
numError++
@@ -324,7 +365,7 @@ func applySpecFile(adminref *sarama.ClusterAdmin) {
324365

325366
// Check the partitions count
326367
if int32(topic.Partitions) != currentTopic.NumPartitions {
327-
err := alterNumPartitions(topic.Name, &admin, topic.Partitions)
368+
err := alterNumPartitions(topic.Name, admin, topic.Partitions)
328369
if err != nil {
329370
printResult(Error, broker, err.Error(), topic)
330371
numError++
@@ -352,7 +393,7 @@ func applySpecFile(adminref *sarama.ClusterAdmin) {
352393
}
353394
}
354395
if topicConfigAlterNeeded {
355-
topic,err = alterTopicConfig(topic, &admin, currentTopic.ConfigEntries)
396+
topic,err = alterTopicConfig(topic, admin, currentTopic.ConfigEntries)
356397
if err != nil {
357398
printResult(Error, broker, err.Error(), topic)
358399
numError++
@@ -376,7 +417,10 @@ func applySpecFile(adminref *sarama.ClusterAdmin) {
376417
}
377418

378419
// Get current ACLs from broker
379-
currentAcls := listAllAcls(&admin)
420+
currentAcls,err := listAllAcls(admin)
421+
if err != nil {
422+
return err
423+
}
380424

381425
// Iterate over ACLs
382426
breakLoop := false
@@ -391,7 +435,7 @@ func applySpecFile(adminref *sarama.ClusterAdmin) {
391435
} else {
392436
permissionType = sarama.AclPermissionDeny
393437
}
394-
result, err := createAclIfNotExists(&admin, &currentAcls, permissionType, principal, resource, OperationHost(rule))
438+
result, err := createAclIfNotExists(admin, &currentAcls, permissionType, principal, resource, OperationHost(rule))
395439
if result == Ok {
396440
printResult(Ok, broker, "", acl)
397441
numOk++
@@ -418,8 +462,9 @@ func applySpecFile(adminref *sarama.ClusterAdmin) {
418462

419463
printSummary(broker, numOk, numChanged, numError)
420464
if numError > 0 {
421-
os.Exit(2)
465+
return errors.New("")
422466
}
467+
return nil
423468
}
424469

425470
func createAclIfNotExists(admin *sarama.ClusterAdmin, acls *[]sarama.ResourceAcls, p sarama.AclPermissionType, principal string, resource Resource, oh OperationHost) (string, error){
@@ -461,33 +506,24 @@ func aclExists(a *[]sarama.ResourceAcls, p sarama.AclPermissionType, principal s
461506
return false
462507
}
463508

464-
func parseSpecFile() Spec {
509+
func parseSpecFile() (Spec,error) {
510+
var spec Spec
465511
specFile, err := ioutil.ReadFile(specfile)
466512
if err != nil {
467-
panic(err)
513+
return spec, err
468514
}
469515

470-
var spec Spec
471516
if (isYAML) {
472517
err = yaml.Unmarshal(specFile, &spec)
473-
if err != nil {
474-
panic(err)
475-
}
476518
} else if (isJSON) {
477519
err = json.Unmarshal(specFile, &spec)
478-
if err != nil {
479-
panic(err)
480-
}
481520
} else {
482521
err = yaml.Unmarshal(specFile, &spec)
483522
if err != nil {
484523
err = json.Unmarshal(specFile, &spec)
485-
if err != nil {
486-
panic(err)
487-
}
488524
}
489525
}
490-
return spec
526+
return spec, err
491527
}
492528

493529
func alterNumPartitions(topic string, clusterAdmin *sarama.ClusterAdmin, count int) error {
@@ -715,7 +751,41 @@ func aclPermissionTypeToString(permissionType sarama.AclPermissionType) string {
715751
}
716752

717753
func usage() {
718-
fmt.Fprintf(os.Stderr, "Usage: %s -b <bootstrap brokers> -s <specfile> [-v]\n", os.Args[0])
719-
flag.PrintDefaults()
754+
usage := `Manage Kafka cluster resources (topics and ACLs)
755+
Usage: %s <action> [<options>] [<broker connection options>]
756+
----------------
757+
Actions
758+
--help Show this help and exit
759+
--dump Dump cluster resources and their configs to stdout
760+
See also --json and --yaml options
761+
--apply Idempotently align cluster resources with the spec manifest
762+
See also --spec, --json and --yaml options
763+
----------------
764+
Options
765+
--spec A path to manifest (specification file) to be used
766+
with --apply action
767+
Can be also set by Env variable KAFKA_SPEC_FILE
768+
--yaml Spec-file is in YAML format
769+
Will try to detect format if none of --yaml or --json is set
770+
--json Spec-file is in JSON format
771+
Will try to detect format if none of --yaml or --json is set
772+
--verbose Verbose output
773+
--stop-on-error Exit on first occurred error
774+
----------------
775+
Broker connection options
776+
--broker Bootstrap-brokers, comma-separated. Default is localhost:9092
777+
Can be also set by Env variable KAFKA_BROKER
778+
--protocol Security protocol. Default is plaintext
779+
Available options: plaintext, sasl_ssl, sasl_plaintext
780+
--mechanism SASL mechanism. Default is scram-sha-256
781+
Available options: scram-sha-256, scram-sha-512
782+
--username Username for authentication
783+
Can be also set by Env variable KAFKA_USERNAME
784+
--password Password for authentication
785+
Can be also set by Env variable KAFKA_PASSWORD
786+
`
787+
788+
fmt.Fprintf(os.Stderr, usage, os.Args[0])
789+
//flag.PrintDefaults()
720790
os.Exit(1)
721791
}

0 commit comments

Comments
 (0)