-
Notifications
You must be signed in to change notification settings - Fork 1.4k
/
receiver.go
226 lines (189 loc) · 7.58 KB
/
receiver.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
// Copyright The OpenTelemetry Authors
// SPDX-License-Identifier: Apache-2.0
package receiver // import "go.opentelemetry.io/collector/receiver"
import (
"context"
"fmt"
"go.opentelemetry.io/collector/component"
"go.opentelemetry.io/collector/consumer"
"go.opentelemetry.io/collector/pipeline"
)
// Traces receiver receives traces.
// Its purpose is to translate data from any format to the collector's internal trace format.
// TracesReceiver feeds a consumer.Traces with data.
//
// For example, it could be Zipkin data source which translates Zipkin spans into ptrace.Traces.
type Traces interface {
component.Component
}
// Metrics receiver receives metrics.
// Its purpose is to translate data from any format to the collector's internal metrics format.
// MetricsReceiver feeds a consumer.Metrics with data.
//
// For example, it could be Prometheus data source which translates Prometheus metrics into pmetric.Metrics.
type Metrics interface {
component.Component
}
// Logs receiver receives logs.
// Its purpose is to translate data from any format to the collector's internal logs data format.
// LogsReceiver feeds a consumer.Logs with data.
//
// For example, it could be a receiver that reads syslogs and convert them into plog.Logs.
type Logs interface {
component.Component
}
// Settings configures Receiver creators.
type Settings struct {
// ID returns the ID of the component that will be created.
ID component.ID
component.TelemetrySettings
// BuildInfo can be used by components for informational purposes.
BuildInfo component.BuildInfo
}
// Factory is a factory interface for receivers.
//
// This interface cannot be directly implemented. Implementations must
// use the NewReceiverFactory to implement it.
type Factory interface {
component.Factory
// CreateTracesReceiver creates a TracesReceiver based on this config.
// If the receiver type does not support traces,
// this function returns the error [pipeline.ErrSignalNotSupported].
// Implementers can assume `nextConsumer` is never nil.
CreateTracesReceiver(ctx context.Context, set Settings, cfg component.Config, nextConsumer consumer.Traces) (Traces, error)
// TracesReceiverStability gets the stability level of the TracesReceiver.
TracesReceiverStability() component.StabilityLevel
// CreateMetricsReceiver creates a MetricsReceiver based on this config.
// If the receiver type does not support metrics,
// this function returns the error [pipeline.ErrSignalNotSupported].
// Implementers can assume `nextConsumer` is never nil.
CreateMetricsReceiver(ctx context.Context, set Settings, cfg component.Config, nextConsumer consumer.Metrics) (Metrics, error)
// MetricsReceiverStability gets the stability level of the MetricsReceiver.
MetricsReceiverStability() component.StabilityLevel
// CreateLogsReceiver creates a LogsReceiver based on this config.
// If the receiver type does not support logs,
// this function returns the error [pipeline.ErrSignalNotSupported].
// Implementers can assume `nextConsumer` is never nil.
CreateLogsReceiver(ctx context.Context, set Settings, cfg component.Config, nextConsumer consumer.Logs) (Logs, error)
// LogsReceiverStability gets the stability level of the LogsReceiver.
LogsReceiverStability() component.StabilityLevel
unexportedFactoryFunc()
}
// FactoryOption apply changes to ReceiverOptions.
type FactoryOption interface {
// applyOption applies the option.
applyOption(o *factory)
}
// factoryOptionFunc is an ReceiverFactoryOption created through a function.
type factoryOptionFunc func(*factory)
func (f factoryOptionFunc) applyOption(o *factory) {
f(o)
}
// CreateTracesFunc is the equivalent of Factory.CreateTraces.
type CreateTracesFunc func(context.Context, Settings, component.Config, consumer.Traces) (Traces, error)
// CreateTracesReceiver implements Factory.CreateTracesReceiver().
func (f CreateTracesFunc) CreateTracesReceiver(
ctx context.Context,
set Settings,
cfg component.Config,
nextConsumer consumer.Traces) (Traces, error) {
if f == nil {
return nil, pipeline.ErrSignalNotSupported
}
return f(ctx, set, cfg, nextConsumer)
}
// CreateMetricsFunc is the equivalent of Factory.CreateMetrics.
type CreateMetricsFunc func(context.Context, Settings, component.Config, consumer.Metrics) (Metrics, error)
// CreateMetricsReceiver implements Factory.CreateMetricsReceiver().
func (f CreateMetricsFunc) CreateMetricsReceiver(
ctx context.Context,
set Settings,
cfg component.Config,
nextConsumer consumer.Metrics,
) (Metrics, error) {
if f == nil {
return nil, pipeline.ErrSignalNotSupported
}
return f(ctx, set, cfg, nextConsumer)
}
// CreateLogsFunc is the equivalent of ReceiverFactory.CreateLogsReceiver().
type CreateLogsFunc func(context.Context, Settings, component.Config, consumer.Logs) (Logs, error)
// CreateLogsReceiver implements Factory.CreateLogsReceiver().
func (f CreateLogsFunc) CreateLogsReceiver(
ctx context.Context,
set Settings,
cfg component.Config,
nextConsumer consumer.Logs,
) (Logs, error) {
if f == nil {
return nil, pipeline.ErrSignalNotSupported
}
return f(ctx, set, cfg, nextConsumer)
}
type factory struct {
cfgType component.Type
component.CreateDefaultConfigFunc
CreateTracesFunc
tracesStabilityLevel component.StabilityLevel
CreateMetricsFunc
metricsStabilityLevel component.StabilityLevel
CreateLogsFunc
logsStabilityLevel component.StabilityLevel
}
func (f *factory) Type() component.Type {
return f.cfgType
}
func (f *factory) unexportedFactoryFunc() {}
func (f *factory) TracesReceiverStability() component.StabilityLevel {
return f.tracesStabilityLevel
}
func (f *factory) MetricsReceiverStability() component.StabilityLevel {
return f.metricsStabilityLevel
}
func (f *factory) LogsReceiverStability() component.StabilityLevel {
return f.logsStabilityLevel
}
// WithTraces overrides the default "error not supported" implementation for CreateTracesReceiver and the default "undefined" stability level.
func WithTraces(createTracesReceiver CreateTracesFunc, sl component.StabilityLevel) FactoryOption {
return factoryOptionFunc(func(o *factory) {
o.tracesStabilityLevel = sl
o.CreateTracesFunc = createTracesReceiver
})
}
// WithMetrics overrides the default "error not supported" implementation for CreateMetricsReceiver and the default "undefined" stability level.
func WithMetrics(createMetricsReceiver CreateMetricsFunc, sl component.StabilityLevel) FactoryOption {
return factoryOptionFunc(func(o *factory) {
o.metricsStabilityLevel = sl
o.CreateMetricsFunc = createMetricsReceiver
})
}
// WithLogs overrides the default "error not supported" implementation for CreateLogsReceiver and the default "undefined" stability level.
func WithLogs(createLogsReceiver CreateLogsFunc, sl component.StabilityLevel) FactoryOption {
return factoryOptionFunc(func(o *factory) {
o.logsStabilityLevel = sl
o.CreateLogsFunc = createLogsReceiver
})
}
// NewFactory returns a Factory.
func NewFactory(cfgType component.Type, createDefaultConfig component.CreateDefaultConfigFunc, options ...FactoryOption) Factory {
f := &factory{
cfgType: cfgType,
CreateDefaultConfigFunc: createDefaultConfig,
}
for _, opt := range options {
opt.applyOption(f)
}
return f
}
// MakeFactoryMap takes a list of receiver factories and returns a map with factory type as keys.
// It returns a non-nil error when there are factories with duplicate type.
func MakeFactoryMap(factories ...Factory) (map[component.Type]Factory, error) {
fMap := map[component.Type]Factory{}
for _, f := range factories {
if _, ok := fMap[f.Type()]; ok {
return fMap, fmt.Errorf("duplicate receiver factory %q", f.Type())
}
fMap[f.Type()] = f
}
return fMap, nil
}