[exporter/datasetexporter]: Initial implementation of logs and traces

.chloggen/issue-20660-datasetexporter-logs-traces-initial-code.yaml

@@ -0,0 +1,16 @@
+# One of 'breaking', 'deprecation', 'new_component', 'enhancement', 'bug_fix'
+change_type: enhancement
+
+# The name of the component, or a single word describing the area of concern, (e.g. filelogreceiver)
+component: datasetexporter
+
+# A brief description of the change. Surround your text with quotes ("") if it needs to start with a backtick (`).
+note: Add support for exporting logs and traces.
+
+# One or more tracking issues related to the change
+issues: [20660]
+
+# (Optional) One or more lines of additional information to render under the primary note.
+# These lines will be padded with 2 spaces and then inserted directly into the document.
+# Use pipe (|) for multiline entries.
+subtext:

cmd/otelcontribcol/go.mod

@@ -318,6 +318,7 @@ require (
  github.com/containerd/ttrpc v1.1.0 // indirect
  github.com/coreos/go-oidc v2.2.1+incompatible // indirect
  github.com/coreos/go-systemd/v22 v22.5.0 // indirect
+ github.com/cskr/pubsub v1.0.2 // indirect
  github.com/cyphar/filepath-securejoin v0.2.3 // indirect
  github.com/danieljoos/wincred v1.1.2 // indirect
  github.com/davecgh/go-spew v1.1.1 // indirect
@@ -550,6 +551,7 @@ require (
  github.com/rs/cors v1.9.0 // indirect
  github.com/samber/lo v1.37.0 // indirect
  github.com/scaleway/scaleway-sdk-go v1.0.0-beta.14 // indirect
+ github.com/scalyr/dataset-go v0.0.7 // indirect
  github.com/seccomp/libseccomp-golang v0.9.2-0.20220502022130-f33da4d89646 // indirect
  github.com/secure-systems-lab/go-securesystemslib v0.4.0 // indirect
  github.com/segmentio/asm v1.2.0 // indirect

exporter/datasetexporter/README.md

@@ -3,7 +3,7 @@
 <!-- status autogenerated section -->
 | Status | |
 | ------------- |-----------|
-| Stability | [development]: traces, logs |
+| Stability | [development]: logs, traces |
 | Distributions | [] |
 
 [development]: https://github.com/open-telemetry/opentelemetry-collector#development
@@ -22,8 +22,15 @@ See the [Getting Started](https://app.scalyr.com/help/getting-started) guide.
 
 ### Optional Settings
 
-- `max_delay_ms` (default = "15000"): The maximum delay between sending batches from the same source.
-- `group_by` (default = []): The list of attributes based on which events should be grouped.
+- `buffer`:
+ - `max_lifetime` (default = 5s): The maximum delay between sending batches from the same source.
+ - `group_by` (default = []): The list of attributes based on which events should be grouped.
+ - `retry_initial_interval` (default = 5s): Time to wait after the first failure before retrying.
+ - `retry_max_interval` (default = 30s): Is the upper bound on backoff.
+ - `max_elapsed_time` (default = 300s): Is the maximum amount of time spent trying to send a buffer.
+- `traces`:
+ - `aggregate` (default = false): Count the number of spans and errors belonging to a trace.
+ - `max_wait` (default = 5s): The maximum waiting for all spans from single trace to arrive; ignored if `aggregate` is false.
 - `retry_on_failure`: See [retry_on_failure](https://github.com/open-telemetry/opentelemetry-collector/blob/main/exporter/exporterhelper/README.md)
 - `sending_queue`: See [sending_queue](https://github.com/open-telemetry/opentelemetry-collector/blob/main/exporter/exporterhelper/README.md)
 - `timeout`: See [timeout](https://github.com/open-telemetry/opentelemetry-collector/blob/main/exporter/exporterhelper/README.md)
@@ -39,16 +46,12 @@ exporters:
  dataset_url: https://app.scalyr.com
  # API Key
  api_key: your_api_key
- # Send batch to the API at least every 15s
- max_delay_ms: 15000
- # Group data based on these attributes
- group_by:
- - attributes.container_id
- - attributes.log.file.path
- - body.map.kubernetes.container_hash
- - body.map.kubernetes.pod_id
- - body.map.kubernetes.docker_id
- - body.map.stream
+ buffer:
+ # Send buffer to the API at least every 10s
+ max_lifetime: 10s
+ # Group data based on these attributes
+ group_by:
+ - attributes.container_id
 
 service:
  pipelines:
@@ -57,4 +60,9 @@ service:
  processors: [batch]
  # add dataset among your exporters
  exporters: [dataset]
+ traces:
+ receivers: [otlp]
+ processors: [batch]
+ # add dataset among your exporters
+ exporters: [dataset]
 ```

exporter/datasetexporter/config.go

@@ -17,19 +17,61 @@ package datasetexporter // import "github.com/open-telemetry/opentelemetry-colle
 import (
  "fmt"
  "os"
- "strconv"
+ "time"
 
+ "github.com/scalyr/dataset-go/pkg/buffer"
+ "github.com/scalyr/dataset-go/pkg/buffer_config"
+ datasetConfig "github.com/scalyr/dataset-go/pkg/config"
+ "go.opentelemetry.io/collector/config/configopaque"
  "go.opentelemetry.io/collector/confmap"
  "go.opentelemetry.io/collector/exporter/exporterhelper"
 )
 
-const maxDelayMs = "15000"
+const tracesMaxWait = 5 * time.Second
+const tracesAggregate = false
+
+type TracesSettings struct {
+ Aggregate bool `mapstructure:"aggregate"`
+ MaxWait time.Duration `mapstructure:"max_wait"`
+}
+
+// newDefaultTracesSettings returns the default settings for TracesSettings.
+func newDefaultTracesSettings() TracesSettings {
+ return TracesSettings{
+ Aggregate: tracesAggregate,
+ MaxWait: tracesMaxWait,
+ }
+}
+
+const bufferMaxLifetime = 5 * time.Second
+const bufferRetryInitialInterval = 5 * time.Second
+const bufferRetryMaxInterval = 30 * time.Second
+const bufferRetryMaxElapsedTime = 300 * time.Second
+
+type BufferSettings struct {
+ MaxLifetime time.Duration `mapstructure:"max_lifetime"`
+ GroupBy []string `mapstructure:"group_by"`
+ RetryInitialInterval time.Duration `mapstructure:"retry_initial_interval"`
+ RetryMaxInterval time.Duration `mapstructure:"retry_max_interval"`
+ RetryMaxElapsedTime time.Duration `mapstructure:"retry_max_elapsed_time"`
+}
+
+// newDefaultBufferSettings returns the default settings for BufferSettings.
+func newDefaultBufferSettings() BufferSettings {
+ return BufferSettings{
+ MaxLifetime: bufferMaxLifetime,
+ GroupBy: []string{},
+ RetryInitialInterval: bufferRetryInitialInterval,
+ RetryMaxInterval: bufferRetryMaxInterval,
+ RetryMaxElapsedTime: bufferRetryMaxElapsedTime,
+ }
+}
 
 type Config struct {
- DatasetURL string `mapstructure:"dataset_url"`
- APIKey string  `mapstructure:"api_key"`
- MaxDelayMs  string `mapstructure:"max_delay_ms"`
- GroupBy  []string `mapstructure:"group_by"`
+ DatasetURL string  `mapstructure:"dataset_url"`
+ APIKey configopaque.String `mapstructure:"api_key"`
+ BufferSettings `mapstructure:"buffer"`
+ TracesSettings `mapstructure:"traces"`
  exporterhelper.RetrySettings `mapstructure:"retry_on_failure"`
  exporterhelper.QueueSettings `mapstructure:"sending_queue"`
  exporterhelper.TimeoutSettings `mapstructure:"timeout"`
@@ -44,11 +86,11 @@ func (c *Config) Unmarshal(conf *confmap.Conf) error {
  c.DatasetURL = os.Getenv("DATASET_URL")
  }
  if len(c.APIKey) == 0 {
- c.APIKey = os.Getenv("DATASET_API_KEY")
+ c.APIKey = configopaque.String(os.Getenv("DATASET_API_KEY"))
  }
 
- if len(c.MaxDelayMs) == 0 {
- c.MaxDelayMs = maxDelayMs
+ if c.TracesSettings.MaxWait == 0 {
+ c.TracesSettings.MaxWait = tracesMaxWait
  }
 
  return nil
@@ -64,15 +106,6 @@ func (c *Config) Validate() error {
  return fmt.Errorf("dataset_url is required")
  }
 
- _, err := strconv.Atoi(c.MaxDelayMs)
- if err != nil {
- return fmt.Errorf(
- "max_delay_ms must be integer, but %s was used: %w",
- c.MaxDelayMs,
- err,
- )
- }
-
  return nil
 }
 
@@ -81,11 +114,40 @@ func (c *Config) Validate() error {
 func (c *Config) String() string {
  s := ""
  s += fmt.Sprintf("%s: %s; ", "DatasetURL", c.DatasetURL)
- s += fmt.Sprintf("%s: %s; ", "MaxDelayMs", c.MaxDelayMs)
- s += fmt.Sprintf("%s: %s; ", "GroupBy", c.GroupBy)
+ s += fmt.Sprintf("%s: %+v; ", "BufferSettings", c.BufferSettings)
+ s += fmt.Sprintf("%s: %+v; ", "TracesSettings", c.TracesSettings)
  s += fmt.Sprintf("%s: %+v; ", "RetrySettings", c.RetrySettings)
  s += fmt.Sprintf("%s: %+v; ", "QueueSettings", c.QueueSettings)
  s += fmt.Sprintf("%s: %+v", "TimeoutSettings", c.TimeoutSettings)
 
  return s
 }
+
+func (c *Config) convert() (*ExporterConfig, error) {
+ err := c.Validate()
+ if err != nil {
+ return nil, fmt.Errorf("config is not valid: %w", err)
+ }
+
+ return &ExporterConfig{
+ datasetConfig: &datasetConfig.DataSetConfig{
+ Endpoint: c.DatasetURL,
+ Tokens: datasetConfig.DataSetTokens{WriteLog: string(c.APIKey)},
+ BufferSettings: buffer_config.DataSetBufferSettings{
+ MaxLifetime: c.BufferSettings.MaxLifetime,
+ MaxSize: buffer.LimitBufferSize,
+ GroupBy: c.BufferSettings.GroupBy,
+ RetryInitialInterval: c.BufferSettings.RetryInitialInterval,
+ RetryMaxInterval: c.BufferSettings.RetryMaxInterval,
+ RetryMaxElapsedTime: c.BufferSettings.RetryMaxElapsedTime,
+ },
+ },
+ tracesSettings: c.TracesSettings,
+ },
+ nil
+}
+
+type ExporterConfig struct {
+ datasetConfig *datasetConfig.DataSetConfig
+ tracesSettings TracesSettings
+}