<commit_before>package main

import (
	"bytes"
	"sort"
	"strconv"
	"strings"
	"time"

	"flag"

	"fmt"

	"os"

	"github.com/Shopify/sarama"
	"github.com/fatih/color"
	"github.com/olekukonko/tablewriter"
	"github.com/samuel/go-zookeeper/zk"
)

var (
	zkAddr            = flag.String("zkAddr", "", "zookeeper address")
	basePath          = flag.String("basePath", "/kafka", "kafka base path in zookeeper")
	brokers           = flag.String("brokers", "localhost:9092", "brokers' address")
	topic             = flag.String("topic", "trigger", "topic name")
	group             = flag.String("group", "default", "consumer group name")
	version           = flag.String("version", "0.10.0.1", "kafka version. min version is 0.8.2.0")
	lagThreshold      = flag.Int("lagThreshold", 1000, "alarm lag threshold for partition")
	totalLagThreshold = flag.Int("totalLagThreshold", 5000, "alarm total lag threshold for topic")
	interval          = flag.Duration("duration", time.Minute, "check interval time")
	informEmail       = flag.String("email", "<EMAIL>", "inform user email")
	smtpHost          = flag.String("smtp.host", "smtp.sina.com", "smtp host for sending alarms")
	smtpPort          = flag.Int("smtp.port", 25, "smtp port for sending alarms")
	smtpUser          = flag.String("smtp.user", "kafka_monitor", "smtp user for sending alarms")
	smtpPassword      = flag.String("smtp.password", "<PASSWORD>", "smtp user password for sending alarms")
)

var (
	maybeProblem       = false
	lastTriggeredTime  time.Time
	mergeAlertDuration = 10 * time.Minute
	kafkaVersions      = kafkaVersion()
)

func kafkaVersion() map[string]sarama.KafkaVersion {
	m := make(map[string]sarama.KafkaVersion)
	m["0.8.2.0"] = sarama.V0_8_2_0
	m["0.8.2.1"] = sarama.V0_8_2_1
	m["0.8.2.2"] = sarama.V0_8_2_2
	m["0.9.0.0"] = sarama.V0_9_0_0
	m["0.9.0.1"] = sarama.V0_9_0_1
	m["0.10.0.0"] = sarama.V0_10_0_0
	m["0.10.0.1"] = sarama.V0_10_0_1
	m["0.10.1.0"] = sarama.V0_10_1_0
	return m
}

func main() {
	flag.Parse()

	for {
		check()
	}
}

func check() {
	defer func() {
		if r := recover(); r != nil {
			fmt.Printf("check error: %v", r)
			alert(*informEmail, *brokers, *topic, []byte(fmt.Sprintf("%v", r)), *smtpHost, *smtpPort, *smtpUser, *smtpPassword)
		}
	}()

	kafkaBrokers := strings.Split(*brokers, ",")
	v := kafkaVersions[*version]
	client := NewSaramaClient(kafkaBrokers, v)

	var buf bytes.Buffer

	ticker := time.NewTicker(*interval)
	for range ticker.C {
		buf.Reset()
		partitions, err := client.Partitions(*topic)
		if err != nil {
			fmt.Printf("failed to get partitions for topic=%s, err=%v\n", *topic, err)
			panic(err)
		}

		writablePartitions, err := client.WritablePartitions(*topic)
		if err != nil {
			fmt.Printf("failed to get writable partitions for topic=%s, err=%v\n", *topic, err)
			panic(err)
		}

		if len(partitions) != len(writablePartitions) {
			buf.WriteString("some partitions are not writable\n")
			buf.WriteString(fmt.Sprintf("all partitions: %v\n", partitions))
			buf.WriteString(fmt.Sprintf("writable partitions: %v\n", writablePartitions))
			//TODO print unwritable partitions
			maybeProblem = true
		}

		fmt.Printf("Time: %s\n", time.Now().Format("2006-01-02 15:04:05"))
		fmt.Printf("Brokers: %s\nVersion: %s\n", color.GreenString(*brokers), color.GreenString(*version))
		fmt.Printf("Topic: %s, Group: %s, Partitions: %s\n\n",
			color.GreenString(*topic), color.GreenString(*group), color.GreenString(strconv.Itoa(len(partitions))))

		//partition info
		var c *zk.Conn
		if *zkAddr != "" {
			c, _, err = zk.Connect(strings.Split(*zkAddr, ","), 5*time.Second)
			if err != nil {
				panic(err)
			}
		}

		infos := GetPartitionInfo(client, *topic, partitions, c, *basePath)<commit_msg>add broker/version info in email<commit_after>package main

import (
	"bytes"
	"sort"
	"strconv"
	"strings"
	"time"

	"flag"

	"fmt"

	"os"

	"github.com/Shopify/sarama"
	"github.com/fatih/color"
	"github.com/olekukonko/tablewriter"
	"github.com/samuel/go-zookeeper/zk"
)

var (
	zkAddr            = flag.String("zkAddr", "", "zookeeper address")
	basePath          = flag.String("basePath", "/kafka", "kafka base path in zookeeper")
	brokers           = flag.String("brokers", "localhost:9092", "brokers' address")
	topic             = flag.String("topic", "trigger", "topic name")
	group             = flag.String("group", "default", "consumer group name")
	version           = flag.String("version", "0.10.0.1", "kafka version. min version is 0.8.2.0")
	lagThreshold      = flag.Int("lagThreshold", 1000, "alarm lag threshold for partition")
	totalLagThreshold = flag.Int("totalLagThreshold", 5000, "alarm total lag threshold for topic")
	interval          = flag.Duration("duration", time.Minute, "check interval time")
	informEmail       = flag.String("email", "<EMAIL>", "inform user email")
	smtpHost          = flag.String("smtp.host", "smtp.sina.com", "smtp host for sending alarms")
	smtpPort          = flag.Int("smtp.port", 25, "smtp port for sending alarms")
	smtpUser          = flag.String("smtp.user", "kafka_monitor", "smtp user for sending alarms")
	smtpPassword      = flag.String("smtp.password", "<PASSWORD>", "smtp user password for sending alarms")
)

var (
	maybeProblem       = false
	lastTriggeredTime  time.Time
	mergeAlertDuration = 10 * time.Minute
	kafkaVersions      = kafkaVersion()
)

func kafkaVersion() map[string]sarama.KafkaVersion {
	m := make(map[string]sarama.KafkaVersion)
	m["0.8.2.0"] = sarama.V0_8_2_0
	m["0.8.2.1"] = sarama.V0_8_2_1
	m["0.8.2.2"] = sarama.V0_8_2_2
	m["0.9.0.0"] = sarama.V0_9_0_0
	m["0.9.0.1"] = sarama.V0_9_0_1
	m["0.10.0.0"] = sarama.V0_10_0_0
	m["0.10.0.1"] = sarama.V0_10_0_1
	m["0.10.1.0"] = sarama.V0_10_1_0
	return m
}

func main() {
	flag.Parse()

	for {
		check()
	}
}

func check() {
	defer func() {
		if r := recover(); r != nil {
			fmt.Printf("check error: %v", r)
			alert(*informEmail, *brokers, *topic, []byte(fmt.Sprintf("%v", r)), *smtpHost, *smtpPort, *smtpUser, *smtpPassword)
		}
	}()

	kafkaBrokers := strings.Split(*brokers, ",")
	v := kafkaVersions[*version]
	client := NewSaramaClient(kafkaBrokers, v)

	var buf bytes.Buffer

	ticker := time.NewTicker(*interval)
	for range ticker.C {
		buf.Reset()
		partitions, err := client.Partitions(*topic)
		if err != nil {
			fmt.Printf("failed to get partitions for topic=%s, err=%v\n", *topic, err)
			panic(err)
		}

		writablePartitions, err := client.WritablePartitions(*topic)
		if err != nil {
			fmt.Printf("failed to get writable partitions for topic=%s, err=%v\n", *topic, err)
			panic(err)
		}

		if len(partitions) != len(writablePartitions) {
			buf.WriteString("some partitions are not writable\n")
			buf.WriteString(fmt.Sprintf("all partitions: %v\n", partitions))
			buf.WriteString(fmt.Sprintf("writable partitions: %v\n", writablePartitions))
			//TODO print unwritable partitions
			maybeProblem = true
		}

		buf.WriteString(fmt.Sprintf("Time: %s\n", time.Now().Format("2006-01-02 15:04:05")))
		buf.WriteString(fmt.Sprintf("Brokers: %s\nVersion: %s\n", color.GreenString(*brokers), color.GreenString(*version)))
		buf.WriteString(fmt.Sprintf("Topic: %s, Group: %s, Partitions: %s\n\n",
			color.GreenString(*topic), color.GreenString(*group), color.GreenString(strconv.Itoa(len(partitions)))))

		fmt.Println(string(buf.Bytes()))

		//partition info
		var c *zk.Conn
		if *zkAddr != "" {
			c, _, err = zk.Connect(strings.Split(*zkAddr, ","), 5*time.Second)
			if err != nil {
				panic(err)
			}
		}

		infos := GetPartitionInfo(client, *topic, partitions, c, *basePath)
{'fraction_non_alphanumeric': 0.17040477770404777, 'fraction_numerical': 0.029197080291970802, 'mean_word_length': 3.8214971209213053, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 20, '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': '3332634', 'n_tokens_mistral': 3091, 'n_tokens_neox': 2800, 'n_words': 574}