<commit_before>package main

import (
	"flag"
	"fmt"
	"math/rand"
	"os"
	"os/signal"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"text/template"

	inKafkaMdm "github.com/grafana/metrictank/input/kafkamdm"
	"github.com/grafana/metrictank/stats"
	"github.com/raintank/worldping-api/pkg/log"
	"github.com/rakyll/globalconf"
	"gopkg.in/raintank/schema.v1"
	"gopkg.in/raintank/schema.v1/msg"
)

var (
	confFile = flag.String("config", "/etc/metrictank/metrictank.ini", "configuration file path")
	formatMd = flag.String("format-md", "{{.Part}} {{.OrgId}} {{.Id}} {{.Name}} {{.Interval}} {{.Value}} {{.Time}} {{.Unit}} {{.Mtype}} {{.Tags}}", "template to render MetricData with")
	formatP  = flag.String("format-point", "{{.Part}} {{.MKey}} {{.Value}} {{.Time}}", "template to render MetricPoint data with")
	prefix   = flag.String("prefix", "", "only show metrics that have this prefix")
	substr   = flag.String("substr", "", "only show metrics that have this substring")
	invalid  = flag.Bool("invalid", false, "only show metrics that are invalid")

	stdoutLock = sync.Mutex{}
)

type DataMd struct {
	Part int32
	schema.MetricData
}

type DataP struct {
	Part int32
	schema.MetricPoint
}

type inputPrinter struct {
	tplMd template.Template
	tplP  template.Template
}

func newInputPrinter(formatMd, formatP string) inputPrinter {
	tplMd := template.Must(template.New("format").Parse(formatMd + "\n"))
	tplP := template.Must(template.New("format").Parse(formatP + "\n"))
	return inputPrinter{
		*tplMd,
		*tplP,
	}
}

func (ip inputPrinter) ProcessMetricData(metric *schema.MetricData, partition int32) {
	if *prefix != "" && !strings.HasPrefix(metric.Name, *prefix) {
		return
	}
	if *substr != "" && !strings.Contains(metric.Name, *substr) {
		return
	}
	if *invalid {
		err := metric.Validate()
		if err == nil && metric.Time != 0 {
			return
		}
	}
	stdoutLock.Lock()
	err := ip.tplMd.Execute(os.Stdout, DataMd{
		partition,
		*metric,
	})

	stdoutLock.Unlock()
	if err != nil {
		log.Error(0, "executing template: %s", err)
	}
}

func (ip inputPrinter) ProcessMetricPoint(point schema.MetricPoint, format msg.Format, partition int32) {

	if *invalid && point.Valid() {
		return
	}

	stdoutLock.Lock()
	err := ip.tplP.Execute(os.Stdout, DataP{
		partition,
		point,
	})
	stdoutLock.Unlock()
	if err != nil {
		log.Error(0, "executing template: %s", err)
	}
}

func main() {
	flag.Usage = func() {
		fmt.Fprintln(os.Stderr, "mt-kafka-mdm-sniff")
		fmt.Fprintln(os.Stderr)
		fmt.Fprintln(os.Stderr, "Inspects what's flowing through kafka (in mdm format) and reports it to you")
		fmt.Fprintf(os.Stderr, "\nFlags:\n\n")
		flag.PrintDefaults()
	}
	flag.Parse()
	log.NewLogger(0, "console", fmt.Sprintf(`{"level": %d, "formatting":false}`, 2))
	instance := "mt-kafka-mdm-sniff" + strconv.Itoa(rand.Int())

	// Only try and parse the conf file if it exists
	path := ""
	if _, err := os.Stat(*confFile); err == nil {
		path = *confFile
	}
	conf, err := globalconf.NewWithOptions(&globalconf.Options{
		Filename:  path,
		EnvPrefix: "MT_",<commit_msg>kafka-mdm-sniff: support printing ts nicely formatted<commit_after>package main

import (
	"flag"
	"fmt"
	"math/rand"
	"os"
	"os/signal"
	"strconv"
	"strings"
	"sync"
	"syscall"
	"text/template"
	"time"

	inKafkaMdm "github.com/grafana/metrictank/input/kafkamdm"
	"github.com/grafana/metrictank/stats"
	"github.com/raintank/worldping-api/pkg/log"
	"github.com/rakyll/globalconf"
	"gopkg.in/raintank/schema.v1"
	"gopkg.in/raintank/schema.v1/msg"
)

var (
	confFile = flag.String("config", "/etc/metrictank/metrictank.ini", "configuration file path")
	formatMd = flag.String("format-md", "{{.Part}} {{.OrgId}} {{.Id}} {{.Name}} {{.Interval}} {{.Value}} {{.Time}} {{.Unit}} {{.Mtype}} {{.Tags}}", "template to render MetricData with")
	formatP  = flag.String("format-point", "{{.Part}} {{.MKey}} {{.Value}} {{.Time}}", "template to render MetricPoint data with")
	prefix   = flag.String("prefix", "", "only show metrics that have this prefix")
	substr   = flag.String("substr", "", "only show metrics that have this substring")
	invalid  = flag.Bool("invalid", false, "only show metrics that are invalid")

	stdoutLock = sync.Mutex{}
)

type DataMd struct {
	Part int32
	schema.MetricData
}

type DataP struct {
	Part int32
	schema.MetricPoint
}

type inputPrinter struct {
	tplMd template.Template
	tplP  template.Template
}

func dateInt64(ts int64) string {
	return time.Unix(ts, 0).Format(time.RFC3339)
}

func dateUint32(ts uint32) string {
	return dateInt64(int64(ts))
}

func newInputPrinter(formatMd, formatP string) inputPrinter {
	funcsMd := map[string]interface{}{
		"date": dateInt64,
	}
	funcsP := map[string]interface{}{
		"date": dateUint32,
	}

	tplMd := template.Must(template.New("format").Funcs(funcsMd).Parse(formatMd + "\n"))
	tplP := template.Must(template.New("format").Funcs(funcsP).Parse(formatP + "\n"))
	return inputPrinter{
		*tplMd,
		*tplP,
	}
}

func (ip inputPrinter) ProcessMetricData(metric *schema.MetricData, partition int32) {
	if *prefix != "" && !strings.HasPrefix(metric.Name, *prefix) {
		return
	}
	if *substr != "" && !strings.Contains(metric.Name, *substr) {
		return
	}
	if *invalid {
		err := metric.Validate()
		if err == nil && metric.Time != 0 {
			return
		}
	}
	stdoutLock.Lock()
	err := ip.tplMd.Execute(os.Stdout, DataMd{
		partition,
		*metric,
	})

	stdoutLock.Unlock()
	if err != nil {
		log.Error(0, "executing template: %s", err)
	}
}

func (ip inputPrinter) ProcessMetricPoint(point schema.MetricPoint, format msg.Format, partition int32) {

	if *invalid && point.Valid() {
		return
	}

	stdoutLock.Lock()
	err := ip.tplP.Execute(os.Stdout, DataP{
		partition,
		point,
	})
	stdoutLock.Unlock()
	if err != nil {
		log.Error(0, "executing template: %s", err)
	}
}

func main() {
	flag.Usage = func() {
		fmt.Fprintln(os.Stderr, "mt-kafka-mdm-sniff")
		fmt.Fprintln(os.Stderr)
		fmt.Fprintln(os.Stderr, "Inspects what's flowing through kafka (in mdm format) and reports it to you")
		fmt.Fprintf(os.Stderr, "\nFlags:\n\n")
		flag.PrintDefaults()
		fmt.Fprintln(os.Stderr, "you can also use functions in templates:")
		fmt.Fprintln(os.Stderr, "date: formats a unix timestamp as a date")
		fmt.Fprintln(os.Stderr, "example: mt-kafka-mdm-sniff -format-point '{{.Time | date}}'")
	}
	flag.Parse()
	log.NewLogger(0, "console", fmt.Sprintf(`{"level": %d, "formatting":false}`, 2))
	instance := "mt-kafka-mdm-sniff" + strconv.Itoa(rand.Int())

	// Only try and parse the conf file if it exists
	path := ""
	if _, err := os.Stat(*confFile); err == nil {
		path = *confFile
	}
	conf, err := globalconf.NewWithOptions(&globalconf.Options{
		Filename:  path,
		EnvPrefix: "MT_",
{'fraction_non_alphanumeric': 0.17409575266396518, 'fraction_numerical': 0.00765420981539847, 'mean_word_length': 4.318435754189944, 'pattern_counts': {'":': 6, '<': 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': '6319517', 'n_tokens_mistral': 2610, 'n_tokens_neox': 2521, 'n_words': 545}