vendor: update buildkit to opentelemetry support

Signed-off-by: Tonis Tiigi <tonistiigi@gmail.com>

vendor/github.com/moby/buildkit/util/appcontext/appcontext.go

@@ -22,7 +22,12 @@ func Context() context.Context {
 		const exitLimit = 3
 		retries := 0
 
-		ctx, cancel := context.WithCancel(context.Background())
+		ctx := context.Background()
+		for _, f := range inits {
+			ctx = f(ctx)
+		}
+
+		ctx, cancel := context.WithCancel(ctx)
 		appContextCache = ctx
 
 		go func() {

vendor/github.com/moby/buildkit/util/appcontext/register.go

@@ -0,0 +1,14 @@
+package appcontext
+
+import (
+	"context"
+)
+
+type Initializer func(context.Context) context.Context
+
+var inits []Initializer
+
+// Register stores a new context initializer that runs on app context creation
+func Register(f Initializer) {
+	inits = append(inits, f)
+}

vendor/github.com/moby/buildkit/util/flightcontrol/flightcontrol.go

@@ -39,7 +39,7 @@ func (g *Group) Do(ctx context.Context, key string, fn func(ctx context.Context)
 			return v, err
 		}
 		// backoff logic
-		if backoff >= 3*time.Second {
+		if backoff >= 15*time.Second {
 			err = errors.Wrapf(errRetryTimeout, "flightcontrol")
 			return v, err
 		}
@@ -132,8 +132,16 @@ func (c *call) wait(ctx context.Context) (v interface{}, err error) {
 	select {
 	case <-c.ready:
 		c.mu.Unlock()
-		<-c.cleaned
-		return nil, errRetry
+		if c.err != nil { // on error retry
+			<-c.cleaned
+			return nil, errRetry
+		}
+		pw, ok, _ := progress.NewFromContext(ctx)
+		if ok {
+			c.progressState.add(pw)
+		}
+		return c.result, nil
+
 	case <-c.ctx.done: // could return if no error
 		c.mu.Unlock()
 		<-c.cleaned
@@ -141,7 +149,7 @@ func (c *call) wait(ctx context.Context) (v interface{}, err error) {
 	default:
 	}
 
-	pw, ok, ctx := progress.FromContext(ctx)
+	pw, ok, ctx := progress.NewFromContext(ctx)
 	if ok {
 		c.progressState.add(pw)
 	}

vendor/github.com/moby/buildkit/util/grpcerrors/grpcerrors.go

@@ -6,7 +6,7 @@ import (
 
 	"github.com/containerd/typeurl"
 	gogotypes "github.com/gogo/protobuf/types"
-	"github.com/golang/protobuf/proto"
+	"github.com/golang/protobuf/proto" // nolint:staticcheck
 	"github.com/golang/protobuf/ptypes/any"
 	"github.com/moby/buildkit/util/stack"
 	"github.com/sirupsen/logrus"
@@ -169,7 +169,7 @@ func FromGRPC(err error) error {
 		}
 	}
 
-	err = status.FromProto(n).Err()
+	err = &grpcStatusErr{st: status.FromProto(n)}
 
 	for _, s := range stacks {
 		if s != nil {
@@ -188,6 +188,21 @@ func FromGRPC(err error) error {
 	return stack.Enable(err)
 }
 
+type grpcStatusErr struct {
+	st *status.Status
+}
+
+func (e *grpcStatusErr) Error() string {
+	if e.st.Code() == codes.OK || e.st.Code() == codes.Unknown {
+		return e.st.Message()
+	}
+	return e.st.Code().String() + ": " + e.st.Message()
+}
+
+func (e *grpcStatusErr) GRPCStatus() *status.Status {
+	return e.st
+}
+
 type withCode struct {
 	code codes.Code
 	error

vendor/github.com/moby/buildkit/util/progress/multireader.go

@@ -28,7 +28,7 @@ func (mr *MultiReader) Reader(ctx context.Context) Reader {
 	defer mr.mu.Unlock()
 
 	pr, ctx, closeWriter := NewContext(ctx)
-	pw, _, ctx := FromContext(ctx)
+	pw, _, ctx := NewFromContext(ctx)
 
 	w := pw.(*progressWriter)
 	mr.writers[w] = closeWriter

vendor/github.com/moby/buildkit/util/progress/progress.go

@@ -18,22 +18,37 @@ type contextKeyT string
 
 var contextKey = contextKeyT("buildkit/util/progress")
 
-// FromContext returns a progress writer from a context.
-func FromContext(ctx context.Context, opts ...WriterOption) (Writer, bool, context.Context) {
+// WriterFactory will generate a new progress Writer and return a new Context
+// with the new Writer stored.  It is the callers responsibility to Close the
+// returned Writer to avoid resource leaks.
+type WriterFactory func(ctx context.Context) (Writer, bool, context.Context)
+
+// FromContext returns a WriterFactory to generate new progress writers based
+// on a Writer previously stored in the Context.
+func FromContext(ctx context.Context, opts ...WriterOption) WriterFactory {
 	v := ctx.Value(contextKey)
-	pw, ok := v.(*progressWriter)
-	if !ok {
-		if pw, ok := v.(*MultiWriter); ok {
-			return pw, true, ctx
+	return func(ctx context.Context) (Writer, bool, context.Context) {
+		pw, ok := v.(*progressWriter)
+		if !ok {
+			if pw, ok := v.(*MultiWriter); ok {
+				return pw, true, ctx
+			}
+			return &noOpWriter{}, false, ctx
 		}
-		return &noOpWriter{}, false, ctx
+		pw = newWriter(pw)
+		for _, o := range opts {
+			o(pw)
+		}
+		ctx = context.WithValue(ctx, contextKey, pw)
+		return pw, true, ctx
 	}
-	pw = newWriter(pw)
-	for _, o := range opts {
-		o(pw)
-	}
-	ctx = context.WithValue(ctx, contextKey, pw)
-	return pw, true, ctx
+}
+
+// NewFromContext creates a new Writer based on a Writer previously stored
+// in the Context and returns a new Context with the new Writer stored.  It is
+// the callers responsibility to Close the returned Writer to avoid resource leaks.
+func NewFromContext(ctx context.Context, opts ...WriterOption) (Writer, bool, context.Context) {
+	return FromContext(ctx, opts...)(ctx)
 }
 
 type WriterOption func(Writer)

vendor/github.com/moby/buildkit/util/progress/progressui/display.go

@@ -2,6 +2,7 @@ package progressui
 
 import (
 	"bytes"
+	"container/ring"
 	"context"
 	"fmt"
 	"io"
@@ -12,11 +13,11 @@ import (
 	"time"
 
 	"github.com/containerd/console"
-	"github.com/jaguilar/vt100"
 	"github.com/moby/buildkit/client"
 	"github.com/morikuni/aec"
 	digest "github.com/opencontainers/go-digest"
 	"github.com/tonistiigi/units"
+	"github.com/tonistiigi/vt100"
 	"golang.org/x/time/rate"
 )
 
@@ -130,6 +131,7 @@ type vertex struct {
 	logs          [][]byte
 	logsPartial   bool
 	logsOffset    int
+	logsBuffer    *ring.Ring // stores last logs to print them on error
 	prev          *client.Vertex
 	events        []string
 	lastBlockTime *time.Time
@@ -295,10 +297,20 @@ func (t *trace) printErrorLogs(f io.Writer) {
 		if v.Error != "" && !strings.HasSuffix(v.Error, context.Canceled.Error()) {
 			fmt.Fprintln(f, "------")
 			fmt.Fprintf(f, " > %s:\n", v.Name)
+			// tty keeps original logs
 			for _, l := range v.logs {
 				f.Write(l)
 				fmt.Fprintln(f)
 			}
+			// printer keeps last logs buffer
+			if v.logsBuffer != nil {
+				for i := 0; i < v.logsBuffer.Len(); i++ {
+					if v.logsBuffer.Value != nil {
+						fmt.Fprintln(f, string(v.logsBuffer.Value.([]byte)))
+					}
+					v.logsBuffer = v.logsBuffer.Next()
+				}
+			}
 			fmt.Fprintln(f, "------")
 		}
 	}

vendor/github.com/moby/buildkit/util/progress/progressui/printer.go

@@ -1,6 +1,7 @@
 package progressui
 
 import (
+	"container/ring"
 	"context"
 	"fmt"
 	"io"
@@ -18,6 +19,8 @@ const maxDelay = 10 * time.Second
 const minTimeDelta = 5 * time.Second
 const minProgressDelta = 0.05 // %
 
+const logsBufferSize = 10
+
 type lastStatus struct {
 	Current   int64
 	Timestamp time.Time
@@ -61,7 +64,6 @@ func (p *textMux) printVtx(t *trace, dgst digest.Digest) {
 			fmt.Fprintf(p.w, "#%d %s\n", v.index, limitString(v.Name, 72))
 		} else {
 			fmt.Fprintf(p.w, "#%d %s\n", v.index, v.Name)
-			fmt.Fprintf(p.w, "#%d %s\n", v.index, v.Digest)
 		}
 
 	}
@@ -131,6 +133,13 @@ func (p *textMux) printVtx(t *trace, dgst digest.Digest) {
 		if i != len(v.logs)-1 || !v.logsPartial {
 			fmt.Fprintln(p.w, "")
 		}
+		if v.logsBuffer == nil {
+			v.logsBuffer = ring.New(logsBufferSize)
+		}
+		v.logsBuffer.Value = l
+		if !v.logsPartial {
+			v.logsBuffer = v.logsBuffer.Next()
+		}
 	}
 
 	if len(v.logs) > 0 {

vendor/github.com/moby/buildkit/util/stack/stack.pb.go

@@ -151,7 +151,9 @@ func init() {
 	proto.RegisterType((*Frame)(nil), "stack.Frame")
 }
 
-func init() { proto.RegisterFile("stack.proto", fileDescriptor_b44c07feb2ca0a5a) }
+func init() {
+	proto.RegisterFile("stack.proto", fileDescriptor_b44c07feb2ca0a5a)
+}
 
 var fileDescriptor_b44c07feb2ca0a5a = []byte{
 	// 185 bytes of a gzipped FileDescriptorProto

vendor/github.com/moby/buildkit/util/tracing/detect/delegated/delegated.go

@@ -0,0 +1,75 @@
+package jaeger
+
+import (
+	"context"
+	"sync"
+
+	"github.com/moby/buildkit/util/tracing/detect"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+)
+
+const maxBuffer = 256
+
+var exp = &Exporter{}
+
+func init() {
+	detect.Register("delegated", func() (sdktrace.SpanExporter, error) {
+		return exp, nil
+	}, 100)
+}
+
+type Exporter struct {
+	mu        sync.Mutex
+	exporters []sdktrace.SpanExporter
+	buffer    []sdktrace.ReadOnlySpan
+}
+
+var _ sdktrace.SpanExporter = &Exporter{}
+
+func (e *Exporter) ExportSpans(ctx context.Context, ss []sdktrace.ReadOnlySpan) error {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+
+	var err error
+	for _, e := range e.exporters {
+		if err1 := e.ExportSpans(ctx, ss); err1 != nil {
+			err = err1
+		}
+	}
+	if err != nil {
+		return err
+	}
+
+	if len(e.buffer) > maxBuffer {
+		return nil
+	}
+
+	e.buffer = append(e.buffer, ss...)
+	return nil
+}
+
+func (e *Exporter) Shutdown(ctx context.Context) error {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+
+	var err error
+	for _, e := range e.exporters {
+		if err1 := e.Shutdown(ctx); err1 != nil {
+			err = err1
+		}
+	}
+
+	return err
+}
+
+func (e *Exporter) SetSpanExporter(ctx context.Context, exp sdktrace.SpanExporter) error {
+	e.mu.Lock()
+	defer e.mu.Unlock()
+
+	e.exporters = append(e.exporters, exp)
+
+	if len(e.buffer) > 0 {
+		return exp.ExportSpans(ctx, e.buffer)
+	}
+	return nil
+}

vendor/github.com/moby/buildkit/util/tracing/detect/detect.go

@@ -0,0 +1,150 @@
+package detect
+
+import (
+	"context"
+	"os"
+	"path/filepath"
+	"sort"
+	"strconv"
+	"sync"
+
+	"github.com/pkg/errors"
+	"go.opentelemetry.io/otel/sdk/resource"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+	semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
+	"go.opentelemetry.io/otel/trace"
+)
+
+type ExporterDetector func() (sdktrace.SpanExporter, error)
+
+type detector struct {
+	f        ExporterDetector
+	priority int
+}
+
+var ServiceName string
+
+var detectors map[string]detector
+var once sync.Once
+var tp trace.TracerProvider
+var exporter sdktrace.SpanExporter
+var closers []func(context.Context) error
+var err error
+
+func Register(name string, exp ExporterDetector, priority int) {
+	if detectors == nil {
+		detectors = map[string]detector{}
+	}
+	detectors[name] = detector{
+		f:        exp,
+		priority: priority,
+	}
+}
+
+func detectExporter() (sdktrace.SpanExporter, error) {
+	if n := os.Getenv("OTEL_TRACES_EXPORTER"); n != "" {
+		d, ok := detectors[n]
+		if !ok {
+			if n == "none" {
+				return nil, nil
+			}
+			return nil, errors.Errorf("unsupported opentelemetry tracer %v", n)
+		}
+		return d.f()
+	}
+	arr := make([]detector, 0, len(detectors))
+	for _, d := range detectors {
+		arr = append(arr, d)
+	}
+	sort.Slice(arr, func(i, j int) bool {
+		return arr[i].priority < arr[j].priority
+	})
+	for _, d := range arr {
+		exp, err := d.f()
+		if err != nil {
+			return nil, err
+		}
+		if exp != nil {
+			return exp, nil
+		}
+	}
+	return nil, nil
+}
+
+func detect() error {
+	tp = trace.NewNoopTracerProvider()
+
+	exp, err := detectExporter()
+	if err != nil {
+		return err
+	}
+
+	if exp == nil {
+		return nil
+	}
+	res, err := resource.Detect(context.Background(), serviceNameDetector{})
+	if err != nil {
+		return err
+	}
+	res, err = resource.Merge(resource.Default(), res)
+	if err != nil {
+		return err
+	}
+
+	sp := sdktrace.NewBatchSpanProcessor(exp)
+
+	sdktp := sdktrace.NewTracerProvider(sdktrace.WithSpanProcessor(sp), sdktrace.WithResource(res))
+	closers = append(closers, sdktp.Shutdown)
+
+	exporter = exp
+	tp = sdktp
+	return nil
+}
+
+func TracerProvider() (trace.TracerProvider, error) {
+	once.Do(func() {
+		if err1 := detect(); err1 != nil {
+			err = err1
+		}
+	})
+	b, _ := strconv.ParseBool(os.Getenv("OTEL_INGORE_ERROR"))
+	if err != nil && !b {
+		return nil, err
+	}
+	return tp, nil
+}
+
+func Exporter() (sdktrace.SpanExporter, error) {
+	_, err := TracerProvider()
+	if err != nil {
+		return nil, err
+	}
+	return exporter, nil
+}
+
+func Shutdown(ctx context.Context) error {
+	for _, c := range closers {
+		if err := c(ctx); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+type serviceNameDetector struct{}
+
+func (serviceNameDetector) Detect(ctx context.Context) (*resource.Resource, error) {
+	return resource.StringDetector(
+		semconv.SchemaURL,
+		semconv.ServiceNameKey,
+		func() (string, error) {
+			if n := os.Getenv("OTEL_SERVICE_NAME"); n != "" {
+				return n, nil
+			}
+			if ServiceName != "" {
+				return ServiceName, nil
+			}
+			return filepath.Base(os.Args[0]), nil
+		},
+	).Detect(ctx)
+}

vendor/github.com/moby/buildkit/util/tracing/detect/otlp.go

@@ -0,0 +1,45 @@
+package detect
+
+import (
+	"context"
+	"os"
+
+	"github.com/pkg/errors"
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace"
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracehttp"
+	sdktrace "go.opentelemetry.io/otel/sdk/trace"
+)
+
+func init() {
+	Register("otlp", otlpExporter, 10)
+}
+
+func otlpExporter() (sdktrace.SpanExporter, error) {
+	set := os.Getenv("OTEL_TRACES_EXPORTER") == "otpl" || os.Getenv("OTEL_EXPORTER_OTLP_ENDPOINT") != "" || os.Getenv("OTEL_EXPORTER_OTLP_TRACES_ENDPOINT") != ""
+	if !set {
+		return nil, nil
+	}
+
+	proto := os.Getenv("OTEL_EXPORTER_OTLP_TRACES_PROTOCOL")
+	if proto == "" {
+		proto = os.Getenv("OTEL_EXPORTER_OTLP_PROTOCOL")
+	}
+	if proto == "" {
+		proto = "grpc"
+	}
+
+	var c otlptrace.Client
+
+	switch proto {
+	case "grpc":
+		c = otlptracegrpc.NewClient()
+	case "http/protobuf":
+		c = otlptracehttp.NewClient()
+	// case "http/json": // unsupported by library
+	default:
+		return nil, errors.Errorf("unsupported otlp protocol %v", proto)
+	}
+
+	return otlptrace.New(context.Background(), c)
+}

vendor/github.com/moby/buildkit/util/tracing/env/traceenv.go

@@ -0,0 +1,54 @@
+package detect
+
+import (
+	"context"
+	"os"
+
+	"github.com/moby/buildkit/util/appcontext"
+	"go.opentelemetry.io/otel/propagation"
+)
+
+const (
+	traceparentHeader = "traceparent"
+	tracestateHeader  = "tracestate"
+)
+
+func init() {
+	appcontext.Register(initContext)
+}
+
+func initContext(ctx context.Context) context.Context {
+	// previously defined in https://github.com/open-telemetry/opentelemetry-swift/blob/4ea467ed4b881d7329bf2254ca7ed7f2d9d6e1eb/Sources/OpenTelemetrySdk/Trace/Propagation/EnvironmentContextPropagator.swift#L14-L15
+	parent := os.Getenv("OTEL_TRACE_PARENT")
+	state := os.Getenv("OTEL_TRACE_STATE")
+
+	if parent == "" {
+		return ctx
+	}
+
+	tc := propagation.TraceContext{}
+	return tc.Extract(ctx, &textMap{parent: parent, state: state})
+}
+
+type textMap struct {
+	parent string
+	state  string
+}
+
+func (tm *textMap) Get(key string) string {
+	switch key {
+	case traceparentHeader:
+		return tm.parent
+	case tracestateHeader:
+		return tm.state
+	default:
+		return ""
+	}
+}
+
+func (tm *textMap) Set(key string, value string) {
+}
+
+func (tm *textMap) Keys() []string {
+	return []string{traceparentHeader, tracestateHeader}
+}

vendor/github.com/moby/buildkit/util/tracing/multispan.go

@@ -0,0 +1,22 @@
+package tracing
+
+import (
+	"go.opentelemetry.io/otel/trace"
+)
+
+// MultiSpan allows shared tracing to multiple spans.
+// TODO: This is a temporary solution and doesn't really support shared tracing yet. Instead the first always wins.
+
+type MultiSpan struct {
+	trace.Span
+}
+
+func NewMultiSpan() *MultiSpan {
+	return &MultiSpan{}
+}
+
+func (ms *MultiSpan) Add(s trace.Span) {
+	if ms.Span == nil {
+		ms.Span = s
+	}
+}

vendor/github.com/moby/buildkit/util/tracing/otlptracegrpc/client.go

@@ -0,0 +1,101 @@
+// 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 otlptracegrpc // import "go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sync"
+	"time"
+
+	"go.opentelemetry.io/otel/exporters/otlp/otlptrace"
+
+	"google.golang.org/grpc"
+
+	coltracepb "go.opentelemetry.io/proto/otlp/collector/trace/v1"
+	tracepb "go.opentelemetry.io/proto/otlp/trace/v1"
+)
+
+type client struct {
+	connection *Connection
+
+	lock         sync.Mutex
+	tracesClient coltracepb.TraceServiceClient
+}
+
+var _ otlptrace.Client = (*client)(nil)
+
+var (
+	errNoClient = errors.New("no client")
+)
+
+// NewClient creates a new gRPC trace client.
+func NewClient(cc *grpc.ClientConn) otlptrace.Client {
+	c := &client{}
+	c.connection = NewConnection(cc, c.handleNewConnection)
+
+	return c
+}
+
+func (c *client) handleNewConnection(cc *grpc.ClientConn) {
+	c.lock.Lock()
+	defer c.lock.Unlock()
+	if cc != nil {
+		c.tracesClient = coltracepb.NewTraceServiceClient(cc)
+	} else {
+		c.tracesClient = nil
+	}
+}
+
+// Start establishes a connection to the collector.
+func (c *client) Start(ctx context.Context) error {
+	return c.connection.StartConnection(ctx)
+}
+
+// Stop shuts down the connection to the collector.
+func (c *client) Stop(ctx context.Context) error {
+	return c.connection.Shutdown(ctx)
+}
+
+// UploadTraces sends a batch of spans to the collector.
+func (c *client) UploadTraces(ctx context.Context, protoSpans []*tracepb.ResourceSpans) error {
+	if !c.connection.Connected() {
+		return fmt.Errorf("traces exporter is disconnected from the server: %w", c.connection.LastConnectError())
+	}
+
+	ctx, cancel := c.connection.ContextWithStop(ctx)
+	defer cancel()
+	ctx, tCancel := context.WithTimeout(ctx, 30*time.Second)
+	defer tCancel()
+
+	ctx = c.connection.ContextWithMetadata(ctx)
+	err := func() error {
+		c.lock.Lock()
+		defer c.lock.Unlock()
+		if c.tracesClient == nil {
+			return errNoClient
+		}
+
+		_, err := c.tracesClient.Export(ctx, &coltracepb.ExportTraceServiceRequest{
+			ResourceSpans: protoSpans,
+		})
+		return err
+	}()
+	if err != nil {
+		c.connection.SetStateDisconnected(err)
+	}
+	return err
+}

vendor/github.com/moby/buildkit/util/tracing/otlptracegrpc/connection.go

@@ -0,0 +1,219 @@
+// 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 otlptracegrpc
+
+import (
+	"context"
+	"math/rand"
+	"sync"
+	"sync/atomic"
+	"time"
+	"unsafe"
+
+	"google.golang.org/grpc"
+	"google.golang.org/grpc/metadata"
+)
+
+type Connection struct {
+	// Ensure pointer is 64-bit aligned for atomic operations on both 32 and 64 bit machines.
+	lastConnectErrPtr unsafe.Pointer
+
+	// mu protects the Connection as it is accessed by the
+	// exporter goroutines and background Connection goroutine
+	mu sync.Mutex
+	cc *grpc.ClientConn
+
+	// these fields are read-only after constructor is finished
+	metadata             metadata.MD
+	newConnectionHandler func(cc *grpc.ClientConn)
+
+	// these channels are created once
+	disconnectedCh             chan bool
+	backgroundConnectionDoneCh chan struct{}
+	stopCh                     chan struct{}
+
+	// this is for tests, so they can replace the closing
+	// routine without a worry of modifying some global variable
+	// or changing it back to original after the test is done
+	closeBackgroundConnectionDoneCh func(ch chan struct{})
+}
+
+func NewConnection(cc *grpc.ClientConn, handler func(cc *grpc.ClientConn)) *Connection {
+	c := new(Connection)
+	c.newConnectionHandler = handler
+	c.cc = cc
+	c.closeBackgroundConnectionDoneCh = func(ch chan struct{}) {
+		close(ch)
+	}
+	return c
+}
+
+func (c *Connection) StartConnection(ctx context.Context) error {
+	c.stopCh = make(chan struct{})
+	c.disconnectedCh = make(chan bool, 1)
+	c.backgroundConnectionDoneCh = make(chan struct{})
+
+	if err := c.connect(ctx); err == nil {
+		c.setStateConnected()
+	} else {
+		c.SetStateDisconnected(err)
+	}
+	go c.indefiniteBackgroundConnection()
+
+	// TODO: proper error handling when initializing connections.
+	// We can report permanent errors, e.g., invalid settings.
+	return nil
+}
+
+func (c *Connection) LastConnectError() error {
+	errPtr := (*error)(atomic.LoadPointer(&c.lastConnectErrPtr))
+	if errPtr == nil {
+		return nil
+	}
+	return *errPtr
+}
+
+func (c *Connection) saveLastConnectError(err error) {
+	var errPtr *error
+	if err != nil {
+		errPtr = &err
+	}
+	atomic.StorePointer(&c.lastConnectErrPtr, unsafe.Pointer(errPtr))
+}
+
+func (c *Connection) SetStateDisconnected(err error) {
+	c.saveLastConnectError(err)
+	select {
+	case c.disconnectedCh <- true:
+	default:
+	}
+	c.newConnectionHandler(nil)
+}
+
+func (c *Connection) setStateConnected() {
+	c.saveLastConnectError(nil)
+}
+
+func (c *Connection) Connected() bool {
+	return c.LastConnectError() == nil
+}
+
+const defaultConnReattemptPeriod = 10 * time.Second
+
+func (c *Connection) indefiniteBackgroundConnection() {
+	defer func() {
+		c.closeBackgroundConnectionDoneCh(c.backgroundConnectionDoneCh)
+	}()
+
+	connReattemptPeriod := defaultConnReattemptPeriod
+
+	// No strong seeding required, nano time can
+	// already help with pseudo uniqueness.
+	rng := rand.New(rand.NewSource(time.Now().UnixNano() + rand.Int63n(1024)))
+
+	// maxJitterNanos: 70% of the connectionReattemptPeriod
+	maxJitterNanos := int64(0.7 * float64(connReattemptPeriod))
+
+	for {
+		// Otherwise these will be the normal scenarios to enable
+		// reconnection if we trip out.
+		// 1. If we've stopped, return entirely
+		// 2. Otherwise block until we are disconnected, and
+		//    then retry connecting
+		select {
+		case <-c.stopCh:
+			return
+
+		case <-c.disconnectedCh:
+			// Quickly check if we haven't stopped at the
+			// same time.
+			select {
+			case <-c.stopCh:
+				return
+
+			default:
+			}
+
+			// Normal scenario that we'll wait for
+		}
+
+		if err := c.connect(context.Background()); err == nil {
+			c.setStateConnected()
+		} else {
+			// this code is unreachable in most cases
+			// c.connect does not establish Connection
+			c.SetStateDisconnected(err)
+		}
+
+		// Apply some jitter to avoid lockstep retrials of other
+		// collector-exporters. Lockstep retrials could result in an
+		// innocent DDOS, by clogging the machine's resources and network.
+		jitter := time.Duration(rng.Int63n(maxJitterNanos))
+		select {
+		case <-c.stopCh:
+			return
+		case <-time.After(connReattemptPeriod + jitter):
+		}
+	}
+}
+
+func (c *Connection) connect(ctx context.Context) error {
+	c.newConnectionHandler(c.cc)
+	return nil
+}
+
+func (c *Connection) ContextWithMetadata(ctx context.Context) context.Context {
+	if c.metadata.Len() > 0 {
+		return metadata.NewOutgoingContext(ctx, c.metadata)
+	}
+	return ctx
+}
+
+func (c *Connection) Shutdown(ctx context.Context) error {
+	close(c.stopCh)
+	// Ensure that the backgroundConnector returns
+	select {
+	case <-c.backgroundConnectionDoneCh:
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+
+	c.mu.Lock()
+	cc := c.cc
+	c.cc = nil
+	c.mu.Unlock()
+
+	if cc != nil {
+		return cc.Close()
+	}
+
+	return nil
+}
+
+func (c *Connection) ContextWithStop(ctx context.Context) (context.Context, context.CancelFunc) {
+	// Unify the parent context Done signal with the Connection's
+	// stop channel.
+	ctx, cancel := context.WithCancel(ctx)
+	go func(ctx context.Context, cancel context.CancelFunc) {
+		select {
+		case <-ctx.Done():
+			// Nothing to do, either cancelled or deadline
+			// happened.
+		case <-c.stopCh:
+			cancel()
+		}
+	}(ctx, cancel)
+	return ctx, cancel
+}

vendor/github.com/moby/buildkit/util/tracing/tracing.go

@@ -0,0 +1,133 @@
+package tracing
+
+import (
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	"go.opentelemetry.io/otel/propagation"
+	semconv "go.opentelemetry.io/otel/semconv/v1.4.0"
+	"go.opentelemetry.io/otel/trace"
+)
+
+// StartSpan starts a new span as a child of the span in context.
+// If there is no span in context then this is a no-op.
+func StartSpan(ctx context.Context, operationName string, opts ...trace.SpanStartOption) (trace.Span, context.Context) {
+	parent := trace.SpanFromContext(ctx)
+	tracer := trace.NewNoopTracerProvider().Tracer("")
+	if parent != nil && parent.SpanContext().IsValid() {
+		tracer = parent.TracerProvider().Tracer("")
+	}
+	ctx, span := tracer.Start(ctx, operationName, opts...)
+	return span, ctx
+}
+
+// FinishWithError finalizes the span and sets the error if one is passed
+func FinishWithError(span trace.Span, err error) {
+	if err != nil {
+		span.RecordError(err)
+		if _, ok := err.(interface {
+			Cause() error
+		}); ok {
+			span.SetAttributes(attribute.String(string(semconv.ExceptionStacktraceKey), fmt.Sprintf("%+v", err)))
+		}
+		span.SetStatus(codes.Error, err.Error())
+	}
+	span.End()
+}
+
+// ContextWithSpanFromContext sets the tracing span of a context from other
+// context if one is not already set. Alternative would be
+// context.WithoutCancel() that would copy the context but reset ctx.Done
+func ContextWithSpanFromContext(ctx, ctx2 context.Context) context.Context {
+	// if already is a span then noop
+	if span := trace.SpanFromContext(ctx); span.SpanContext().IsValid() {
+		return ctx
+	}
+	if span := trace.SpanFromContext(ctx2); span != nil && span.SpanContext().IsValid() {
+		return trace.ContextWithSpan(ctx, span)
+	}
+	return ctx
+}
+
+var DefaultTransport http.RoundTripper = &Transport{
+	RoundTripper: NewTransport(http.DefaultTransport),
+}
+
+var DefaultClient = &http.Client{
+	Transport: DefaultTransport,
+}
+
+var propagators = propagation.NewCompositeTextMapPropagator(propagation.TraceContext{}, propagation.Baggage{})
+
+type Transport struct {
+	http.RoundTripper
+}
+
+func NewTransport(rt http.RoundTripper) http.RoundTripper {
+	// TODO: switch to otelhttp. needs upstream updates to avoid transport-global tracer
+	return &Transport{
+		RoundTripper: rt,
+	}
+}
+
+func (t *Transport) RoundTrip(req *http.Request) (*http.Response, error) {
+	span := trace.SpanFromContext(req.Context())
+	if !span.SpanContext().IsValid() { // no tracer connected with either request or transport
+		return t.RoundTripper.RoundTrip(req)
+	}
+
+	ctx, span := span.TracerProvider().Tracer("").Start(req.Context(), req.Method)
+
+	req = req.WithContext(ctx)
+	span.SetAttributes(semconv.HTTPClientAttributesFromHTTPRequest(req)...)
+	propagators.Inject(ctx, propagation.HeaderCarrier(req.Header))
+
+	resp, err := t.RoundTripper.RoundTrip(req)
+	if err != nil {
+		span.RecordError(err)
+		span.End()
+		return resp, err
+	}
+
+	span.SetAttributes(semconv.HTTPAttributesFromHTTPStatusCode(resp.StatusCode)...)
+	span.SetStatus(semconv.SpanStatusFromHTTPStatusCode(resp.StatusCode))
+
+	if req.Method == "HEAD" {
+		span.End()
+	} else {
+		resp.Body = &wrappedBody{ctx: ctx, span: span, body: resp.Body}
+	}
+
+	return resp, err
+}
+
+type wrappedBody struct {
+	ctx  context.Context
+	span trace.Span
+	body io.ReadCloser
+}
+
+var _ io.ReadCloser = &wrappedBody{}
+
+func (wb *wrappedBody) Read(b []byte) (int, error) {
+	n, err := wb.body.Read(b)
+
+	switch err {
+	case nil:
+		// nothing to do here but fall through to the return
+	case io.EOF:
+		wb.span.End()
+	default:
+		wb.span.RecordError(err)
+	}
+	return n, err
+}
+
+func (wb *wrappedBody) Close() error {
+	wb.span.End()
+	return wb.body.Close()
+}