minor fixes in benhcmark rally

Author: Andrea Spacca

internal/benchrunner/runners/rally/options.go

@@ -67,18 +67,6 @@ func WithBenchmarkName(name string) OptionFunc {
 	}
 }
 
-func WithDeferCleanup(d time.Duration) OptionFunc {
-	return func(opts *Options) {
-		opts.DeferCleanup = d
-	}
-}
-
-func WithMetricsInterval(d time.Duration) OptionFunc {
-	return func(opts *Options) {
-		opts.MetricsInterval = d
-	}
-}
-
 func WithDataReindexing(b bool) OptionFunc {
 	return func(opts *Options) {
 		opts.ReindexData = b

internal/benchrunner/runners/rally/runner.go

@@ -16,13 +16,13 @@ import (
 	"os/exec"
 	"path/filepath"
 	"strings"
+	"text/template"
 	"time"
 
 	"github.com/elastic/elastic-package/internal/packages/installer"
 
 	"github.com/magefile/mage/sh"
 
-	"github.com/elastic/elastic-package/internal/corpusgenerator"
 	"github.com/elastic/elastic-package/internal/stack"
 
 	"github.com/google/uuid"
@@ -50,6 +50,42 @@ const (
 
 	// BenchType defining rally benchmark
 	BenchType benchrunner.Type = "rally"
+
+	rallyTrackTemplate = `{% import "rally.helpers" as rally with context %}
+{
+  "version": 2,
+  "description": "Track for [[.DataStream]]",
+  "datastream": [
+    {
+      "name": "[[.DataStream]]",
+      "body": "[[.CorpusFilename]]"
+    }
+  ],
+  "corpora": [
+    {
+      "name": "[[.CorpusFilename]]",
+      "documents": [
+        {
+          "target-data-stream": "[[.DataStream]]",
+          "source-file": "[[.CorpusFilename]]",
+          "document-count": [[.CorpusDocsCount]],
+          "uncompressed-bytes": [[.CorpusSizeInBytes]]
+        }
+      ]
+    }
+  ],
+  "schedule": [
+    {
+      "operation": {
+        "operation-type": "bulk",
+        "bulk-size": {{bulk_size | default(5000)}},
+        "ingest-percentage": {{ingest_percentage | default(100)}}
+      },
+      "clients": {{bulk_indexing_clients | default(8)}}
+    }
+  ]
+}
+`
 )
 
 var ErrDryRun = errors.New("dry run: rally benchmark not executed")
@@ -513,7 +549,7 @@ func (r *runner) runGenerator(destDir string) error {
 		return fmt.Errorf("cannot not create rally track file: %w", err)
 	}
 	r.trackFile = trackFile.Name()
-	rallyTrackContent, err := corpusgenerator.GenerateRallyTrack(r.runtimeDataStream, corpusFile, corpusDocsCount)
+	rallyTrackContent, err := generateRallyTrack(r.runtimeDataStream, corpusFile, corpusDocsCount)
 	if err != nil {
 		return fmt.Errorf("cannot not generate rally track content: %w", err)
 	}
@@ -871,3 +907,34 @@ func createRunID() string {
 func getDataStreamPath(packageRoot, dataStream string) string {
 	return filepath.Join(packageRoot, "data_stream", dataStream)
 }
+
+func generateRallyTrack(dataStream string, corpusFile *os.File, corpusDocsCount uint64) ([]byte, error) {
+	t := template.New("rallytrack")
+
+	parsedTpl, err := t.Delims("[[", "]]").Parse(rallyTrackTemplate)
+	if err != nil {
+		return nil, fmt.Errorf("error while parsing rally track template: %w", err)
+	}
+
+	fi, err := corpusFile.Stat()
+	if err != nil {
+		return nil, fmt.Errorf("error with stat on rally corpus file: %w", err)
+	}
+
+	corpusSizeInBytes := fi.Size()
+
+	buf := new(bytes.Buffer)
+	templateData := map[string]any{
+		"DataStream":        dataStream,
+		"CorpusFilename":    filepath.Base(corpusFile.Name()),
+		"CorpusDocsCount":   corpusDocsCount,
+		"CorpusSizeInBytes": corpusSizeInBytes,
+	}
+
+	err = parsedTpl.Execute(buf, templateData)
+	if err != nil {
+		return nil, fmt.Errorf("error on parsin on rally track template: %w", err)
+	}
+
+	return buf.Bytes(), nil
+}

internal/benchrunner/runners/rally/scenario.go

@@ -7,7 +7,6 @@ package rally
 import (
 	"errors"
 	"fmt"
-	"os"
 	"path/filepath"
 
 	"github.com/elastic/go-ucfg/yaml"
@@ -56,7 +55,7 @@ func readConfig(path, scenario, packageName, packageVersion string) (*scenario,
 	configPath := filepath.Join(path, devPath, fmt.Sprintf("%s.yml", scenario))
 	c := defaultConfig()
 	cfg, err := yaml.NewConfigWithFile(configPath)
-	if err != nil && !errors.Is(err, os.ErrNotExist) {
+	if err != nil {
 		return nil, fmt.Errorf("can't load benchmark configuration: %s: %w", configPath, err)
 	}
 
@@ -69,5 +68,9 @@ func readConfig(path, scenario, packageName, packageVersion string) (*scenario,
 	c.Package = packageName
 	c.Version = packageVersion
 
+	if c.DataStream.Name == "" {
+		return nil, errors.New("can't read data stream name from benchmark configuration: empty")
+	}
+
 	return c, nil
 }