go.temporal.io/server/common/telemetry/config.go
422 LOC · 136 covered · 286 uncovered · 35 ranges · 1016 concepts · 9 introducers · 503 tests
File neighbourhood
The centred file is linked to every concept that introduces one of its ranges, every test that runs code from the file, and the gray connector concepts standing between those tests and the file's own introducer concepts. Undirected links join concepts to every file where they introduce source and concepts to the tests they introduce; arrows show specialization between the displayed concepts and bridge only concepts omitted from this view. Concept colors match the source ranges below; connector concepts have no source color and are shown in gray.
Focused file, its introducer and connector concepts, their introduced files, and tests that run code from the file
In the embedded map, ordinary wheel input scrolls the page; use the visible controls to zoom and drag to pan. Open the full-screen map for canvas navigation: wheel pans, Ctrl/Command plus wheel zooms, and arrow keys pan when this region is focused. On touch screens, open the full-screen map to pan or pinch. If JavaScript or WebGL is unavailable, use the related-file, concept, and source links on this page.
Graph controls are ready.
Interactive rendering requires JavaScript and WebGL. Use the related-file, concept, and source links on this page while the interactive map is unavailable.
1
package telemetry
2
3
import (
4
"cmp"
5
"context"
6
"fmt"
7
"os"
8
"strconv"
9
"strings"
10
"sync"
11
"time"
12
13
"go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc"
14
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
15
"go.opentelemetry.io/otel/sdk/metric"
16
otelsdktrace "go.opentelemetry.io/otel/sdk/trace"
17
"go.opentelemetry.io/otel/trace"
18
otelnoop "go.opentelemetry.io/otel/trace/noop"
19
"google.golang.org/grpc"
20
"google.golang.org/grpc/backoff"
21
"google.golang.org/grpc/credentials/insecure"
22
"gopkg.in/yaml.v3"
23
)
24
25
const (
26
debugModeEnvVar = "TEMPORAL_OTEL_DEBUG"
27
28
// the following defaults were taken from the grpc docs as of grpc v1.46.
29
// they are not available programmatically
30
31
defaultReadBufferSize = 32 * 1024
32
defaultWriteBufferSize = 32 * 1024
33
defaultMinConnectTimeout = 10 * time.Second
34
35
// the following defaults were taken from the otel library as of v1.7.
36
// they are not available programmatically
37
38
retryDefaultEnabled = true
39
retryDefaultInitialInterval = 5 * time.Second
40
retryDefaultMaxInterval = 30 * time.Second
41
retryDefaultMaxElapsedTime = 1 * time.Minute
42
)
43
44
var (
45
NoopTracerProvider = otelnoop.NewTracerProvider()
46
NoopTracer = NoopTracerProvider.Tracer("")
47
)
48
49
type (
50
metadata struct {
51
Name string
52
Labels map[string]string
53
}
54
55
connection struct {
56
Kind string
57
Metadata metadata
58
Spec any `yaml:"-"`
59
}
60
61
grpcconn struct {
62
Endpoint string
63
Block bool
64
ConnectParams struct {
65
MinConnectTimeout time.Duration `yaml:"min_connect_timeout"`
66
Backoff struct {
67
BaseDelay time.Duration `yaml:"base_delay"`
68
Multiplier float64
69
Jitter float64
70
MaxDelay time.Duration `yaml:"max_delay"`
71
}
72
} `yaml:"connect_params"`
73
UserAgent string `yaml:"user_agent"`
74
ReadBufferSize int `yaml:"read_buffer_size"`
75
WriteBufferSize int `yaml:"write_buffer_size"`
76
Authority string
77
Insecure bool
78
79
cc *grpc.ClientConn
80
}
81
82
exporter struct {
83
Kind struct {
84
Signal string
85
Model string
86
Protocol string
87
}
88
Metadata metadata
89
Spec any `yaml:"-"`
90
}
91
92
otlpGrpcExporter struct {
93
ConnectionName string `yaml:"connection_name"`
94
Connection grpcconn
95
Headers map[string]string
96
Timeout time.Duration
97
Retry struct {
98
Enabled bool
99
InitialInterval time.Duration `yaml:"initial_interval"`
100
MaxInterval time.Duration `yaml:"max_interval"`
101
MaxElapsedTime time.Duration `yaml:"max_elapsed_time"`
102
}
103
}
104
105
otlpGrpcSpanExporter struct {
106
otlpGrpcExporter `yaml:",inline"`
107
}
108
otlpGrpcMetricExporter struct {
109
otlpGrpcExporter `yaml:",inline"`
110
}
111
112
exportConfig struct {
113
Connections []connection
114
Exporters []exporter
115
}
116
117
// sharedConnSpanExporter and sharedConnMetricExporter exist to wrap a span
118
// exporter that uses a shared *grpc.ClientConn so that the grpc.Dial call
119
// doesn't happen until Start() is called. Without this wrapper the
120
// grpc.ClientConn (which can only be created via grpc.Dial or
121
// grpc.DialContext) would need to exist at _construction_ time, meaning
122
// that we would need to dial at construction rather then during the start
123
// phase.
124
125
sharedConnSpanExporter struct {
126
baseOpts []otlptracegrpc.Option
127
dialer interface {
128
Dial() (*grpc.ClientConn, error)
129
}
130
startOnce sync.Once
131
otelsdktrace.SpanExporter
132
}
133
134
sharedConnMetricExporter struct {
135
baseOpts []otlpmetricgrpc.Option
136
dialer interface {
137
Dial() (*grpc.ClientConn, error)
138
}
139
startOnce sync.Once
140
metric.Exporter
141
}
142
143
// ExportConfig represents YAML structured configuration for a set of OTEL
144
// trace/span/log exporters.
145
ExportConfig struct {
146
inner exportConfig `yaml:",inline"`
147
// CustomExporters is for testing.
148
CustomExporters map[SpanExporterType]otelsdktrace.SpanExporter `yaml:"-"`
149
}
150
151
SpanExporterType string
152
)
153
154
// UnmarshalYAML loads the state of an ExportConfig from parsed YAML
155
func (ec *ExportConfig) UnmarshalYAML(n *yaml.Node) error {
156
return n.Decode(&ec.inner)
157
}
158
159
>
func (ec *ExportConfig) SpanExporters() (map[SpanExporterType]otelsdktrace.SpanExporter, error) {
config.go ×1
160
>
return ec.inner.SpanExporters()
161
>
}
162
163
func (ec *ExportConfig) MetricExporters() ([]metric.Exporter, error) {
164
return ec.inner.MetricExporters()
165
}
166
167
// Dial returns the cached *grpc.ClientConn instance or creates a new one,
168
// caches and then returns it. This function is not threadsafe.
169
func (g *grpcconn) Dial() (*grpc.ClientConn, error) {
170
var err error
171
if g.cc == nil {
172
g.cc, err = grpc.NewClient(g.Endpoint, g.dialOpts()...)
173
}
174
return g.cc, err
175
}
176
178
>
out := []grpc.DialOption{
179
>
grpc.WithReadBufferSize(cmp.Or(g.ReadBufferSize, defaultReadBufferSize)),
180
>
grpc.WithWriteBufferSize(cmp.Or(g.WriteBufferSize, defaultWriteBufferSize)),
181
>
grpc.WithUserAgent(g.UserAgent),
182
>
grpc.WithConnectParams(grpc.ConnectParams{
183
>
MinConnectTimeout: cmp.Or(g.ConnectParams.MinConnectTimeout, defaultMinConnectTimeout),
184
>
Backoff: backoff.Config{
185
>
BaseDelay: cmp.Or(g.ConnectParams.Backoff.BaseDelay, backoff.DefaultConfig.BaseDelay),
186
>
MaxDelay: cmp.Or(g.ConnectParams.Backoff.MaxDelay, backoff.DefaultConfig.MaxDelay),
187
>
Jitter: cmp.Or(g.ConnectParams.Backoff.Jitter, backoff.DefaultConfig.Jitter),
188
>
Multiplier: cmp.Or(g.ConnectParams.Backoff.Multiplier, backoff.DefaultConfig.Multiplier),
189
>
},
190
>
}),
191
>
}
192
>
if g.Insecure {
193
out = append(out, grpc.WithTransportCredentials(insecure.NewCredentials()))
194
}
196
out = append(out, grpc.WithBlock())
197
}
199
out = append(out, grpc.WithAuthority(g.Authority))
200
}
202
}
203
204
// SpanExporters builds the set of OTEL SpanExporter objects defined by the YAML
205
// unmarshalled into this ExportConfig object. The returned SpanExporters have
206
// not been started.
207
>
func (ec *exportConfig) SpanExporters() (map[SpanExporterType]otelsdktrace.SpanExporter, error) {
config.go ×2
208
>
out := make(map[SpanExporterType]otelsdktrace.SpanExporter, len(ec.Exporters))
209
>
for _, expcfg := range ec.Exporters {
211
>
continue
212
}
214
>
case *otlpGrpcSpanExporter:
215
>
spanexp, err := ec.buildOtlpGrpcSpanExporter(spec)
216
>
if err != nil {
217
return nil, err
218
}
220
default:
221
return nil, fmt.Errorf("unsupported span exporter type: %T", spec)
222
}
223
}
225
}
226
228
>
out := make([]metric.Exporter, 0, len(ec.Exporters))
229
>
for _, expcfg := range ec.Exporters {
230
>
if !strings.HasPrefix(expcfg.Kind.Signal, "metric") {
231
>
continue
232
}
234
>
case *otlpGrpcMetricExporter:
235
>
metricexp, err := ec.buildOtlpGrpcMetricExporter(spec)
236
>
if err != nil {
237
return nil, err
238
}
240
default:
241
return nil, fmt.Errorf("unsupported metric exporter type: %T", spec)
242
}
243
}
245
246
}
247
248
func (ec *exportConfig) buildOtlpGrpcMetricExporter(
249
cfg *otlpGrpcMetricExporter,
251
>
dopts := cfg.Connection.dialOpts()
252
>
opts := []otlpmetricgrpc.Option{
253
>
otlpmetricgrpc.WithEndpoint(cfg.Connection.Endpoint),
254
>
otlpmetricgrpc.WithHeaders(cfg.Headers),
255
>
otlpmetricgrpc.WithTimeout(cmp.Or(cfg.Timeout, 10*time.Second)),
256
>
otlpmetricgrpc.WithDialOption(dopts...),
257
>
otlpmetricgrpc.WithRetry(otlpmetricgrpc.RetryConfig{
258
>
Enabled: cmp.Or(cfg.Retry.Enabled, retryDefaultEnabled),
259
>
InitialInterval: cmp.Or(cfg.Retry.InitialInterval, retryDefaultInitialInterval),
260
>
MaxInterval: cmp.Or(cfg.Retry.MaxInterval, retryDefaultMaxInterval),
261
>
MaxElapsedTime: cmp.Or(cfg.Retry.MaxElapsedTime, retryDefaultMaxElapsedTime),
262
>
}),
263
>
}
264
>
265
>
// work around https://github.com/open-telemetry/opentelemetry-go/issues/2940
266
>
if cfg.Connection.Insecure {
267
opts = append(opts, otlpmetricgrpc.WithInsecure())
268
}
269
271
return otlpmetricgrpc.New(context.Background(), opts...)
272
}
273
275
>
if !ok {
276
return nil, fmt.Errorf("OTEL exporter connection %q not found", cfg.ConnectionName)
277
}
279
>
baseOpts: opts,
280
>
dialer: conncfg,
281
>
}, nil
282
}
283
284
func (ec *exportConfig) buildOtlpGrpcSpanExporter(
285
cfg *otlpGrpcSpanExporter,
287
>
opts := []otlptracegrpc.Option{
288
>
otlptracegrpc.WithEndpoint(cfg.Connection.Endpoint),
289
>
otlptracegrpc.WithHeaders(cfg.Headers),
290
>
otlptracegrpc.WithTimeout(cmp.Or(cfg.Timeout, 10*time.Second)),
291
>
otlptracegrpc.WithDialOption(cfg.Connection.dialOpts()...),
292
>
otlptracegrpc.WithRetry(otlptracegrpc.RetryConfig{
293
>
Enabled: cmp.Or(cfg.Retry.Enabled, retryDefaultEnabled),
294
>
InitialInterval: cmp.Or(cfg.Retry.InitialInterval, retryDefaultInitialInterval),
295
>
MaxInterval: cmp.Or(cfg.Retry.MaxInterval, retryDefaultMaxInterval),
296
>
MaxElapsedTime: cmp.Or(cfg.Retry.MaxElapsedTime, retryDefaultMaxElapsedTime),
297
>
}),
298
>
}
299
>
300
>
// work around https://github.com/open-telemetry/opentelemetry-go/issues/2940
301
>
if cfg.Connection.Insecure {
302
opts = append(opts, otlptracegrpc.WithInsecure())
303
}
304
306
return otlptracegrpc.NewUnstarted(opts...), nil
307
}
308
310
>
if !ok {
311
return nil, fmt.Errorf("OTEL exporter connection %q not found", cfg.ConnectionName)
312
}
314
>
baseOpts: opts,
315
>
dialer: conncfg,
316
>
}, nil
317
}
318
319
// Start initiates the connection to an upstream grpc OTLP server
320
func (scse *sharedConnSpanExporter) Start(ctx context.Context) error {
321
var err error
322
scse.startOnce.Do(func() {
323
var cc *grpc.ClientConn
324
cc, err = scse.dialer.Dial()
325
if err != nil {
326
return
327
}
328
opts := append(scse.baseOpts, otlptracegrpc.WithGRPCConn(cc))
329
scse.SpanExporter, err = otlptracegrpc.New(ctx, opts...)
330
})
331
return err
332
}
333
334
// Start initiates the connection to an upstream grpc OTLP server
335
func (scme *sharedConnMetricExporter) Start(ctx context.Context) error {
336
var err error
337
scme.startOnce.Do(func() {
338
var cc *grpc.ClientConn
339
cc, err = scme.dialer.Dial()
340
if err != nil {
341
return
342
}
343
opts := append(scme.baseOpts, otlpmetricgrpc.WithGRPCConn(cc))
344
scme.Exporter, err = otlpmetricgrpc.New(ctx, opts...)
345
})
346
return err
347
}
348
350
>
if name == "" {
351
return nil, false
352
}
354
>
if gconn, ok := conn.Spec.(*grpcconn); ok && conn.Metadata.Name == name {
355
>
return gconn, true
356
>
}
357
}
358
return nil, false
359
}
360
361
// UnmarshalYAML loads the state of a generic connection from parsed YAML
363
>
type conn connection
364
>
type overlay struct {
365
>
*conn `yaml:",inline"`
366
>
Spec yaml.Node `yaml:"spec"`
367
>
}
368
>
obj := overlay{conn: (*conn)(c)}
369
>
err := n.Decode(&obj)
370
>
if err != nil {
371
return err
372
}
374
>
case "grpc":
375
>
c.Spec = &grpcconn{}
376
default:
377
return fmt.Errorf("unsupported connection kind: %q", c.Kind)
378
}
380
}
381
382
// UnmarshalYAML loads the state of a generic exporter from parsed YAML
384
>
type exp exporter
385
>
type overlay struct {
386
>
*exp `yaml:",inline"`
387
>
Spec yaml.Node `yaml:"spec"`
388
>
}
389
>
obj := overlay{exp: (*exp)(e)}
390
>
err := n.Decode(&obj)
391
>
if err != nil {
392
return err
393
}
394
>
descriptor := fmt.Sprintf("%v+%v+%v", e.Kind.Signal, e.Kind.Model, e.Kind.Protocol)
config.go ×3
395
>
switch descriptor {
396
>
case "traces+otlp+grpc", "trace+otlp+grpc":
397
>
e.Spec = new(otlpGrpcSpanExporter)
399
>
e.Spec = new(otlpGrpcMetricExporter)
400
default:
401
return fmt.Errorf(
402
"unsupported exporter kind: signal=%q; model=%q; protocol=%q",
403
e.Kind.Signal,
404
e.Kind.Model,
405
e.Kind.Protocol,
406
)
407
}
409
}
410
412
>
isDebug, err := strconv.ParseBool(os.Getenv(debugModeEnvVar))
413
>
if err != nil {
415
>
}
417
}
418
420
>
_, isNoop := t.(otelnoop.Tracer)
421
>
return !isNoop
422
>
}