func (this *Start) Help() string {
	help := fmt.Sprintf(`
Usage: %s start [options]

    Start %s system on localhost

Options:

    -z zone
      Default %s

    -forwardfor
      Default false.
      If true, haproxy will add X-Forwarded-For http header.

    -pub pub server listen port

    -sub sub server listen port

    -man manager server listen port

    -p directory prefix
      Default %s

    -log log file
      Default %s

`, this.Cmd, this.Cmd, ctx.ZkDefaultZone(), defaultPrefix, defaultLogfile)
	return strings.TrimSpace(help)
}

func (this *TopBroker) Help() string {
	help := fmt.Sprintf(`
Usage: %s topbroker [options]

    Unix “top” like utility for kafka brokers

Options:

    -z zone
      Default %s

    -c cluster pattern

    -t topic pattern  

    -i interval
      Refresh interval in seconds.
      e,g. 5s    

    -shortip
	  Print ending portion of broker ip address.

`, this.Cmd, ctx.ZkDefaultZone())
	return strings.TrimSpace(help)
}

func (this *Stat) Run(args []string) (exitCode int) {
	cmdFlags := flag.NewFlagSet("stat", flag.ContinueOnError)
	cmdFlags.Usage = func() { this.Ui.Output(this.Help()) }
	cmdFlags.StringVar(&this.zone, "z", ctx.ZkDefaultZone(), "")
	if err := cmdFlags.Parse(args); err != nil {
		return 1
	}

	if this.zone == "" {
		this.Ui.Error("unknown zone")
		return 2
	}

	if len(args) == 0 {
		this.Ui.Error("missing path")
		return 2
	}

	this.path = args[len(args)-1]

	zkzone := gzk.NewZkZone(gzk.DefaultConfig(this.zone, ctx.ZoneZkAddrs(this.zone)))
	defer zkzone.Close()
	conn := zkzone.Conn()

	_, stat, err := conn.Get(this.path)
	must(err)
	this.Ui.Output(fmt.Sprintf("%# v", pretty.Formatter(*stat)))
	ctime := gzk.ZkTimestamp(stat.Ctime).Time()
	mtime := gzk.ZkTimestamp(stat.Mtime).Time()
	this.Ui.Output(fmt.Sprintf("ctime: %s, mtime: %s",
		gofmt.PrettySince(ctime), gofmt.PrettySince(mtime)))
	return
}

func (this *Topology) Run(args []string) (exitCode int) {
	cmdFlags := flag.NewFlagSet("topology", flag.ContinueOnError)
	cmdFlags.Usage = func() { this.Ui.Output(this.Help()) }
	cmdFlags.StringVar(&this.zone, "z", ctx.ZkDefaultZone(), "")
	cmdFlags.StringVar(&this.cluster, "c", "", "")
	cmdFlags.StringVar(&this.hostPattern, "host", "", "")
	cmdFlags.BoolVar(&this.verbose, "l", false, "")
	cmdFlags.BoolVar(&this.watchMode, "w", false, "")
	cmdFlags.BoolVar(&this.maxPort, "maxport", false, "")
	if err := cmdFlags.Parse(args); err != nil {
		return 1
	}

	// a single zone
	ensureZoneValid(this.zone)
	zkzone := zk.NewZkZone(zk.DefaultConfig(this.zone, ctx.ZoneZkAddrs(this.zone)))
	for {
		if this.maxPort {
			this.displayZoneMaxPort(zkzone)
		} else {
			this.displayZoneTopology(zkzone)
		}

		if !this.watchMode {
			return
		}
		time.Sleep(time.Second * 5)
		refreshScreen()
	}

	return
}

func (this *Kguard) Run(args []string) (exitCode int) {
	cmdFlags := flag.NewFlagSet("kguard", flag.ContinueOnError)
	cmdFlags.Usage = func() { this.Ui.Output(this.Help()) }
	cmdFlags.StringVar(&this.zone, "z", ctx.ZkDefaultZone(), "")
	cmdFlags.BoolVar(&this.longFmt, "l", false, "")
	if err := cmdFlags.Parse(args); err != nil {
		return 2
	}

	zkzone := zk.NewZkZone(zk.DefaultConfig(this.zone, ctx.ZoneZkAddrs(this.zone)))
	kguards, err := zkzone.KguardInfos()
	if err != nil {
		this.Ui.Error(fmt.Sprintf("%s %v", zk.KguardLeaderPath, err.Error()))
		return
	}

	leader := kguards[0]
	this.Ui.Output(fmt.Sprintf("%s(out of %d candidates) up: %s",
		color.Green(leader.Host), leader.Candidates,
		gofmt.PrettySince(leader.Ctime)))

	if this.longFmt {
		this.showKguardVersion(leader.Host)
		this.showStats(leader.Host)
	}

	return
}

func (this *Zones) Run(args []string) (exitCode int) {
	// header
	this.Ui.Output(fmt.Sprintf("%10s %-70s", "zone", "zookeeper ensemble"))
	this.Ui.Output(fmt.Sprintf("%s %s",
		strings.Repeat("-", 10),
		strings.Repeat("-", 70)))

	if len(args) > 0 {
		// user specified the zones to print
		for _, zone := range args {
			if zk, present := ctx.Zones()[zone]; present {
				this.Ui.Output(fmt.Sprintf("%10s %s", zone, zk))
			} else {
				this.Ui.Output(fmt.Sprintf("%10s not defined", zone))
			}
		}

		return
	}

	// print all by default
	defaultZone := ctx.ZkDefaultZone()
	for _, zone := range ctx.SortedZones() {
		if defaultZone == zone {
			this.Ui.Output(fmt.Sprintf("%10s %s", zone+"*", ctx.ZoneZkAddrs(zone)))
			continue
		}

		this.Ui.Output(fmt.Sprintf("%10s %s", zone, ctx.ZoneZkAddrs(zone)))
	}

	return
}

func (this *Mount) Help() string {
	help := fmt.Sprintf(`
Usage: %s mount [options] mount_point

    %s

Pre-requisites:
    yum install -y fuse
    modprobe fuse
    lsmod | grep fuse

Options:

    -z zone
      Default '%s'.

    -c cluster

    -l log level
      Default 'info'.
      

`, this.Cmd, this.Synopsis(), ctx.ZkDefaultZone())
	return strings.TrimSpace(help)
}

func (this *Lags) Help() string {
	help := fmt.Sprintf(`
Usage: %s lags [options]

    Display online high level consumers lag on a topic

Options:

    -z zone
      Default %s

    -c cluster

    -g group name pattern

    -t topic name pattern

    -w
      Watch mode. Keep printing out lags with 60s interval refresh.

    -l
      Only show online consumers lag.

    -lag threshold
      Default 5000.

    -p
      Only show problematic consumers.

    -table
      Display in table format.

`, this.Cmd, ctx.ZkDefaultZone())
	return strings.TrimSpace(help)
}

// TODO run 3 nodes in a zone to monitor as daemon
// register the 3 nodes as host service tag.
func (this *Ping) Run(args []string) (exitCode int) {
	cmdFlags := flag.NewFlagSet("ping", flag.ContinueOnError)
	cmdFlags.Usage = func() { this.Ui.Output(this.Help()) }
	cmdFlags.StringVar(&this.zone, "z", ctx.ZkDefaultZone(), "")
	cmdFlags.DurationVar(&this.interval, "interval", time.Minute*5, "")
	cmdFlags.StringVar(&this.logfile, "logfile", "stdout", "")
	cmdFlags.BoolVar(&this.problematicMode, "p", false, "")
	if err := cmdFlags.Parse(args); err != nil {
		return 1
	}

	this.setupLog()
	this.zkzone = zk.NewZkZone(zk.DefaultConfig(this.zone, ctx.ZoneZkAddrs(this.zone)))

	for {
		this.diagnose()
		if this.logfile == "stdout" {
			break
		}

		time.Sleep(this.interval)
	}

	return
}

func (this *Create) Run(args []string) (exitCode int) {
	cmdFlags := flag.NewFlagSet("create", flag.ContinueOnError)
	cmdFlags.Usage = func() { this.Ui.Output(this.Help()) }
	cmdFlags.StringVar(&this.zone, "z", ctx.ZkDefaultZone(), "")
	cmdFlags.StringVar(&this.path, "p", "", "")
	if err := cmdFlags.Parse(args); err != nil {
		return 1
	}

	if validateArgs(this, this.Ui).
		require("-p").
		requireAdminRights("-p").
		invalid(args) {
		return 2
	}

	if this.zone == "" {
		this.Ui.Error("unknown zone")
		return 2
	}

	zkzone := gzk.NewZkZone(gzk.DefaultConfig(this.zone, ctx.ZoneZkAddrs(this.zone)))
	defer zkzone.Close()
	conn := zkzone.Conn()

	data := inData()
	flags := int32(0)
	acl := zk.WorldACL(zk.PermAll)
	_, err := conn.Create(this.path, data, flags, acl)
	must(err)

	return
}

func (this *Consumers) Run(args []string) (exitCode int) {
	var (
		cluster string
		zone    string
	)
	cmdFlags := flag.NewFlagSet("consumers", flag.ContinueOnError)
	cmdFlags.Usage = func() { this.Ui.Output(this.Help()) }
	cmdFlags.StringVar(&zone, "z", ctx.ZkDefaultZone(), "")
	cmdFlags.StringVar(&cluster, "c", "", "")
	cmdFlags.StringVar(&this.groupPattern, "g", "", "")
	cmdFlags.BoolVar(&this.onlineOnly, "online", false, "")
	cmdFlags.BoolVar(&this.byHost, "byhost", false, "")
	cmdFlags.StringVar(&this.topicPattern, "t", "", "")
	cmdFlags.BoolVar(&this.warnOnly, "warn", false, "")
	cmdFlags.BoolVar(&this.ownerOnly, "own", false, "")
	cmdFlags.BoolVar(&this.cleanup, "cleanup", false, "")
	cmdFlags.BoolVar(&this.confirmYes, "yes", false, "")
	if err := cmdFlags.Parse(args); err != nil {
		return 1
	}

	if validateArgs(this, this.Ui).
		requireAdminRights("-cleanup").
		invalid(args) {
		return 2
	}

	if this.ownerOnly {
		this.onlineOnly = true
	}

	if zone == "" {
		forSortedZones(func(zkzone *zk.ZkZone) {
			switch {
			case this.cleanup:
				this.cleanupStaleConsumerGroups(zkzone, cluster)
			case this.byHost:
				this.printConsumersByHost(zkzone, cluster)
			default:
				this.printConsumersByGroupTable(zkzone, cluster)
			}
		})

		return
	}

	zkzone := zk.NewZkZone(zk.DefaultConfig(zone, ctx.ZoneZkAddrs(zone)))
	switch {
	case this.cleanup:
		this.cleanupStaleConsumerGroups(zkzone, cluster)
	case this.byHost:
		this.printConsumersByHost(zkzone, cluster)
	default:
		this.printConsumersByGroupTable(zkzone, cluster)
	}

	return
}

func (this *Peek) Help() string {
	help := fmt.Sprintf(`
Usage: %s peek [options]

    %s

Options:

    -z zone
      Default %s

    -c cluster

    -t topic pattern
   
    -p partition id
      -1 will peek all partitions of a topic

    -beep
      Make a beep sound for each message

    -pretty
      Pretty print the json message body

    -col json column name
      Will json decode message and extract specified column value only

    -last n
      Peek the most recent N messages

    -offset message offset value
      -1 OffsetNewest, -2 OffsetOldest. 
      You can specify your own offset.
      Default -1(OffsetNewest)

    -n count
      Limit how many messages to consume

    -d duration
      Limit how long to keep peeking
      e,g. -d 5m

    -body
      Only display message body

    -s
      Silence mode, only display statastics instead of message content

    -color
      Enable colorized output
`, this.Cmd, this.Synopsis(), ctx.ZkDefaultZone())
	return strings.TrimSpace(help)
}

func (this *Zones) Run(args []string) (exitCode int) {
	cmdFlags := flag.NewFlagSet("zones", flag.ContinueOnError)
	cmdFlags.Usage = func() { this.Ui.Output(this.Help()) }
	cmdFlags.BoolVar(&this.ipInNumber, "n", false, "")
	cmdFlags.BoolVar(&this.plain, "plain", false, "")
	if err := cmdFlags.Parse(args); err != nil {
		return 2
	}

	// print all by default
	zones := make([][]string, 0)
	defaultZone := ctx.ZkDefaultZone()
	for _, zone := range ctx.SortedZones() {
		if defaultZone == zone {
			if this.ipInNumber {
				zones = append(zones, []string{zone + "*", ctx.ZoneZkAddrs(zone)})
			} else {
				zones = append(zones, []string{zone + "*", ctx.NamedZoneZkAddrs(zone)})
			}

			continue
		}

		if this.ipInNumber {
			zones = append(zones, []string{zone, ctx.ZoneZkAddrs(zone)})
		} else {
			zones = append(zones, []string{zone, ctx.NamedZoneZkAddrs(zone)})
		}

	}

	if this.plain {
		for _, z := range zones {
			this.Ui.Output(fmt.Sprintf("%s:", z[0]))
			this.Ui.Output(fmt.Sprintf("%s\n", z[1]))
		}
		return
	}

	table := tablewriter.NewWriter(os.Stdout)
	for _, z := range zones {
		table.Append(z)
	}
	table.SetHeader([]string{"Zone", "ZK ensemble"})
	table.SetFooter([]string{"Total", fmt.Sprintf("%d", len(zones))})
	table.Render() // Send output

	return

}

func (this *Offset) Run(args []string) (exitCode int) {
	var (
		zone      string
		cluster   string
		topic     string
		group     string
		partition string
		offset    int64
	)
	cmdFlags := flag.NewFlagSet("offset", flag.ContinueOnError)
	cmdFlags.Usage = func() { this.Ui.Output(this.Help()) }
	cmdFlags.StringVar(&zone, "z", ctx.ZkDefaultZone(), "")
	cmdFlags.StringVar(&cluster, "c", "", "")
	cmdFlags.StringVar(&topic, "t", "", "")
	cmdFlags.StringVar(&group, "g", "", "")
	cmdFlags.Int64Var(&offset, "offset", -1, "")
	cmdFlags.StringVar(&partition, "p", "", "")
	if err := cmdFlags.Parse(args); err != nil {
		return 1
	}

	if validateArgs(this, this.Ui).
		require("-z", "-c", "-t", "-g", "-p", "-offset").
		requireAdminRights("-z").
		invalid(args) {
		return 2
	}

	if offset < 0 {
		this.Ui.Error("offset must be positive")
		return
	}

	if p, err := strconv.Atoi(partition); err != nil {
		this.Ui.Error("invalid partition")
		return
	} else if p < 0 || p > 100 {
		this.Ui.Error("invalid partition")
		return
	}

	zkzone := zk.NewZkZone(zk.DefaultConfig(zone, ctx.ZoneZkAddrs(zone)))
	zkcluster := zkzone.NewCluster(cluster)
	zkcluster.ResetConsumerGroupOffset(topic, group, partition, offset)
	this.Ui.Output("done")
	return
}

func (this *Rm) Run(args []string) (exitCode int) {
	cmdFlags := flag.NewFlagSet("rm", flag.ContinueOnError)
	cmdFlags.Usage = func() { this.Ui.Output(this.Help()) }
	cmdFlags.StringVar(&this.zone, "z", ctx.ZkDefaultZone(), "")
	cmdFlags.StringVar(&this.path, "p", "", "")
	cmdFlags.BoolVar(&this.recursive, "R", false, "")
	if err := cmdFlags.Parse(args); err != nil {
		return 1
	}

	if validateArgs(this, this.Ui).
		require("-p").
		requireAdminRights("-p").
		invalid(args) {
		return 2
	}

	if this.zone == "" {
		this.Ui.Error("unknown zone")
		return 2
	}

	zkzone := gzk.NewZkZone(gzk.DefaultConfig(this.zone, ctx.ZoneZkAddrs(this.zone)))
	defer zkzone.Close()
	if this.recursive {
		err := zkzone.DeleteRecursive(this.path)
		if err != nil {
			this.Ui.Error(err.Error())
		} else {
			this.Ui.Info(fmt.Sprintf("%s deleted ok", this.path))
		}

		return 0
	}

	// remove a single znode
	conn := zkzone.Conn()
	err := conn.Delete(this.path, -1)
	if err != nil {
		this.Ui.Error(err.Error())
	} else {
		this.Ui.Info(fmt.Sprintf("%s deleted ok", this.path))
	}

	return
}

func (this *LsZk) Help() string {
	help := fmt.Sprintf(`
Usage: %s lszk [options] <path>

    List kafka related zookeepeer znode children

Options:

    -z zone
      Default %s

    -c cluster

    -R
      recursive.    

`, this.Cmd, ctx.ZkDefaultZone())
	return strings.TrimSpace(help)
}

func (this *Ls) Run(args []string) (exitCode int) {
	cmdFlags := flag.NewFlagSet("ls", flag.ContinueOnError)
	cmdFlags.Usage = func() { this.Ui.Output(this.Help()) }
	cmdFlags.StringVar(&this.zone, "z", ctx.ZkDefaultZone(), "")
	cmdFlags.BoolVar(&this.recursive, "R", false, "")
	cmdFlags.StringVar(&this.likePattern, "like", "", "")
	if err := cmdFlags.Parse(args); err != nil {
		return 1
	}

	if len(args) == 0 {
		this.Ui.Error("missing path")
		return 2
	}

	this.path = args[len(args)-1]

	if this.zone == "" {
		this.Ui.Error("unknown zone")
		return 2
	}

	zkzone := gzk.NewZkZone(gzk.DefaultConfig(this.zone, ctx.ZoneZkAddrs(this.zone)))
	defer zkzone.Close()
	conn := zkzone.Conn()

	if this.recursive {
		this.showChildrenRecursively(conn, this.path)
		return
	}

	children, _, err := conn.Children(this.path)
	must(err)
	sort.Strings(children)
	if this.path == "/" {
		this.path = ""
	}
	for _, child := range children {
		this.Ui.Output(this.path + "/" + child)
	}

	return
}

func (this *LsZk) Help() string {
	help := fmt.Sprintf(`
Usage: %s lszk [options] <path>

    %s

Options:

    -z zone
      Default %s

    -c cluster

    -R
      recursive.    

`, this.Cmd, this.Synopsis(), ctx.ZkDefaultZone())
	return strings.TrimSpace(help)
}

func (this *Brokers) Run(args []string) (exitCode int) {
	var (
		zone  string
		debug bool
	)
	cmdFlags := flag.NewFlagSet("brokers", flag.ContinueOnError)
	cmdFlags.Usage = func() { this.Ui.Output(this.Help()) }
	cmdFlags.StringVar(&zone, "z", ctx.ZkDefaultZone(), "")
	cmdFlags.StringVar(&this.cluster, "c", "", "")
	cmdFlags.BoolVar(&debug, "debug", false, "")
	cmdFlags.BoolVar(&this.ipInNumber, "n", false, "")
	cmdFlags.BoolVar(&this.staleOnly, "stale", false, "")
	cmdFlags.BoolVar(&this.showVersions, "versions", false, "")
	if err := cmdFlags.Parse(args); err != nil {
		return 1
	}

	if this.showVersions {
		this.doShowVersions()
		return
	}

	if debug {
		sarama.Logger = log.New(os.Stderr, color.Magenta("[sarama]"), log.LstdFlags)
	}

	if zone != "" {
		ensureZoneValid(zone)

		zkzone := zk.NewZkZone(zk.DefaultConfig(zone, ctx.ZoneZkAddrs(zone)))
		this.displayZoneBrokers(zkzone)

		return
	}

	// print all brokers on all zones by default
	forSortedZones(func(zkzone *zk.ZkZone) {
		this.displayZoneBrokers(zkzone)
	})

	return
}

func (this *Top) Help() string {
	help := fmt.Sprintf(`
Usage: %s top [options]

    %s

Options:

    -z zone
      Default %s

    -c cluster pattern

    -t topic pattern    

    -i interval
      Refresh interval in seconds.
      e,g. 5s

    -n limit

    -l
      Display long format. Print broker host.

    -shortip
      Used with -l, broker host 10.20.30.40 will display as 30.40

    -d
      Draw dashboard in graph.

    -for duration
      Run top for how long.

    -b 
      Batch mode operation. 
      Could be useful for sending output from top to other programs or to a file.

    -who <%s%s|%s%s>
`, this.Cmd, this.Synopsis(), ctx.ZkDefaultZone(), color.Colorize([]string{color.Underscore}, "p"), "roducer",
		color.Colorize([]string{color.Underscore}, "c"), "onsumer")
	return strings.TrimSpace(help)
}