func (this *Mirror) Main() (exitCode int) {
	this.quit = make(chan struct{})
	signal.RegisterHandler(func(sig os.Signal) {
		log.Info("received signal: %s", strings.ToUpper(sig.String()))
		log.Info("quiting...")

		this.once.Do(func() {
			close(this.quit)
		})
	}, syscall.SIGINT, syscall.SIGTERM)

	limit := (1 << 20) * this.BandwidthLimit / 8
	if this.BandwidthLimit > 0 {
		this.bandwidthRateLimiter = ratelimiter.NewLeakyBucket(limit*10, time.Second*10)
	}

	log.Info("starting [email protected]%s", gafka.BuildId)

	// pprof
	debugAddr := ":10009"
	go http.ListenAndServe(debugAddr, nil)
	log.Info("pprof ready on %s", debugAddr)

	z1 := zk.NewZkZone(zk.DefaultConfig(this.Z1, ctx.ZoneZkAddrs(this.Z1)))
	z2 := zk.NewZkZone(zk.DefaultConfig(this.Z2, ctx.ZoneZkAddrs(this.Z2)))
	c1 := z1.NewCluster(this.C1)
	c2 := z2.NewCluster(this.C2)

	this.runMirror(c1, c2, limit)

	log.Info("bye [email protected]%s, %s", gafka.BuildId, time.Since(this.startedAt))
	log.Close()

	return
}

func (this *Mirror) Run(args []string) (exitCode int) {
	cmdFlags := flag.NewFlagSet("mirror", flag.ContinueOnError)
	cmdFlags.Usage = func() { this.Ui.Output(this.Help()) }
	cmdFlags.StringVar(&this.zone1, "z1", "", "")
	cmdFlags.StringVar(&this.zone2, "z2", "", "")
	cmdFlags.StringVar(&this.cluster1, "c1", "", "")
	cmdFlags.StringVar(&this.cluster2, "c2", "", "")
	cmdFlags.StringVar(&this.excludes, "excluded", "", "")
	cmdFlags.BoolVar(&this.debug, "debug", false, "")
	cmdFlags.StringVar(&this.compress, "compress", "", "")
	cmdFlags.Int64Var(&this.bandwidthLimit, "net", 100, "")
	cmdFlags.BoolVar(&this.autoCommit, "commit", true, "")
	cmdFlags.Int64Var(&this.progressStep, "step", 5000, "")
	if err := cmdFlags.Parse(args); err != nil {
		return 1
	}

	if validateArgs(this, this.Ui).
		require("-z1", "-z2", "-c1", "-c2").
		invalid(args) {
		return 2
	}

	this.topicsExcluded = make(map[string]struct{})
	for _, e := range strings.Split(this.excludes, ",") {
		this.topicsExcluded[e] = struct{}{}
	}

	log.SetOutput(os.Stdout)
	this.quit = make(chan struct{})
	limit := (1 << 20) * this.bandwidthLimit / 8
	this.bandwidthRateLimiter = ratelimiter.NewLeakyBucket(limit, time.Second)
	log.Printf("[%s]%s -> [%s]%s with bandwidth %sbps",
		this.zone1, this.cluster1,
		this.zone2, this.cluster2,
		gofmt.Comma(int64(limit*8)))
	signal.RegisterSignalsHandler(func(sig os.Signal) {
		log.Printf("received signal: %s", strings.ToUpper(sig.String()))
		log.Println("quiting...")

		this.once.Do(func() {
			close(this.quit)
		})
	}, syscall.SIGINT, syscall.SIGTERM)

	z1 := zk.NewZkZone(zk.DefaultConfig(this.zone1, ctx.ZoneZkAddrs(this.zone1)))
	z2 := zk.NewZkZone(zk.DefaultConfig(this.zone2, ctx.ZoneZkAddrs(this.zone2)))
	c1 := z1.NewCluster(this.cluster1)
	c2 := z2.NewCluster(this.cluster2)
	this.makeMirror(c1, c2)

	return
}

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 *Partition) Run(args []string) (exitCode int) {
	var (
		zone       string
		topic      string
		cluster    string
		partitions int
	)
	cmdFlags := flag.NewFlagSet("partition", flag.ContinueOnError)
	cmdFlags.Usage = func() { this.Ui.Output(this.Help()) }
	cmdFlags.StringVar(&zone, "z", "", "")
	cmdFlags.StringVar(&cluster, "c", "", "")
	cmdFlags.StringVar(&topic, "t", "", "")
	cmdFlags.IntVar(&partitions, "n", 1, "")
	if err := cmdFlags.Parse(args); err != nil {
		return 1
	}

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

	zkzone := zk.NewZkZone(zk.DefaultConfig(zone, ctx.ZoneZkAddrs(zone)))
	zkcluster := zkzone.NewCluster(cluster)
	this.addPartition(zkcluster.ZkConnectAddr(), topic, partitions)
	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 *Job) Run(args []string) (exitCode int) {
	var (
		zone    string
		appid   string
		initJob string
	)
	cmdFlags := flag.NewFlagSet("job", flag.ContinueOnError)
	cmdFlags.Usage = func() { this.Ui.Output(this.Help()) }
	cmdFlags.StringVar(&zone, "z", ctx.DefaultZone(), "")
	cmdFlags.StringVar(&appid, "app", "", "")
	cmdFlags.IntVar(&this.due, "d", 0, "")
	cmdFlags.StringVar(&initJob, "init", "", "")
	if err := cmdFlags.Parse(args); err != nil {
		return 1
	}

	this.zkzone = zk.NewZkZone(zk.DefaultConfig(zone, ctx.ZoneZkAddrs(zone)))
	if initJob != "" {
		this.initializeJob(initJob)
		return
	}

	if appid != "" {
		this.displayAppJobs(appid)
		return
	}

	this.printResourcesAndActors()

	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 *Controllers) Run(args []string) (exitCode int) {
	var (
		zone string
	)
	cmdFlags := flag.NewFlagSet("controllers", flag.ContinueOnError)
	cmdFlags.Usage = func() { this.Ui.Output(this.Help()) }
	cmdFlags.StringVar(&zone, "z", "", "")
	cmdFlags.StringVar(&this.cluster, "c", "", "") // TODO not used
	if err := cmdFlags.Parse(args); err != nil {
		return 1
	}

	if zone == "" {
		forSortedZones(func(zkzone *zk.ZkZone) {
			this.printControllers(zkzone)
			printSwallowedErrors(this.Ui, zkzone)
		})

		return
	}

	zkzone := zk.NewZkZone(zk.DefaultConfig(zone, ctx.ZoneZkAddrs(zone)))
	this.printControllers(zkzone)
	printSwallowedErrors(this.Ui, zkzone)

	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 *Rebalance) Run(args []string) (exitCode int) {
	cmdFlags := flag.NewFlagSet("rebalance", flag.ContinueOnError)
	cmdFlags.Usage = func() { this.Ui.Output(this.Help()) }
	cmdFlags.StringVar(&this.zone, "z", "", "")
	cmdFlags.StringVar(&this.cluster, "c", "", "")
	cmdFlags.StringVar(&this.topic, "t", "", "")
	cmdFlags.StringVar(&this.partition, "p", "", "comma seperated ids")
	if err := cmdFlags.Parse(args); err != nil {
		return 1
	}

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

	zkzone := zk.NewZkZone(zk.DefaultConfig(this.zone, ctx.ZoneZkAddrs(this.zone)))
	this.zkcluster = zkzone.NewCluster(this.cluster)

	data := this.generateReassignFile()
	this.Ui.Output(data)
	yes, _ := this.Ui.Ask("Are you sure to execute? [Y/N]")
	if yes == "Y" {
		this.executeReassignment()
	} else {
		this.Ui.Output("bye")
	}

	return
}

// 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 *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 New(zone string, id string, data []byte) *zkreg {
	this := &zkreg{
		zkzone:     zk.NewZkZone(zk.DefaultConfig(zone, ctx.ZoneZkAddrs(zone))),
		id:         id,
		data:       data,
		shutdownCh: make(chan struct{}),
	}

	return this
}

func (this *Migrate) Run(args []string) (exitCode int) {
	cmdFlags := flag.NewFlagSet("migrate", flag.ContinueOnError)
	cmdFlags.Usage = func() { this.Ui.Output(this.Help()) }
	cmdFlags.StringVar(&this.zone, "z", "", "")
	cmdFlags.StringVar(&this.cluster, "c", "", "")
	cmdFlags.StringVar(&this.topic, "t", "", "")
	cmdFlags.StringVar(&this.partition, "p", "", "")
	cmdFlags.StringVar(&this.brokerId, "brokers", "", "")
	cmdFlags.BoolVar(&this.verifyMode, "verify", false, "")
	if err := cmdFlags.Parse(args); err != nil {
		return 1
	}

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

	zkzone := zk.NewZkZone(zk.DefaultConfig(this.zone, ctx.ZoneZkAddrs(this.zone)))
	this.zkcluster = zkzone.NewCluster(this.cluster)

	if this.verifyMode {
		this.Ui.Info(fmt.Sprintf("You MUST manually remove the %s after migration is done.", reassignNodeFilename))
		this.Ui.Info(fmt.Sprintf("After verify ok, modify producer/consumer to point to new brokers!"))

		for {
			this.Ui.Output(fmt.Sprintf("%s", time.Now().String()))
			this.verify()

			time.Sleep(time.Second * 5)
		}

		return
	}

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

	//this.ensureBrokersAreAlive()
	data := this.generateReassignFile()
	this.Ui.Output(data)
	yes, _ := this.Ui.Ask("Are you sure to execute the migration? [Y/N]")
	if yes == "Y" {
		this.executeReassignment()
	} else {
		this.Ui.Output("bye")
	}

	return
}

func forAllSortedZones(fn func(zkzone *zk.ZkZone)) {
	for _, zone := range ctx.SortedZones() {
		zkAddrs := ctx.ZoneZkAddrs(zone)
		if strings.TrimSpace(zkAddrs) == "" {
			continue
		}

		zkzone := zk.NewZkZone(zk.DefaultConfig(zone, zkAddrs))
		fn(zkzone)
	}
}

func New(zone, cluster string) *KafkaFS {
	log.Info("starting kfs zone:%s, cluster:%s", zone, cluster)

	this := &KafkaFS{}

	ctx.LoadFromHome()
	zkzone := zk.NewZkZone(zk.DefaultConfig(zone, ctx.ZoneZkAddrs(zone)))
	this.zkcluster = zkzone.NewCluster(cluster) // panic if invalid cluster

	this.root = this.newDir(os.FileMode(0777))
	return this
}

func (this *Start) main() {
	ctx.LoadFromHome()
	this.zkzone = zk.NewZkZone(zk.DefaultConfig(this.zone, ctx.ZoneZkAddrs(this.zone)))
	zkConnEvt, ok := this.zkzone.SessionEvents()
	if !ok {
		panic("someone stealing my events")
	}

	registry.Default = zkr.New(this.zkzone)

	log.Info("ehaproxy[%s] starting...", gafka.BuildId)
	go this.runMonitorServer(this.httpAddr)

	zkConnected := false
	for {
		instances, instancesChange, err := registry.Default.WatchInstances()
		if err != nil {
			log.Error("zone[%s] %s", this.zkzone.Name(), err)
			time.Sleep(time.Second)
			continue
		}

		if zkConnected {
			if len(instances) > 0 {
				this.reload(instances)
			} else {
				// resilience to zk problem by local cache
				log.Warn("backend all shutdown? skip this change")
				time.Sleep(time.Second)
				continue
			}
		}

		select {
		case <-this.quitCh:
			return

		case evt := <-zkConnEvt:
			if evt.State == zklib.StateHasSession && !zkConnected {
				log.Info("zk connected")
				zkConnected = true
			} else if zkConnected {
				log.Warn("zk jitter: %+v", evt)
			}

		case <-instancesChange:
			log.Info("instances changed!!")
		}
	}

}

func TestAll(t *testing.T) {
	zkzone := zk.NewZkZone(zk.DefaultConfig(ctx.DefaultZone(), ctx.ZoneZkAddrs(ctx.DefaultZone())))
	defer zkzone.Close()
	z := New(DefaultConfig(), zkzone)
	z.Start()

	assert.Equal(t, "/kafka_pubsub", z.ZkChroot("me"))
	assert.Equal(t, []string{"localhost:2181"}, z.ZkAddrs())

	t.Logf("%+v", z.BrokerList("me"))
	t.Logf("%+v", z.TopicPartitions("me", "app1.foobar.v1"))
	t.Logf("%+v", z.ZkCluster("me"))
	n, err := z.OnlineConsumersCount("me", "app1.foobar.v1", "group")
	t.Logf("%d %+v", n, err)
}

func TestZkPath(t *testing.T) {
	zkzone := zk.NewZkZone(zk.DefaultConfig(ctx.DefaultZone(), ctx.ZoneZkAddrs(ctx.DefaultZone())))
	defer zkzone.Close()
	zk := New(zkzone)
	id := "1"
	assert.Equal(t, "/_kateway/ids/local/1", zk.mypath(id))

	data := []byte("foo, bar")
	err := zk.Register(id, data)
	assert.Equal(t, nil, err)
	defer zk.Deregister(id, data)

	ok, err := zk.Registered(id)
	assert.Equal(t, true, ok)
	assert.Equal(t, nil, err)
}

func forSortedZones(fn func(zkzone *zk.ZkZone)) {
	for _, zone := range ctx.SortedZones() {
		if strings.HasPrefix(zone, "z_") {
			// zk only
			continue
		}

		zkAddrs := ctx.ZoneZkAddrs(zone)
		if strings.TrimSpace(zkAddrs) == "" {
			continue
		}

		zkzone := zk.NewZkZone(zk.DefaultConfig(zone, zkAddrs))
		fn(zkzone)
	}
}