-
Notifications
You must be signed in to change notification settings - Fork 1.1k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
* Add the metric.Exporter interface * Move the reader errors to reader.go * Update Reader.Collect docs Remove TODO being addressed in this PR and restate purpose of method. * Initial draft of the periodic reader * Refer to correct config in periodic reader opts * Refactor reader testing into a harness * Move wait group handling out of run * Refactor ticker creation to allow testing * Honor export timeout in run * Fix wait group wait bug * Add periodic reader tests * Fix lint * Update periodic reader comments * Add concurrency test for readers * Simplify the ticker stop deferral * Only register once * Restrict build of metric sdk to Go>1.16 * Clean up ShutdownBeforeRegister test
- Loading branch information
Showing
15 changed files
with
556 additions
and
13 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
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,61 @@ | ||
// 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. | ||
|
||
//go:build go1.17 | ||
// +build go1.17 | ||
|
||
package metric // import "go.opentelemetry.io/otel/sdk/metric" | ||
|
||
import ( | ||
"context" | ||
"fmt" | ||
|
||
"go.opentelemetry.io/otel/sdk/metric/export" | ||
) | ||
|
||
// ErrExporterShutdown is returned if Export or Shutdown are called after an | ||
// Exporter has been Shutdown. | ||
var ErrExporterShutdown = fmt.Errorf("exporter is shutdown") | ||
|
||
// Exporter handles the delivery of metric data to external receivers. This is | ||
// the final component in the metric push pipeline. | ||
type Exporter interface { | ||
// Export serializes and transmits metric data to a receiver. | ||
// | ||
// This is called synchronously, there is no concurrency safety | ||
// requirement. Because of this, it is critical that all timeouts and | ||
// cancellations of the passed context be honored. | ||
// | ||
// All retry logic must be contained in this function. The SDK does not | ||
// implement any retry logic. All errors returned by this function are | ||
// considered unrecoverable and will be reported to a configured error | ||
// Handler. | ||
Export(context.Context, export.Metrics) error | ||
|
||
// ForceFlush flushes any metric data held by an exporter. | ||
// | ||
// The deadline or cancellation of the passed context must be honored. An | ||
// appropriate error should be returned in these situations. | ||
ForceFlush(context.Context) error | ||
|
||
// Shutdown flushes all metric data held by an exporter and releases any | ||
// held computational resources. | ||
// | ||
// The deadline or cancellation of the passed context must be honored. An | ||
// appropriate error should be returned in these situations. | ||
// | ||
// After Shutdown is called, calls to Export will perform no operation and | ||
// instead will return an error indicating the shutdown state. | ||
Shutdown(context.Context) error | ||
} |
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
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
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,230 @@ | ||
// 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. | ||
|
||
//go:build go1.17 | ||
// +build go1.17 | ||
|
||
package metric // import "go.opentelemetry.io/otel/sdk/metric" | ||
|
||
import ( | ||
"context" | ||
"fmt" | ||
"sync" | ||
"sync/atomic" | ||
"time" | ||
|
||
"go.opentelemetry.io/otel" | ||
"go.opentelemetry.io/otel/sdk/metric/export" | ||
) | ||
|
||
// Default periodic reader timing. | ||
const ( | ||
defaultTimeout = time.Millisecond * 30000 | ||
defaultInterval = time.Millisecond * 60000 | ||
) | ||
|
||
// periodicReaderConfig contains configuration options for a PeriodicReader. | ||
type periodicReaderConfig struct { | ||
interval time.Duration | ||
timeout time.Duration | ||
} | ||
|
||
// newPeriodicReaderConfig returns a periodicReaderConfig configured with | ||
// options. | ||
func newPeriodicReaderConfig(options []PeriodicReaderOption) periodicReaderConfig { | ||
c := periodicReaderConfig{ | ||
interval: defaultInterval, | ||
timeout: defaultTimeout, | ||
} | ||
for _, o := range options { | ||
c = o.apply(c) | ||
} | ||
return c | ||
} | ||
|
||
// PeriodicReaderOption applies a configuration option value to a PeriodicReader. | ||
type PeriodicReaderOption interface { | ||
apply(periodicReaderConfig) periodicReaderConfig | ||
} | ||
|
||
// periodicReaderOptionFunc applies a set of options to a periodicReaderConfig. | ||
type periodicReaderOptionFunc func(periodicReaderConfig) periodicReaderConfig | ||
|
||
// apply returns a periodicReaderConfig with option(s) applied. | ||
func (o periodicReaderOptionFunc) apply(conf periodicReaderConfig) periodicReaderConfig { | ||
return o(conf) | ||
} | ||
|
||
// WithTimeout configures the time a PeriodicReader waits for an export to | ||
// complete before canceling it. | ||
// | ||
// If this option is not used or d is less than or equal to zero, 30 seconds | ||
// is used as the default. | ||
func WithTimeout(d time.Duration) PeriodicReaderOption { | ||
return periodicReaderOptionFunc(func(conf periodicReaderConfig) periodicReaderConfig { | ||
if d <= 0 { | ||
return conf | ||
} | ||
conf.timeout = d | ||
return conf | ||
}) | ||
} | ||
|
||
// WithInterval configures the intervening time between exports for a | ||
// PeriodicReader. | ||
// | ||
// If this option is not used or d is less than or equal to zero, 60 seconds | ||
// is used as the default. | ||
func WithInterval(d time.Duration) PeriodicReaderOption { | ||
return periodicReaderOptionFunc(func(conf periodicReaderConfig) periodicReaderConfig { | ||
if d <= 0 { | ||
return conf | ||
} | ||
conf.interval = d | ||
return conf | ||
}) | ||
} | ||
|
||
// NewPeriodicReader returns a Reader that collects and exports metric data to | ||
// the exporter at a defined interval. By default, the returned Reader will | ||
// collect and export data every 60 seconds, and will cancel export attempts | ||
// that exceed 30 seconds. The export time is not counted towards the interval | ||
// between attempts. | ||
// | ||
// The Collect method of the returned Reader continues to gather and return | ||
// metric data to the user. It will not automatically send that data to the | ||
// exporter. That is left to the user to accomplish. | ||
func NewPeriodicReader(exporter Exporter, options ...PeriodicReaderOption) Reader { | ||
conf := newPeriodicReaderConfig(options) | ||
ctx, cancel := context.WithCancel(context.Background()) | ||
r := &periodicReader{ | ||
timeout: conf.timeout, | ||
exporter: exporter, | ||
cancel: cancel, | ||
} | ||
|
||
r.wg.Add(1) | ||
go func() { | ||
defer r.wg.Done() | ||
r.run(ctx, conf.interval) | ||
}() | ||
|
||
return r | ||
} | ||
|
||
// periodicReader is a Reader that continuously collects and exports metric | ||
// data at a set interval. | ||
type periodicReader struct { | ||
producer atomic.Value | ||
|
||
timeout time.Duration | ||
exporter Exporter | ||
|
||
wg sync.WaitGroup | ||
cancel context.CancelFunc | ||
shutdownOnce sync.Once | ||
} | ||
|
||
// newTicker allows testing override. | ||
var newTicker = time.NewTicker | ||
|
||
// run continuously collects and exports metric data at the specified | ||
// interval. This will run until ctx is canceled or times out. | ||
func (r *periodicReader) run(ctx context.Context, interval time.Duration) { | ||
ticker := newTicker(interval) | ||
defer ticker.Stop() | ||
|
||
for { | ||
select { | ||
case <-ticker.C: | ||
m, err := r.Collect(ctx) | ||
if err == nil { | ||
c, cancel := context.WithTimeout(ctx, r.timeout) | ||
err = r.exporter.Export(c, m) | ||
cancel() | ||
} | ||
if err != nil { | ||
otel.Handle(err) | ||
} | ||
case <-ctx.Done(): | ||
return | ||
} | ||
} | ||
} | ||
|
||
// register registers p as the producer of this reader. | ||
func (r *periodicReader) register(p producer) { | ||
// Only register once. If producer is already set, do nothing. | ||
r.producer.CompareAndSwap(nil, produceHolder{produce: p.produce}) | ||
} | ||
|
||
// Collect gathers and returns all metric data related to the Reader from | ||
// the SDK. The returned metric data is not exported to the configured | ||
// exporter, it is left to the caller to handle that if desired. | ||
// | ||
// An error is returned if this is called after Shutdown. | ||
func (r *periodicReader) Collect(ctx context.Context) (export.Metrics, error) { | ||
p := r.producer.Load() | ||
if p == nil { | ||
return export.Metrics{}, ErrReaderNotRegistered | ||
} | ||
|
||
ph, ok := p.(produceHolder) | ||
if !ok { | ||
// The atomic.Value is entirely in the periodicReader's control so | ||
// this should never happen. In the unforeseen case that this does | ||
// happen, return an error instead of panicking so a users code does | ||
// not halt in the processes. | ||
err := fmt.Errorf("periodic reader: invalid producer: %T", p) | ||
return export.Metrics{}, err | ||
} | ||
return ph.produce(ctx) | ||
} | ||
|
||
// ForceFlush flushes the Exporter. | ||
func (r *periodicReader) ForceFlush(ctx context.Context) error { | ||
return r.exporter.ForceFlush(ctx) | ||
} | ||
|
||
// Shutdown stops the export pipeline. | ||
func (r *periodicReader) Shutdown(ctx context.Context) error { | ||
err := ErrReaderShutdown | ||
r.shutdownOnce.Do(func() { | ||
// Stop the run loop. | ||
r.cancel() | ||
r.wg.Wait() | ||
|
||
// Any future call to Collect will now return ErrReaderShutdown. | ||
r.producer.Store(produceHolder{ | ||
produce: shutdownProducer{}.produce, | ||
}) | ||
|
||
err = r.exporter.Shutdown(ctx) | ||
}) | ||
return err | ||
} | ||
|
||
// produceHolder is used as an atomic.Value to wrap the non-concrete producer | ||
// type. | ||
type produceHolder struct { | ||
produce func(context.Context) (export.Metrics, error) | ||
} | ||
|
||
// shutdownProducer produces an ErrReaderShutdown error always. | ||
type shutdownProducer struct{} | ||
|
||
// produce returns an ErrReaderShutdown error. | ||
func (p shutdownProducer) produce(context.Context) (export.Metrics, error) { | ||
return export.Metrics{}, ErrReaderShutdown | ||
} |
Oops, something went wrong.