-
Notifications
You must be signed in to change notification settings - Fork 1.4k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
Showing
8 changed files
with
159 additions
and
48 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
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,47 @@ | ||
// Copyright The OpenTelemetry Authors | ||
// SPDX-License-Identifier: Apache-2.0 | ||
|
||
package consumerprofiles // import "go.opentelemetry.io/collector/consumer/consumerprofiles" | ||
|
||
import ( | ||
"context" | ||
"errors" | ||
|
||
"go.opentelemetry.io/collector/consumer" | ||
"go.opentelemetry.io/collector/consumer/internal" | ||
"go.opentelemetry.io/collector/pdata/pprofile" | ||
) | ||
|
||
var errNilFunc = errors.New("nil consumer func") | ||
|
||
// Profiles is an interface that receives pprofile.Profiles, processes it | ||
// as needed, and sends it to the next processing node if any or to the destination. | ||
type Profiles interface { | ||
internal.BaseConsumer | ||
// ConsumeProfiles receives pprofile.Profiles for consumption. | ||
ConsumeProfiles(ctx context.Context, td pprofile.Profiles) error | ||
} | ||
|
||
// ConsumeProfilesFunc is a helper function that is similar to ConsumeProfiles. | ||
type ConsumeProfilesFunc func(ctx context.Context, td pprofile.Profiles) error | ||
|
||
// ConsumeProfiles calls f(ctx, td). | ||
func (f ConsumeProfilesFunc) ConsumeProfiles(ctx context.Context, td pprofile.Profiles) error { | ||
return f(ctx, td) | ||
} | ||
|
||
type baseProfiles struct { | ||
*internal.BaseImpl | ||
ConsumeProfilesFunc | ||
} | ||
|
||
// NewProfiles returns a Profiles configured with the provided options. | ||
func NewProfiles(consume ConsumeProfilesFunc, options ...consumer.Option) (Profiles, error) { | ||
if consume == nil { | ||
return nil, errNilFunc | ||
} | ||
return &baseProfiles{ | ||
BaseImpl: internal.NewBaseImpl(options...), | ||
ConsumeProfilesFunc: consume, | ||
}, nil | ||
} |
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,51 @@ | ||
// Copyright The OpenTelemetry Authors | ||
// SPDX-License-Identifier: Apache-2.0 | ||
|
||
package consumerprofiles | ||
|
||
import ( | ||
"context" | ||
"errors" | ||
"testing" | ||
|
||
"github.com/stretchr/testify/assert" | ||
|
||
"go.opentelemetry.io/collector/consumer" | ||
"go.opentelemetry.io/collector/pdata/pprofile" | ||
) | ||
|
||
func TestDefaultProfiles(t *testing.T) { | ||
cp, err := NewProfiles(func(context.Context, pprofile.Profiles) error { return nil }) | ||
assert.NoError(t, err) | ||
assert.NoError(t, cp.ConsumeProfiles(context.Background(), pprofile.NewProfiles())) | ||
assert.Equal(t, consumer.Capabilities{MutatesData: false}, cp.Capabilities()) | ||
} | ||
|
||
func TestNilFuncProfiles(t *testing.T) { | ||
_, err := NewProfiles(nil) | ||
assert.Equal(t, errNilFunc, err) | ||
} | ||
|
||
func TestWithCapabilitiesProfiles(t *testing.T) { | ||
cp, err := NewProfiles( | ||
func(context.Context, pprofile.Profiles) error { return nil }, | ||
consumer.WithCapabilities(consumer.Capabilities{MutatesData: true})) | ||
assert.NoError(t, err) | ||
assert.NoError(t, cp.ConsumeProfiles(context.Background(), pprofile.NewProfiles())) | ||
assert.Equal(t, consumer.Capabilities{MutatesData: true}, cp.Capabilities()) | ||
} | ||
|
||
func TestConsumeProfiles(t *testing.T) { | ||
consumeCalled := false | ||
cp, err := NewProfiles(func(context.Context, pprofile.Profiles) error { consumeCalled = true; return nil }) | ||
assert.NoError(t, err) | ||
assert.NoError(t, cp.ConsumeProfiles(context.Background(), pprofile.NewProfiles())) | ||
assert.True(t, consumeCalled) | ||
} | ||
|
||
func TestConsumeProfiles_ReturnError(t *testing.T) { | ||
want := errors.New("my_error") | ||
cp, err := NewProfiles(func(context.Context, pprofile.Profiles) error { return want }) | ||
assert.NoError(t, err) | ||
assert.Equal(t, want, cp.ConsumeProfiles(context.Background(), pprofile.NewProfiles())) | ||
} |
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,42 @@ | ||
// Copyright The OpenTelemetry Authors | ||
// SPDX-License-Identifier: Apache-2.0 | ||
|
||
package internal // import "go.opentelemetry.io/collector/consumer/internal" | ||
|
||
// Capabilities describes the capabilities of a Processor. | ||
type Capabilities struct { | ||
// MutatesData is set to true if Consume* function of the | ||
// processor modifies the input Traces, Logs or Metrics argument. | ||
// Processors which modify the input data MUST set this flag to true. If the processor | ||
// does not modify the data it MUST set this flag to false. If the processor creates | ||
// a copy of the data before modifying then this flag can be safely set to false. | ||
MutatesData bool | ||
} | ||
|
||
type BaseConsumer interface { | ||
Capabilities() Capabilities | ||
} | ||
|
||
type BaseImpl struct { | ||
Cap Capabilities | ||
} | ||
|
||
// Option to construct new consumers. | ||
type Option func(*BaseImpl) | ||
|
||
// Capabilities returns the capabilities of the component | ||
func (bs BaseImpl) Capabilities() Capabilities { | ||
return bs.Cap | ||
} | ||
|
||
func NewBaseImpl(options ...Option) *BaseImpl { | ||
bs := &BaseImpl{ | ||
Cap: Capabilities{MutatesData: false}, | ||
} | ||
|
||
for _, op := range options { | ||
op(bs) | ||
} | ||
|
||
return bs | ||
} |
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