Add tool for reporting out of order/duplicate metrics grouped by name or tag

cmd/mt-kafka-mdm-report-out-of-order/flags.go

@@ -0,0 +1,129 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	"os"
+	"time"
+
+	"github.com/grafana/globalconf"
+	"github.com/grafana/metrictank/idx/cassandra"
+	inKafkaMdm "github.com/grafana/metrictank/input/kafkamdm"
+	log "github.com/sirupsen/logrus"
+)
+
+type Flags struct {
+	flagSet *flag.FlagSet
+
+	RunDuration   time.Duration
+	Config        string
+	PartitionFrom int
+	PartitionTo   int
+	ReorderWindow uint
+	Prefix        string
+	Substr        string
+	GroupByName   bool
+	GroupByTag    string
+}
+
+func NewFlags() *Flags {
+	var flags Flags
+
+	flags.flagSet = flag.NewFlagSet("application flags", flag.ExitOnError)
+	flags.flagSet.DurationVar(&flags.RunDuration, "run-duration", 5*time.Minute, "the duration of time to run the program")
+	flags.flagSet.StringVar(&flags.Config, "config", "/etc/metrictank/metrictank.ini", "configuration file path")
+	flags.flagSet.IntVar(&flags.PartitionFrom, "partition-from", 0, "the partition to load the index from")
+	flags.flagSet.IntVar(&flags.PartitionTo, "partition-to", -1, "load the index from all partitions up to this one (exclusive). If unset, only the partition defined with \"--partition-from\" is loaded from")
+	flags.flagSet.UintVar(&flags.ReorderWindow, "reorder-window", 1, "the size of the reorder buffer window")
+	flags.flagSet.StringVar(&flags.Prefix, "prefix", "", "only report metrics with a name that has this prefix")
+	flags.flagSet.StringVar(&flags.Substr, "substr", "", "only report metrics with a name that has this substring")
+	flags.flagSet.BoolVar(&flags.GroupByName, "group-by-name", false, "group out-of-order metrics by name")
+	flags.flagSet.StringVar(&flags.GroupByTag, "group-by-tag", "", "group out-of-order metrics by the specified tag")
+
+	flags.flagSet.Usage = flags.Usage
+	return &flags
+}
+
+func (flags *Flags) Parse(args []string) {
+	err := flags.flagSet.Parse(args)
+	if err != nil {
+		log.Fatalf("failed to parse application flags %v: %s", args, err.Error)
+		os.Exit(1)
+	}
+
+	path := ""
+	if _, err := os.Stat(flags.Config); err == nil {
+		path = flags.Config
+	}
+	config, err := globalconf.NewWithOptions(&globalconf.Options{
+		Filename:  path,
+		EnvPrefix: "MT_",
+	})
+	if err != nil {
+		log.Fatalf("error with configuration file: %s", err.Error())
+		os.Exit(1)
+	}
+	_ = cassandra.ConfigSetup()
+	inKafkaMdm.ConfigSetup()
+	config.Parse()
+
+	if flags.GroupByName == false && flags.GroupByTag == "" {
+		log.Fatalf("must specify at least one of -group-by-name or -group-by-tag")
+		os.Exit(1)
+	}
+
+	if flags.ReorderWindow < 1 {
+		log.Fatalf("-reorder-window must be greater than zero")
+		os.Exit(1)
+	}
+}
+
+func (flags *Flags) Usage() {
+	fmt.Fprintln(os.Stderr, "mt-kafka-mdm-report-out-of-order")
+	fmt.Fprintln(os.Stderr)
+	fmt.Fprintln(os.Stderr, "Inspects what's flowing through kafka (in mdm format) and reports out of order data grouped by metric name or tag, taking into account the reorder buffer)")
+	fmt.Fprintln(os.Stderr)
+	fmt.Fprintln(os.Stderr, "# Mechanism")
+	fmt.Fprintln(os.Stderr, "* it sniffs points being added on a per-series (metric Id) level")
+	fmt.Fprintln(os.Stderr, "* for every series, tracks the last 'correct' point.  E.g. a point that was able to be added to the series because its timestamp is higher than any previous timestamp")
+	fmt.Fprintln(os.Stderr, "* if for any series, a point comes in with a timestamp equal or lower than the last point correct point - which metrictank would not add unless it falls within the reorder buffer - it triggers an event for this out-of-order point")
+	fmt.Fprintln(os.Stderr, "* the reorder buffer is described by the window size")
+	fmt.Fprintln(os.Stderr, "Usage:")
+	fmt.Fprintln(os.Stderr, "  mt-kafka-mdm-report-out-of-order [flags]")
+	fmt.Fprintln(os.Stderr)
+	fmt.Fprintln(os.Stderr, "out-of-order metrics are printed in the following format based on grouping")
+	fmt.Fprintln(os.Stderr)
+	fmt.Fprintln(os.Stderr, "out-of-order metrics grouped by name:")
+	fmt.Fprintln(os.Stderr, "out-of-order name=<name1> count=<count1>")
+	fmt.Fprintln(os.Stderr, "out-of-order name=<name1> count=<count1>")
+	fmt.Fprintln(os.Stderr, "...")
+	fmt.Fprintln(os.Stderr, "duplicate metrics grouped by name:")
+	fmt.Fprintln(os.Stderr, "duplicate name=<name1> count=<count1>")
+	fmt.Fprintln(os.Stderr, "duplicate name=<name1> count=<count1>")
+	fmt.Fprintln(os.Stderr, "...")
+	fmt.Fprintln(os.Stderr, "out-of-order metrics grouped by tag:")
+	fmt.Fprintln(os.Stderr, "out-of-order tag=<tag1> count=<count1>")
+	fmt.Fprintln(os.Stderr, "out-of-order tag=<tag2> count=<count2>")
+	fmt.Fprintln(os.Stderr, "...")
+	fmt.Fprintln(os.Stderr, "duplicate metrics grouped by tag:")
+	fmt.Fprintln(os.Stderr, "duplicate tag=<tag1> count=<count1>")
+	fmt.Fprintln(os.Stderr, "duplicate tag=<tag2> count=<count2>")
+	fmt.Fprintln(os.Stderr, "...")
+	fmt.Fprintln(os.Stderr)
+	fmt.Fprintf(os.Stderr, "flags:")
+	flags.flagSet.PrintDefaults()
+	fmt.Fprintln(os.Stderr)
+	fmt.Fprintln(os.Stderr, "EXAMPLES:")
+	fmt.Fprintln(os.Stderr, "  mt-kafka-mdm-report-out-of-order -group-by-name -config metrictank.ini -partition-from 0")
+	fmt.Fprintln(os.Stderr, "  mt-kafka-mdm-report-out-of-order -group-by-name -group-by-tag namespace -config metrictank.ini -partition-from 0 -partition-to 3 -reorder-window 5 -run-duration 5m")
+}
+
+func ParseFlags() Flags {
+	flags := NewFlags()
+
+	flag.Usage = flags.Usage
+
+	flags.Parse(os.Args[1:])
+
+	return *flags
+}

cmd/mt-kafka-mdm-report-out-of-order/inputooofinder.go

@@ -0,0 +1,159 @@
+package main
+
+import (
+	"os"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/grafana/metrictank/idx/cassandra"
+	"github.com/grafana/metrictank/mdata"
+	"github.com/grafana/metrictank/mdata/errors"
+	"github.com/grafana/metrictank/schema"
+	"github.com/grafana/metrictank/schema/msg"
+	log "github.com/sirupsen/logrus"
+)
+
+type Track struct {
+	Name string
+	Tags []string
+
+	reorderBuffer *mdata.ReorderBuffer
+}
+
+type Tracker map[schema.MKey]Track
+
+// find out-of-order and duplicate metrics
+type inputOOOFinder struct {
+	prefix string
+	substr string
+
+	reorderWindow uint32
+	tracker       Tracker
+
+	groupByName             bool
+	outOfOrderGroupedByName *map[string]int
+	duplicatesGroupedByName *map[string]int
+	groupByTag              string
+	outOfOrderGroupedByTag  *map[string]int
+	duplicatesGroupedByTag  *map[string]int
+
+	lock sync.Mutex
+}
+
+func newInputOOOFinder(prefix string, substr string, partitionFrom int, partitionTo int, reorderWindow uint32, groupByName bool, outOfOrderGroupedByName *map[string]int, duplicatesGroupedByName *map[string]int, groupByTag string, outOfOrderGroupedByTag *map[string]int, duplicatesGroupedByTag *map[string]int) *inputOOOFinder {
+	cassandraIndex := cassandra.New(cassandra.CliConfig)
+	err := cassandraIndex.InitBare()
+	if err != nil {
+		log.Fatalf("error initializing cassandra index: %s", err.Error())
+		os.Exit(1)
+	}
+
+	metricDefinitions := make([]schema.MetricDefinition, 0)
+	for partition := partitionFrom; (partitionTo == -1 && partition == partitionFrom) || (partitionTo > 0 && partition < partitionTo); partition++ {
+		metricDefinitions = cassandraIndex.LoadPartitions([]int32{int32(partition)}, metricDefinitions, time.Now())
+	}
+
+	tracker := Tracker{}
+	for _, metricDefinition := range metricDefinitions {
+		tracker[metricDefinition.Id] = Track{
+			Name:          metricDefinition.Name,
+			Tags:          metricDefinition.Tags,
+			reorderBuffer: mdata.NewReorderBuffer(reorderWindow, uint32(metricDefinition.Interval), false),
+		}
+	}
+
+	return &inputOOOFinder{
+		prefix: prefix,
+		substr: substr,
+
+		reorderWindow: reorderWindow,
+		tracker:       tracker,
+
+		groupByName:             groupByName,
+		outOfOrderGroupedByName: outOfOrderGroupedByName,
+		duplicatesGroupedByName: duplicatesGroupedByName,
+		groupByTag:              groupByTag,
+		outOfOrderGroupedByTag:  outOfOrderGroupedByTag,
+		duplicatesGroupedByTag:  duplicatesGroupedByTag,
+
+		lock: sync.Mutex{},
+	}
+}
+
+func (ip *inputOOOFinder) incrementGroupings(groupedByName *map[string]int, groupedByTag *map[string]int, track Track) {
+	if ip.groupByName == true {
+		(*groupedByName)[track.Name]++
+	}
+
+	if ip.groupByTag != "" {
+		for _, tag := range track.Tags {
+			kv := strings.Split(tag, "=")
+			if len(kv) != 2 {
+				log.Errorf("unexpected tag encoding tag=%q", tag)
+				continue
+			}
+			if kv[0] == ip.groupByTag {
+				(*groupedByTag)[kv[1]]++
+			}
+		}
+	}
+}
+
+func (ip *inputOOOFinder) processTrack(metricKey schema.MKey, metricTime int64, track Track, partition int32) {
+	if ip.prefix != "" && !strings.HasPrefix(track.Name, ip.prefix) {
+		return
+	}
+	if ip.substr != "" && !strings.Contains(track.Name, ip.substr) {
+		return
+	}
+
+	_, err := track.reorderBuffer.Add(uint32(metricTime), 0) // ignore value
+	if err == errors.ErrMetricTooOld {
+		ip.incrementGroupings(ip.outOfOrderGroupedByName, ip.outOfOrderGroupedByTag, track)
+	} else if err == errors.ErrMetricNewValueForTimestamp {
+		ip.incrementGroupings(ip.duplicatesGroupedByName, ip.duplicatesGroupedByTag, track)
+	} else if err != nil {
+		log.Errorf("failed to add metric with name=%q and timestamp=%d from partition=%d to reorder buffer: %s", track.Name, metricTime, partition, err)
+	}
+}
+
+func (ip *inputOOOFinder) ProcessMetricData(metric *schema.MetricData, partition int32) {
+	metricKey, err := schema.MKeyFromString(metric.Id)
+	if err != nil {
+		log.Errorf("failed to get metric key from id=%v: %s", metric.Id, err.Error())
+		return
+	}
+
+	ip.lock.Lock()
+	defer ip.lock.Unlock()
+
+	track, exists := ip.tracker[metricKey]
+	if !exists {
+		ip.tracker[metricKey] = Track{
+			Name:          metric.Name,
+			Tags:          metric.Tags,
+			reorderBuffer: mdata.NewReorderBuffer(ip.reorderWindow, uint32(metric.Interval), false),
+		}
+		return
+	}
+
+	ip.processTrack(metricKey, metric.Time, track, partition)
+}
+
+func (ip *inputOOOFinder) ProcessMetricPoint(mp schema.MetricPoint, format msg.Format, partition int32) {
+	ip.lock.Lock()
+	defer ip.lock.Unlock()
+
+	track, exists := ip.tracker[mp.MKey]
+	if !exists {
+		log.Errorf("track for metric with key=%v from partition=%d not found", mp.MKey, partition)
+		return
+	}
+
+	ip.processTrack(mp.MKey, int64(mp.Time), track, partition)
+}
+
+func (ip *inputOOOFinder) ProcessIndexControlMsg(msg schema.ControlMsg, partition int32) {
+
+}

cmd/mt-kafka-mdm-report-out-of-order/main.go

@@ -0,0 +1,87 @@
+package main
+
+import (
+	"context"
+	"math/rand"
+	"os"
+	"os/signal"
+	"strconv"
+	"syscall"
+	"time"
+
+	inKafkaMdm "github.com/grafana/metrictank/input/kafkamdm"
+	"github.com/grafana/metrictank/logger"
+	log "github.com/sirupsen/logrus"
+)
+
+func configureLogging() {
+	formatter := &logger.TextFormatter{}
+	formatter.TimestampFormat = "2006-01-02 15:04:05.000"
+	log.SetFormatter(formatter)
+	log.SetLevel(log.InfoLevel)
+}
+
+func main() {
+	configureLogging()
+
+	flags := ParseFlags()
+
+	inKafkaMdm.ConfigProcess("mt-kafka-mdm-report-out-of-order" + strconv.Itoa(rand.Int()))
+	kafkaMdm := inKafkaMdm.New()
+
+	outOfOrderGroupedByName := map[string]int{}
+	duplicatesGroupedByName := map[string]int{}
+	outOfOrderGroupedByTag := map[string]int{}
+	duplicatesGroupedByTag := map[string]int{}
+	inputOOOFinder := newInputOOOFinder(
+		flags.Prefix,
+		flags.Substr,
+		flags.PartitionFrom,
+		flags.PartitionTo,
+		uint32(flags.ReorderWindow),
+		flags.GroupByName,
+		&outOfOrderGroupedByName,
+		&duplicatesGroupedByName,
+		flags.GroupByTag,
+		&outOfOrderGroupedByTag,
+		&duplicatesGroupedByTag,
+	)
+
+	sigChan := make(chan os.Signal, 1)
+	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	kafkaMdm.Start(inputOOOFinder, cancel)
+	select {
+	case sig := <-sigChan:
+		log.Infof("Received signal %q. Shutting down", sig)
+	case <-ctx.Done():
+		log.Info("Mdm input plugin signalled a fatal error. Shutting down")
+	case <-time.After(flags.RunDuration):
+		log.Infof("Finished scanning")
+	}
+	kafkaMdm.Stop()
+
+	if flags.GroupByName {
+		log.Info("out-of-order metrics grouped by name:")
+		for key, value := range outOfOrderGroupedByName {
+			log.Infof("out-of-order name=%q count=%d", key, value)
+		}
+
+		log.Info("duplicate metrics grouped by name:")
+		for key, value := range duplicatesGroupedByName {
+			log.Infof("duplicate name=%q count=%d", key, value)
+		}
+	}
+	if flags.GroupByTag != "" {
+		log.Info("out-of-order metrics grouped by tag:")
+		for key, value := range outOfOrderGroupedByTag {
+			log.Infof("out-of-order tag=%q count=%d", key, value)
+		}
+
+		log.Info("duplicate metrics grouped by tag:")
+		for key, value := range duplicatesGroupedByTag {
+			log.Infof("duplicate tag=%q count=%d", key, value)
+		}
+	}
+}