<commit_before>package main

import (
	"sync"
	"time"

	"github.com/Shopify/sarama"
	"github.com/funkygao/gafka/zk"
	"github.com/funkygao/go-metrics"
	log "github.com/funkygao/log4go"
)

// MonitorTopics montor total msg count over time.
type MonitorTopics struct {
	zkzone *zk.ZkZone
	stop   chan struct{}
	tick   time.Duration
	wg     *sync.WaitGroup
}

func (this *MonitorTopics) Run() {
	defer this.wg.Done()

	ticker := time.NewTicker(this.tick)
	defer ticker.Stop()

	pubQps := metrics.NewRegisteredMeter("pub.qps", nil)
	offsets := metrics.NewRegisteredGauge("msg.cum", nil)
	topics := metrics.NewRegisteredGauge("topics", nil)
	partitions := metrics.NewRegisteredGauge("partitions", nil)
	brokers := metrics.NewRegisteredGauge("brokers", nil)
	var lastTotalOffsets int64
	for {

		select {
		case <-this.stop:
			return

		case <-ticker.C:
			o, t, p, b := this.report()
			offsets.Update(o)
			topics.Update(t)
			partitions.Update(p)
			brokers.Update(b)

			if lastTotalOffsets > 0 {
				pubQps.Mark(o - lastTotalOffsets)
			}
			lastTotalOffsets = o
		}
	}

}

func (this *MonitorTopics) report() (totalOffsets int64, topicsN int64,
	partitionN int64, brokersN int64) {
	this.zkzone.ForSortedClusters(func(zkcluster *zk.ZkCluster) {
		brokerList := zkcluster.BrokerList()
		kfk, err := sarama.NewClient(brokerList, sarama.NewConfig())
		if err != nil {
			log.Error("cluster[%s] %v", zkcluster.Name(), err)
			return
		}
		defer kfk.Close()

		brokersN += int64(len(brokerList))

		topics, err := kfk.Topics()
		if err != nil {<commit_msg>wonder: why sometimes total offset goes backwards<commit_after>package main

import (
	"sync"
	"time"

	"github.com/Shopify/sarama"
	"github.com/funkygao/gafka/zk"
	"github.com/funkygao/go-metrics"
	log "github.com/funkygao/log4go"
)

// MonitorTopics montor total msg count over time.
type MonitorTopics struct {
	zkzone *zk.ZkZone
	stop   chan struct{}
	tick   time.Duration
	wg     *sync.WaitGroup
}

func (this *MonitorTopics) Run() {
	defer this.wg.Done()

	ticker := time.NewTicker(this.tick)
	defer ticker.Stop()

	pubQps := metrics.NewRegisteredMeter("pub.qps", nil)
	offsets := metrics.NewRegisteredGauge("msg.cum", nil)
	topics := metrics.NewRegisteredGauge("topics", nil)
	partitions := metrics.NewRegisteredGauge("partitions", nil)
	brokers := metrics.NewRegisteredGauge("brokers", nil)
	var lastTotalOffsets int64
	for {

		select {
		case <-this.stop:
			return

		case <-ticker.C:
			o, t, p, b := this.report()
			offsets.Update(o)
			topics.Update(t)
			partitions.Update(p)
			brokers.Update(b)

			if lastTotalOffsets > 0 {
				if o-lastTotalOffsets > 0 {
					pubQps.Mark(o - lastTotalOffsets)
				} else {
					log.Warn("offset backwards: %d %d", o, lastTotalOffsets)
				}
			}
			lastTotalOffsets = o
		}
	}

}

func (this *MonitorTopics) report() (totalOffsets int64, topicsN int64,
	partitionN int64, brokersN int64) {
	this.zkzone.ForSortedClusters(func(zkcluster *zk.ZkCluster) {
		brokerList := zkcluster.BrokerList()
		kfk, err := sarama.NewClient(brokerList, sarama.NewConfig())
		if err != nil {
			log.Error("cluster[%s] %v", zkcluster.Name(), err)
			return
		}
		defer kfk.Close()

		brokersN += int64(len(brokerList))

		topics, err := kfk.Topics()
		if err != nil {
{'fraction_non_alphanumeric': 0.13745387453874539, 'fraction_numerical': 0.00891758917589176, 'mean_word_length': 3.9287878787878787, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 6, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '308196', 'n_tokens_mistral': 1381, 'n_tokens_neox': 1242, 'n_words': 252}