-
Notifications
You must be signed in to change notification settings - Fork 187
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat: Add intervalprocessor component (#1484)
* feat: Add intervalprocessor component Signed-off-by: Arthur Silva Sens <arthursens2005@gmail.com> * Apply suggestions from code review Co-authored-by: Paulin Todev <paulin.todev@gmail.com> * Update outdated error in testdata Signed-off-by: Arthur Silva Sens <arthursens2005@gmail.com> * Apply suggestions from code review Co-authored-by: Clayton Cornell <131809008+clayton-cornell@users.noreply.github.com> --------- Signed-off-by: Arthur Silva Sens <arthursens2005@gmail.com> Co-authored-by: Paulin Todev <paulin.todev@gmail.com> Co-authored-by: Clayton Cornell <131809008+clayton-cornell@users.noreply.github.com>
- Loading branch information
3 people
authored
Aug 27, 2024
1 parent
e1d3892
commit 9b8b81b
Showing
11 changed files
with
372 additions
and
6 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
166 changes: 166 additions & 0 deletions
166
docs/sources/reference/components/otelcol/otelcol.processor.interval.md
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,166 @@ | ||
| --- | ||
| canonical: https://grafana.com/docs/alloy/latest/reference/components/otelcol/otelcol.processor.interval/ | ||
| description: Learn about otelcol.processor.interval | ||
| title: otelcol.processor.interval | ||
| --- | ||
|
|
||
| <span class="badge docs-labels__stage docs-labels__item">Experimental</span> | ||
|
|
||
| # otelcol.processor.interval | ||
|
|
||
| {{< docs/shared lookup="stability/experimental.md" source="alloy" version="<ALLOY_VERSION>" >}} | ||
|
|
||
| `otelcol.processor.interval` aggregates metrics and periodically forwards the latest values to the next component in the pipeline. | ||
| The processor supports aggregating the following metric types: | ||
|
|
||
| * Monotonically increasing, cumulative sums | ||
| * Monotonically increasing, cumulative histograms | ||
| * Monotonically increasing, cumulative exponential histograms | ||
|
|
||
| The following metric types will *not* be aggregated and will instead be passed, unchanged, to the next component in the pipeline: | ||
|
|
||
| * All delta metrics | ||
| * Non-monotonically increasing sums | ||
| * Gauges | ||
| * Summaries | ||
|
|
||
| {{< admonition type="warning" >}} | ||
| After exporting, any internal state is cleared. If no new metrics come in, the next interval will export nothing. | ||
| {{< /admonition >}} | ||
|
|
||
| {{< admonition type="note" >}} | ||
| `otelcol.processor.interval` is a wrapper over the upstream OpenTelemetry Collector `interval` processor. | ||
| Bug reports or feature requests will be redirected to the upstream repository, if necessary. | ||
| {{< /admonition >}} | ||
|
|
||
| ## Usage | ||
|
|
||
| ```alloy | ||
| otelcol.processor.interval "LABEL" { | ||
| output { | ||
| metrics = [...] | ||
| } | ||
| } | ||
| ``` | ||
|
|
||
| ## Arguments | ||
|
|
||
| `otelcol.processor.interval` supports the following arguments: | ||
|
|
||
| Name | Type | Description | Default | Required | ||
| ------------- | ---------- | ------------------------------------------------------------------- | ------- | -------- | ||
| `interval` | `duration` | The interval in the processor should export the aggregated metrics. | `"60s"` | no | ||
|
|
||
| ## Blocks | ||
|
|
||
| The following blocks are supported inside the definition of `otelcol.processor.interval`: | ||
|
|
||
| Hierarchy | Block | Description | Required | ||
| ------------- | ----------------- | -------------------------------------------------------------------------- | -------- | ||
| output | [output][] | Configures where to send received telemetry data. | yes | ||
| debug_metrics | [debug_metrics][] | Configures the metrics that this component generates to monitor its state. | no | ||
|
|
||
| [output]: #output-block | ||
| [debug_metrics]: #debug_metrics-block | ||
|
|
||
| ### output block | ||
|
|
||
| {{< docs/shared lookup="reference/components/output-block.md" source="alloy" version="<ALLOY_VERSION>" >}} | ||
|
|
||
| ### debug_metrics block | ||
|
|
||
| {{< docs/shared lookup="reference/components/otelcol-debug-metrics-block.md" source="alloy" version="<ALLOY_VERSION>" >}} | ||
|
|
||
| ## Exported fields | ||
|
|
||
| The following fields are exported and can be referenced by other components: | ||
|
|
||
| Name | Type | Description | ||
| --------|--------------------|----------------------------------------------------------------- | ||
| `input` | `otelcol.Consumer` | A value that other components can use to send telemetry data to. | ||
|
|
||
| `input` accepts `otelcol.Consumer` data for metrics. | ||
|
|
||
| ## Component health | ||
|
|
||
| `otelcol.processor.interval` is only reported as unhealthy if given an invalid configuration. | ||
|
|
||
| ## Debug information | ||
|
|
||
| `otelcol.processor.interval` does not expose any component-specific debug information. | ||
|
|
||
| ## Example | ||
|
|
||
| This example receives OTLP metrics and aggregates them for 30s before sending to the next exporter. | ||
|
|
||
| ```alloy | ||
| otelcol.receiver.otlp "default" { | ||
| grpc { ... } | ||
| http { ... } | ||
| output { | ||
| metrics = [otelcol.processor.interval.default.input] | ||
| } | ||
| } | ||
| otelcol.processor.interval "default" { | ||
| interval = "30s" | ||
| output { | ||
| metrics = [otelcol.exporter.otlphttp.grafana_cloud.input] | ||
| } | ||
| } | ||
| otelcol.exporter.otlphttp "grafana_cloud" { | ||
| client { | ||
| endpoint = "https://otlp-gateway-prod-gb-south-0.grafana.net/otlp" | ||
| auth = otelcol.auth.basic.grafana_cloud.handler | ||
| } | ||
| } | ||
| otelcol.auth.basic "grafana_cloud" { | ||
| username = env("GRAFANA_CLOUD_USERNAME") | ||
| password = env("GRAFANA_CLOUD_API_KEY") | ||
| } | ||
| ``` | ||
|
|
||
| | Timestamp | Metric Name | Aggregation Temporarility | Attributes | Value | | ||
| | --------- | ------------ | ------------------------- | ----------------- | ----: | | ||
| | 0 | test_metric | Cumulative | labelA: foo | 4.0 | | ||
| | 2 | test_metric | Cumulative | labelA: bar | 3.1 | | ||
| | 4 | other_metric | Delta | fruitType: orange | 77.4 | | ||
| | 6 | test_metric | Cumulative | labelA: foo | 8.2 | | ||
| | 8 | test_metric | Cumulative | labelA: foo | 12.8 | | ||
| | 10 | test_metric | Cumulative | labelA: bar | 6.4 | | ||
|
|
||
| The processor immediately passes the following metric to the next processor in the chain because it is a Delta metric. | ||
|
|
||
| | Timestamp | Metric Name | Aggregation Temporarility | Attributes | Value | | ||
| | --------- | ------------ | ------------------------- | ----------------- | ----: | | ||
| | 4 | other_metric | Delta | fruitType: orange | 77.4 | | ||
|
|
||
|
|
||
| At the next `interval` (15s by default), the processor passed the following metrics to the next processor in the chain. | ||
|
|
||
| | Timestamp | Metric Name | Aggregation Temporarility | Attributes | Value | | ||
| | --------- | ----------- | ------------------------- | ----------- | ----: | | ||
| | 8 | test_metric | Cumulative | labelA: foo | 12.8 | | ||
| | 10 | test_metric | Cumulative | labelA: bar | 6.4 | | ||
|
|
||
| <!-- START GENERATED COMPATIBLE COMPONENTS --> | ||
|
|
||
| ## Compatible components | ||
|
|
||
| `otelcol.processor.interval` can accept arguments from the following components: | ||
|
|
||
| - Components that export [OpenTelemetry `otelcol.Consumer`](../../../compatibility/#opentelemetry-otelcolconsumer-exporters) | ||
|
|
||
| `otelcol.processor.interval` has exports that can be consumed by the following components: | ||
|
|
||
| - Components that consume [OpenTelemetry `otelcol.Consumer`](../../../compatibility/#opentelemetry-otelcolconsumer-consumers) | ||
|
|
||
| {{< admonition type="note" >}} | ||
| Connecting some components may not be sensible or components may require further configuration to make the connection work correctly. | ||
| Refer to the linked documentation for more details. | ||
| {{< /admonition >}} | ||
|
|
||
| <!-- END GENERATED COMPATIBLE COMPONENTS --> |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,88 @@ | ||
| // Package interval provides an otelcol.processor.interval component. | ||
| package interval | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "time" | ||
|
|
||
| "github.com/grafana/alloy/internal/component" | ||
| "github.com/grafana/alloy/internal/component/otelcol" | ||
| otelcolCfg "github.com/grafana/alloy/internal/component/otelcol/config" | ||
| "github.com/grafana/alloy/internal/component/otelcol/processor" | ||
| "github.com/grafana/alloy/internal/featuregate" | ||
| "github.com/open-telemetry/opentelemetry-collector-contrib/processor/intervalprocessor" | ||
| otelcomponent "go.opentelemetry.io/collector/component" | ||
| otelextension "go.opentelemetry.io/collector/extension" | ||
| ) | ||
|
|
||
| func init() { | ||
| component.Register(component.Registration{ | ||
| Name: "otelcol.processor.interval", | ||
| Stability: featuregate.StabilityExperimental, | ||
| Args: Arguments{}, | ||
| Exports: otelcol.ConsumerExports{}, | ||
| Build: func(opts component.Options, args component.Arguments) (component.Component, error) { | ||
| return processor.New(opts, intervalprocessor.NewFactory(), args.(Arguments)) | ||
| }, | ||
| }) | ||
| } | ||
|
|
||
| type Arguments struct { | ||
| // The interval in which the processor should export the aggregated metrics. Default: 60s. | ||
| Interval time.Duration `alloy:"interval,attr,optional"` | ||
|
|
||
| // Output configures where to send processed data. Required. | ||
| Output *otelcol.ConsumerArguments `alloy:"output,block"` | ||
|
|
||
| // DebugMetrics configures component internal metrics. Optional. | ||
| DebugMetrics otelcolCfg.DebugMetricsArguments `alloy:"debug_metrics,block,optional"` | ||
| } | ||
|
|
||
| var ( | ||
| _ processor.Arguments = Arguments{} | ||
| ) | ||
|
|
||
| // DefaultArguments holds default settings for Arguments. | ||
| var DefaultArguments = Arguments{ | ||
| Interval: 60 * time.Second, | ||
| } | ||
|
|
||
| // SetToDefault implements syntax.Defaulter. | ||
| func (args *Arguments) SetToDefault() { | ||
| *args = DefaultArguments | ||
| } | ||
|
|
||
| // Validate implements syntax.Validator. | ||
| func (args *Arguments) Validate() error { | ||
| if args.Interval <= 0 { | ||
| return fmt.Errorf("interval must be greater than 0") | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| // Convert implements processor.Arguments. | ||
| func (args Arguments) Convert() (otelcomponent.Config, error) { | ||
| return &intervalprocessor.Config{ | ||
| Interval: args.Interval, | ||
| }, nil | ||
| } | ||
|
|
||
| // Extensions implements processor.Arguments. | ||
| func (args Arguments) Extensions() map[otelcomponent.ID]otelextension.Extension { | ||
| return nil | ||
| } | ||
|
|
||
| // Exporters implements processor.Arguments. | ||
| func (args Arguments) Exporters() map[otelcomponent.DataType]map[otelcomponent.ID]otelcomponent.Component { | ||
| return nil | ||
| } | ||
|
|
||
| // NextConsumers implements processor.Arguments. | ||
| func (args Arguments) NextConsumers() *otelcol.ConsumerArguments { | ||
| return args.Output | ||
| } | ||
|
|
||
| // DebugMetricsConfig implements processor.Arguments. | ||
| func (args Arguments) DebugMetricsConfig() otelcolCfg.DebugMetricsArguments { | ||
| return args.DebugMetrics | ||
| } |
61 changes: 61 additions & 0 deletions
61
internal/converter/internal/otelcolconvert/converter_intervalprocessor.go
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,61 @@ | ||
| package otelcolconvert | ||
|
|
||
| import ( | ||
| "fmt" | ||
|
|
||
| "github.com/grafana/alloy/internal/component/otelcol" | ||
| "github.com/grafana/alloy/internal/component/otelcol/processor/interval" | ||
| "github.com/grafana/alloy/internal/converter/diag" | ||
| "github.com/grafana/alloy/internal/converter/internal/common" | ||
| "github.com/open-telemetry/opentelemetry-collector-contrib/processor/intervalprocessor" | ||
| "go.opentelemetry.io/collector/component" | ||
| ) | ||
|
|
||
| func init() { | ||
| converters = append(converters, intervalProcessorConverter{}) | ||
| } | ||
|
|
||
| type intervalProcessorConverter struct{} | ||
|
|
||
| func (intervalProcessorConverter) Factory() component.Factory { | ||
| return intervalprocessor.NewFactory() | ||
| } | ||
|
|
||
| func (intervalProcessorConverter) InputComponentName() string { | ||
| return "otelcol.processor.interval" | ||
| } | ||
|
|
||
| func (intervalProcessorConverter) ConvertAndAppend(state *State, id component.InstanceID, cfg component.Config) diag.Diagnostics { | ||
| var diags diag.Diagnostics | ||
|
|
||
| label := state.AlloyComponentLabel() | ||
|
|
||
| args := toIntervalProcessor(state, id, cfg.(*intervalprocessor.Config)) | ||
| block := common.NewBlockWithOverride([]string{"otelcol", "processor", "interval"}, label, args) | ||
|
|
||
| diags.Add( | ||
| diag.SeverityLevelInfo, | ||
| fmt.Sprintf("Converted %s into %s", StringifyInstanceID(id), StringifyBlock(block)), | ||
| ) | ||
|
|
||
| state.Body().AppendBlock(block) | ||
| return diags | ||
| } | ||
|
|
||
| func toIntervalProcessor(state *State, id component.InstanceID, cfg *intervalprocessor.Config) *interval.Arguments { | ||
| var ( | ||
| nextMetrics = state.Next(id, component.DataTypeMetrics) | ||
| nextLogs = state.Next(id, component.DataTypeLogs) | ||
| nextTraces = state.Next(id, component.DataTypeTraces) | ||
| ) | ||
|
|
||
| return &interval.Arguments{ | ||
| Interval: cfg.Interval, | ||
| Output: &otelcol.ConsumerArguments{ | ||
| Metrics: ToTokenizedConsumers(nextMetrics), | ||
| Logs: ToTokenizedConsumers(nextLogs), | ||
| Traces: ToTokenizedConsumers(nextTraces), | ||
| }, | ||
| DebugMetrics: common.DefaultValue[interval.Arguments]().DebugMetrics, | ||
| } | ||
| } |
Oops, something went wrong.