<commit_before>package monitor

import (
	"flag"
	"net"
	"net/http"
	"os"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/docker/leadership"
	"github.com/docker/libkv/store"
	"github.com/docker/libkv/store/zookeeper"
	"github.com/funkygao/gafka"
	"github.com/funkygao/gafka/ctx"
	"github.com/funkygao/gafka/telemetry"
	"github.com/funkygao/gafka/telemetry/influxdb"
	"github.com/funkygao/gafka/zk"
	"github.com/funkygao/go-metrics"
	"github.com/funkygao/golib/signal"
	log "github.com/funkygao/log4go"
	"github.com/julienschmidt/httprouter"
)

// Monitor is the engine that will start/stop plugin watchers.
// It itself is an implementation of Context.
type Monitor struct {
	influxdbAddr   string
	influxdbDbName string
	apiAddr        string
	externalDir    string

	router *httprouter.Router
	zkzone *zk.ZkZone

	candidate *leadership.Candidate

	inflight *sync.WaitGroup
	stop     chan struct{} // broadcast to all watchers to stop, but might restart again
	quit     chan struct{}
	leader   bool
}

func (this *Monitor) Init() {
	var logFile, zone string
	flag.StringVar(&logFile, "log", "stdout", "log filename")
	flag.StringVar(&zone, "z", "", "zone, required")
	flag.StringVar(&this.apiAddr, "http", ":10025", "api http server addr")
	flag.StringVar(&this.influxdbAddr, "influxAddr", "", "influxdb addr, required")
	flag.StringVar(&this.influxdbDbName, "db", "", "influxdb db name, required")
	flag.StringVar(&this.externalDir, "confd", "", "external script config dir")
	flag.Parse()

	if zone == "" || this.influxdbDbName == "" || this.influxdbAddr == "" {
		panic("zone or influxdb empty, run help ")
	}

	ctx.LoadFromHome()
	this.zkzone = zk.NewZkZone(zk.DefaultConfig(zone, ctx.ZoneZkAddrs(zone)))
	this.quit = make(chan struct{})

	// export RESTful api
	this.setupRoutes()

	if logFile == "stdout" {
		log.AddFilter("stdout", log.INFO, log.NewConsoleLogWriter())
	} else {
		log.DeleteFilter("stdout")

		filer := log.NewFileLogWriter(logFile, true, false, 0644)
		filer.SetRotateDaily(true)
		filer.SetFormat("[%d %T] [%L] (%S) %M")
		log.AddFilter("file", log.INFO, filer)
	}

	rc, err := influxdb.NewConfig(this.influxdbAddr, this.influxdbDbName, "", "", time.Minute)
	if err != nil {
		panic(err)
	}
	telemetry.Default = influxdb.New(metrics.DefaultRegistry, rc)
}

func (this *Monitor) Stop() {
	if this.leader {
		this.leader = false

		log.Info("stopping all watchers ...")
		close(this.stop)

		log.Info("stopping reporter and flush all metrics...")
		telemetry.Default.Stop()

		this.candidate.Stop()
		log.Info("election stopped")<commit_msg>kguard use TRACE log level<commit_after>package monitor

import (
	"flag"
	"net"
	"net/http"
	"os"
	"strings"
	"sync"
	"syscall"
	"time"

	"github.com/docker/leadership"
	"github.com/docker/libkv/store"
	"github.com/docker/libkv/store/zookeeper"
	"github.com/funkygao/gafka"
	"github.com/funkygao/gafka/ctx"
	"github.com/funkygao/gafka/telemetry"
	"github.com/funkygao/gafka/telemetry/influxdb"
	"github.com/funkygao/gafka/zk"
	"github.com/funkygao/go-metrics"
	"github.com/funkygao/golib/signal"
	log "github.com/funkygao/log4go"
	"github.com/julienschmidt/httprouter"
)

// Monitor is the engine that will start/stop plugin watchers.
// It itself is an implementation of Context.
type Monitor struct {
	influxdbAddr   string
	influxdbDbName string
	apiAddr        string
	externalDir    string

	router *httprouter.Router
	zkzone *zk.ZkZone

	candidate *leadership.Candidate

	inflight *sync.WaitGroup
	stop     chan struct{} // broadcast to all watchers to stop, but might restart again
	quit     chan struct{}
	leader   bool
}

func (this *Monitor) Init() {
	var logFile, zone string
	flag.StringVar(&logFile, "log", "stdout", "log filename")
	flag.StringVar(&zone, "z", "", "zone, required")
	flag.StringVar(&this.apiAddr, "http", ":10025", "api http server addr")
	flag.StringVar(&this.influxdbAddr, "influxAddr", "", "influxdb addr, required")
	flag.StringVar(&this.influxdbDbName, "db", "", "influxdb db name, required")
	flag.StringVar(&this.externalDir, "confd", "", "external script config dir")
	flag.Parse()

	if zone == "" || this.influxdbDbName == "" || this.influxdbAddr == "" {
		panic("zone or influxdb empty, run help ")
	}

	ctx.LoadFromHome()
	this.zkzone = zk.NewZkZone(zk.DefaultConfig(zone, ctx.ZoneZkAddrs(zone)))
	this.quit = make(chan struct{})

	// export RESTful api
	this.setupRoutes()

	if logFile == "stdout" {
		log.AddFilter("stdout", log.TRACE, log.NewConsoleLogWriter())
	} else {
		log.DeleteFilter("stdout")

		filer := log.NewFileLogWriter(logFile, true, false, 0644)
		filer.SetRotateDaily(true)
		filer.SetFormat("[%d %T] [%L] (%S) %M")
		log.AddFilter("file", log.TRACE, filer)
	}

	rc, err := influxdb.NewConfig(this.influxdbAddr, this.influxdbDbName, "", "", time.Minute)
	if err != nil {
		panic(err)
	}
	telemetry.Default = influxdb.New(metrics.DefaultRegistry, rc)
}

func (this *Monitor) Stop() {
	if this.leader {
		this.leader = false

		log.Info("stopping all watchers ...")
		close(this.stop)

		log.Info("stopping reporter and flush all metrics...")
		telemetry.Default.Stop()

		this.candidate.Stop()
		log.Info("election stopped")
{'fraction_non_alphanumeric': 0.15324927255092144, 'fraction_numerical': 0.0038797284190106693, 'mean_word_length': 4.387669801462905, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 3, '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': '217123', 'n_tokens_mistral': 1991, 'n_tokens_neox': 1892, 'n_words': 395}