<commit_before>package main

import (
	"flag"
	"fmt"
	"math/rand"
	"os"
	"strings"

	"time"

	"github.com/grafana/grafana/pkg/log"
	"github.com/nsqio/go-nsq"
	"github.com/raintank/met"
	"github.com/raintank/met/helper"
	"github.com/raintank/raintank-metric/dur"
	"github.com/raintank/raintank-metric/msg"
	"github.com/raintank/raintank-metric/schema"
)

const maxMpubSize = 5 * 1024 * 1024 // nsq errors if more. not sure if can be changed
const maxMetricPerMsg = 1000        // emperically found through benchmarks (should result in 64~128k messages)
var (
	producer          *nsq.Producer
	metricsPublished  met.Count
	messagesPublished met.Count
	messagesSize      met.Meter
	metricsPerMessage met.Meter
	publishDuration   met.Timer
)

var (
	showVersion = flag.Bool("version", false, "print version string")

	topic = flag.String("topic", "metrics", "NSQ topic")

	//producerOpts     = flag.String("producer-opt", "", "option to passthrough to nsq.Producer (may be given multiple times as comma-separated list, see http://godoc.org/github.com/nsqio/go-nsq#Config)")
	nsqdTCPAddr  = flag.String("nsqd-tcp-address", "localhost:4150", "nsqd TCP address")
	logLevel     = flag.Int("log-level", 2, "log level. 0=TRACE|1=DEBUG|2=INFO|3=WARN|4=ERROR|5=CRITICAL|6=FATAL")
	orgs         = flag.Int("orgs", 2000, "how many orgs to simulate")
	keysPerOrg   = flag.Int("keys-per-org", 100, "how many metrics per orgs to simulate")
	speedup      = flag.Int("speedup", 1, "for each advancement of real time, how many advancements of fake data to simulate")
	metricPeriod = flag.Int("metricPeriod", 1, "period in seconds between metric points")
	flushPeriod  = flag.Int("flushPeriod", 100, "period in ms between flushes. metricPeriod must be cleanly divisible by flushPeriod. does not affect volume/throughput per se. the message is adjusted as to keep the volume/throughput constant")
	offset       = flag.String("offset", "0", "offset duration expression. (how far back in time to start. e.g. 1month, 6h, etc")
	stopAtNow    = flag.Bool("stop-at-now", false, "stop program instead of starting to write data with future timestamps")
	statsdAddr   = flag.String("statsd-addr", "localhost:8125", "statsd address")
	statsdType   = flag.String("statsd-type", "standard", "statsd type: standard or datadog")
)

func main() {
	flag.Parse()

	log.NewLogger(0, "console", fmt.Sprintf(`{"level": %d, "formatting":true}`, *logLevel))

	if *showVersion {
		fmt.Println("fake_metrics_to_nsq")
		return
	}

	if *topic == "" {
		log.Fatal(4, "--topic is required")
	}

	hostname, err := os.Hostname()
	if err != nil {
		log.Fatal(4, "failed to lookup hostname. %s", err)
	}
	stats, err := helper.New(true, *statsdAddr, *statsdType, "fake_metrics_to_nsq", strings.Replace(hostname, ".", "_", -1))
	if err != nil {
		log.Fatal(4, "failed to initialize statsd. %s", err)
	}

	cfg := nsq.NewConfig()
	cfg.UserAgent = "fake_metrics_to_nsq"
	producer, err = nsq.NewProducer(*nsqdTCPAddr, cfg)
	if err != nil {
		log.Fatal(0, "failed to initialize nsq producer.", err)
	}
	err = producer.Ping()
	if err != nil {
		log.Fatal(0, "can't connect to nsqd: %s", err)
	}
	metricsPublished = stats.NewCount("metricpublisher.metrics-published")
	messagesPublished = stats.NewCount("metricpublisher.messages-published")
	messagesSize = stats.NewMeter("metricpublisher.message_size", 0)
	metricsPerMessage = stats.NewMeter("metricpublisher.metrics_per_message", 0)
	publishDuration = stats.NewTimer("metricpublisher.publish_duration", 0)
	if (1000**metricPeriod)%*flushPeriod != 0 {
		panic("metricPeriod must be cleanly divisible by flushPeriod")
	}
	off := int(dur.MustParseUsec("offset", *offset))
	run(*orgs, *keysPerOrg, *metricPeriod, *flushPeriod, off, *speedup)
}

func Reslice(in []*schema.MetricData, size int) [][]*schema.MetricData {
	numSubSlices := len(in) / size
	if len(in)%size > 0 {
		numSubSlices += 1
	}
	out := make([][]*schema.MetricData, numSubSlices)
	for i := 0; i < numSubSlices; i++ {
		start := i * size
		end := (i + 1) * size
		if end > len(in) {
			out[i] = in[start:]
		} else {
			out[i] = in[start:end]
		}
	}
	return out
}

func Publish(metrics []*schema.MetricData) error {
	if len(metrics) == 0 {
		return nil
	}
	// typical metrics seem to be around 300B
	// nsqd allows <= 10MiB messages.
	// we ideally have 64kB ~ 1MiB messages (see benchmark https://gist.github.com/Dieterbe/604232d35494eae73f15)
	// at 300B, about 3500 msg fit in 1MiB
	// in worst case, this allows messages up to 2871B
	// this could be made more robust of course

	// real world findings in dev-stack with env-load:
	// 159569B msg /795  metrics per msg = 200B per msg
	// so peak message size is about 3500*200 = 700k (seen 711k)

	subslices := Reslice(metrics, 3500)

	for _, subslice := range subslices {
		id := time.Now().UnixNano()
		data, err := msg.CreateMsg(subslice, id, msg.FormatMetricDataArrayMsgp)
		if err != nil {
			log.Fatal(0, "Fatal error creating metric message: %s", err)
		}
		metricsPublished.Inc(int64(len(subslice)))
		messagesPublished.Inc(1)
		messagesSize.Value(int64(len(data)))
		metricsPerMessage.Value(int64(len(subslice)))
		pre := time.Now()
		err = producer.Publish(*topic, data)
		publishDuration.Value(time.Since(pre))
		if err != nil {
			log.Fatal(0, "can't publish to nsqd: %s", err)
		}
		//log.Info("published metrics %d size=%d", id, len(data))
	}
	return nil
}

func run(orgs, keysPerOrg, metricPeriod, flushPeriod, offset, speedup int) {
	// examples:
	// flushperiod - metricperiod - speedup -> data to flush each flush
	// 100           1              1          1/10 -> flush 1 of 10 fractions
	// 100           1              2          1/5 -> flush 1 of 5 fractions
	// 1000          1              2          2x -> flush ratio 2
	// 2000          1              1          2x -> flush ratio 2
	// 2000          1              2          4x -> flush ratio 4

	if flushPeriod*speedup >= 1000*metricPeriod {
		runMultiplied(orgs, keysPerOrg, metricPeriod, flushPeriod, offset, speedup)
	} else {
		runDivided(orgs, keysPerOrg, metricPeriod, flushPeriod, offset, speedup)
	}
}

// we need to send multiples of the data at every flush
func runMultiplied(orgs, keysPerOrg, metricPeriod, flushPeriod, offset, speedup int) {<commit_msg>explain better<commit_after>package main

import (
	"flag"
	"fmt"
	"math/rand"
	"os"
	"strings"

	"time"

	"github.com/grafana/grafana/pkg/log"
	"github.com/nsqio/go-nsq"
	"github.com/raintank/met"
	"github.com/raintank/met/helper"
	"github.com/raintank/raintank-metric/dur"
	"github.com/raintank/raintank-metric/msg"
	"github.com/raintank/raintank-metric/schema"
)

const maxMpubSize = 5 * 1024 * 1024 // nsq errors if more. not sure if can be changed
const maxMetricPerMsg = 1000        // emperically found through benchmarks (should result in 64~128k messages)
var (
	producer          *nsq.Producer
	metricsPublished  met.Count
	messagesPublished met.Count
	messagesSize      met.Meter
	metricsPerMessage met.Meter
	publishDuration   met.Timer
)

var (
	showVersion = flag.Bool("version", false, "print version string")

	topic = flag.String("topic", "metrics", "NSQ topic")

	//producerOpts     = flag.String("producer-opt", "", "option to passthrough to nsq.Producer (may be given multiple times as comma-separated list, see http://godoc.org/github.com/nsqio/go-nsq#Config)")
	nsqdTCPAddr  = flag.String("nsqd-tcp-address", "localhost:4150", "nsqd TCP address")
	logLevel     = flag.Int("log-level", 2, "log level. 0=TRACE|1=DEBUG|2=INFO|3=WARN|4=ERROR|5=CRITICAL|6=FATAL")
	orgs         = flag.Int("orgs", 2000, "how many orgs to simulate")
	keysPerOrg   = flag.Int("keys-per-org", 100, "how many metrics per orgs to simulate")
	speedup      = flag.Int("speedup", 1, "for each advancement of real time, how many advancements of fake data to simulate")
	metricPeriod = flag.Int("metricPeriod", 1, "period in seconds between metric points")
	flushPeriod  = flag.Int("flushPeriod", 100, "period in ms between flushes. metricPeriod must be cleanly divisible by flushPeriod. does not affect volume/throughput per se. the message is adjusted as to keep the volume/throughput constant")
	offset       = flag.String("offset", "0", "offset duration expression. (how far back in time to start. e.g. 1month, 6h, etc")
	stopAtNow    = flag.Bool("stop-at-now", false, "stop program instead of starting to write data with future timestamps")
	statsdAddr   = flag.String("statsd-addr", "localhost:8125", "statsd address")
	statsdType   = flag.String("statsd-type", "standard", "statsd type: standard or datadog")
)

func main() {
	flag.Parse()

	log.NewLogger(0, "console", fmt.Sprintf(`{"level": %d, "formatting":true}`, *logLevel))

	if *showVersion {
		fmt.Println("fake_metrics_to_nsq")
		return
	}

	if *topic == "" {
		log.Fatal(4, "--topic is required")
	}

	hostname, err := os.Hostname()
	if err != nil {
		log.Fatal(4, "failed to lookup hostname. %s", err)
	}
	stats, err := helper.New(true, *statsdAddr, *statsdType, "fake_metrics_to_nsq", strings.Replace(hostname, ".", "_", -1))
	if err != nil {
		log.Fatal(4, "failed to initialize statsd. %s", err)
	}

	cfg := nsq.NewConfig()
	cfg.UserAgent = "fake_metrics_to_nsq"
	producer, err = nsq.NewProducer(*nsqdTCPAddr, cfg)
	if err != nil {
		log.Fatal(0, "failed to initialize nsq producer.", err)
	}
	err = producer.Ping()
	if err != nil {
		log.Fatal(0, "can't connect to nsqd: %s", err)
	}
	metricsPublished = stats.NewCount("metricpublisher.metrics-published")
	messagesPublished = stats.NewCount("metricpublisher.messages-published")
	messagesSize = stats.NewMeter("metricpublisher.message_size", 0)
	metricsPerMessage = stats.NewMeter("metricpublisher.metrics_per_message", 0)
	publishDuration = stats.NewTimer("metricpublisher.publish_duration", 0)
	if (1000**metricPeriod)%*flushPeriod != 0 {
		panic("metricPeriod must be cleanly divisible by flushPeriod")
	}
	off := int(dur.MustParseUsec("offset", *offset))
	run(*orgs, *keysPerOrg, *metricPeriod, *flushPeriod, off, *speedup)
}

func Reslice(in []*schema.MetricData, size int) [][]*schema.MetricData {
	numSubSlices := len(in) / size
	if len(in)%size > 0 {
		numSubSlices += 1
	}
	out := make([][]*schema.MetricData, numSubSlices)
	for i := 0; i < numSubSlices; i++ {
		start := i * size
		end := (i + 1) * size
		if end > len(in) {
			out[i] = in[start:]
		} else {
			out[i] = in[start:end]
		}
	}
	return out
}

func Publish(metrics []*schema.MetricData) error {
	if len(metrics) == 0 {
		return nil
	}
	// typical metrics seem to be around 300B
	// nsqd allows <= 10MiB messages.
	// we ideally have 64kB ~ 1MiB messages (see benchmark https://gist.github.com/Dieterbe/604232d35494eae73f15)
	// at 300B, about 3500 msg fit in 1MiB
	// in worst case, this allows messages up to 2871B
	// this could be made more robust of course

	// real world findings in dev-stack with env-load:
	// 159569B msg /795  metrics per msg = 200B per msg
	// so peak message size is about 3500*200 = 700k (seen 711k)

	subslices := Reslice(metrics, 3500)

	for _, subslice := range subslices {
		id := time.Now().UnixNano()
		data, err := msg.CreateMsg(subslice, id, msg.FormatMetricDataArrayMsgp)
		if err != nil {
			log.Fatal(0, "Fatal error creating metric message: %s", err)
		}
		metricsPublished.Inc(int64(len(subslice)))
		messagesPublished.Inc(1)
		messagesSize.Value(int64(len(data)))
		metricsPerMessage.Value(int64(len(subslice)))
		pre := time.Now()
		err = producer.Publish(*topic, data)
		publishDuration.Value(time.Since(pre))
		if err != nil {
			log.Fatal(0, "can't publish to nsqd: %s", err)
		}
		//log.Info("published metrics %d size=%d", id, len(data))
	}
	return nil
}

func run(orgs, keysPerOrg, metricPeriod, flushPeriod, offset, speedup int) {
	// examples:
	// flushperiod - metricperiod - speedup -> data to flush each flush
	// 100     ms    1 s            1          1/10 -> flush 1 of 10 fractions runDivided <-- default
	// 100     ms    1 s            2          1/5 -> flush 1 of 5 fractions   runDivided
	// 100     ms    1 s            100        1/5 -> flush 1x                 runMultiplied
	// 1000    ms    1 s            2          2x -> flush 2x                  runMultiplied
	// 2000    ms    1 s            1          2x -> flush 2x                  runMultiplied
	// 2000    ms    1 s            2          4x -> flush 4x                  runMultiplied

	if flushPeriod*speedup >= 1000*metricPeriod {
		runMultiplied(orgs, keysPerOrg, metricPeriod, flushPeriod, offset, speedup)
	} else {
		runDivided(orgs, keysPerOrg, metricPeriod, flushPeriod, offset, speedup)
	}
}

// we need to send multiples of the data at every flush
func runMultiplied(orgs, keysPerOrg, metricPeriod, flushPeriod, offset, speedup int) {
{'fraction_non_alphanumeric': 0.13331234752498622, 'fraction_numerical': 0.030691744707641456, 'mean_word_length': 3.5434393993564535, 'pattern_counts': {'":': 4, '<': 8, '<?xml version=': 0, '>': 22, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 11, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2000005', 'n_tokens_mistral': 4794, 'n_tokens_neox': 4334, 'n_words': 1207}