<commit_before>package influx

import (
	"encoding/json"
	"sync"
	"time"

	"github.com/funkygao/gafka/cmd/kguard/monitor"
	"github.com/funkygao/gafka/zk"
	log "github.com/funkygao/log4go"
	"github.com/influxdata/influxdb/client/v2"
	"github.com/rcrowley/go-metrics"
)

func init() {
	monitor.RegisterWatcher("influx.query", func() monitor.Watcher {
		return &WatchInfluxDB{
			Tick: time.Minute,
		}
	})
}

// WatchInfluxDB continuously query InfluxDB for major metrics.
//
// kateway itself will report metrics to influxdb, but there are
// several kateway instances in a cluster, we need kguard to aggregate them.
type WatchInfluxDB struct {
	Zkzone *zk.ZkZone
	Stop   <-chan struct{}
	Tick   time.Duration
	Wg     *sync.WaitGroup

	addr string
	db   string
	cli  client.Client
}

func (this *WatchInfluxDB) Init(ctx monitor.Context) {
	this.Zkzone = ctx.ZkZone()
	this.Stop = ctx.StopChan()
	this.Wg = ctx.WaitGroup()
	this.addr = ctx.InfluxAddr()
	this.db = ctx.InfluxDB()
}

func (this *WatchInfluxDB) Run() {
	defer this.Wg.Done()

	if this.addr == "" || this.db == "" {
		log.Warn("empty addr or db, quit...")
		return
	}

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

	// SELECT mean("p99") FROM "pub.latency.histogram" WHERE $timeFilter GROUP BY time(10s)
	// DB: pubsub
	pubLatency := metrics.NewRegisteredGauge("_pub.latency.99", nil) // private metric name

	for {
		select {
		case <-this.Stop:
			log.Info("influx.query stopped")
			return

		case <-ticker.C:
			p99, err := this.pubLatency()
			if err != nil {
				log.Error(err.Error())
			} else {
				pubLatency.Update(int64(p99))
			}
		}
	}
}

func (this *WatchInfluxDB) pubLatency() (float64, error) {
	res, err := this.queryDB(`SELECT mean("p99") FROM "pub.latency.histogram" where time > now() - 1m`)<commit_msg>kill the useless comment<commit_after>package influx

import (
	"encoding/json"
	"sync"
	"time"

	"github.com/funkygao/gafka/cmd/kguard/monitor"
	"github.com/funkygao/gafka/zk"
	log "github.com/funkygao/log4go"
	"github.com/influxdata/influxdb/client/v2"
	"github.com/rcrowley/go-metrics"
)

func init() {
	monitor.RegisterWatcher("influx.query", func() monitor.Watcher {
		return &WatchInfluxDB{
			Tick: time.Minute,
		}
	})
}

// WatchInfluxDB continuously query InfluxDB for major metrics.
//
// kateway itself will report metrics to influxdb, but there are
// several kateway instances in a cluster, we need kguard to aggregate them.
type WatchInfluxDB struct {
	Zkzone *zk.ZkZone
	Stop   <-chan struct{}
	Tick   time.Duration
	Wg     *sync.WaitGroup

	addr string
	db   string
	cli  client.Client
}

func (this *WatchInfluxDB) Init(ctx monitor.Context) {
	this.Zkzone = ctx.ZkZone()
	this.Stop = ctx.StopChan()
	this.Wg = ctx.WaitGroup()
	this.addr = ctx.InfluxAddr()
	this.db = ctx.InfluxDB()
}

func (this *WatchInfluxDB) Run() {
	defer this.Wg.Done()

	if this.addr == "" || this.db == "" {
		log.Warn("empty addr or db, quit...")
		return
	}

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

	pubLatency := metrics.NewRegisteredGauge("_pub.latency.99", nil) // private metric name
	for {
		select {
		case <-this.Stop:
			log.Info("influx.query stopped")
			return

		case <-ticker.C:
			p99, err := this.pubLatency()
			if err != nil {
				log.Error(err.Error())
			} else {
				pubLatency.Update(int64(p99))
			}
		}
	}
}

func (this *WatchInfluxDB) pubLatency() (float64, error) {
	res, err := this.queryDB(`SELECT mean("p99") FROM "pub.latency.histogram" where time > now() - 1m`)
{'fraction_non_alphanumeric': 0.14620217018846374, 'fraction_numerical': 0.009708737864077669, 'mean_word_length': 3.5850785340314135, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 5, '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': '7231635', 'n_tokens_mistral': 1441, 'n_tokens_neox': 1324, 'n_words': 297}