/
controller.go
312 lines (274 loc) · 8.53 KB
/
controller.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
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
// 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 basic // import "go.opentelemetry.io/otel/sdk/metric/controller/basic"
import (
"context"
"fmt"
"sync"
"time"
"go.opentelemetry.io/otel"
"go.opentelemetry.io/otel/metric"
"go.opentelemetry.io/otel/metric/registry"
export "go.opentelemetry.io/otel/sdk/export/metric"
sdk "go.opentelemetry.io/otel/sdk/metric"
controllerTime "go.opentelemetry.io/otel/sdk/metric/controller/time"
"go.opentelemetry.io/otel/sdk/resource"
)
// DefaultPeriod is used for:
//
// - the minimum time between calls to Collect()
// - the timeout for Export()
// - the timeout for Collect().
const DefaultPeriod = 10 * time.Second
// ErrControllerStarted indicates that a controller was started more
// than once.
var ErrControllerStarted = fmt.Errorf("controller already started")
// Controller organizes and synchronizes collection of metric data in
// both "pull" and "push" configurations. This supports two distinct
// modes:
//
// - Push and Pull: Start() must be called to begin calling the pusher;
// Collect() is called periodically by a background thread after starting
// the controller.
// - Pull-Only: Start() is optional in this case, to call Collect periodically.
// If Start() is not called, Collect() can be called manually to initiate
// collection
//
// The controller supports mixing push and pull access to metric data
// using the export.CheckpointSet RWLock interface. Collection will
// be blocked by a pull request in the basic controller.
type Controller struct {
lock sync.Mutex
accumulator *sdk.Accumulator
provider *registry.MeterProvider
checkpointer export.Checkpointer
pusher export.Exporter
wg sync.WaitGroup
stopCh chan struct{}
clock controllerTime.Clock
ticker controllerTime.Ticker
collectPeriod time.Duration
collectTimeout time.Duration
pushTimeout time.Duration
// collectedTime is used only in configurations with no
// pusher, when ticker != nil.
collectedTime time.Time
}
// New constructs a Controller using the provided checkpointer and
// options (including optional Pusher) to configure a metric
// export pipeline.
func New(checkpointer export.Checkpointer, opts ...Option) *Controller {
c := &Config{
CollectPeriod: DefaultPeriod,
CollectTimeout: DefaultPeriod,
PushTimeout: DefaultPeriod,
}
for _, opt := range opts {
opt.Apply(c)
}
if c.Resource == nil {
c.Resource = resource.Default()
}
impl := sdk.NewAccumulator(
checkpointer,
c.Resource,
)
return &Controller{
provider: registry.NewMeterProvider(impl),
accumulator: impl,
checkpointer: checkpointer,
pusher: c.Pusher,
stopCh: nil,
clock: controllerTime.RealClock{},
collectPeriod: c.CollectPeriod,
collectTimeout: c.CollectTimeout,
pushTimeout: c.PushTimeout,
}
}
// SetClock supports setting a mock clock for testing. This must be
// called before Start().
func (c *Controller) SetClock(clock controllerTime.Clock) {
c.lock.Lock()
defer c.lock.Unlock()
c.clock = clock
}
// MeterProvider returns a MeterProvider instance for this controller.
func (c *Controller) MeterProvider() metric.MeterProvider {
return c.provider
}
// Start begins a ticker that periodically collects and exports
// metrics with the configured interval. This is required for calling
// a configured Exporter (see WithExporter) and is otherwise optional
// when only pulling metric data.
//
// The passed context is passed to Collect() and subsequently to
// asynchronous instrument callbacks. Returns an error when the
// controller was already started.
//
// Note that it is not necessary to Start a controller when only
// pulling data; use the Collect() and ForEach() methods directly in
// this case.
func (c *Controller) Start(ctx context.Context) error {
c.lock.Lock()
defer c.lock.Unlock()
if c.stopCh != nil {
return ErrControllerStarted
}
c.wg.Add(1)
c.stopCh = make(chan struct{})
c.ticker = c.clock.Ticker(c.collectPeriod)
go c.runTicker(ctx, c.stopCh)
return nil
}
// Stop waits for the background goroutine to return and then collects
// and exports metrics one last time before returning. The passed
// context is passed to the final Collect() and subsequently to the
// final asynchronous instruments.
//
// Note that Stop() will not cancel an ongoing collection or export.
func (c *Controller) Stop(ctx context.Context) error {
c.lock.Lock()
defer c.lock.Unlock()
if c.stopCh == nil {
return nil
}
close(c.stopCh)
c.stopCh = nil
c.wg.Wait()
c.ticker.Stop()
c.ticker = nil
return c.collect(ctx)
}
// runTicker collection on ticker events until the stop channel is closed.
func (c *Controller) runTicker(ctx context.Context, stopCh chan struct{}) {
defer c.wg.Done()
for {
select {
case <-stopCh:
return
case <-c.ticker.C():
if err := c.collect(ctx); err != nil {
otel.Handle(err)
}
}
}
}
// collect computes a checkpoint and optionally exports it.
func (c *Controller) collect(ctx context.Context) error {
if err := c.checkpoint(ctx, func() bool {
return true
}); err != nil {
return err
}
if c.pusher == nil {
return nil
}
// Note: this is not subject to collectTimeout. This blocks the next
// collection despite collectTimeout because it holds a lock.
if err := c.export(ctx); err != nil {
return err
}
return nil
}
// checkpoint calls the Accumulator and Checkpointer interfaces to
// compute the CheckpointSet. This applies the configured collection
// timeout. Note that this does not try to cancel a Collect or Export
// when Stop() is called.
func (c *Controller) checkpoint(ctx context.Context, cond func() bool) error {
ckpt := c.checkpointer.CheckpointSet()
ckpt.Lock()
defer ckpt.Unlock()
if !cond() {
return nil
}
c.checkpointer.StartCollection()
if c.collectTimeout > 0 {
var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, c.collectTimeout)
defer cancel()
}
_ = c.accumulator.Collect(ctx)
var err error
select {
case <-ctx.Done():
err = ctx.Err()
default:
// The context wasn't done, ok.
}
// Finish the checkpoint whether the accumulator timed out or not.
if cerr := c.checkpointer.FinishCollection(); cerr != nil {
if err == nil {
err = cerr
} else {
err = fmt.Errorf("%s: %w", cerr.Error(), err)
}
}
return err
}
// export calls the exporter with a read lock on the CheckpointSet,
// applying the configured export timeout.
func (c *Controller) export(ctx context.Context) error {
ckpt := c.checkpointer.CheckpointSet()
ckpt.RLock()
defer ckpt.RUnlock()
if c.pushTimeout > 0 {
var cancel context.CancelFunc
ctx, cancel = context.WithTimeout(ctx, c.pushTimeout)
defer cancel()
}
return c.pusher.Export(ctx, ckpt)
}
// Foreach gives the caller read-locked access to the current
// export.CheckpointSet.
func (c *Controller) ForEach(ks export.ExportKindSelector, f func(export.Record) error) error {
ckpt := c.checkpointer.CheckpointSet()
ckpt.RLock()
defer ckpt.RUnlock()
return ckpt.ForEach(ks, f)
}
// IsRunning returns true if the controller was started via Start(),
// indicating that the current export.CheckpointSet is being kept
// up-to-date.
func (c *Controller) IsRunning() bool {
c.lock.Lock()
defer c.lock.Unlock()
return c.ticker != nil
}
// Collect requests a collection. The collection will be skipped if
// the last collection is aged less than the configured collection
// period.
func (c *Controller) Collect(ctx context.Context) error {
if c.IsRunning() {
// When there's a non-nil ticker, there's a goroutine
// computing checkpoints with the collection period.
return ErrControllerStarted
}
return c.checkpoint(ctx, c.shouldCollect)
}
// shouldCollect returns true if the collector should collect now,
// based on the timestamp, the last collection time, and the
// configured period.
func (c *Controller) shouldCollect() bool {
// This is called with the CheckpointSet exclusive
// lock held.
if c.collectPeriod == 0 {
return true
}
now := c.clock.Now()
if now.Sub(c.collectedTime) < c.collectPeriod {
return false
}
c.collectedTime = now
return true
}