Introduce additional consumer packages split by telemetry type

CHANGELOG.md

@@ -48,6 +48,8 @@
   In case of type mismatch, they don't panic right away but return an invalid zero-initialized
   instance for consistency with other OneOf field accessors (#5034)
 - Update OTLP to v0.15.0 (#5064)
+- Create additional pdata packages separated by type for further split of pdata (#4918)
+- Create additional consumer packages split by telemetry type (#5086)
 
 ### 🧰 Bug fixes 🧰
 

consumer/consumer.go

@@ -15,53 +15,23 @@
 package consumer // import "go.opentelemetry.io/collector/consumer"
 
 import (
-	"errors"
+	"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 TraceData or MetricsData 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
-}
-
-var errNilFunc = errors.New("nil consumer func")
-
-type baseImpl struct {
-	capabilities Capabilities
-}
+type Capabilities = internal.Capabilities
 
 // Option to construct new consumers.
-type Option func(*baseImpl)
+type Option = internal.Option
 
 // WithCapabilities overrides the default GetCapabilities function for a processor.
 // The default GetCapabilities function returns mutable capabilities.
-func WithCapabilities(capabilities Capabilities) Option {
-	return func(o *baseImpl) {
-		o.capabilities = capabilities
-	}
-}
+var WithCapabilities = internal.WithCapabilities
 
-// Capabilities implementation of the base
-func (bs baseImpl) Capabilities() Capabilities {
-	return bs.capabilities
-}
+type baseConsumer = internal.BaseConsumer
 
-func newBaseImpl(options ...Option) *baseImpl {
-	bs := &baseImpl{
-		capabilities: Capabilities{MutatesData: false},
-	}
+var errNilFunc = internal.ErrNilFunc
 
-	for _, op := range options {
-		op(bs)
-	}
+type baseImpl = internal.BaseImpl
 
-	return bs
-}
+var newBaseImpl = internal.NewBaseImpl

consumer/internal/consumer.go

@@ -0,0 +1,67 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package internal // import "go.opentelemetry.io/collector/consumer/internal"
+
+import (
+	"errors"
+)
+
+// Capabilities describes the capabilities of a Processor.
+type Capabilities struct {
+	// MutatesData is set to true if Consume* function of the
+	// processor modifies the input TraceData or MetricsData 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
+}
+
+var ErrNilFunc = errors.New("nil consumer func")
+
+type BaseImpl struct {
+	Cap Capabilities
+}
+
+// Option to construct new consumers.
+type Option func(*BaseImpl)
+
+// WithCapabilities overrides the default GetCapabilities function for a processor.
+// The default GetCapabilities function returns mutable capabilities.
+func WithCapabilities(capabilities Capabilities) Option {
+	return func(o *BaseImpl) {
+		o.Cap = capabilities
+	}
+}
+
+// Capabilities implementation of the base
+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
+}

consumer/logs/consumer.go

@@ -0,0 +1,54 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package logs // import "go.opentelemetry.io/collector/consumer/logs"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/consumer/internal"
+	"go.opentelemetry.io/collector/model/pdata/logs"
+)
+
+// Consumer is an interface that receives logs.Logs, processes it
+// as needed, and sends it to the next processing node if any or to the destination.
+type Consumer interface {
+	internal.BaseConsumer
+	// Consume receives logs.Logs for consumption.
+	Consume(context.Context, logs.Logs) error
+}
+
+// ConsumeFunc is a helper function that is similar to Consume.
+type ConsumeFunc func(context.Context, logs.Logs) error
+
+// Consume calls f(ctx, l).
+func (f ConsumeFunc) Consume(ctx context.Context, l logs.Logs) error {
+	return f(ctx, l)
+}
+
+type base struct {
+	*internal.BaseImpl
+	ConsumeFunc
+}
+
+// NewConsumer returns a Logs configured with the provided options.
+func NewConsumer(consume ConsumeFunc, options ...internal.Option) (Consumer, error) {
+	if consume == nil {
+		return nil, internal.ErrNilFunc
+	}
+	return &base{
+		BaseImpl:    internal.NewBaseImpl(options...),
+		ConsumeFunc: consume,
+	}, nil
+}

consumer/metrics/consumer.go

@@ -0,0 +1,54 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package metrics // import "go.opentelemetry.io/collector/consumer/metrics"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/consumer/internal"
+	"go.opentelemetry.io/collector/model/pdata/metrics"
+)
+
+// Consumer is an interface that receives metrics.Metrics, processes it
+// as needed, and sends it to the next processing node if any or to the destination.
+type Consumer interface {
+	internal.BaseConsumer
+	// Consume receives metrics.Metrics for consumption.
+	Consume(context.Context, metrics.Metrics) error
+}
+
+// ConsumeFunc is a helper function that is similar to Consume.
+type ConsumeFunc func(context.Context, metrics.Metrics) error
+
+// Consume calls f(ctx, m).
+func (f ConsumeFunc) Consume(ctx context.Context, m metrics.Metrics) error {
+	return f(ctx, m)
+}
+
+type base struct {
+	*internal.BaseImpl
+	ConsumeFunc
+}
+
+// NewConsumer returns a Metrics configured with the provided options.
+func NewConsumer(consume ConsumeFunc, options ...internal.Option) (Consumer, error) {
+	if consume == nil {
+		return nil, internal.ErrNilFunc
+	}
+	return &base{
+		BaseImpl:    internal.NewBaseImpl(options...),
+		ConsumeFunc: consume,
+	}, nil
+}

consumer/traces/consumer.go

@@ -0,0 +1,54 @@
+// Copyright The OpenTelemetry Authors
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//       http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package traces // import "go.opentelemetry.io/collector/consumer/traces"
+
+import (
+	"context"
+
+	"go.opentelemetry.io/collector/consumer/internal"
+	"go.opentelemetry.io/collector/model/pdata/traces"
+)
+
+// Consumer is an interface that receives traces.Traces, processes it
+// as needed, and sends it to the next processing node if any or to the destination.
+type Consumer interface {
+	internal.BaseConsumer
+	// Consume receives traces.Traces for consumption.
+	Consume(context.Context, traces.Traces) error
+}
+
+// ConsumeFunc is a helper function that is similar to Consume.
+type ConsumeFunc func(context.Context, traces.Traces) error
+
+// Consume calls f(ctx, t).
+func (f ConsumeFunc) Consume(ctx context.Context, t traces.Traces) error {
+	return f(ctx, t)
+}
+
+type base struct {
+	*internal.BaseImpl
+	ConsumeFunc
+}
+
+// NewConsumer returns a Traces configured with the provided options.
+func NewConsumer(consume ConsumeFunc, options ...internal.Option) (Consumer, error) {
+	if consume == nil {
+		return nil, internal.ErrNilFunc
+	}
+	return &base{
+		BaseImpl:    internal.NewBaseImpl(options...),
+		ConsumeFunc: consume,
+	}, nil
+}

model/internal/cmd/pdatagen/internal/log_structs.go

@@ -15,7 +15,7 @@
 package internal // import "go.opentelemetry.io/collector/model/internal/cmd/pdatagen/internal"
 
 var logFile = &File{
-	Name: "log",
+	Name: "logs",
 	imports: []string{
 		`"sort"`,
 		``,

model/internal/cmd/pdatagen/internal/trace_structs.go

@@ -15,7 +15,7 @@
 package internal // import "go.opentelemetry.io/collector/model/internal/cmd/pdatagen/internal"
 
 var traceFile = &File{
-	Name: "trace",
+	Name: "traces",
 	imports: []string{
 		`"sort"`,
 		``,