Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Benchmark stream command #1584

Merged
merged 13 commits into from
Dec 8, 2023
Merged

Benchmark stream command #1584

Show file tree
Hide file tree
Changes from 12 commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
a9f3f27
benchmark stream command scaffold
Dec 5, 2023
b3a776c
benchmark stream command implementation
Dec 5, 2023
9d12b2d
docs
Dec 5, 2023
f29ef5f
TEMPORARY: use tool generator branch in order to set field in config
Dec 5, 2023
75937b4
fix bug
Dec 5, 2023
9e750a3
io.EOF is err != nil
Dec 5, 2023
6794803
ping generator v0.9.0
Dec 5, 2023
6cf9f12
cr fixes
Dec 6, 2023
537fd4b
fix flags description
Dec 6, 2023
6398ec6
add remove ES cluster docs for benchmark stream
Dec 7, 2023
52e3c59
Update cmd/benchmark.go
aspacca Dec 8, 2023
9017170
make backfill a positive duration
Dec 8, 2023
045d42b
Update cmd/benchmark.go
jsoriano Dec 8, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -159,6 +159,16 @@ These benchmarks allow you to benchmark an integration corpus with rally.

For details on how to configure rally benchmarks for a package, review the [HOWTO guide](./docs/howto/rally_benchmarking.md).

#### Stream Benchmarks

These benchmarks allow you to benchmark ingesting real time data.
You can stream data to a remote ES cluster setting the following environment variables:

ELASTIC_PACKAGE_ELASTICSEARCH_HOST=https://my-deployment.es.eu-central-1.aws.foundit.no
ELASTIC_PACKAGE_ELASTICSEARCH_USERNAME=elastic
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Do API keys work here? Serverless has mostly API keys

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

no, it doesn't https://github.com/elastic/elastic-package/blob/main/internal/stack/clients.go#L22-L25

but it should be possible to run something like:

elastic-package stack up --provider serverless
$(elastic-package stack shellinit)

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@jsoriano Seems like a missing feature in elastic-package? Should we open an issue?

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes, this is missing, please create an issue.

ELASTIC_PACKAGE_ELASTICSEARCH_PASSWORD=changeme
ELASTIC_PACKAGE_KIBANA_HOST=https://my-deployment.kb.eu-central-1.aws.foundit.no:9243

#### System Benchmarks

These benchmarks allow you to benchmark an integration end to end.
Expand All @@ -177,6 +187,12 @@ _Context: package_

Run rally benchmarks for the package (esrally needs to be installed in the path of the system).

### `elastic-package benchmark stream`

_Context: package_

Run stream benchmarks for the package.

### `elastic-package benchmark system`

_Context: package_
Expand Down
137 changes: 137 additions & 0 deletions cmd/benchmark.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ import (
benchcommon "github.com/elastic/elastic-package/internal/benchrunner/runners/common"
"github.com/elastic/elastic-package/internal/benchrunner/runners/pipeline"
"github.com/elastic/elastic-package/internal/benchrunner/runners/rally"
"github.com/elastic/elastic-package/internal/benchrunner/runners/stream"
"github.com/elastic/elastic-package/internal/benchrunner/runners/system"
"github.com/elastic/elastic-package/internal/cobraext"
"github.com/elastic/elastic-package/internal/common"
Expand All @@ -47,6 +48,16 @@ These benchmarks allow you to benchmark an integration corpus with rally.

For details on how to configure rally benchmarks for a package, review the [HOWTO guide](./docs/howto/rally_benchmarking.md).

#### Stream Benchmarks

These benchmarks allow you to benchmark ingesting real time data.
You can stream data to a remote ES cluster setting the following environment variables:

ELASTIC_PACKAGE_ELASTICSEARCH_HOST=https://my-deployment.es.eu-central-1.aws.foundit.no
ELASTIC_PACKAGE_ELASTICSEARCH_USERNAME=elastic
ELASTIC_PACKAGE_ELASTICSEARCH_PASSWORD=changeme
ELASTIC_PACKAGE_KIBANA_HOST=https://my-deployment.kb.eu-central-1.aws.foundit.no:9243

#### System Benchmarks

These benchmarks allow you to benchmark an integration end to end.
Expand All @@ -68,6 +79,9 @@ func setupBenchmarkCommand() *cobraext.Command {
rallyCmd := getRallyCommand()
cmd.AddCommand(rallyCmd)

streamCmd := getStreamCommand()
cmd.AddCommand(streamCmd)

systemCmd := getSystemCommand()
cmd.AddCommand(systemCmd)

Expand Down Expand Up @@ -385,6 +399,129 @@ func getPackageNameAndVersion(packageFromRegistry string) (string, string, error
return packageName, packageVersion, nil
}

func getStreamCommand() *cobra.Command {
cmd := &cobra.Command{
Use: "stream",
Short: "Run stream benchmarks",
Long: "Run stream benchmarks for the package",
Args: cobra.NoArgs,
RunE: streamCommandAction,
}

cmd.Flags().StringP(cobraext.BenchNameFlagName, "", "", cobraext.BenchNameFlagDescription)
cmd.Flags().String(cobraext.VariantFlagName, "", cobraext.VariantFlagDescription)
cmd.Flags().DurationP(cobraext.BenchStreamBackFillFlagName, "", 15*time.Minute, cobraext.BenchStreamBackFillFlagDescription)
cmd.Flags().Uint64P(cobraext.BenchStreamEventsPerPeriodFlagName, "", 10, cobraext.BenchStreamEventsPerPeriodFlagDescription)
cmd.Flags().DurationP(cobraext.BenchStreamPeriodDurationFlagName, "", 10*time.Second, cobraext.BenchStreamPeriodDurationFlagDescription)
cmd.Flags().BoolP(cobraext.BenchStreamPerformCleanupFlagName, "", false, cobraext.BenchStreamPerformCleanupFlagDescription)
cmd.Flags().StringP(cobraext.BenchStreamTimestampFieldFlagName, "", "timestamp", cobraext.BenchStreamTimestampFieldFlagDescription)

return cmd
}

func streamCommandAction(cmd *cobra.Command, args []string) error {
cmd.Println("Run stream benchmarks for the package")

variant, err := cmd.Flags().GetString(cobraext.VariantFlagName)
if err != nil {
return cobraext.FlagParsingError(err, cobraext.VariantFlagName)
}

benchName, err := cmd.Flags().GetString(cobraext.BenchNameFlagName)
if err != nil {
return cobraext.FlagParsingError(err, cobraext.VariantFlagName)
jsoriano marked this conversation as resolved.
Show resolved Hide resolved
}

backFill, err := cmd.Flags().GetDuration(cobraext.BenchStreamBackFillFlagName)
if err != nil {
return cobraext.FlagParsingError(err, cobraext.BenchStreamBackFillFlagName)
}

if backFill < 0 {
return cobraext.FlagParsingError(errors.New("cannot be a negative duration"), cobraext.BenchStreamBackFillFlagName)
}

eventsPerPeriod, err := cmd.Flags().GetUint64(cobraext.BenchStreamEventsPerPeriodFlagName)
if err != nil {
return cobraext.FlagParsingError(err, cobraext.BenchStreamEventsPerPeriodFlagName)
}

if eventsPerPeriod <= 0 {
return cobraext.FlagParsingError(errors.New("cannot be zero or negative"), cobraext.BenchStreamEventsPerPeriodFlagName)
}

periodDuration, err := cmd.Flags().GetDuration(cobraext.BenchStreamPeriodDurationFlagName)
if err != nil {
return cobraext.FlagParsingError(err, cobraext.BenchStreamPeriodDurationFlagName)
}

if periodDuration < time.Nanosecond {
return cobraext.FlagParsingError(errors.New("cannot be a negative duration"), cobraext.BenchStreamPeriodDurationFlagName)
}

performCleanup, err := cmd.Flags().GetBool(cobraext.BenchStreamPerformCleanupFlagName)
if err != nil {
return cobraext.FlagParsingError(err, cobraext.BenchStreamPerformCleanupFlagName)
}

timestampField, err := cmd.Flags().GetString(cobraext.BenchStreamTimestampFieldFlagName)
if err != nil {
return cobraext.FlagParsingError(err, cobraext.BenchStreamTimestampFieldFlagName)
}

packageRootPath, found, err := packages.FindPackageRoot()
if !found {
return errors.New("package root not found")
}
if err != nil {
return fmt.Errorf("locating package root failed: %w", err)
}

profile, err := cobraext.GetProfileFlag(cmd)
aspacca marked this conversation as resolved.
Show resolved Hide resolved
if err != nil {
return err
}

signal.Enable()

esClient, err := stack.NewElasticsearchClientFromProfile(profile)
if err != nil {
return fmt.Errorf("can't create Elasticsearch client: %w", err)
}
err = esClient.CheckHealth(cmd.Context())
if err != nil {
return err
}

kc, err := stack.NewKibanaClientFromProfile(profile)
aspacca marked this conversation as resolved.
Show resolved Hide resolved
if err != nil {
return fmt.Errorf("can't create Kibana client: %w", err)
}

withOpts := []stream.OptionFunc{
stream.WithVariant(variant),
stream.WithBenchmarkName(benchName),
stream.WithBackFill(backFill),
stream.WithEventsPerPeriod(eventsPerPeriod),
stream.WithPeriodDuration(periodDuration),
stream.WithPerformCleanup(performCleanup),
stream.WithTimestampField(timestampField),
stream.WithPackageRootPath(packageRootPath),
stream.WithESAPI(esClient.API),
stream.WithKibanaClient(kc),
stream.WithProfile(profile),
}

runner := stream.NewStreamBenchmark(stream.NewOptions(withOpts...))

_, err = benchrunner.Run(runner)
if err != nil {
return fmt.Errorf("error running package stream benchmarks: %w", err)
}

return nil
}

func getSystemCommand() *cobra.Command {
cmd := &cobra.Command{
Use: "system",
Expand Down
9 changes: 5 additions & 4 deletions go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@ require (
github.com/cbroglie/mustache v1.4.0
github.com/cespare/xxhash/v2 v2.2.0
github.com/dustin/go-humanize v1.0.1
github.com/elastic/elastic-integration-corpus-generator-tool v0.8.0
github.com/elastic/elastic-integration-corpus-generator-tool v0.9.0
github.com/elastic/go-elasticsearch/v7 v7.17.10
github.com/elastic/go-licenser v0.4.1
github.com/elastic/go-resource v0.1.1
Expand Down Expand Up @@ -95,6 +95,7 @@ require (
github.com/gregjones/httpcache v0.0.0-20190611155906-901d90724c79 // indirect
github.com/hashicorp/errwrap v1.1.0 // indirect
github.com/hashicorp/go-cleanhttp v0.5.2 // indirect
github.com/hashicorp/go-hclog v1.5.0 // indirect
github.com/hashicorp/go-multierror v1.1.1 // indirect
github.com/huandu/xstrings v1.4.0 // indirect
github.com/imdario/mergo v0.3.16 // indirect
Expand Down Expand Up @@ -138,7 +139,7 @@ require (
github.com/shoenig/go-m1cpu v0.1.6 // indirect
github.com/shopspring/decimal v1.3.1 // indirect
github.com/skeema/knownhosts v1.2.1 // indirect
github.com/spf13/afero v1.10.0 // indirect
github.com/spf13/afero v1.11.0 // indirect
github.com/spf13/cast v1.5.1 // indirect
github.com/spf13/pflag v1.0.5 // indirect
github.com/stretchr/objx v0.5.1 // indirect
Expand All @@ -157,12 +158,12 @@ require (
golang.org/x/exp/typeparams v0.0.0-20221208152030-732eee02a75a // indirect
golang.org/x/mod v0.14.0 // indirect
golang.org/x/net v0.19.0 // indirect
golang.org/x/oauth2 v0.12.0 // indirect
golang.org/x/oauth2 v0.15.0 // indirect
golang.org/x/sync v0.5.0 // indirect
golang.org/x/sys v0.15.0 // indirect
golang.org/x/term v0.15.0 // indirect
golang.org/x/text v0.14.0 // indirect
golang.org/x/time v0.3.0 // indirect
golang.org/x/time v0.5.0 // indirect
google.golang.org/appengine v1.6.7 // indirect
google.golang.org/protobuf v1.31.0 // indirect
gopkg.in/inf.v0 v0.9.1 // indirect
Expand Down
Loading