Add blockgen command to generate blocks of mock data (#191)

Generate blocks of mock data

Author: replay

CHANGELOG.md

@@ -4,6 +4,7 @@ Order should be `CHANGE`, `FEATURE`, `ENHANCEMENT`, and `BUGFIX`
 
 ## unreleased/master
 
+* [FEATURE] Blockgen: adding a new tool to generate blocks of mock data.
 * [BUGFIX] Benchtool: avoid duplicate DNS metrics registration when enabling both query and write benchmarking. #188
 * [ENHANCEMENT] Added the ability to set an explicit user when Cortex is behind basic auth. #187
 

cmd/blockgen/main.go

@@ -0,0 +1,18 @@
+package main
+
+import (
+	"os"
+
+	"gopkg.in/alecthomas/kingpin.v2"
+
+	"github.com/grafana/cortex-tools/pkg/commands"
+)
+
+var blockGenCommand commands.BlockGenCommand
+
+func main() {
+	kingpin.Version("0.0.1")
+	app := kingpin.New("blockgen", "A command-line tool to generate cortex blocks.")
+	blockGenCommand.Register(app)
+	kingpin.MustParse(app.Parse(os.Args[1:]))
+}

docs/blockgen.md

@@ -0,0 +1,73 @@
+# Blockgen
+
+The `blockgen` is a tool which generates Prometheus blocks of mock data. These
+blocks can be uploaded to an object store used by Cortex, which may be useful
+to profile and benchmark Cortex components when querying or compacting the 
+generated blocks.
+
+## Config file
+
+The blockgen command requires a config file to be specified. 
+It uses the same syntax to define what metrics should be generated like the
+`benchtool` which is documented [here](benchtool.md). 
+ 
+This is an example config file:
+
+```yaml
+replicas: 3
+block_gen:
+  interval: 15s
+  block_size: 2h
+  block_dir: /tmp/mock_blocks
+  max_t: 1623422958000
+  min_t: 1623336558000
+series:
+  - labels:
+      - name: label_01
+        unique_values: 100
+        value_prefix: label_value_01
+      - name: label_02
+        unique_values: 10
+        value_prefix: label_value_02
+    name: metric_gauge_random_01
+    static_labels:
+        static: "true"
+    type: gauge-random
+  - labels:
+      - name: label_01
+        unique_values: 100
+        value_prefix: label_value_01
+      - name: label_02
+        unique_values: 10
+        value_prefix: label_value_02
+    name: metric_gauge_zero_01
+    static_labels:
+        static: "true"
+    type: gauge-zero
+```
+
+
+## Running it
+
+Assuming the config is in a file named `blockgen.conf` the command could 
+be run like this:
+
+```
+$ ./blockgen  --config.file=./blockgen.conf
+level=info msg="Generating data" minT=1623336558000 maxT=1623422958000 interval=15000
+level=info msg="starting new block" block_id=225463 blocks_left=13
+level=info msg="starting new block" block_id=225464 blocks_left=12
+level=info msg="starting new block" block_id=225465 blocks_left=11
+level=info msg="starting new block" block_id=225466 blocks_left=10
+level=info msg="starting new block" block_id=225467 blocks_left=9
+level=info msg="starting new block" block_id=225468 blocks_left=8
+level=info msg="starting new block" block_id=225469 blocks_left=7
+level=info msg="starting new block" block_id=225470 blocks_left=6
+level=info msg="starting new block" block_id=225471 blocks_left=5
+level=info msg="starting new block" block_id=225472 blocks_left=4
+level=info msg="starting new block" block_id=225473 blocks_left=3
+level=info msg="starting new block" block_id=225474 blocks_left=2
+level=info msg="starting new block" block_id=225475 blocks_left=1
+level=info msg=finished block_dir=/tmp/mock_blocks
+$
+```

pkg/bench/ring_check.go

@@ -40,11 +40,11 @@ type RingChecker struct {
 
 	Ring         *ring.Ring
 	MemberlistKV *memberlist.KVInitService
-	workload     *writeWorkload
+	workload     *WriteWorkload
 	logger       log.Logger
 }
 
-func NewRingChecker(id string, instanceName string, cfg RingCheckConfig, workload *writeWorkload, logger log.Logger) (*RingChecker, error) {
+func NewRingChecker(id string, instanceName string, cfg RingCheckConfig, workload *WriteWorkload, logger log.Logger) (*RingChecker, error) {
 	r := RingChecker{
 		id:           id,
 		instanceName: instanceName,
@@ -86,7 +86,7 @@ func (r *RingChecker) Run(ctx context.Context) error {
 }
 
 func (r *RingChecker) check() {
-	timeseries := r.workload.generateTimeSeries(r.id, time.Now())
+	timeseries := r.workload.GenerateTimeSeries(r.id, time.Now())
 
 	addrMap := map[string]int{}
 	for _, s := range timeseries {
@@ -97,7 +97,6 @@ func (r *RingChecker) check() {
 		token := shardByAllLabels(r.instanceName, s.Labels)
 
 		rs, err := r.Ring.Get(token, ring.Write, []ring.InstanceDesc{}, nil, nil)
-
 		if err != nil {
 			level.Warn(r.logger).Log("msg", "unable to get token for metric", "err", err)
 			continue

pkg/bench/workload.go

@@ -61,17 +61,17 @@ type WorkloadDesc struct {
 	Write     WriteDesc    `yaml:"write_options"`
 }
 
-type timeseries struct {
+type Timeseries struct {
 	labelSets  [][]prompb.Label
 	lastValue  float64
 	seriesType SeriesType
 }
 
-type writeWorkload struct {
-	replicas           int
-	series             []*timeseries
-	totalSeries        int
-	totalSeriesTypeMap map[SeriesType]int
+type WriteWorkload struct {
+	Replicas           int
+	Series             []*Timeseries
+	TotalSeries        int
+	TotalSeriesTypeMap map[SeriesType]int
 
 	missedIterations prometheus.Counter
 
@@ -80,18 +80,56 @@ type writeWorkload struct {
 	seriesBufferChan chan []prompb.TimeSeries
 }
 
-func newWriteWorkload(workloadDesc WorkloadDesc, reg prometheus.Registerer) *writeWorkload {
+func newWriteWorkload(workloadDesc WorkloadDesc, reg prometheus.Registerer) *WriteWorkload {
+	series, totalSeriesTypeMap := SeriesDescToSeries(workloadDesc.Series)
+
 	totalSeries := 0
+	for _, typeTotal := range totalSeriesTypeMap {
+		totalSeries += typeTotal
+	}
+
+	// Set batch size to 500 samples if not set
+	if workloadDesc.Write.BatchSize == 0 {
+		workloadDesc.Write.BatchSize = 500
+	}
+
+	// Set the write interval to 15 seconds if not set
+	if workloadDesc.Write.Interval == 0 {
+		workloadDesc.Write.Interval = time.Second * 15
+	}
+
+	// Set the write timeout to 15 seconds if not set
+	if workloadDesc.Write.Timeout == 0 {
+		workloadDesc.Write.Timeout = time.Second * 15
+	}
+
+	return &WriteWorkload{
+		Replicas:           workloadDesc.Replicas,
+		Series:             series,
+		TotalSeries:        totalSeries,
+		TotalSeriesTypeMap: totalSeriesTypeMap,
+		options:            workloadDesc.Write,
+
+		missedIterations: promauto.With(reg).NewCounter(
+			prometheus.CounterOpts{
+				Namespace: "benchtool",
+				Name:      "write_iterations_late_total",
+				Help:      "Number of write intervals started late because the previous interval did not complete in time.",
+			},
+		),
+	}
+}
+
+func SeriesDescToSeries(seriesDescs []SeriesDesc) ([]*Timeseries, map[SeriesType]int) {
+	series := []*Timeseries{}
 	totalSeriesTypeMap := map[SeriesType]int{
 		GaugeZero:     0,
 		GaugeRandom:   0,
 		CounterOne:    0,
 		CounterRandom: 0,
 	}
 
-	series := []*timeseries{}
-
-	for _, seriesDesc := range workloadDesc.Series {
+	for _, seriesDesc := range seriesDescs {
 		// Create the metric with a name value
 		labelSets := [][]prompb.Label{
 			{
@@ -109,45 +147,15 @@ func newWriteWorkload(workloadDesc WorkloadDesc, reg prometheus.Registerer) *wri
 			labelSets = addLabelToLabelSet(labelSets, lbl)
 		}
 
-		series = append(series, &timeseries{
+		series = append(series, &Timeseries{
 			labelSets:  labelSets,
 			seriesType: seriesDesc.Type,
 		})
 		numSeries := len(labelSets)
-		totalSeries += numSeries
 		totalSeriesTypeMap[seriesDesc.Type] += numSeries
 	}
 
-	// Set batch size to 500 samples if not set
-	if workloadDesc.Write.BatchSize == 0 {
-		workloadDesc.Write.BatchSize = 500
-	}
-
-	// Set the write interval to 15 seconds if not set
-	if workloadDesc.Write.Interval == 0 {
-		workloadDesc.Write.Interval = time.Second * 15
-	}
-
-	// Set the write timeout to 15 seconds if not set
-	if workloadDesc.Write.Timeout == 0 {
-		workloadDesc.Write.Timeout = time.Second * 15
-	}
-
-	return &writeWorkload{
-		replicas:           workloadDesc.Replicas,
-		series:             series,
-		totalSeries:        totalSeries,
-		totalSeriesTypeMap: totalSeriesTypeMap,
-		options:            workloadDesc.Write,
-
-		missedIterations: promauto.With(reg).NewCounter(
-			prometheus.CounterOpts{
-				Namespace: "benchtool",
-				Name:      "write_iterations_late_total",
-				Help:      "Number of write intervals started late because the previous interval did not complete in time.",
-			},
-		),
-	}
+	return series, totalSeriesTypeMap
 }
 
 func addLabelToLabelSet(labelSets [][]prompb.Label, lbl LabelDesc) [][]prompb.Label {
@@ -167,14 +175,14 @@ func addLabelToLabelSet(labelSets [][]prompb.Label, lbl LabelDesc) [][]prompb.La
 	return newLabelSets
 }
 
-func (w *writeWorkload) generateTimeSeries(id string, t time.Time) []prompb.TimeSeries {
+func (w *WriteWorkload) GenerateTimeSeries(id string, t time.Time) []prompb.TimeSeries {
 	now := t.UnixNano() / int64(time.Millisecond)
 
-	timeseries := make([]prompb.TimeSeries, 0, w.replicas*w.totalSeries)
-	for replicaNum := 0; replicaNum < w.replicas; replicaNum++ {
+	timeseries := make([]prompb.TimeSeries, 0, w.Replicas*w.TotalSeries)
+	for replicaNum := 0; replicaNum < w.Replicas; replicaNum++ {
 		replicaLabel := prompb.Label{Name: "bench_replica", Value: fmt.Sprintf("replica-%05d", replicaNum)}
 		idLabel := prompb.Label{Name: "bench_id", Value: id}
-		for _, series := range w.series {
+		for _, series := range w.Series {
 			var value float64
 			switch series.seriesType {
 			case GaugeZero:
@@ -215,7 +223,7 @@ type batchReq struct {
 	putBack chan []prompb.TimeSeries
 }
 
-func (w *writeWorkload) getSeriesBuffer(ctx context.Context) []prompb.TimeSeries {
+func (w *WriteWorkload) getSeriesBuffer(ctx context.Context) []prompb.TimeSeries {
 	select {
 	case <-ctx.Done():
 		return nil
@@ -224,7 +232,7 @@ func (w *writeWorkload) getSeriesBuffer(ctx context.Context) []prompb.TimeSeries
 	}
 }
 
-func (w *writeWorkload) generateWriteBatch(ctx context.Context, id string, numBuffers int, seriesChan chan batchReq) error {
+func (w *WriteWorkload) generateWriteBatch(ctx context.Context, id string, numBuffers int, seriesChan chan batchReq) error {
 	w.seriesBufferChan = make(chan []prompb.TimeSeries, numBuffers)
 	for i := 0; i < numBuffers; i++ {
 		w.seriesBufferChan <- make([]prompb.TimeSeries, 0, w.options.BatchSize)
@@ -250,10 +258,10 @@ func (w *writeWorkload) generateWriteBatch(ctx context.Context, id string, numBu
 		timeNow := time.Now()
 		timeNowMillis := timeNow.UnixNano() / int64(time.Millisecond)
 		wg := &sync.WaitGroup{}
-		for replicaNum := 0; replicaNum < w.replicas; replicaNum++ {
+		for replicaNum := 0; replicaNum < w.Replicas; replicaNum++ {
 			replicaLabel := prompb.Label{Name: "bench_replica", Value: fmt.Sprintf("replica-%05d", replicaNum)}
 			idLabel := prompb.Label{Name: "bench_id", Value: id}
-			for _, series := range w.series {
+			for _, series := range w.Series {
 				var value float64
 				switch series.seriesType {
 				case GaugeZero:

pkg/bench/workload_test.go

@@ -131,7 +131,7 @@ func TestWorkload_generateTimeSeries(t *testing.T) {
 	for _, testCase := range testCases {
 		t.Run(testCase.name, func(t *testing.T) {
 			w := newWriteWorkload(testCase.workloadDesc, prometheus.NewRegistry())
-			generatedSeries := w.generateTimeSeries("test-id", time.Now())
+			generatedSeries := w.GenerateTimeSeries("test-id", time.Now())
 			require.Equal(t, testCase.numSeries, countUniqueTimeSeries(generatedSeries))
 		})
 	}

pkg/bench/write_runner.go

@@ -50,15 +50,15 @@ type WriteBenchmarkRunner struct {
 
 	dnsProvider *dns.Provider
 
-	workload *writeWorkload
+	workload *WriteWorkload
 
 	reg    prometheus.Registerer
 	logger log.Logger
 
 	requestDuration *prometheus.HistogramVec
 }
 
-func NewWriteBenchmarkRunner(id string, tenantName string, cfg WriteBenchConfig, workload *writeWorkload, logger log.Logger, reg prometheus.Registerer) (*WriteBenchmarkRunner, error) {
+func NewWriteBenchmarkRunner(id string, tenantName string, cfg WriteBenchConfig, workload *WriteWorkload, logger log.Logger, reg prometheus.Registerer) (*WriteBenchmarkRunner, error) {
 	writeBench := &WriteBenchmarkRunner{
 		id:         id,
 		tenantName: tenantName,
@@ -137,7 +137,7 @@ func (w *WriteBenchmarkRunner) Run(ctx context.Context) error {
 
 	// Run replicas * 10 write client workers.
 	// This number will also be used for the number of series buffers to store at once.
-	numWorkers := w.workload.replicas * 10
+	numWorkers := w.workload.Replicas * 10
 
 	batchChan := make(chan batchReq, 10)
 	for i := 0; i < numWorkers; i++ {