From a01c01610478b88b711bec78534abb4215364ce5 Mon Sep 17 00:00:00 2001 From: Zeying Zhu Date: Wed, 10 Dec 2025 16:42:33 -0500 Subject: [PATCH] add ddsketch metric type in telemetrygen --- .../cmd/telemetrygen/README.md | 127 +++ .../cmd/telemetrygen/go.mod | 76 ++ .../cmd/telemetrygen/go.sum | 121 +++ .../cmd/telemetrygen/pkg/metrics/config.go | 102 +++ .../telemetrygen/pkg/metrics/metrics_test.go | 334 ++++++++ .../telemetrygen/pkg/metrics/metrics_types.go | 39 + .../cmd/telemetrygen/pkg/metrics/worker.go | 340 ++++++++ .../telemetrygen/pkg/metrics/worker_test.go | 779 ++++++++++++++++++ .../cmd/telemetrygen/run.sh | 2 + 9 files changed, 1920 insertions(+) create mode 100644 opentelemetry-collector-contrib-patch/cmd/telemetrygen/README.md create mode 100644 opentelemetry-collector-contrib-patch/cmd/telemetrygen/go.mod create mode 100644 opentelemetry-collector-contrib-patch/cmd/telemetrygen/go.sum create mode 100644 opentelemetry-collector-contrib-patch/cmd/telemetrygen/pkg/metrics/config.go create mode 100644 opentelemetry-collector-contrib-patch/cmd/telemetrygen/pkg/metrics/metrics_test.go create mode 100644 opentelemetry-collector-contrib-patch/cmd/telemetrygen/pkg/metrics/metrics_types.go create mode 100644 opentelemetry-collector-contrib-patch/cmd/telemetrygen/pkg/metrics/worker.go create mode 100644 opentelemetry-collector-contrib-patch/cmd/telemetrygen/pkg/metrics/worker_test.go create mode 100644 opentelemetry-collector-contrib-patch/cmd/telemetrygen/run.sh diff --git a/opentelemetry-collector-contrib-patch/cmd/telemetrygen/README.md b/opentelemetry-collector-contrib-patch/cmd/telemetrygen/README.md new file mode 100644 index 00000000..6205f2ed --- /dev/null +++ b/opentelemetry-collector-contrib-patch/cmd/telemetrygen/README.md @@ -0,0 +1,127 @@ +# Telemetry generator for OpenTelemetry + + +| Status | | +| ------------- |-----------| +| Stability | [development]: metrics, logs | +| | [alpha]: traces | +| Issues | [![Open issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aopen%20label%3Acmd%2Ftelemetrygen%20&label=open&color=orange&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aopen+is%3Aissue+label%3Acmd%2Ftelemetrygen) [![Closed issues](https://img.shields.io/github/issues-search/open-telemetry/opentelemetry-collector-contrib?query=is%3Aissue%20is%3Aclosed%20label%3Acmd%2Ftelemetrygen%20&label=closed&color=blue&logo=opentelemetry)](https://github.com/open-telemetry/opentelemetry-collector-contrib/issues?q=is%3Aclosed+is%3Aissue+label%3Acmd%2Ftelemetrygen) | +| [Code Owners](https://github.com/open-telemetry/opentelemetry-collector-contrib/blob/main/CONTRIBUTING.md#becoming-a-code-owner) | [@mx-psi](https://www.github.com/mx-psi), [@codeboten](https://www.github.com/codeboten), [@Erog38](https://www.github.com/Erog38), [@bogdan-st](https://www.github.com/bogdan-st) | + +[development]: https://github.com/open-telemetry/opentelemetry-collector/blob/main/docs/component-stability.md#development +[alpha]: https://github.com/open-telemetry/opentelemetry-collector/blob/main/docs/component-stability.md#alpha + + +This utility simulates a client generating **traces**, **metrics**, and **logs**. It is useful for testing and demonstration purposes. + +## Installing + +To install the latest version run the following command: + +```console +go install github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen@latest +``` + +Check the [`go install` reference](https://go.dev/ref/mod#go-install) to install specific versions. + +### Docker build + +You can build locally the Docker image with: + +```bash +make docker-telemetrygen +``` + +Using github actions, we also push a docker image on commit to main or on release to this [Github docker registry](https://github.com/orgs/open-telemetry/packages/container/package/opentelemetry-collector-contrib%2Ftelemetrygen). + +## Running + +First, you'll need an OpenTelemetry Collector to receive the telemetry data. Follow the project's instructions for a detailed setting up guide. The following configuration file should be sufficient: + +config.yaml: +```yaml +receivers: + otlp: + protocols: + grpc: + endpoint: 0.0.0.0:4317 + +exporters: + debug: + verbosity: detailed + +service: + pipelines: + logs: + receivers: [otlp] + exporters: [debug] + metrics: + receivers: [otlp] + exporters: [debug] + traces: + receivers: [otlp] + exporters: [debug] +``` + +Starting OpenTelemetry collector via docker: +``` +docker run -p 4317:4317 -v $(pwd)/config.yaml:/etc/otelcol-contrib/config.yaml ghcr.io/open-telemetry/opentelemetry-collector-releases/opentelemetry-collector-contrib:0.86.0 +``` + +Other options for running the collector are documented here https://opentelemetry.io/docs/collector/getting-started/ + +Once the OpenTelemetry Collector instance is up and running, run `telemetrygen` for your desired telemetry: + +### Traces + +```console +telemetrygen traces --otlp-insecure --duration 5s +``` + +Or, to generate a specific number of traces: + +```console +telemetrygen traces --otlp-insecure --traces 1 +``` + +Or, to generate traces continuously every 10 seconds: + +```console +telemetrygen traces --otlp-insecure --continuous --rate 0.1 +``` + +To send traces in secure connection, see [examples/secure-tracing](../../examples/secure-tracing/) + +Check `telemetrygen traces --help` for all the options. + +### Logs + +```console +telemetrygen logs --duration 5s --otlp-insecure +``` + +### Metrics + +```console +telemetrygen metrics --duration 5s --otlp-insecure +``` + +Generate DDSketch-encoded metrics: + +```console +./telemetrygen metrics \ + --workers 1 \ + --duration 5m \ + --metric-type DDSketch \ + --unique-timeseries \ + --unique-timeseries-duration 5m \ + --otlp-endpoint localhost:4317 \ + --otlp-insecure + +# or + +go run . metrics --metric-type DDSketch --duration 5s --otlp-insecure +``` + +Setting --rate 0 disables throttling entirely. +Use --unique-timeseries together with --unique-timeseries-duration to control how many distinct metric series are produced within a window. diff --git a/opentelemetry-collector-contrib-patch/cmd/telemetrygen/go.mod b/opentelemetry-collector-contrib-patch/cmd/telemetrygen/go.mod new file mode 100644 index 00000000..63d04b4d --- /dev/null +++ b/opentelemetry-collector-contrib-patch/cmd/telemetrygen/go.mod @@ -0,0 +1,76 @@ +module github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen + +go 1.24.0 + +require ( + github.com/DataDog/sketches-go v1.4.7 + github.com/lightstep/go-expohisto v1.0.0 + github.com/spf13/cobra v1.10.1 + github.com/spf13/pflag v1.0.10 + github.com/stretchr/testify v1.11.1 + go.opentelemetry.io/collector/pdata v1.47.0 + go.opentelemetry.io/otel v1.38.0 + go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc v0.14.0 + go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.14.0 + go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.38.0 + go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.38.0 + go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.38.0 + go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.38.0 + go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.38.0 + go.opentelemetry.io/otel/log v0.14.0 + go.opentelemetry.io/otel/sdk v1.38.0 + go.opentelemetry.io/otel/sdk/log v0.14.0 + go.opentelemetry.io/otel/sdk/log/logtest v0.14.0 + go.opentelemetry.io/otel/sdk/metric v1.38.0 + go.opentelemetry.io/otel/trace v1.38.0 + go.opentelemetry.io/proto/otlp v1.9.0 + go.uber.org/goleak v1.3.0 + go.uber.org/zap v1.27.1 + golang.org/x/time v0.13.0 + google.golang.org/grpc v1.77.0 + google.golang.org/protobuf v1.36.10 +) + +require ( + github.com/cenkalti/backoff/v5 v5.0.3 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect + github.com/davecgh/go-spew v1.1.1 // indirect + github.com/go-logr/logr v1.4.3 // indirect + github.com/go-logr/stdr v1.2.2 // indirect + github.com/google/uuid v1.6.0 // indirect + github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.3 // indirect + github.com/hashicorp/go-version v1.7.0 // indirect + github.com/inconshreveable/mousetrap v1.1.0 // indirect + github.com/json-iterator/go v1.1.12 // indirect + github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect + github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect + github.com/pmezard/go-difflib v1.0.0 // indirect + go.opentelemetry.io/auto/sdk v1.2.1 // indirect + go.opentelemetry.io/collector/featuregate v1.47.0 // indirect + go.opentelemetry.io/otel/metric v1.38.0 // indirect + go.uber.org/multierr v1.11.0 // indirect + golang.org/x/net v0.47.0 // indirect + golang.org/x/sys v0.38.0 // indirect + golang.org/x/text v0.31.0 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20251202230838-ff82c1b0f217 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217 // indirect + gopkg.in/yaml.v3 v3.0.1 // indirect +) + +retract ( + v0.76.2 + v0.76.1 + v0.65.0 +) + +replace ( + go.opentelemetry.io/otel => ../../../opentelemetry-go + go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc => ../../../opentelemetry-go/exporters/otlp/otlpmetric/otlpmetricgrpc + go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp => ../../../opentelemetry-go/exporters/otlp/otlpmetric/otlpmetrichttp + go.opentelemetry.io/otel/sdk => ../../../opentelemetry-go/sdk + go.opentelemetry.io/otel/sdk/metric => ../../../opentelemetry-go/sdk/metric + go.opentelemetry.io/proto/otlp => ../../../opentelemetry-proto/gen/go/go.opentelemetry.io/proto/otlp +) + +// IMPORTANT NOTE: Do not add replace statements to this go.mod. This will break go install. +// See https://github.com/open-telemetry/opentelemetry-collector-contrib/issues/27855. diff --git a/opentelemetry-collector-contrib-patch/cmd/telemetrygen/go.sum b/opentelemetry-collector-contrib-patch/cmd/telemetrygen/go.sum new file mode 100644 index 00000000..01abc657 --- /dev/null +++ b/opentelemetry-collector-contrib-patch/cmd/telemetrygen/go.sum @@ -0,0 +1,121 @@ +github.com/DataDog/sketches-go v1.4.7 h1:eHs5/0i2Sdf20Zkj0udVFWuCrXGRFig2Dcfm5rtcTxc= +github.com/DataDog/sketches-go v1.4.7/go.mod h1:eAmQ/EBmtSO+nQp7IZMZVRPT4BQTmIc5RZQ+deGlTPM= +github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM= +github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= +github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= +github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= +github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= +github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= +github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= +github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= +github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= +github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= +github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/google/gofuzz v1.2.0 h1:xRy4A+RhZaiKjJ1bPfwQ8sedCA+YS2YcCHW6ec7JMi0= +github.com/google/gofuzz v1.2.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg= +github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= +github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.3 h1:NmZ1PKzSTQbuGHw9DGPFomqkkLWMC+vZCkfs+FHv1Vg= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.27.3/go.mod h1:zQrxl1YP88HQlA6i9c63DSVPFklWpGX4OWAc9bFuaH4= +github.com/hashicorp/go-version v1.7.0 h1:5tqGy27NaOTB8yJKUZELlFAS/LTKJkrmONwQKeRZfjY= +github.com/hashicorp/go-version v1.7.0/go.mod h1:fltr4n8CU8Ke44wwGCBoEymUuxUHl09ZGVZPK5anwXA= +github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= +github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= +github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM= +github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +github.com/lightstep/go-expohisto v1.0.0 h1:UPtTS1rGdtehbbAF7o/dhkWLTDI73UifG8LbfQI7cA4= +github.com/lightstep/go-expohisto v1.0.0/go.mod h1:xDXD0++Mu2FOaItXtdDfksfgxfV0z1TMPa+e/EUd0cs= +github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= +github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= +github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee h1:W5t00kpgFdJifH4BDsTlE89Zl93FEloxaWZfGcifgq8= +github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= +github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/rogpeppe/go-internal v1.14.1 h1:UQB4HGPB6osV0SQTLymcB4TgvyWu6ZyliaW0tI/otEQ= +github.com/rogpeppe/go-internal v1.14.1/go.mod h1:MaRKkUm5W0goXpeCfT7UZI6fk/L7L7so1lCWt35ZSgc= +github.com/russross/blackfriday/v2 v2.1.0/go.mod h1:+Rmxgy9KzJVeS9/2gXHxylqXiyQDYRxCVz55jmeOWTM= +github.com/spf13/cobra v1.10.1 h1:lJeBwCfmrnXthfAupyUTzJ/J4Nc1RsHC/mSRU2dll/s= +github.com/spf13/cobra v1.10.1/go.mod h1:7SmJGaTHFVBY0jW4NXGluQoLvhqFQM+6XSKD+P4XaB0= +github.com/spf13/pflag v1.0.9/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= +github.com/spf13/pflag v1.0.10 h1:4EBh2KAYBwaONj6b2Ye1GiHfwjqyROoF4RwYO+vPwFk= +github.com/spf13/pflag v1.0.10/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg= +github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME= +github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= +go.opentelemetry.io/collector/featuregate v1.47.0 h1:LuJnDngViDzPKds5QOGxVYNL1QCCVWN/m61lHTV8Pf4= +go.opentelemetry.io/collector/featuregate v1.47.0/go.mod h1:d0tiRzVYrytB6LkcYgz2ESFTv7OktRPQe0QEQcPt1L4= +go.opentelemetry.io/collector/internal/testutil v0.141.0 h1:/rUGApojPtUPMN3rFfApNgEjAt03rCGt2qxNxGGs/4A= +go.opentelemetry.io/collector/internal/testutil v0.141.0/go.mod h1:YAD9EAkwh/l5asZNbEBEUCqEjoL1OKMjAMoPjPqH76c= +go.opentelemetry.io/collector/pdata v1.47.0 h1:4Mk0mo2RlKCUPomV8ISm+Yx/STFtuSn88yjiCePHkGA= +go.opentelemetry.io/collector/pdata v1.47.0/go.mod h1:yMdjdWZBNA8wLFCQXOCLb0RfcpZOxp7exH+bN7udWO0= +go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc v0.14.0 h1:OMqPldHt79PqWKOMYIAQs3CxAi7RLgPxwfFSwr4ZxtM= +go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploggrpc v0.14.0/go.mod h1:1biG4qiqTxKiUCtoWDPpL3fB3KxVwCiGw81j3nKMuHE= +go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.14.0 h1:QQqYw3lkrzwVsoEX0w//EhH/TCnpRdEenKBOOEIMjWc= +go.opentelemetry.io/otel/exporters/otlp/otlplog/otlploghttp v0.14.0/go.mod h1:gSVQcr17jk2ig4jqJ2DX30IdWH251JcNAecvrqTxH1s= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.38.0 h1:GqRJVj7UmLjCVyVJ3ZFLdPRmhDUp2zFmQe3RHIOsw24= +go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.38.0/go.mod h1:ri3aaHSmCTVYu2AWv44YMauwAQc0aqI9gHKIcSbI1pU= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.38.0 h1:lwI4Dc5leUqENgGuQImwLo4WnuXFPetmPpkLi2IrX54= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.38.0/go.mod h1:Kz/oCE7z5wuyhPxsXDuaPteSWqjSBD5YaSdbxZYGbGk= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.38.0 h1:aTL7F04bJHUlztTsNGJ2l+6he8c+y/b//eR0jjjemT4= +go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp v1.38.0/go.mod h1:kldtb7jDTeol0l3ewcmd8SDvx3EmIE7lyvqbasU3QC4= +go.opentelemetry.io/otel/log v0.14.0 h1:2rzJ+pOAZ8qmZ3DDHg73NEKzSZkhkGIua9gXtxNGgrM= +go.opentelemetry.io/otel/log v0.14.0/go.mod h1:5jRG92fEAgx0SU/vFPxmJvhIuDU9E1SUnEQrMlJpOno= +go.opentelemetry.io/otel/metric v1.38.0 h1:Kl6lzIYGAh5M159u9NgiRkmoMKjvbsKtYRwgfrA6WpA= +go.opentelemetry.io/otel/metric v1.38.0/go.mod h1:kB5n/QoRM8YwmUahxvI3bO34eVtQf2i4utNVLr9gEmI= +go.opentelemetry.io/otel/sdk/log v0.14.0 h1:JU/U3O7N6fsAXj0+CXz21Czg532dW2V4gG1HE/e8Zrg= +go.opentelemetry.io/otel/sdk/log v0.14.0/go.mod h1:imQvII+0ZylXfKU7/wtOND8Hn4OpT3YUoIgqJVksUkM= +go.opentelemetry.io/otel/sdk/log/logtest v0.14.0 h1:Ijbtz+JKXl8T2MngiwqBlPaHqc4YCaP/i13Qrow6gAM= +go.opentelemetry.io/otel/sdk/log/logtest v0.14.0/go.mod h1:dCU8aEL6q+L9cYTqcVOk8rM9Tp8WdnHOPLiBgp0SGOA= +go.opentelemetry.io/otel/trace v1.38.0 h1:Fxk5bKrDZJUH+AMyyIXGcFAPah0oRcT+LuNtJrmcNLE= +go.opentelemetry.io/otel/trace v1.38.0/go.mod h1:j1P9ivuFsTceSWe1oY+EeW3sc+Pp42sO++GHkg4wwhs= +go.opentelemetry.io/proto/slim/otlp v1.9.0 h1:fPVMv8tP3TrsqlkH1HWYUpbCY9cAIemx184VGkS6vlE= +go.opentelemetry.io/proto/slim/otlp v1.9.0/go.mod h1:xXdeJJ90Gqyll+orzUkY4bOd2HECo5JofeoLpymVqdI= +go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.2.0 h1:o13nadWDNkH/quoDomDUClnQBpdQQ2Qqv0lQBjIXjE8= +go.opentelemetry.io/proto/slim/otlp/collector/profiles/v1development v0.2.0/go.mod h1:Gyb6Xe7FTi/6xBHwMmngGoHqL0w29Y4eW8TGFzpefGA= +go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.2.0 h1:EiUYvtwu6PMrMHVjcPfnsG3v+ajPkbUeH+IL93+QYyk= +go.opentelemetry.io/proto/slim/otlp/profiles/v1development v0.2.0/go.mod h1:mUUHKFiN2SST3AhJ8XhJxEoeVW12oqfXog0Bo8W3Ec4= +go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= +go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= +go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0= +go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y= +go.uber.org/zap v1.27.1 h1:08RqriUEv8+ArZRYSTXy1LeBScaMpVSTBhCeaZYfMYc= +go.uber.org/zap v1.27.1/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E= +golang.org/x/net v0.47.0 h1:Mx+4dIFzqraBXUugkia1OOvlD6LemFo1ALMHjrXDOhY= +golang.org/x/net v0.47.0/go.mod h1:/jNxtkgq5yWUGYkaZGqo27cfGZ1c5Nen03aYrrKpVRU= +golang.org/x/sys v0.38.0 h1:3yZWxaJjBmCWXqhN1qh02AkOnCQ1poK6oF+a7xWL6Gc= +golang.org/x/sys v0.38.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= +golang.org/x/text v0.31.0 h1:aC8ghyu4JhP8VojJ2lEHBnochRno1sgL6nEi9WGFGMM= +golang.org/x/text v0.31.0/go.mod h1:tKRAlv61yKIjGGHX/4tP1LTbc13YSec1pxVEWXzfoeM= +golang.org/x/time v0.13.0 h1:eUlYslOIt32DgYD6utsuUeHs4d7AsEYLuIAdg7FlYgI= +golang.org/x/time v0.13.0/go.mod h1:eL/Oa2bBBK0TkX57Fyni+NgnyQQN4LitPmob2Hjnqw4= +gonum.org/v1/gonum v0.16.0 h1:5+ul4Swaf3ESvrOnidPp4GZbzf0mxVQpDCYUQE7OJfk= +gonum.org/v1/gonum v0.16.0/go.mod h1:fef3am4MQ93R2HHpKnLk4/Tbh/s0+wqD5nfa6Pnwy4E= +google.golang.org/genproto/googleapis/api v0.0.0-20251202230838-ff82c1b0f217 h1:fCvbg86sFXwdrl5LgVcTEvNC+2txB5mgROGmRL5mrls= +google.golang.org/genproto/googleapis/api v0.0.0-20251202230838-ff82c1b0f217/go.mod h1:+rXWjjaukWZun3mLfjmVnQi18E1AsFbDN9QdJ5YXLto= +google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217 h1:gRkg/vSppuSQoDjxyiGfN4Upv/h/DQmIR10ZU8dh4Ww= +google.golang.org/genproto/googleapis/rpc v0.0.0-20251202230838-ff82c1b0f217/go.mod h1:7i2o+ce6H/6BluujYR+kqX3GKH+dChPTQU19wjRPiGk= +google.golang.org/grpc v1.77.0 h1:wVVY6/8cGA6vvffn+wWK5ToddbgdU3d8MNENr4evgXM= +google.golang.org/grpc v1.77.0/go.mod h1:z0BY1iVj0q8E1uSQCjL9cppRj+gnZjzDnzV0dHhrNig= +google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE= +google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= +gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= +gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= +gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/opentelemetry-collector-contrib-patch/cmd/telemetrygen/pkg/metrics/config.go b/opentelemetry-collector-contrib-patch/cmd/telemetrygen/pkg/metrics/config.go new file mode 100644 index 00000000..e254988b --- /dev/null +++ b/opentelemetry-collector-contrib-patch/cmd/telemetrygen/pkg/metrics/config.go @@ -0,0 +1,102 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +package metrics + +import ( + "errors" + "fmt" + "time" + + "github.com/spf13/pflag" + "go.opentelemetry.io/otel/sdk/metric/metricdata" + + "github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen/internal/config" + "github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen/internal/validate" + types "github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen/pkg" +) + +// Config describes the test scenario. +type Config struct { + config.Config + NumMetrics int + MetricName string + MetricType MetricType + AggregationTemporality AggregationTemporality + SpanID string + TraceID string + EnforceUniqueTimeseries bool + UniqueTimelimit time.Duration +} + +// NewConfig creates a new Config with default values. +func NewConfig() *Config { + cfg := &Config{} + cfg.SetDefaults() + return cfg +} + +// Flags registers config flags. +func (c *Config) Flags(fs *pflag.FlagSet) { + c.CommonFlags(fs) + + fs.StringVar(&c.HTTPPath, "otlp-http-url-path", c.HTTPPath, "Which URL path to write to") + + fs.IntVar(&c.NumMetrics, "metrics", c.NumMetrics, "Number of metrics to generate in each worker (ignored if duration is provided)") + fs.StringVar(&c.MetricName, "otlp-metric-name", c.MetricName, "Metric name of the exported metric") + + fs.StringVar(&c.TraceID, "trace-id", c.TraceID, "TraceID to use as exemplar") + fs.StringVar(&c.SpanID, "span-id", c.SpanID, "SpanID to use as exemplar") + + fs.Var(&c.MetricType, "metric-type", "Metric type enum. must be one of 'Gauge', 'Sum', 'Histogram', 'ExponentialHistogram', or 'DDSketch'") + fs.Var(&c.AggregationTemporality, "aggregation-temporality", "aggregation-temporality for metrics. Must be one of 'delta' or 'cumulative'") + fs.BoolVar(&c.EnforceUniqueTimeseries, "unique-timeseries", c.EnforceUniqueTimeseries, "Enforce unique timeseries within unique-timeseries-timelimit, performance impacting") + fs.DurationVar(&c.UniqueTimelimit, "unique-timeseries-duration", c.UniqueTimelimit, "Time limit for unique timeseries generation, timeseries generated within this time will be unique") +} + +// SetDefaults sets the default values for the configuration +// This is called before parsing the command line flags and when +// calling NewConfig() +func (c *Config) SetDefaults() { + c.Config.SetDefaults() + c.HTTPPath = "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/v1/metrics" + c.Rate = 1 + c.TotalDuration = types.DurationWithInf(0) + + c.MetricName = "gen" + // Use Gauge as default metric type. + c.MetricType = MetricTypeGauge + // Use cumulative temporality as default. + c.AggregationTemporality = AggregationTemporality(metricdata.CumulativeTemporality) + + c.EnforceUniqueTimeseries = false + c.UniqueTimelimit = time.Second + + c.TraceID = "" + c.SpanID = "" +} + +// Validate validates the test scenario parameters. +func (c *Config) Validate() error { + if !c.TotalDuration.IsInf() && c.TotalDuration.Duration() <= 0 && c.NumMetrics <= 0 { + return errors.New("either `metrics` or `duration` must be greater than 0") + } + + if c.LoadSize < 0 { + return fmt.Errorf("load size must be non-negative, found %d", c.LoadSize) + } + + if c.TraceID != "" { + if err := validate.TraceID(c.TraceID); err != nil { + return err + } + } + + if c.SpanID != "" { + if err := validate.SpanID(c.SpanID); err != nil { + return err + } + } + + return nil +} diff --git a/opentelemetry-collector-contrib-patch/cmd/telemetrygen/pkg/metrics/metrics_test.go b/opentelemetry-collector-contrib-patch/cmd/telemetrygen/pkg/metrics/metrics_test.go new file mode 100644 index 00000000..1af28788 --- /dev/null +++ b/opentelemetry-collector-contrib-patch/cmd/telemetrygen/pkg/metrics/metrics_test.go @@ -0,0 +1,334 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +package metrics + +import ( + "encoding/hex" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/sdk/metric/metricdata" + "go.opentelemetry.io/otel/sdk/metric/metricdata/metricdatatest" + + "github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen/internal/config" + types "github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen/pkg" +) + +// TestDurationAndMetricsInteraction tests the interaction between duration and metrics parameters +func TestDurationAndMetricsInteraction(t *testing.T) { + tests := []struct { + name string + config Config + expectedMetrics int + description string + }{ + { + name: "Default behavior - respects metrics parameter", + config: Config{ + Config: config.Config{ + WorkerCount: 1, + }, + NumMetrics: 3, + }, + expectedMetrics: 3, + description: "By default, TotalDuration is 0, so NumMetrics should be respected", + }, + { + name: "Finite duration overrides metrics", + config: Config{ + Config: config.Config{ + WorkerCount: 1, + TotalDuration: types.DurationWithInf(100 * time.Millisecond), + }, + NumMetrics: 100, + }, + expectedMetrics: 0, + description: "Finite duration should override NumMetrics (set to 0)", + }, + { + name: "Infinite duration overrides metrics", + config: Config{ + Config: config.Config{ + WorkerCount: 1, + TotalDuration: types.MustDurationWithInf("Inf"), + }, + NumMetrics: 50, + }, + expectedMetrics: 0, + description: "Infinite duration should override NumMetrics (set to 0)", + }, + { + name: "Zero duration with metrics", + config: Config{ + Config: config.Config{ + WorkerCount: 1, + TotalDuration: types.DurationWithInf(0), + }, + NumMetrics: 5, + }, + expectedMetrics: 5, + description: "Zero duration should not override NumMetrics", + }, + { + name: "Negative duration with metrics", + config: Config{ + Config: config.Config{ + WorkerCount: 1, + TotalDuration: types.DurationWithInf(-100 * time.Millisecond), + }, + NumMetrics: 10, + }, + expectedMetrics: 10, + description: "Negative duration should not override NumMetrics", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + cfg := tt.config + + if cfg.TotalDuration.Duration() > 0 || cfg.TotalDuration.IsInf() { + cfg.NumMetrics = 0 + } + + assert.Equal(t, tt.expectedMetrics, cfg.NumMetrics, tt.description) + }) + } +} + +// TestDefaultConfiguration tests that the default configuration is correct +func TestDefaultConfiguration(t *testing.T) { + cfg := NewConfig() + + assert.Equal(t, types.DurationWithInf(0), cfg.TotalDuration, "Default TotalDuration should be 0") + assert.Equal(t, 0, cfg.NumMetrics, "Default NumMetrics should be 0") + assert.Equal(t, float64(1), cfg.Rate, "Default Rate should be 1") + assert.Equal(t, "gen", cfg.MetricName, "Default MetricName should be 'gen'") + assert.Equal(t, MetricTypeGauge, cfg.MetricType, "Default MetricType should be Gauge") +} + +// TestConfigValidation tests the validation logic +func TestConfigValidation(t *testing.T) { + tests := []struct { + name string + config Config + expectError bool + description string + }{ + { + name: "Valid config with metrics", + config: Config{ + Config: config.Config{ + WorkerCount: 1, + }, + NumMetrics: 5, + }, + expectError: false, + description: "Config with NumMetrics > 0 should be valid", + }, + { + name: "Valid config with finite duration", + config: Config{ + Config: config.Config{ + WorkerCount: 1, + TotalDuration: types.DurationWithInf(1 * time.Second), + }, + NumMetrics: 0, + }, + expectError: false, + description: "Config with finite duration > 0 should be valid", + }, + { + name: "Valid config with infinite duration", + config: Config{ + Config: config.Config{ + WorkerCount: 1, + TotalDuration: types.MustDurationWithInf("Inf"), + }, + NumMetrics: 0, + }, + expectError: false, + description: "Config with infinite duration should be valid", + }, + { + name: "Invalid config - no metrics and no duration", + config: Config{ + Config: config.Config{ + WorkerCount: 1, + }, + NumMetrics: 0, + }, + expectError: true, + description: "Config with no metrics and no duration should be invalid", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + err := tt.config.Validate() + if tt.expectError { + assert.Error(t, err, tt.description) + } else { + assert.NoError(t, err, tt.description) + } + }) + } +} + +// TestWorkerBehavior tests that workers behave correctly with different configurations +func TestWorkerBehavior(t *testing.T) { + tests := []struct { + name string + config Config + expectedMetrics int + description string + }{ + { + name: "Worker with finite metrics and no duration", + config: Config{ + Config: config.Config{ + WorkerCount: 1, + }, + NumMetrics: 2, + }, + expectedMetrics: 2, + description: "Worker should generate exactly the specified number of metrics", + }, + { + name: "Worker with infinite duration", + config: Config{ + Config: config.Config{ + WorkerCount: 1, + TotalDuration: types.MustDurationWithInf("Inf"), + }, + NumMetrics: 0, // This will be set by the run logic + }, + expectedMetrics: 0, + description: "Worker with infinite duration should have NumMetrics set to 0", + }, + { + name: "Worker with finite duration", + config: Config{ + Config: config.Config{ + WorkerCount: 1, + TotalDuration: types.DurationWithInf(100 * time.Millisecond), + }, + NumMetrics: 10, // This will be set to 0 by the run logic + }, + expectedMetrics: 0, + description: "Worker with finite duration should have NumMetrics set to 0", + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + if tt.config.TotalDuration.Duration() > 0 || tt.config.TotalDuration.IsInf() { + tt.config.NumMetrics = 0 + } + + assert.Equal(t, tt.expectedMetrics, tt.config.NumMetrics, tt.description) + }) + } +} + +func Test_exemplarsFromConfig(t *testing.T) { + traceID, err := hex.DecodeString("ae87dadd90e9935a4bc9660628efd569") + require.NoError(t, err) + + spanID, err := hex.DecodeString("5828fa4960140870") + require.NoError(t, err) + + tests := []struct { + name string + c *Config + validateFunc func(t *testing.T, got []metricdata.Exemplar[int64]) + }{ + { + name: "no exemplars", + c: &Config{}, + validateFunc: func(t *testing.T, got []metricdata.Exemplar[int64]) { + assert.Nil(t, got) + }, + }, + { + name: "both-traceID-and-spanID", + c: &Config{ + TraceID: "ae87dadd90e9935a4bc9660628efd569", + SpanID: "5828fa4960140870", + }, + validateFunc: func(t *testing.T, got []metricdata.Exemplar[int64]) { + require.Len(t, got, 1) + metricdatatest.AssertEqual[metricdata.Exemplar[int64]](t, got[0], metricdata.Exemplar[int64]{ + TraceID: traceID, + SpanID: spanID, + }, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreValue()) + }, + }, + { + name: "only-traceID", + c: &Config{ + TraceID: "ae87dadd90e9935a4bc9660628efd569", + }, + validateFunc: func(t *testing.T, got []metricdata.Exemplar[int64]) { + require.Len(t, got, 1) + metricdatatest.AssertEqual[metricdata.Exemplar[int64]](t, got[0], metricdata.Exemplar[int64]{ + TraceID: traceID, + SpanID: nil, + }, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreValue()) + }, + }, + { + name: "only-spanID", + c: &Config{ + SpanID: "5828fa4960140870", + }, + validateFunc: func(t *testing.T, got []metricdata.Exemplar[int64]) { + require.Len(t, got, 1) + metricdatatest.AssertEqual[metricdata.Exemplar[int64]](t, got[0], metricdata.Exemplar[int64]{ + TraceID: nil, + SpanID: spanID, + }, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreValue()) + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + tt.validateFunc(t, exemplarsFromConfig(tt.c)) + }) + } +} + +func TestNewMetricTypes(t *testing.T) { + tests := []struct { + name string + metricType MetricType + validateFunc func(t *testing.T, config Config) + }{ + { + name: "ExponentialHistogram metric type", + metricType: MetricTypeExponentialHistogram, + validateFunc: func(t *testing.T, config Config) { + assert.Equal(t, MetricTypeExponentialHistogram, config.MetricType) + }, + }, + { + name: "DDSketch metric type", + metricType: MetricTypeDDSketch, + validateFunc: func(t *testing.T, config Config) { + assert.Equal(t, MetricTypeDDSketch, config.MetricType) + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + config := Config{ + MetricType: tt.metricType, + } + tt.validateFunc(t, config) + }) + } +} diff --git a/opentelemetry-collector-contrib-patch/cmd/telemetrygen/pkg/metrics/metrics_types.go b/opentelemetry-collector-contrib-patch/cmd/telemetrygen/pkg/metrics/metrics_types.go new file mode 100644 index 00000000..4bab6bb9 --- /dev/null +++ b/opentelemetry-collector-contrib-patch/cmd/telemetrygen/pkg/metrics/metrics_types.go @@ -0,0 +1,39 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +package metrics + +import ( + "errors" +) + +type MetricType string + +const ( + MetricTypeGauge MetricType = "Gauge" + MetricTypeSum MetricType = "Sum" + MetricTypeHistogram MetricType = "Histogram" + MetricTypeExponentialHistogram MetricType = "ExponentialHistogram" + MetricTypeDDSketch MetricType = "DDSketch" +) + +// String is used both by fmt.Print and by Cobra in help text +func (e *MetricType) String() string { + return string(*e) +} + +// Set must have pointer receiver so it doesn't change the value of a copy +func (e *MetricType) Set(v string) error { + switch v { + case "Gauge", "Sum", "Histogram", "ExponentialHistogram", "DDSketch": + *e = MetricType(v) + return nil + default: + return errors.New(`must be one of "Gauge", "Sum", "Histogram", "ExponentialHistogram", "DDSketch"`) + } +} + +// Type is only used in help text +func (*MetricType) Type() string { + return "MetricType" +} diff --git a/opentelemetry-collector-contrib-patch/cmd/telemetrygen/pkg/metrics/worker.go b/opentelemetry-collector-contrib-patch/cmd/telemetrygen/pkg/metrics/worker.go new file mode 100644 index 00000000..6259d451 --- /dev/null +++ b/opentelemetry-collector-contrib-patch/cmd/telemetrygen/pkg/metrics/worker.go @@ -0,0 +1,340 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +package metrics + +import ( + "context" + "fmt" + "math" + "math/rand/v2" + "sync" + "sync/atomic" + "time" + + "github.com/DataDog/sketches-go/ddsketch" + "github.com/lightstep/go-expohisto/structure" + "go.opentelemetry.io/otel/attribute" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" + "go.opentelemetry.io/otel/sdk/resource" + "go.uber.org/zap" + "golang.org/x/time/rate" + "google.golang.org/protobuf/proto" + + "github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen/internal/config" + types "github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen/pkg" +) + +type worker struct { + running *atomic.Bool // pointer to shared flag that indicates it's time to stop the test + metricName string // name of metric to generate + metricType MetricType // type of metric to generate + aggregationTemporality AggregationTemporality // Temporality type to use + exemplars []metricdata.Exemplar[int64] // exemplars to attach to the metric + numMetrics int // how many metrics the worker has to generate (only when duration==0) + enforceUnique bool // if true, the worker will generate unique timeseries + totalDuration types.DurationWithInf // how long to run the test for (overrides `numMetrics`) + limitPerSecond rate.Limit // how many metrics per second to generate + wg *sync.WaitGroup // notify when done + logger *zap.Logger // logger + index int // worker index + clock Clock // clock + loadSize int // desired minimum size in MB of string data for each generated metric + allowFailures bool // whether to continue on export failures + rand *rand.Rand // random number generator for exponential histogram generation +} + +// We use a 15-element bounds slice for histograms below, so there must be 16 buckets here. +// From metrics.proto: +// The number of elements in bucket_counts array must be by one greater than +// the number of elements in explicit_bounds array. +var histogramBucketSamples = []struct { + bucketCounts []uint64 + sum int64 +}{ + { + []uint64{0, 0, 1, 0, 0, 0, 3, 4, 1, 1, 0, 0, 0, 0, 0, 0}, + 3940, + }, + { + []uint64{0, 0, 0, 0, 0, 0, 2, 4, 4, 0, 0, 0, 0, 0, 0, 0}, + 4455, + }, + { + []uint64{0, 0, 0, 0, 0, 0, 1, 4, 3, 2, 0, 0, 0, 0, 0, 0}, + 5337, + }, + { + []uint64{0, 0, 1, 0, 1, 0, 2, 2, 1, 3, 0, 0, 0, 0, 0, 0}, + 4477, + }, + { + []uint64{0, 0, 0, 0, 0, 1, 3, 2, 2, 2, 0, 0, 0, 0, 0, 0}, + 4670, + }, + { + []uint64{0, 0, 0, 1, 1, 0, 1, 1, 1, 5, 0, 0, 0, 0, 0, 0}, + 5670, + }, + { + []uint64{0, 0, 0, 0, 0, 2, 1, 1, 4, 2, 0, 0, 0, 0, 0, 0}, + 5091, + }, + { + []uint64{0, 0, 2, 0, 0, 0, 2, 4, 1, 1, 0, 0, 0, 0, 0, 0}, + 3420, + }, + { + []uint64{0, 0, 0, 0, 0, 0, 1, 3, 2, 4, 0, 0, 0, 0, 0, 0}, + 5917, + }, + { + []uint64{0, 0, 1, 0, 1, 0, 0, 4, 4, 0, 0, 0, 0, 0, 0, 0}, + 3988, + }, +} + +var ddsketchValueSamples = [][]float64{ + {120, 135, 150, 160, 175, 190, 205, 220, 235, 250}, + {80, 95, 110, 130, 155, 180, 210, 240, 270, 300}, + {60, 75, 90, 105, 135, 165, 195, 225, 255, 285}, + {200, 215, 230, 245, 260, 275, 290, 305, 320, 335}, +} + +func (w worker) simulateMetrics(res *resource.Resource, exporter sdkmetric.Exporter, signalAttrs []attribute.KeyValue, tb *timeBox) { + limiter := rate.NewLimiter(w.limitPerSecond, 1) + + startTime := w.clock.Now() + + var i int64 + for w.running.Load() { + if w.enforceUnique { + signalAttrs = append(signalAttrs, tb.getAttribute()) + } + + // Add load size attributes if specified + loadAttrs := signalAttrs + if w.loadSize > 0 { + for j := 0; j < w.loadSize; j++ { + loadAttrs = append(loadAttrs, config.CreateLoadAttribute(fmt.Sprintf("load-%v", j), 1)) + } + } + var metrics []metricdata.Metrics + now := w.clock.Now() + if w.aggregationTemporality.AsTemporality() == metricdata.DeltaTemporality { + startTime = now.Add(-1 * time.Second) + } + + switch w.metricType { + case MetricTypeGauge: + metrics = append(metrics, metricdata.Metrics{ + Name: w.metricName, + Data: metricdata.Gauge[int64]{ + DataPoints: []metricdata.DataPoint[int64]{ + { + Time: now, + Value: i, + Attributes: attribute.NewSet(loadAttrs...), + Exemplars: w.exemplars, + }, + }, + }, + }) + case MetricTypeSum: + metrics = append(metrics, metricdata.Metrics{ + Name: w.metricName, + Data: metricdata.Sum[int64]{ + IsMonotonic: true, + Temporality: w.aggregationTemporality.AsTemporality(), + DataPoints: []metricdata.DataPoint[int64]{ + { + StartTime: startTime, + Time: now, + Value: i, + Attributes: attribute.NewSet(loadAttrs...), + Exemplars: w.exemplars, + }, + }, + }, + }) + case MetricTypeHistogram: + var totalCount uint64 + iteration := uint64(i) % 10 + sum := histogramBucketSamples[iteration].sum + bucketCounts := histogramBucketSamples[iteration].bucketCounts + for _, count := range bucketCounts { + totalCount += count + } + metrics = append(metrics, metricdata.Metrics{ + Name: w.metricName, + Data: metricdata.Histogram[int64]{ + Temporality: w.aggregationTemporality.AsTemporality(), + DataPoints: []metricdata.HistogramDataPoint[int64]{ + { + StartTime: startTime, + Time: now, + Attributes: attribute.NewSet(loadAttrs...), + Exemplars: w.exemplars, + Count: totalCount, + Sum: sum, + // Bounds from https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/metrics/sdk.md#explicit-bucket-histogram-aggregation + Bounds: []float64{0, 5, 10, 25, 50, 75, 100, 250, 500, 750, 1000, 2500, 5000, 7500, 10000}, + BucketCounts: bucketCounts, + }, + }, + }, + }) + case MetricTypeExponentialHistogram: + // Generate realistic exponential histogram data using go-expohisto + cfg := structure.NewConfig(structure.WithMaxSize(8)) + hist := structure.NewFloat64(cfg) + + // Add random values to the histogram + count := 10 + w.rand.IntN(20) // Random count between 10-30 + for range count { + value := float64(w.rand.IntN(1000)) + hist.Update(value) + } + + // Create the data point and convert using utility function + dp := &metricdata.ExponentialHistogramDataPoint[int64]{ + StartTime: startTime, + Time: now, + Attributes: attribute.NewSet(signalAttrs...), + Exemplars: w.exemplars, + } + expoHistToSDKExponentialDataPoint(hist, dp) + + metrics = append(metrics, metricdata.Metrics{ + Name: w.metricName, + Data: metricdata.ExponentialHistogram[int64]{ + Temporality: w.aggregationTemporality.AsTemporality(), + DataPoints: []metricdata.ExponentialHistogramDataPoint[int64]{*dp}, + }, + }) + case MetricTypeDDSketch: + metric, err := w.buildDDSketchMetric(startTime, now, loadAttrs, int(i)) + if err != nil { + w.logger.Fatal("failed to build DDSketch metric", zap.Error(err)) + } + metrics = append(metrics, metric) + default: + w.logger.Fatal("unknown metric type") + } + + rm := metricdata.ResourceMetrics{ + Resource: res, + ScopeMetrics: []metricdata.ScopeMetrics{{Metrics: metrics}}, + } + + if err := limiter.Wait(context.Background()); err != nil { + w.logger.Fatal("limiter wait failed, retry", zap.Error(err)) + } + + if err := exporter.Export(context.Background(), &rm); err != nil { + if w.allowFailures { + w.logger.Error("exporter failed, continuing due to --allow-export-failures", zap.Error(err)) + } else { + w.logger.Fatal("exporter failed", zap.Error(err)) + } + } + + i++ + if w.numMetrics != 0 && i >= int64(w.numMetrics) { + break + } + } + + w.logger.Info("metrics generated", zap.Int64("metrics", i)) + w.wg.Done() +} + +// expoHistToSDKExponentialDataPoint copies `lightstep/go-expohisto` structure.Histogram to +// metricdata.ExponentialHistogramDataPoint +func expoHistToSDKExponentialDataPoint(agg *structure.Histogram[float64], dp *metricdata.ExponentialHistogramDataPoint[int64]) { + dp.Count = agg.Count() + dp.Sum = int64(agg.Sum()) + dp.ZeroCount = agg.ZeroCount() + dp.Scale = agg.Scale() + dp.ZeroThreshold = 0.0 // go-expohisto doesn't expose ZeroThreshold, use default + + // Convert positive buckets + posBuckets := agg.Positive() + dp.PositiveBucket.Offset = posBuckets.Offset() + dp.PositiveBucket.Counts = make([]uint64, posBuckets.Len()) + for i := uint32(0); i < posBuckets.Len(); i++ { + dp.PositiveBucket.Counts[i] = posBuckets.At(i) + } + + // Convert negative buckets + negBuckets := agg.Negative() + dp.NegativeBucket.Offset = negBuckets.Offset() + dp.NegativeBucket.Counts = make([]uint64, negBuckets.Len()) + for i := uint32(0); i < negBuckets.Len(); i++ { + dp.NegativeBucket.Counts[i] = negBuckets.At(i) + } +} + +func (w worker) buildDDSketchMetric(startTime, now time.Time, attrs []attribute.KeyValue, iteration int) (metricdata.Metrics, error) { + values := ddsketchValueSamples[iteration%len(ddsketchValueSamples)] + dp, err := buildDDSketchDataPoint(values, startTime, now, attrs) + if err != nil { + return metricdata.Metrics{}, err + } + + return metricdata.Metrics{ + Name: w.metricName, + Data: metricdata.DDSketch[float64]{ + Temporality: w.aggregationTemporality.AsTemporality(), + DataPoints: []metricdata.DDSketchDataPoint[float64]{dp}, + }, + }, nil +} + +func buildDDSketchDataPoint(values []float64, startTime, now time.Time, attrs []attribute.KeyValue) (metricdata.DDSketchDataPoint[float64], error) { + dp := metricdata.DDSketchDataPoint[float64]{ + StartTime: startTime, + Time: now, + Attributes: attribute.NewSet(attrs...), + Encoding: metricdata.DDSketchEncodingProto, + } + + if len(values) == 0 { + return dp, fmt.Errorf("no values provided to build DDSketch data point") + } + + sketch, err := ddsketch.NewDefaultDDSketch(0.01) + if err != nil { + return dp, err + } + + min := math.MaxFloat64 + max := -math.MaxFloat64 + var sum float64 + for _, v := range values { + if v < min { + min = v + } + if v > max { + max = v + } + sum += v + if err := sketch.Add(v); err != nil { + return dp, err + } + } + + payload, err := proto.Marshal(sketch.ToProto()) + if err != nil { + return dp, err + } + + dp.Count = uint64(len(values)) + dp.Sum = sum + dp.Min = metricdata.NewExtrema(min) + dp.Max = metricdata.NewExtrema(max) + dp.Sketch = payload + + return dp, nil +} diff --git a/opentelemetry-collector-contrib-patch/cmd/telemetrygen/pkg/metrics/worker_test.go b/opentelemetry-collector-contrib-patch/cmd/telemetrygen/pkg/metrics/worker_test.go new file mode 100644 index 00000000..e6d78549 --- /dev/null +++ b/opentelemetry-collector-contrib-patch/cmd/telemetrygen/pkg/metrics/worker_test.go @@ -0,0 +1,779 @@ +// Copyright The OpenTelemetry Authors +// SPDX-License-Identifier: Apache-2.0 + +package metrics + +import ( + "context" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "go.opentelemetry.io/otel/attribute" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/metric/metricdata" + "go.opentelemetry.io/otel/sdk/resource" + "go.uber.org/zap" + "golang.org/x/time/rate" + + "github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen/internal/config" + types "github.com/open-telemetry/opentelemetry-collector-contrib/cmd/telemetrygen/pkg" +) + +const ( + telemetryAttrKeyOne = "key1" + telemetryAttrKeyTwo = "key2" + telemetryAttrValueOne = "value1" + telemetryAttrValueTwo = "value2" + telemetryAttrIntKeyOne = "intKey1" + telemetryAttrIntValueOne = 1 + telemetryAttrBoolKeyOne = "boolKey1" + telemetryAttrBoolValueOne = true +) + +type mockExporter struct { + rms []*metricdata.ResourceMetrics +} + +func (*mockExporter) Temporality(sdkmetric.InstrumentKind) metricdata.Temporality { + return metricdata.DeltaTemporality +} + +func (*mockExporter) Aggregation(sdkmetric.InstrumentKind) sdkmetric.Aggregation { + return sdkmetric.AggregationDefault{} +} + +func (m *mockExporter) Export(_ context.Context, metrics *metricdata.ResourceMetrics) error { + m.rms = append(m.rms, metrics) + return nil +} + +func (*mockExporter) ForceFlush(context.Context) error { + return nil +} + +func (mockExporter) Shutdown(context.Context) error { + return nil +} + +func checkMetricTemporality(t *testing.T, ms metricdata.Metrics, metricType MetricType, expectedAggregationTemporality metricdata.Temporality) { + switch metricType { + case MetricTypeSum: + sumData, ok := ms.Data.(metricdata.Sum[int64]) + require.True(t, ok, "expected Sum data type") + assert.Equal(t, expectedAggregationTemporality, sumData.Temporality) + case MetricTypeHistogram: + histogramData, ok := ms.Data.(metricdata.Histogram[int64]) + require.True(t, ok, "expected Histogram data type") + assert.Equal(t, expectedAggregationTemporality, histogramData.Temporality) + case MetricTypeExponentialHistogram: + histogramData, ok := ms.Data.(metricdata.ExponentialHistogram[int64]) + require.True(t, ok, "expected ExponentialHistogram data type") + assert.Equal(t, expectedAggregationTemporality, histogramData.Temporality) + case MetricTypeDDSketch: + ddsData, ok := ms.Data.(metricdata.DDSketch[float64]) + require.True(t, ok, "expected DDSketch data type") + assert.Equal(t, expectedAggregationTemporality, ddsData.Temporality) + default: + t.Fatalf("unsupported metric type: %v", metricType) + } +} + +func TestFixedNumberOfMetrics(t *testing.T) { + // arrange + cfg := &Config{ + Config: config.Config{ + WorkerCount: 1, + }, + NumMetrics: 5, + MetricType: MetricTypeSum, + } + m := &mockExporter{} + expFunc := func() (sdkmetric.Exporter, error) { + return m, nil + } + + // act + logger, _ := zap.NewDevelopment() + require.NoError(t, run(cfg, expFunc, logger)) + time.Sleep(1 * time.Second) + + // assert + require.Len(t, m.rms, 5) +} + +func TestDurationInf(t *testing.T) { + cfg := &Config{ + Config: config.Config{ + TotalDuration: types.DurationWithInf(-1), + }, + MetricType: MetricTypeSum, + } + m := &mockExporter{} + expFunc := func() (sdkmetric.Exporter, error) { + return m, nil + } + + // test + require.NoError(t, run(cfg, expFunc, zap.NewNop())) +} + +func TestRateOfMetrics(t *testing.T) { + // arrange + cfg := &Config{ + Config: config.Config{ + Rate: 10, + TotalDuration: types.DurationWithInf(time.Second / 2), + WorkerCount: 1, + }, + MetricType: MetricTypeSum, + } + m := &mockExporter{} + expFunc := func() (sdkmetric.Exporter, error) { + return m, nil + } + + // act + require.NoError(t, run(cfg, expFunc, zap.NewNop())) + + // assert + // the minimum acceptable number of metrics for the rate of 10/sec for half a second + assert.GreaterOrEqual(t, len(m.rms), 6, "there should have been more than 6 metrics, had %d", len(m.rms)) + // the maximum acceptable number of metrics for the rate of 10/sec for half a second + assert.LessOrEqual(t, len(m.rms), 20, "there should have been less than 20 metrics, had %d", len(m.rms)) +} + +func TestMetricsWithTemporality(t *testing.T) { + tests := []struct { + name string + metricType MetricType + aggregationTemporality AggregationTemporality + expectedAggregationTemporality metricdata.Temporality + }{ + { + name: "Sum: delta temporality", + metricType: MetricTypeSum, + aggregationTemporality: AggregationTemporality(metricdata.DeltaTemporality), + expectedAggregationTemporality: metricdata.DeltaTemporality, + }, + { + name: "Sum: cumulative temporality", + metricType: MetricTypeSum, + aggregationTemporality: AggregationTemporality(metricdata.CumulativeTemporality), + expectedAggregationTemporality: metricdata.CumulativeTemporality, + }, + { + name: "Histogram: delta temporality", + metricType: MetricTypeHistogram, + aggregationTemporality: AggregationTemporality(metricdata.DeltaTemporality), + expectedAggregationTemporality: metricdata.DeltaTemporality, + }, + { + name: "Histogram: cumulative temporality", + metricType: MetricTypeHistogram, + aggregationTemporality: AggregationTemporality(metricdata.CumulativeTemporality), + expectedAggregationTemporality: metricdata.CumulativeTemporality, + }, + { + name: "DDSketch: delta temporality", + metricType: MetricTypeDDSketch, + aggregationTemporality: AggregationTemporality(metricdata.DeltaTemporality), + expectedAggregationTemporality: metricdata.DeltaTemporality, + }, + { + name: "DDSketch: cumulative temporality", + metricType: MetricTypeDDSketch, + aggregationTemporality: AggregationTemporality(metricdata.CumulativeTemporality), + expectedAggregationTemporality: metricdata.CumulativeTemporality, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + // arrange + cfg := &Config{ + Config: config.Config{ + WorkerCount: 1, + }, + NumMetrics: 1, + MetricName: "test", + MetricType: tt.metricType, + AggregationTemporality: tt.aggregationTemporality, + } + m := &mockExporter{} + expFunc := func() (sdkmetric.Exporter, error) { + return m, nil + } + + // act + logger, _ := zap.NewDevelopment() + require.NoError(t, run(cfg, expFunc, logger)) + + time.Sleep(1 * time.Second) + + // assert + require.Len(t, m.rms, 1) + ms := m.rms[0].ScopeMetrics[0].Metrics[0] + assert.Equal(t, "test", ms.Name) + + checkMetricTemporality(t, ms, tt.metricType, tt.expectedAggregationTemporality) + }) + } +} + +func TestUnthrottled(t *testing.T) { + // arrange + cfg := &Config{ + Config: config.Config{ + TotalDuration: types.DurationWithInf(1 * time.Second), + WorkerCount: 1, + }, + MetricType: MetricTypeSum, + } + m := &mockExporter{} + expFunc := func() (sdkmetric.Exporter, error) { + return m, nil + } + + // act + logger, _ := zap.NewDevelopment() + require.NoError(t, run(cfg, expFunc, logger)) + + // assert + assert.Greater(t, len(m.rms), 100, "there should have been more than 100 metrics, had %d", len(m.rms)) +} + +func TestSumNoTelemetryAttrs(t *testing.T) { + // arrange + qty := 2 + cfg := configWithNoAttributes(MetricTypeSum, qty) + m := &mockExporter{} + expFunc := func() (sdkmetric.Exporter, error) { + return m, nil + } + + // act + logger, _ := zap.NewDevelopment() + require.NoError(t, run(cfg, expFunc, logger)) + + time.Sleep(1 * time.Second) + + // asserts + require.Len(t, m.rms, qty) + + rms := m.rms + for i := range qty { + ms := rms[i].ScopeMetrics[0].Metrics[0] + assert.Equal(t, "test", ms.Name) + // @note update when telemetrygen allow other metric types + attr := ms.Data.(metricdata.Sum[int64]).DataPoints[0].Attributes + assert.Equal(t, 0, attr.Len(), "it shouldn't have attrs here") + } +} + +func TestGaugeNoTelemetryAttrs(t *testing.T) { + // arrange + qty := 2 + cfg := configWithNoAttributes(MetricTypeGauge, qty) + m := &mockExporter{} + expFunc := func() (sdkmetric.Exporter, error) { + return m, nil + } + + // act + logger, _ := zap.NewDevelopment() + require.NoError(t, run(cfg, expFunc, logger)) + + time.Sleep(1 * time.Second) + + // asserts + require.Len(t, m.rms, qty) + + rms := m.rms + for i := range qty { + ms := rms[i].ScopeMetrics[0].Metrics[0] + assert.Equal(t, "test", ms.Name) + // @note update when telemetrygen allow other metric types + attr := ms.Data.(metricdata.Gauge[int64]).DataPoints[0].Attributes + assert.Equal(t, 0, attr.Len(), "it shouldn't have attrs here") + } +} + +func TestSumSingleTelemetryAttr(t *testing.T) { + // arrange + qty := 2 + cfg := configWithOneAttribute(MetricTypeSum, qty) + m := &mockExporter{} + expFunc := func() (sdkmetric.Exporter, error) { + return m, nil + } + + // act + logger, _ := zap.NewDevelopment() + require.NoError(t, run(cfg, expFunc, logger)) + + time.Sleep(1 * time.Second) + + // asserts + require.Len(t, m.rms, qty) + + rms := m.rms + for i := range qty { + ms := rms[i].ScopeMetrics[0].Metrics[0] + assert.Equal(t, "test", ms.Name) + // @note update when telemetrygen allow other metric types + attr := ms.Data.(metricdata.Sum[int64]).DataPoints[0].Attributes + assert.Equal(t, 1, attr.Len(), "it must have a single attribute here") + actualValue, _ := attr.Value(telemetryAttrKeyOne) + assert.Equal(t, telemetryAttrValueOne, actualValue.AsString(), "it should be "+telemetryAttrValueOne) + } +} + +func TestGaugeSingleTelemetryAttr(t *testing.T) { + // arrange + qty := 2 + cfg := configWithOneAttribute(MetricTypeGauge, qty) + m := &mockExporter{} + expFunc := func() (sdkmetric.Exporter, error) { + return m, nil + } + + // act + logger, _ := zap.NewDevelopment() + require.NoError(t, run(cfg, expFunc, logger)) + + time.Sleep(1 * time.Second) + + // asserts + require.Len(t, m.rms, qty) + + rms := m.rms + for i := range qty { + ms := rms[i].ScopeMetrics[0].Metrics[0] + assert.Equal(t, "test", ms.Name) + // @note update when telemetrygen allow other metric types + attr := ms.Data.(metricdata.Gauge[int64]).DataPoints[0].Attributes + assert.Equal(t, 1, attr.Len(), "it must have a single attribute here") + actualValue, _ := attr.Value(telemetryAttrKeyOne) + assert.Equal(t, telemetryAttrValueOne, actualValue.AsString(), "it should be "+telemetryAttrValueOne) + } +} + +func TestSumMultipleTelemetryAttr(t *testing.T) { + // arrange + qty := 2 + cfg := configWithMultipleAttributes(MetricTypeSum, qty) + m := &mockExporter{} + expFunc := func() (sdkmetric.Exporter, error) { + return m, nil + } + + // act + logger, _ := zap.NewDevelopment() + require.NoError(t, run(cfg, expFunc, logger)) + + time.Sleep(1 * time.Second) + + // asserts + require.Len(t, m.rms, qty) + + rms := m.rms + var actualValue attribute.Value + for i := range qty { + ms := rms[i].ScopeMetrics[0].Metrics[0] + // @note update when telemetrygen allow other metric types + attr := ms.Data.(metricdata.Sum[int64]).DataPoints[0].Attributes + assert.Equal(t, 4, attr.Len(), "it must have multiple attributes here") + actualValue, _ = attr.Value(telemetryAttrKeyOne) + assert.Equal(t, telemetryAttrValueOne, actualValue.AsString(), "it should be %s", telemetryAttrValueOne) + actualValue, _ = attr.Value(telemetryAttrKeyTwo) + assert.Equal(t, telemetryAttrValueTwo, actualValue.AsString(), "it should be %s", telemetryAttrValueTwo) + actualValue, _ = attr.Value(telemetryAttrIntKeyOne) + assert.Equal(t, int64(telemetryAttrIntValueOne), actualValue.AsInt64(), "it should be %d", telemetryAttrIntValueOne) + actualValue, _ = attr.Value(telemetryAttrBoolKeyOne) + assert.Equal(t, telemetryAttrBoolValueOne, actualValue.AsBool(), "it should be %t", telemetryAttrBoolValueOne) + } +} + +func TestGaugeMultipleTelemetryAttr(t *testing.T) { + // arrange + qty := 2 + cfg := configWithMultipleAttributes(MetricTypeGauge, qty) + m := &mockExporter{} + expFunc := func() (sdkmetric.Exporter, error) { + return m, nil + } + + // act + logger, _ := zap.NewDevelopment() + require.NoError(t, run(cfg, expFunc, logger)) + + time.Sleep(1 * time.Second) + + // asserts + require.Len(t, m.rms, qty) + + rms := m.rms + var actualValue attribute.Value + for i := range qty { + ms := rms[i].ScopeMetrics[0].Metrics[0] + // @note update when telemetrygen allow other metric types + attr := ms.Data.(metricdata.Gauge[int64]).DataPoints[0].Attributes + assert.Equal(t, 4, attr.Len(), "it must have multiple attributes here") + actualValue, _ = attr.Value(telemetryAttrKeyOne) + assert.Equal(t, telemetryAttrValueOne, actualValue.AsString(), "it should be "+telemetryAttrValueOne) + actualValue, _ = attr.Value(telemetryAttrKeyTwo) + assert.Equal(t, telemetryAttrValueTwo, actualValue.AsString(), "it should be "+telemetryAttrValueTwo) + actualValue, _ = attr.Value(telemetryAttrIntKeyOne) + assert.Equal(t, int64(telemetryAttrIntValueOne), actualValue.AsInt64(), "it should be %d", telemetryAttrIntValueOne) + actualValue, _ = attr.Value(telemetryAttrBoolKeyOne) + assert.Equal(t, telemetryAttrBoolValueOne, actualValue.AsBool(), "it should be %t", telemetryAttrBoolValueOne) + } +} + +func TestValidate(t *testing.T) { + tests := []struct { + name string + cfg *Config + wantErrMessage string + }{ + { + name: "No duration, NumMetrics, or Continuous", + cfg: &Config{ + Config: config.Config{ + WorkerCount: 1, + }, + MetricType: MetricTypeSum, + TraceID: "123", + }, + wantErrMessage: "either `metrics` or `duration` must be greater than 0", + }, + { + name: "TraceID invalid", + cfg: &Config{ + Config: config.Config{ + WorkerCount: 1, + }, + NumMetrics: 5, + MetricType: MetricTypeSum, + TraceID: "123", + }, + wantErrMessage: "TraceID must be a 32 character hex string, like: 'ae87dadd90e9935a4bc9660628efd569'", + }, + { + name: "SpanID invalid", + cfg: &Config{ + Config: config.Config{ + WorkerCount: 1, + }, + NumMetrics: 5, + MetricType: MetricTypeSum, + TraceID: "ae87dadd90e9935a4bc9660628efd569", + SpanID: "123", + }, + wantErrMessage: "SpanID must be a 16 character hex string, like: '5828fa4960140870'", + }, + { + name: "LoadSize negative", + cfg: &Config{ + Config: config.Config{ + WorkerCount: 1, + LoadSize: -1, + }, + NumMetrics: 5, + MetricType: MetricTypeSum, + }, + wantErrMessage: "load size must be non-negative, found -1", + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + m := &mockExporter{} + expFunc := func() (sdkmetric.Exporter, error) { + return m, nil + } + logger, _ := zap.NewDevelopment() + require.EqualError(t, run(tt.cfg, expFunc, logger), tt.wantErrMessage) + }) + } +} + +func configWithNoAttributes(metric MetricType, qty int) *Config { + return &Config{ + Config: config.Config{ + WorkerCount: 1, + TelemetryAttributes: nil, + }, + NumMetrics: qty, + MetricName: "test", + MetricType: metric, + } +} + +func configWithOneAttribute(metric MetricType, qty int) *Config { + return &Config{ + Config: config.Config{ + WorkerCount: 1, + TelemetryAttributes: config.KeyValue{telemetryAttrKeyOne: telemetryAttrValueOne}, + }, + NumMetrics: qty, + MetricName: "test", + MetricType: metric, + } +} + +func configWithMultipleAttributes(metric MetricType, qty int) *Config { + kvs := config.KeyValue{ + telemetryAttrKeyOne: telemetryAttrValueOne, + telemetryAttrKeyTwo: telemetryAttrValueTwo, + telemetryAttrIntKeyOne: telemetryAttrIntValueOne, + telemetryAttrBoolKeyOne: telemetryAttrBoolValueOne, + } + return &Config{ + Config: config.Config{ + WorkerCount: 1, + TelemetryAttributes: kvs, + }, + NumMetrics: qty, + MetricType: metric, + } +} + +func configWithEnabledUnique(metric MetricType, qty int) *Config { + return &Config{ + Config: config.Config{ + WorkerCount: 1, + TelemetryAttributes: nil, + }, + EnforceUniqueTimeseries: true, + NumMetrics: qty, + MetricName: "test", + MetricType: metric, + } +} + +func TestTemporalityStartTimes(t *testing.T) { + tests := []struct { + name string + temporality AggregationTemporality + checkTimes func(t *testing.T, firstTime, secondTime time.Time) + }{ + { + name: "Cumulative temporality keeps same start timestamp", + temporality: AggregationTemporality(metricdata.CumulativeTemporality), + checkTimes: func(t *testing.T, firstTime, secondTime time.Time) { + if !assert.Equal(t, firstTime, secondTime, + "cumulative metrics should have same start time") { + logTimestampDiff(t, firstTime, secondTime) + } + }, + }, + { + name: "Delta temporality has different start timestamps", + temporality: AggregationTemporality(metricdata.DeltaTemporality), + checkTimes: func(t *testing.T, firstTime, secondTime time.Time) { + if !assert.True(t, secondTime.After(firstTime), + "delta metrics should have increasing start times") { + logTimestampDiff(t, firstTime, secondTime) + } + }, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + m := &mockExporter{} + clock := &mockClock{ + now: time.Now(), + } + + running := &atomic.Bool{} + running.Store(true) + + wg := &sync.WaitGroup{} + wg.Add(1) + + w := worker{ + metricName: "test_metric", + metricType: MetricTypeSum, + aggregationTemporality: tt.temporality, + numMetrics: 2, + running: running, + limitPerSecond: rate.Inf, + logger: zap.NewNop(), + wg: wg, + clock: clock, + } + + w.simulateMetrics(resource.Default(), m, nil, nil) + + wg.Wait() + + require.GreaterOrEqual(t, len(m.rms), 2, "should have at least 2 metric points") + + firstMetric := m.rms[0].ScopeMetrics[0].Metrics[0] + secondMetric := m.rms[1].ScopeMetrics[0].Metrics[0] + + firstStartTime := firstMetric.Data.(metricdata.Sum[int64]).DataPoints[0].StartTime + secondStartTime := secondMetric.Data.(metricdata.Sum[int64]).DataPoints[0].StartTime + + tt.checkTimes(t, firstStartTime, secondStartTime) + }) + } +} + +func logTimestampDiff(t *testing.T, firstTime, secondTime time.Time) { + t.Logf("Timestamp debug logging:\n"+ + "First start time: %s\n"+ + "Second start time: %s\n"+ + "Difference: %v", + firstTime.String(), + secondTime.String(), + secondTime.Sub(firstTime)) +} + +func TestUniqueSumTimeseries(t *testing.T) { + // arrange + qty := 4 + cfg := configWithEnabledUnique(MetricTypeSum, qty) + m := &mockExporter{} + expFunc := func() (sdkmetric.Exporter, error) { + return m, nil + } + + // act + logger, _ := zap.NewDevelopment() + require.NoError(t, run(cfg, expFunc, logger)) + + time.Sleep(1 * time.Second) + + // asserts + require.Len(t, m.rms, qty) + + rms := m.rms + var actualValue attribute.Value + var exist bool + for i := range qty { + ms := rms[i].ScopeMetrics[0].Metrics[0] + // @note update when telemetrygen allow other metric types + attr := ms.Data.(metricdata.Sum[int64]).DataPoints[0].Attributes + assert.Equal(t, 1, attr.Len(), "it must have one attribute here") + actualValue, exist = attr.Value(timeBoxAttributeName) + assert.True(t, exist, "it should have the timebox attribute") + assert.LessOrEqual(t, actualValue.AsInt64(), int64(4), "it should be between 0 and 4") + } +} + +func TestMetricsWithLoadSize(t *testing.T) { + // arrange + cfg := &Config{ + Config: config.Config{ + WorkerCount: 1, + LoadSize: 2, // 2MB of load data + }, + NumMetrics: 1, + MetricName: "test", + MetricType: MetricTypeSum, + } + m := &mockExporter{} + expFunc := func() (sdkmetric.Exporter, error) { + return m, nil + } + + // act + logger, _ := zap.NewDevelopment() + require.NoError(t, run(cfg, expFunc, logger)) + + time.Sleep(1 * time.Second) + + // assert + require.Len(t, m.rms, 1) + ms := m.rms[0].ScopeMetrics[0].Metrics[0] + attr := ms.Data.(metricdata.Sum[int64]).DataPoints[0].Attributes + + // Should have 2 load attributes (load-0 and load-1) each with 1MB of data + assert.Equal(t, 2, attr.Len(), "should have 2 load attributes") + + // Check that load attributes exist and have the expected size + load0Value, exists := attr.Value("load-0") + assert.True(t, exists, "should have load-0 attribute") + assert.Len(t, load0Value.AsString(), config.CharactersPerMB, "load-0 should have 1MB of data") + + load1Value, exists := attr.Value("load-1") + assert.True(t, exists, "should have load-1 attribute") + assert.Len(t, load1Value.AsString(), config.CharactersPerMB, "load-1 should have 1MB of data") +} + +func TestMetricsWithDefaultLoadSize(t *testing.T) { + // arrange + cfg := NewConfig() + cfg.NumMetrics = 1 + cfg.MetricName = "test" + cfg.MetricType = MetricTypeSum + // LoadSize should default to 0 + + m := &mockExporter{} + expFunc := func() (sdkmetric.Exporter, error) { + return m, nil + } + + // act + logger, _ := zap.NewDevelopment() + require.NoError(t, run(cfg, expFunc, logger)) + + time.Sleep(1 * time.Second) + + // assert + require.Len(t, m.rms, 1) + ms := m.rms[0].ScopeMetrics[0].Metrics[0] + attr := ms.Data.(metricdata.Sum[int64]).DataPoints[0].Attributes + + // Should have no load attributes by default (LoadSize = 0) + assert.Equal(t, 0, attr.Len(), "should have no load attributes by default") +} + +// TestExponentialHistogramMetricGeneration tests ExponentialHistogram metric generation +func TestExponentialHistogramMetricGeneration(t *testing.T) { + // arrange + m := &mockExporter{} + cfg := Config{ + Config: config.Config{ + WorkerCount: 1, + }, + NumMetrics: 1, + MetricName: "test_exp_hist", + MetricType: MetricTypeExponentialHistogram, + } + logger := zap.NewNop() + + // act + expFunc := func() (sdkmetric.Exporter, error) { + return m, nil + } + require.NoError(t, run(&cfg, expFunc, logger)) + + time.Sleep(100 * time.Millisecond) + + // assert + require.Len(t, m.rms, 1) + ms := m.rms[0].ScopeMetrics[0].Metrics[0] + + // Verify it's an ExponentialHistogram + expHist, ok := ms.Data.(metricdata.ExponentialHistogram[int64]) + require.True(t, ok, "Expected ExponentialHistogram metric type") + + // Verify data point structure + require.Len(t, expHist.DataPoints, 1) + dp := expHist.DataPoints[0] + assert.Equal(t, "test_exp_hist", ms.Name) + assert.Positive(t, dp.Count) + assert.Positive(t, dp.Sum) + assert.Equal(t, 0.0, dp.ZeroThreshold) + + // Verify buckets exist + assert.NotNil(t, dp.PositiveBucket) + assert.NotNil(t, dp.NegativeBucket) +} diff --git a/opentelemetry-collector-contrib-patch/cmd/telemetrygen/run.sh b/opentelemetry-collector-contrib-patch/cmd/telemetrygen/run.sh new file mode 100644 index 00000000..c515f243 --- /dev/null +++ b/opentelemetry-collector-contrib-patch/cmd/telemetrygen/run.sh @@ -0,0 +1,2 @@ +go run . metrics --metric-type DDSketch --duration 100s --otlp-insecure +