// Copyright (c) 2017-2018 Uber Technologies, Inc. // // 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 jaeger import ( "fmt" "io" "math/rand" "os" "reflect" "strconv" "sync" "time" "github.com/opentracing/opentracing-go" "github.com/opentracing/opentracing-go/ext" "github.com/uber/jaeger-client-go/internal/baggage" "github.com/uber/jaeger-client-go/internal/throttler" "github.com/uber/jaeger-client-go/log" "github.com/uber/jaeger-client-go/utils" ) // Tracer implements opentracing.Tracer. type Tracer struct { serviceName string hostIPv4 uint32 // this is for zipkin endpoint conversion sampler SamplerV2 reporter Reporter metrics Metrics logger log.DebugLogger timeNow func() time.Time randomNumber func() uint64 options struct { gen128Bit bool // whether to generate 128bit trace IDs zipkinSharedRPCSpan bool highTraceIDGenerator func() uint64 // custom high trace ID generator maxTagValueLength int noDebugFlagOnForcedSampling bool maxLogsPerSpan int // more options to come } // allocator of Span objects spanAllocator SpanAllocator injectors map[interface{}]Injector extractors map[interface{}]Extractor observer compositeObserver tags []Tag process Process baggageRestrictionManager baggage.RestrictionManager baggageSetter *baggageSetter debugThrottler throttler.Throttler } // NewTracer creates Tracer implementation that reports tracing to Jaeger. // The returned io.Closer can be used in shutdown hooks to ensure that the internal // queue of the Reporter is drained and all buffered spans are submitted to collectors. // TODO (breaking change) return *Tracer only, without closer. func NewTracer( serviceName string, sampler Sampler, reporter Reporter, options ...TracerOption, ) (opentracing.Tracer, io.Closer) { t := &Tracer{ serviceName: serviceName, sampler: samplerV1toV2(sampler), reporter: reporter, injectors: make(map[interface{}]Injector), extractors: make(map[interface{}]Extractor), metrics: *NewNullMetrics(), spanAllocator: simpleSpanAllocator{}, } for _, option := range options { option(t) } // register default injectors/extractors unless they are already provided via options textPropagator := NewTextMapPropagator(getDefaultHeadersConfig(), t.metrics) t.addCodec(opentracing.TextMap, textPropagator, textPropagator) httpHeaderPropagator := NewHTTPHeaderPropagator(getDefaultHeadersConfig(), t.metrics) t.addCodec(opentracing.HTTPHeaders, httpHeaderPropagator, httpHeaderPropagator) binaryPropagator := NewBinaryPropagator(t) t.addCodec(opentracing.Binary, binaryPropagator, binaryPropagator) // TODO remove after TChannel supports OpenTracing interopPropagator := &jaegerTraceContextPropagator{tracer: t} t.addCodec(SpanContextFormat, interopPropagator, interopPropagator) zipkinPropagator := &zipkinPropagator{tracer: t} t.addCodec(ZipkinSpanFormat, zipkinPropagator, zipkinPropagator) if t.baggageRestrictionManager != nil { t.baggageSetter = newBaggageSetter(t.baggageRestrictionManager, &t.metrics) } else { t.baggageSetter = newBaggageSetter(baggage.NewDefaultRestrictionManager(0), &t.metrics) } if t.debugThrottler == nil { t.debugThrottler = throttler.DefaultThrottler{} } if t.randomNumber == nil { seedGenerator := utils.NewRand(time.Now().UnixNano()) pool := sync.Pool{ New: func() interface{} { return rand.NewSource(seedGenerator.Int63()) }, } t.randomNumber = func() uint64 { generator := pool.Get().(rand.Source) number := uint64(generator.Int63()) pool.Put(generator) return number } } if t.timeNow == nil { t.timeNow = time.Now } if t.logger == nil { t.logger = log.NullLogger } // Set tracer-level tags t.tags = append(t.tags, Tag{key: JaegerClientVersionTagKey, value: JaegerClientVersion}) if hostname, err := os.Hostname(); err == nil { t.tags = append(t.tags, Tag{key: TracerHostnameTagKey, value: hostname}) } if ipval, ok := t.getTag(TracerIPTagKey); ok { ipv4, err := utils.ParseIPToUint32(ipval.(string)) if err != nil { t.hostIPv4 = 0 t.logger.Error("Unable to convert the externally provided ip to uint32: " + err.Error()) } else { t.hostIPv4 = ipv4 } } else if ip, err := utils.HostIP(); err == nil { t.tags = append(t.tags, Tag{key: TracerIPTagKey, value: ip.String()}) t.hostIPv4 = utils.PackIPAsUint32(ip) } else { t.logger.Error("Unable to determine this host's IP address: " + err.Error()) } if t.options.gen128Bit { if t.options.highTraceIDGenerator == nil { t.options.highTraceIDGenerator = t.randomNumber } } else if t.options.highTraceIDGenerator != nil { t.logger.Error("Overriding high trace ID generator but not generating " + "128 bit trace IDs, consider enabling the \"Gen128Bit\" option") } if t.options.maxTagValueLength == 0 { t.options.maxTagValueLength = DefaultMaxTagValueLength } t.process = Process{ Service: serviceName, UUID: strconv.FormatUint(t.randomNumber(), 16), Tags: t.tags, } if throttler, ok := t.debugThrottler.(ProcessSetter); ok { throttler.SetProcess(t.process) } return t, t } // addCodec adds registers injector and extractor for given propagation format if not already defined. func (t *Tracer) addCodec(format interface{}, injector Injector, extractor Extractor) { if _, ok := t.injectors[format]; !ok { t.injectors[format] = injector } if _, ok := t.extractors[format]; !ok { t.extractors[format] = extractor } } // StartSpan implements StartSpan() method of opentracing.Tracer. func (t *Tracer) StartSpan( operationName string, options ...opentracing.StartSpanOption, ) opentracing.Span { sso := opentracing.StartSpanOptions{} for _, o := range options { o.Apply(&sso) } return t.startSpanWithOptions(operationName, sso) } func (t *Tracer) startSpanWithOptions( operationName string, options opentracing.StartSpanOptions, ) opentracing.Span { if options.StartTime.IsZero() { options.StartTime = t.timeNow() } // Predicate whether the given span context is a valid reference // which may be used as parent / debug ID / baggage items source isValidReference := func(ctx SpanContext) bool { return ctx.IsValid() || ctx.isDebugIDContainerOnly() || len(ctx.baggage) != 0 } var references []Reference var parent SpanContext var hasParent bool // need this because `parent` is a value, not reference var ctx SpanContext var isSelfRef bool for _, ref := range options.References { ctxRef, ok := ref.ReferencedContext.(SpanContext) if !ok { t.logger.Error(fmt.Sprintf( "Reference contains invalid type of SpanReference: %s", reflect.ValueOf(ref.ReferencedContext))) continue } if !isValidReference(ctxRef) { continue } if ref.Type == selfRefType { isSelfRef = true ctx = ctxRef continue } references = append(references, Reference{Type: ref.Type, Context: ctxRef}) if !hasParent { parent = ctxRef hasParent = ref.Type == opentracing.ChildOfRef } } if !hasParent && isValidReference(parent) { // If ChildOfRef wasn't found but a FollowFromRef exists, use the context from // the FollowFromRef as the parent hasParent = true } rpcServer := false if v, ok := options.Tags[ext.SpanKindRPCServer.Key]; ok { rpcServer = (v == ext.SpanKindRPCServerEnum || v == string(ext.SpanKindRPCServerEnum)) } var internalTags []Tag newTrace := false if !isSelfRef { if !hasParent || !parent.IsValid() { newTrace = true ctx.traceID.Low = t.randomID() if t.options.gen128Bit { ctx.traceID.High = t.options.highTraceIDGenerator() } ctx.spanID = SpanID(ctx.traceID.Low) ctx.parentID = 0 ctx.samplingState = &samplingState{ localRootSpan: ctx.spanID, } if hasParent && parent.isDebugIDContainerOnly() && t.isDebugAllowed(operationName) { ctx.samplingState.setDebugAndSampled() internalTags = append(internalTags, Tag{key: JaegerDebugHeader, value: parent.debugID}) } } else { ctx.traceID = parent.traceID if rpcServer && t.options.zipkinSharedRPCSpan { // Support Zipkin's one-span-per-RPC model ctx.spanID = parent.spanID ctx.parentID = parent.parentID } else { ctx.spanID = SpanID(t.randomID()) ctx.parentID = parent.spanID } ctx.samplingState = parent.samplingState if parent.remote { ctx.samplingState.setFinal() ctx.samplingState.localRootSpan = ctx.spanID } } if hasParent { // copy baggage items if l := len(parent.baggage); l > 0 { ctx.baggage = make(map[string]string, len(parent.baggage)) for k, v := range parent.baggage { ctx.baggage[k] = v } } } } sp := t.newSpan() sp.context = ctx sp.tracer = t sp.operationName = operationName sp.startTime = options.StartTime sp.duration = 0 sp.references = references sp.firstInProcess = rpcServer || sp.context.parentID == 0 if !sp.isSamplingFinalized() { decision := t.sampler.OnCreateSpan(sp) sp.applySamplingDecision(decision, false) } sp.observer = t.observer.OnStartSpan(sp, operationName, options) if tagsTotalLength := len(options.Tags) + len(internalTags); tagsTotalLength > 0 { if sp.tags == nil || cap(sp.tags) < tagsTotalLength { sp.tags = make([]Tag, 0, tagsTotalLength) } sp.tags = append(sp.tags, internalTags...) for k, v := range options.Tags { sp.setTagInternal(k, v, false) } } t.emitNewSpanMetrics(sp, newTrace) return sp } // Inject implements Inject() method of opentracing.Tracer func (t *Tracer) Inject(ctx opentracing.SpanContext, format interface{}, carrier interface{}) error { c, ok := ctx.(SpanContext) if !ok { return opentracing.ErrInvalidSpanContext } if injector, ok := t.injectors[format]; ok { return injector.Inject(c, carrier) } return opentracing.ErrUnsupportedFormat } // Extract implements Extract() method of opentracing.Tracer func (t *Tracer) Extract( format interface{}, carrier interface{}, ) (opentracing.SpanContext, error) { if extractor, ok := t.extractors[format]; ok { spanCtx, err := extractor.Extract(carrier) if err != nil { return nil, err // ensure returned spanCtx is nil } spanCtx.remote = true return spanCtx, nil } return nil, opentracing.ErrUnsupportedFormat } // Close releases all resources used by the Tracer and flushes any remaining buffered spans. func (t *Tracer) Close() error { t.logger.Debugf("closing tracer") t.reporter.Close() t.sampler.Close() if mgr, ok := t.baggageRestrictionManager.(io.Closer); ok { _ = mgr.Close() } if throttler, ok := t.debugThrottler.(io.Closer); ok { _ = throttler.Close() } return nil } // Tags returns a slice of tracer-level tags. func (t *Tracer) Tags() []opentracing.Tag { tags := make([]opentracing.Tag, len(t.tags)) for i, tag := range t.tags { tags[i] = opentracing.Tag{Key: tag.key, Value: tag.value} } return tags } // getTag returns the value of specific tag, if not exists, return nil. // TODO only used by tests, move there. func (t *Tracer) getTag(key string) (interface{}, bool) { for _, tag := range t.tags { if tag.key == key { return tag.value, true } } return nil, false } // newSpan returns an instance of a clean Span object. // If options.PoolSpans is true, the spans are retrieved from an object pool. func (t *Tracer) newSpan() *Span { return t.spanAllocator.Get() } // emitNewSpanMetrics generates metrics on the number of started spans and traces. // newTrace param: we cannot simply check for parentID==0 because in Zipkin model the // server-side RPC span has the exact same trace/span/parent IDs as the // calling client-side span, but obviously the server side span is // no longer a root span of the trace. func (t *Tracer) emitNewSpanMetrics(sp *Span, newTrace bool) { if !sp.isSamplingFinalized() { t.metrics.SpansStartedDelayedSampling.Inc(1) if newTrace { t.metrics.TracesStartedDelayedSampling.Inc(1) } // joining a trace is not possible, because sampling decision inherited from upstream is final } else if sp.context.IsSampled() { t.metrics.SpansStartedSampled.Inc(1) if newTrace { t.metrics.TracesStartedSampled.Inc(1) } else if sp.firstInProcess { t.metrics.TracesJoinedSampled.Inc(1) } } else { t.metrics.SpansStartedNotSampled.Inc(1) if newTrace { t.metrics.TracesStartedNotSampled.Inc(1) } else if sp.firstInProcess { t.metrics.TracesJoinedNotSampled.Inc(1) } } } func (t *Tracer) reportSpan(sp *Span) { if !sp.isSamplingFinalized() { t.metrics.SpansFinishedDelayedSampling.Inc(1) } else if sp.context.IsSampled() { t.metrics.SpansFinishedSampled.Inc(1) } else { t.metrics.SpansFinishedNotSampled.Inc(1) } // Note: if the reporter is processing Span asynchronously then it needs to Retain() the span, // and then Release() it when no longer needed. // Otherwise, the span may be reused for another trace and its data may be overwritten. if sp.context.IsSampled() { t.reporter.Report(sp) } sp.Release() } // randomID generates a random trace/span ID, using tracer.random() generator. // It never returns 0. func (t *Tracer) randomID() uint64 { val := t.randomNumber() for val == 0 { val = t.randomNumber() } return val } // (NB) span must hold the lock before making this call func (t *Tracer) setBaggage(sp *Span, key, value string) { t.baggageSetter.setBaggage(sp, key, value) } // (NB) span must hold the lock before making this call func (t *Tracer) isDebugAllowed(operation string) bool { return t.debugThrottler.IsAllowed(operation) } // Sampler returns the sampler given to the tracer at creation. func (t *Tracer) Sampler() SamplerV2 { return t.sampler } // SelfRef creates an opentracing compliant SpanReference from a jaeger // SpanContext. This is a factory function in order to encapsulate jaeger specific // types. func SelfRef(ctx SpanContext) opentracing.SpanReference { return opentracing.SpanReference{ Type: selfRefType, ReferencedContext: ctx, } }