github/containerd
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases 2 Wiki Activity

Changes to configuring otel from env only

Browse Source 
These are standard environment variables described by the otel spec in
https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/.

The old config options are removed

Also since otel will by default try to connect to https://localhost:4318
if no endpoint is set, this will also just disable the otlp plugin when
there is no endpoint so we don't have otel continuously trying to
connect to the default endpoint, littering the logs with connection
failure messages and collecting traces that won't go anywhere.

Signed-off-by: Brian Goff <cpuguy83@gmail.com>
This commit is contained in:
Brian Goff 2023-08-15 19:14:01 +00:00
parent 753a525b3b
commit 4fbc9842d5
4 changed files with 99 additions and 211 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

16
integration/client/testdata/default-1.6.toml vendored
View File

@ -168,9 +168,10 @@ version = 2
[plugins."io.containerd.internal.v1.restart"]
interval = "10s"
[plugins."io.containerd.internal.v1.tracing"]
sampling_ratio = 1.0
service_name = "containerd"
# Removed in latest
#[plugins."io.containerd.internal.v1.tracing"]
# sampling_ratio = 1.0
# service_name = "containerd"
[plugins."io.containerd.metadata.v1.bolt"]
content_sharing_policy = "shared"
@ -225,10 +226,11 @@ version = 2
#[plugins."io.containerd.snapshotter.v1.zfs"]
# root_path = ""
[plugins."io.containerd.tracing.processor.v1.otlp"]
endpoint = ""
insecure = false
protocol = ""
# Removed in latest
#[plugins."io.containerd.tracing.processor.v1.otlp"]
# endpoint = ""
# insecure = false
# protocol = ""
[proxy_plugins]

16
integration/client/testdata/default-1.7.toml vendored
View File

@ -185,9 +185,10 @@ version = 2
[plugins."io.containerd.internal.v1.restart"]
interval = "10s"
[plugins."io.containerd.internal.v1.tracing"]
sampling_ratio = 1.0
service_name = "containerd"
# Removed in latest
#[plugins."io.containerd.internal.v1.tracing"]
# sampling_ratio = 1.0
# service_name = "containerd"
[plugins."io.containerd.metadata.v1.bolt"]
content_sharing_policy = "shared"
@ -258,10 +259,11 @@ version = 2
#[plugins."io.containerd.snapshotter.v1.zfs"]
# root_path = ""
[plugins."io.containerd.tracing.processor.v1.otlp"]
endpoint = ""
insecure = false
protocol = ""
# Removed in latest
#[plugins."io.containerd.tracing.processor.v1.otlp"]
# endpoint = ""
# insecure = false
# protocol = ""
[plugins."io.containerd.transfer.v1.local"]
config_path = ""

160
pkg/tracing/plugin/otlp.go
View File

@ -18,10 +18,10 @@ package plugin
import (
"context"
"errors"
"fmt"
"io"
"net/url"
"os"
"strconv"
"time"
"github.com/containerd/containerd/v2/pkg/deprecation"
@ -37,13 +37,25 @@ import (
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
"go.opentelemetry.io/otel/propagation"
"go.opentelemetry.io/otel/sdk/resource"
"go.opentelemetry.io/otel/sdk/trace"
semconv "go.opentelemetry.io/otel/semconv/v1.21.0"
)
const exporterPlugin = "otlp"
// OTEL and OTLP standard env vars
// See https://opentelemetry.io/docs/specs/otel/configuration/sdk-environment-variables/
const (
sdkDisabledEnv = "OTEL_SDK_DISABLED"
otlpEndpointEnv = "OTEL_EXPORTER_OTLP_ENDPOINT"
otlpTracesEndpointEnv = "OTEL_EXPORTER_OTLP_TRACES_ENDPOINT"
otlpProtocolEnv = "OTEL_EXPORTER_OTLP_PROTOCOL"
otlpTracesProtocolEnv = "OTEL_EXPORTER_OTLP_TRACES_PROTOCOL"
otelTracesExporterEnv = "OTEL_TRACES_EXPORTER"
otelServiceNameEnv = "OTEL_SERVICE_NAME"
)
func init() {
registry.Register(&plugin.Registration{
ID: exporterPlugin,
@ -53,8 +65,16 @@ func init() {
if err := warnOTLPConfig(ic); err != nil {
return nil, err
}
cfg := ic.Config.(*OTLPConfig)
exp, err := newExporter(ic.Context, cfg)
if err := checkDisabled(); err != nil {
return nil, err
}
// If OTEL_TRACES_EXPORTER is set, it must be "otlp"
if v := os.Getenv(otelTracesExporterEnv); v != "" && v != "otlp" {
return nil, fmt.Errorf("unsupported traces exporter %q: %w", v, errdefs.ErrInvalidArgument)
}
exp, err := newExporter(ic.Context)
if err != nil {
return nil, err
}
@ -62,29 +82,32 @@ func init() {
},
})
registry.Register(&plugin.Registration{
ID: "tracing",
Type: plugins.InternalPlugin,
ID: "tracing",
Type: plugins.InternalPlugin,
Config: &TraceConfig{},
Requires: []plugin.Type{
plugins.TracingProcessorPlugin,
},
Config: &TraceConfig{
ServiceName: "containerd",
TraceSamplingRatio: 1.0,
},
InitFn: func(ic *plugin.InitContext) (interface{}, error) {
if err := warnTraceConfig(ic); err != nil {
return nil, err
}
// get TracingProcessorPlugin which is a dependency
if err := checkDisabled(); err != nil {
return nil, err
}
//get TracingProcessorPlugin which is a dependency
plugins, err := ic.GetByType(plugins.TracingProcessorPlugin)
if err != nil {
return nil, fmt.Errorf("failed to get tracing processors: %w", err)
}
procs := make([]trace.SpanProcessor, 0, len(plugins))
for _, p := range plugins {
procs = append(procs, p.(trace.SpanProcessor))
}
return newTracer(ic.Context, ic.Config.(*TraceConfig), procs)
return newTracer(ic.Context, procs)
},
})
@ -94,61 +117,63 @@ func init() {
// OTLPConfig holds the configurations for the built-in otlp span processor
type OTLPConfig struct {
Endpoint string `toml:"endpoint"`
Protocol string `toml:"protocol"`
Insecure bool `toml:"insecure"`
Endpoint string `toml:"endpoint,omitempty"`
Protocol string `toml:"protocol,omitempty"`
Insecure bool `toml:"insecure,omitempty"`
}
// TraceConfig is the common configuration for open telemetry.
type TraceConfig struct {
ServiceName string `toml:"service_name"`
TraceSamplingRatio float64 `toml:"sampling_ratio"`
ServiceName string `toml:"service_name,omitempty"`
TraceSamplingRatio float64 `toml:"sampling_ratio,omitempty"`
}
type closer struct {
close func() error
func checkDisabled() error {
v := os.Getenv(sdkDisabledEnv)
if v != "" {
disable, err := strconv.ParseBool(v)
if err != nil {
return fmt.Errorf("invalid value for %s: %w: %w", sdkDisabledEnv, err, errdefs.ErrInvalidArgument)
}
if disable {
return fmt.Errorf("%w: tracing disabled by env %s=%s", plugin.ErrSkipPlugin, sdkDisabledEnv, v)
}
}
if os.Getenv(otlpEndpointEnv) == "" && os.Getenv(otlpTracesEndpointEnv) == "" {
return fmt.Errorf("%w: tracing endpoint not configured", plugin.ErrSkipPlugin)
}
return nil
}
func (c *closer) Close() error {
return c.close()
type closerFunc func() error
func (f closerFunc) Close() error {
return f()
}
// newExporter creates an exporter based on the given configuration.
//
// The default protocol is http/protobuf since it is recommended by
// https://github.com/open-telemetry/opentelemetry-specification/blob/v1.8.0/specification/protocol/exporter.md#specify-protocol.
func newExporter(ctx context.Context, cfg *OTLPConfig) (*otlptrace.Exporter, error) {
func newExporter(ctx context.Context) (*otlptrace.Exporter, error) {
const timeout = 5 * time.Second
v := os.Getenv(otlpTracesProtocolEnv)
if v == "" {
v = os.Getenv(otlpProtocolEnv)
}
ctx, cancel := context.WithTimeout(ctx, timeout)
defer cancel()
switch cfg.Protocol {
switch v {
case "", "http/protobuf":
var opts []otlptracehttp.Option
if cfg.Endpoint != "" {
u, err := url.Parse(cfg.Endpoint)
if err != nil {
return nil, fmt.Errorf("OpenTelemetry endpoint %q %w : %v", cfg.Endpoint, errdefs.ErrInvalidArgument, err)
}
opts = append(opts, otlptracehttp.WithEndpoint(u.Host))
if u.Scheme == "http" {
opts = append(opts, otlptracehttp.WithInsecure())
}
}
return otlptracehttp.New(ctx, opts...)
return otlptracehttp.New(ctx)
case "grpc":
var opts []otlptracegrpc.Option
if cfg.Endpoint != "" {
opts = append(opts, otlptracegrpc.WithEndpoint(cfg.Endpoint))
}
if cfg.Insecure {
opts = append(opts, otlptracegrpc.WithInsecure())
}
return otlptracegrpc.New(ctx, opts...)
return otlptracegrpc.New(ctx)
default:
// Other protocols such as "http/json" are not supported.
return nil, fmt.Errorf("OpenTelemetry protocol %q : %w", cfg.Protocol, errdefs.ErrNotImplemented)
return nil, fmt.Errorf("OpenTelemetry protocol %q : %w", v, errdefs.ErrNotImplemented)
}
}
@ -156,48 +181,25 @@ func newExporter(ctx context.Context, cfg *OTLPConfig) (*otlptrace.Exporter, err
// its sampling ratio and returns io.Closer.
//
// Note that this function sets process-wide tracing configuration.
func newTracer(ctx context.Context, config *TraceConfig, procs []trace.SpanProcessor) (io.Closer, error) {
res, err := resource.New(ctx,
resource.WithHost(),
resource.WithAttributes(
// Service name used to displace traces in backends
semconv.ServiceNameKey.String(config.ServiceName),
),
)
if err != nil {
return nil, fmt.Errorf("failed to create resource: %w", err)
func newTracer(ctx context.Context, procs []trace.SpanProcessor) (io.Closer, error) {
// Let otel configure the service name from env
if os.Getenv(otelServiceNameEnv) == "" {
os.Setenv(otelServiceNameEnv, "containerd")
}
sampler := trace.ParentBased(trace.TraceIDRatioBased(config.TraceSamplingRatio))
opts := []trace.TracerProviderOption{
trace.WithSampler(sampler),
trace.WithResource(res),
}
otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(propagation.TraceContext{}, propagation.Baggage{}))
opts := make([]trace.TracerProviderOption, 0, len(procs))
for _, proc := range procs {
opts = append(opts, trace.WithSpanProcessor(proc))
}
provider := trace.NewTracerProvider(opts...)
otel.SetTracerProvider(provider)
otel.SetTextMapPropagator(propagators())
return closerFunc(func() error {
return provider.Shutdown(ctx)
}), nil
return &closer{close: func() error {
for _, p := range procs {
if err := p.Shutdown(ctx); err != nil && !errors.Is(err, context.Canceled) {
return err
}
}
return nil
}}, nil
}
// Returns a composite TestMap propagator
func propagators() propagation.TextMapPropagator {
return propagation.NewCompositeTextMapPropagator(propagation.TraceContext{}, propagation.Baggage{})
}
func warnTraceConfig(ic *plugin.InitContext) error {

118
pkg/tracing/plugin/otlp_test.go
View File

@ -1,118 +0,0 @@
/*
Copyright The containerd 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 plugin
import (
"context"
"errors"
"testing"
"github.com/containerd/errdefs"
"go.opentelemetry.io/otel/sdk/trace"
"go.opentelemetry.io/otel/sdk/trace/tracetest"
)
// TestNewExporter runs tests with different combinations of configuration for NewExporter function
func TestNewExporter(t *testing.T) {
for _, testcase := range []struct {
name string
input OTLPConfig
output error
}{
{
name: "Test http/protobuf protocol, expect no error",
input: OTLPConfig{Endpoint: "http://localhost:4318",
Protocol: "http/protobuf",
Insecure: false},
output: nil,
},
{
name: "Test invalid endpoint, expect ErrInvalidArgument error",
input: OTLPConfig{Endpoint: "http://localhost\n:4318",
Protocol: "http/protobuf",
Insecure: false},
output: errdefs.ErrInvalidArgument,
},
{
name: "Test default protocol, expect no error",
input: OTLPConfig{Endpoint: "http://localhost:4318",
Protocol: "",
Insecure: false},
output: nil,
},
{
name: "Test grpc protocol, expect no error",
input: OTLPConfig{Endpoint: "http://localhost:4317",
Protocol: "grpc",
Insecure: false},
output: nil,
},
{
name: "Test http/json protocol which is not supported, expect not implemented error",
input: OTLPConfig{Endpoint: "http://localhost:4318",
Protocol: "http/json",
Insecure: false},
output: errdefs.ErrNotImplemented,
},
} {
testcase := testcase
t.Run(testcase.name, func(t *testing.T) {
t.Logf("input: %v", testcase.input)
ctx := context.TODO()
exp, err := newExporter(ctx, &testcase.input)
t.Logf("output: %v", err)
if err == nil {
if err != testcase.output {
t.Fatalf("Expect to get error: %v, however no error got\n", testcase.output)
} else if exp == nil {
t.Fatalf("Something went wrong, Exporter not created as expected\n")
}
} else {
if !errors.Is(err, testcase.output) {
t.Fatalf("Expect to get error: %v, however error %v returned\n", testcase.output, err)
}
}
})
}
}
// TestNewTracer runs test for NewTracer function
func TestNewTracer(t *testing.T) {
config := &TraceConfig{ServiceName: "containerd", TraceSamplingRatio: 1.0}
t.Logf("config: %v", config)
procs := make([]trace.SpanProcessor, 0, 1)
//Create a dummy in memory exporter for test
exp := tracetest.NewInMemoryExporter()
proc := trace.NewBatchSpanProcessor(exp)
procs = append(procs, proc)
ctx := context.TODO()
tracerCloser, err := newTracer(ctx, config, procs)
if err != nil {
t.Fatalf("Something went wrong, Tracer not created as expected\n")
}
defer tracerCloser.Close()
}