Skip to content
New issue

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

Sign up for GitHub

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

Already on GitHub? Sign in to your account

[exporter/awss3] Implement sending queue #5

Closed
wants to merge 2 commits into from
Closed

[exporter/awss3] Implement sending queue #5

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
d67d631
[exporter/awss3] Implement sending queue
lopes-felipe Jan 15, 2025
308e07e
lint
lopes-felipe Jan 15, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions exporter/awss3exporter/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@ import (

"go.opentelemetry.io/collector/component"
"go.opentelemetry.io/collector/config/configcompression"
"go.opentelemetry.io/collector/exporter/exporterhelper"
"go.uber.org/multierr"
)

Expand Down Expand Up @@ -49,6 +50,8 @@ const (

// Config contains the main configuration options for the s3 exporter
type Config struct {
QueueSettings exporterhelper.QueueConfig `mapstructure:"sending_queue"`

S3Uploader S3UploaderConfig `mapstructure:"s3uploader"`
MarshalerName MarshalerType `mapstructure:"marshaler"`

Expand Down
34 changes: 34 additions & 0 deletions exporter/awss3exporter/config_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,7 @@ import (
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
"go.opentelemetry.io/collector/component"
"go.opentelemetry.io/collector/exporter/exporterhelper"
"go.opentelemetry.io/collector/otelcol/otelcoltest"
"go.uber.org/multierr"

Expand All @@ -32,7 +33,13 @@ func TestLoadConfig(t *testing.T) {

e := cfg.Exporters[component.MustNewID("awss3")].(*Config)
encoding := component.MustNewIDWithName("foo", "bar")

queueCfg := exporterhelper.NewDefaultQueueConfig()
queueCfg.Enabled = false

assert.Equal(t, &Config{
QueueSettings: queueCfg,

Encoding: &encoding,
EncodingFileExtension: "baz",
S3Uploader: S3UploaderConfig{
Expand All @@ -59,9 +66,17 @@ func TestConfig(t *testing.T) {
require.NoError(t, err)
require.NotNil(t, cfg)

queueCfg := exporterhelper.QueueConfig{
Enabled: true,
NumConsumers: 23,
QueueSize: 42,
}

e := cfg.Exporters[component.MustNewID("awss3")].(*Config)

assert.Equal(t, &Config{
QueueSettings: queueCfg,

S3Uploader: S3UploaderConfig{
Region: "us-east-1",
S3Bucket: "foo",
Expand All @@ -88,9 +103,14 @@ func TestConfigForS3CompatibleSystems(t *testing.T) {
require.NoError(t, err)
require.NotNil(t, cfg)

queueCfg := exporterhelper.NewDefaultQueueConfig()
queueCfg.Enabled = false

e := cfg.Exporters[component.MustNewID("awss3")].(*Config)

assert.Equal(t, &Config{
QueueSettings: queueCfg,

S3Uploader: S3UploaderConfig{
Region: "us-east-1",
S3Bucket: "foo",
Expand Down Expand Up @@ -200,9 +220,14 @@ func TestMarshallerName(t *testing.T) {
require.NoError(t, err)
require.NotNil(t, cfg)

queueCfg := exporterhelper.NewDefaultQueueConfig()
queueCfg.Enabled = false

e := cfg.Exporters[component.MustNewID("awss3")].(*Config)

assert.Equal(t, &Config{
QueueSettings: queueCfg,

S3Uploader: S3UploaderConfig{
Region: "us-east-1",
S3Bucket: "foo",
Expand All @@ -215,6 +240,8 @@ func TestMarshallerName(t *testing.T) {
e = cfg.Exporters[component.MustNewIDWithName("awss3", "proto")].(*Config)

assert.Equal(t, &Config{
QueueSettings: queueCfg,

S3Uploader: S3UploaderConfig{
Region: "us-east-1",
S3Bucket: "bar",
Expand All @@ -239,9 +266,14 @@ func TestCompressionName(t *testing.T) {
require.NoError(t, err)
require.NotNil(t, cfg)

queueCfg := exporterhelper.NewDefaultQueueConfig()
queueCfg.Enabled = false

e := cfg.Exporters[component.MustNewID("awss3")].(*Config)

assert.Equal(t, &Config{
QueueSettings: queueCfg,

S3Uploader: S3UploaderConfig{
Region: "us-east-1",
S3Bucket: "foo",
Expand All @@ -255,6 +287,8 @@ func TestCompressionName(t *testing.T) {
e = cfg.Exporters[component.MustNewIDWithName("awss3", "proto")].(*Config)

assert.Equal(t, &Config{
QueueSettings: queueCfg,

S3Uploader: S3UploaderConfig{
Region: "us-east-1",
S3Bucket: "bar",
Expand Down
47 changes: 41 additions & 6 deletions exporter/awss3exporter/factory.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,7 +26,12 @@ func NewFactory() exporter.Factory {
}

func createDefaultConfig() component.Config {
queueCfg := exporterhelper.NewDefaultQueueConfig()
queueCfg.Enabled = false

return &Config{
QueueSettings: queueCfg,

S3Uploader: S3UploaderConfig{
Region: "us-east-1",
S3Partition: "minute",
Expand All @@ -39,19 +44,31 @@ func createLogsExporter(ctx context.Context,
params exporter.Settings,
config component.Config,
) (exporter.Logs, error) {
s3Exporter := newS3Exporter(config.(*Config), "logs", params)
cfg, err := checkAndCastConfig(config)
if err != nil {
return nil, err
}

s3Exporter := newS3Exporter(cfg, "logs", params)

return exporterhelper.NewLogs(ctx, params,
config,
s3Exporter.ConsumeLogs,
exporterhelper.WithStart(s3Exporter.start))
exporterhelper.WithStart(s3Exporter.start),
exporterhelper.WithQueue(cfg.QueueSettings),
)
}

func createMetricsExporter(ctx context.Context,
params exporter.Settings,
config component.Config,
) (exporter.Metrics, error) {
s3Exporter := newS3Exporter(config.(*Config), "metrics", params)
cfg, err := checkAndCastConfig(config)
if err != nil {
return nil, err
}

s3Exporter := newS3Exporter(cfg, "metrics", params)

if config.(*Config).MarshalerName == SumoIC {
return nil, fmt.Errorf("metrics are not supported by sumo_ic output format")
Expand All @@ -60,14 +77,21 @@ func createMetricsExporter(ctx context.Context,
return exporterhelper.NewMetrics(ctx, params,
config,
s3Exporter.ConsumeMetrics,
exporterhelper.WithStart(s3Exporter.start))
exporterhelper.WithStart(s3Exporter.start),
exporterhelper.WithQueue(cfg.QueueSettings),
)
}

func createTracesExporter(ctx context.Context,
params exporter.Settings,
config component.Config,
) (exporter.Traces, error) {
s3Exporter := newS3Exporter(config.(*Config), "traces", params)
cfg, err := checkAndCastConfig(config)
if err != nil {
return nil, err
}

s3Exporter := newS3Exporter(cfg, "traces", params)

if config.(*Config).MarshalerName == SumoIC {
return nil, fmt.Errorf("traces are not supported by sumo_ic output format")
Expand All @@ -77,5 +101,16 @@ func createTracesExporter(ctx context.Context,
params,
config,
s3Exporter.ConsumeTraces,
exporterhelper.WithStart(s3Exporter.start))
exporterhelper.WithStart(s3Exporter.start),
exporterhelper.WithQueue(cfg.QueueSettings),
)
}

// checkAndCastConfig checks the configuration type and casts it to the S3 exporter Config struct.
func checkAndCastConfig(c component.Config) (*Config, error) {
cfg, ok := c.(*Config)
if !ok {
return nil, fmt.Errorf("config structure is not of type *awss3exporter.Config")
}
return cfg, nil
}
5 changes: 5 additions & 0 deletions exporter/awss3exporter/testdata/config.yaml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,11 @@ receivers:

exporters:
awss3:
sending_queue:
enabled: true
num_consumers: 23
queue_size: 42

s3uploader:
region: 'us-east-1'
s3_bucket: 'foo'
Expand Down
4 changes: 1 addition & 3 deletions receiver/datadogreceiver/go.mod
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,8 +1,6 @@
module github.com/open-telemetry/opentelemetry-collector-contrib/receiver/datadogreceiver

go 1.22.7

toolchain go1.23.0
go 1.22.0

require (
github.com/DataDog/agent-payload/v5 v5.0.140
Expand Down
Loading