// StartSpanFromHeaders will extract a span from the signature headers
// and start a new span with the given operation name.
489577
func StartSpanFromHeaders(headers tasks.Headers, operationName string) opentracing.Span {
	// Try to extract the span context from the carrier.
	spanContext, err := opentracing.GlobalTracer().Extract(opentracing.TextMap, headers)

	// Create a new span from the span context if found or start a new trace with the function name.
	// For clarity add the machinery component tag.
	span := opentracing.StartSpan(
		operationName,
		ConsumerOption(spanContext),
		MachineryTag,
	)

	// Log any error but don't fail
	if err != nil {
		span.LogFields(opentracing_log.Error(err))
	}

	return span
}
394220
func (h *HTTPSink) AddSpans(ctx context.Context, traces []*trace.Span) (err error) {
	if len(traces) == 0 || h.TraceEndpoint == "" {
		return nil
	}
	return h.doBottom(ctx, func() (io.Reader, bool, error) {
		b, err := h.jsonMarshal(traces)
		if err != nil {
			return nil, false, errors.Annotate(err, "cannot encode traces into json")
		}
		return h.getReader(b)
	}, "application/json", h.TraceEndpoint, spanResponseValidator)
}
1322209
func parseHeader(b *bytes.Reader, shortHeaderConnIDLen int) (*Header, error) {
	startLen := b.Len()
	h, err := parseHeaderImpl(b, shortHeaderConnIDLen)
	if err != nil {
		return h, err
	}
	h.parsedLen = protocol.ByteCount(startLen - b.Len())
	return h, err
}
1129587
func (t *http2Server) operateHeaders(frame *http2.MetaHeadersFrame, handle func(*Stream), traceCtx func(context.Context, string) context.Context) (fatal bool) {
	streamID := frame.Header().StreamID
	state := &decodeState{
		serverSide: true,
	}
	if err := state.decodeHeader(frame); err != nil {
		if se, ok := status.FromError(err); ok {
			t.controlBuf.put(&cleanupStream{
				streamID: streamID,
				rst:      true,
				rstCode:  statusCodeConvTab[se.Code()],
				onWrite:  func() {},
			})
		}
		return false
	}

	buf := newRecvBuffer()
	s := &Stream{
		id:             streamID,
		st:             t,
		buf:            buf,
		fc:             &inFlow{limit: uint32(t.initialWindowSize)},
		recvCompress:   state.data.encoding,
		method:         state.data.method,
		contentSubtype: state.data.contentSubtype,
	}
	if frame.StreamEnded() {
		// s is just created by the caller. No lock needed.
		s.state = streamReadDone
	}
	if state.data.timeoutSet {
		s.ctx, s.cancel = context.WithTimeout(t.ctx, state.data.timeout)
	} else {
		s.ctx, s.cancel = context.WithCancel(t.ctx)
	}
	pr := &peer.Peer{
		Addr: t.remoteAddr,
	}
	// Attach Auth info if there is any.
	if t.authInfo != nil {
		pr.AuthInfo = t.authInfo
	}
	s.ctx = peer.NewContext(s.ctx, pr)
	// Attach the received metadata to the context.
	if len(state.data.mdata) > 0 {
		s.ctx = metadata.NewIncomingContext(s.ctx, state.data.mdata)
	}
	if state.data.statsTags != nil {
		s.ctx = stats.SetIncomingTags(s.ctx, state.data.statsTags)
	}
	if state.data.statsTrace != nil {
		s.ctx = stats.SetIncomingTrace(s.ctx, state.data.statsTrace)
	}
	if t.inTapHandle != nil {
		var err error
		info := &tap.Info{
			FullMethodName: state.data.method,
		}
		s.ctx, err = t.inTapHandle(s.ctx, info)
		if err != nil {
			warningf("transport: http2Server.operateHeaders got an error from InTapHandle: %v", err)
			t.controlBuf.put(&cleanupStream{
				streamID: s.id,
				rst:      true,
				rstCode:  http2.ErrCodeRefusedStream,
				onWrite:  func() {},
			})
			return false
		}
	}
	t.mu.Lock()
	if t.state != reachable {
		t.mu.Unlock()
		return false
	}
	if uint32(len(t.activeStreams)) >= t.maxStreams {
		t.mu.Unlock()
		t.controlBuf.put(&cleanupStream{
			streamID: streamID,
			rst:      true,
			rstCode:  http2.ErrCodeRefusedStream,
			onWrite:  func() {},
		})
		return false
	}
	if streamID%2 != 1 || streamID <= t.maxStreamID {
		t.mu.Unlock()
		// illegal gRPC stream id.
		errorf("transport: http2Server.HandleStreams received an illegal stream id: %v", streamID)
		return true
	}
	t.maxStreamID = streamID
	t.activeStreams[streamID] = s
	if len(t.activeStreams) == 1 {
		t.idle = time.Time{}
	}
	t.mu.Unlock()
	if channelz.IsOn() {
		atomic.AddInt64(&t.czData.streamsStarted, 1)
		atomic.StoreInt64(&t.czData.lastStreamCreatedTime, time.Now().UnixNano())
	}
	s.requestRead = func(n int) {
		t.adjustWindow(s, uint32(n))
	}
	s.ctx = traceCtx(s.ctx, s.method)
	if t.stats != nil {
		s.ctx = t.stats.TagRPC(s.ctx, &stats.RPCTagInfo{FullMethodName: s.method})
		inHeader := &stats.InHeader{
			FullMethod:  s.method,
			RemoteAddr:  t.remoteAddr,
			LocalAddr:   t.localAddr,
			Compression: s.recvCompress,
			WireLength:  int(frame.Header().Length),
		}
		t.stats.HandleRPC(s.ctx, inHeader)
	}
	s.ctxDone = s.ctx.Done()
	s.wq = newWriteQuota(defaultWriteQuota, s.ctxDone)
	s.trReader = &transportReader{
		reader: &recvBufferReader{
			ctx:     s.ctx,
			ctxDone: s.ctxDone,
			recv:    s.buf,
		},
		windowHandler: func(n int) {
			t.updateWindow(s, uint32(n))
		},
	}
	// Register the stream with loopy.
	t.controlBuf.put(&registerStream{
		streamID: s.id,
		wq:       s.wq,
	})
	handle(s)
	return false
}
374406
func (connector *Forwarder) AddSpans(ctx context.Context, spans []*trace.Span) error {
	atomic.AddInt64(&connector.stats.pipeline, int64(len(spans)))
	defer atomic.AddInt64(&connector.stats.pipeline, -int64(len(spans)))
	atomic.AddInt64(&connector.stats.totalSpansForwarded, int64(len(spans)))
	// could filter here
	if len(spans) == 0 {
		return nil
	}
	return connector.sampler.AddSpans(ctx, spans, connector.sink)
}
1106188
func (n *Node) start(nsMap map[string]client.MessageSet) error {
	n.l.Infoln("adaptor Starting...")

	s, err := n.c.Connect()
	if err != nil {
		return err
	}
	if closer, ok := s.(client.Closer); ok {
		defer func() {
			n.l.Debugln("closing session...")
			closer.Close()
			n.l.Debugln("session closed...")
		}()
	}
	readFunc := n.reader.Read(nsMap, func(check string) bool { return n.nsFilter.MatchString(check) })
	msgChan, err := readFunc(s, n.done)
	if err != nil {
		return err
	}
	var logOffset int64
	for msg := range msgChan {
		if n.clog != nil {
			d, _ := mejson.Marshal(msg.Msg.Data().AsMap())
			b, _ := json.Marshal(d)
			o, err := n.clog.Append(
				commitlog.NewLogFromEntry(
					commitlog.LogEntry{
						Key:       []byte(msg.Msg.Namespace()),
						Mode:      msg.Mode,
						Op:        msg.Msg.OP(),
						Timestamp: uint64(msg.Timestamp),
						Value:     b,
					}))
			if err != nil {
				return err
			}
			logOffset = o
			n.l.With("offset", logOffset).Debugln("attaching offset to message")
		}
		n.pipe.Send(msg.Msg, offset.Offset{
			Namespace: msg.Msg.Namespace(),
			LogOffset: uint64(logOffset),
			Timestamp: time.Now().Unix(),
		})
	}

	n.l.Infoln("adaptor Start finished...")
	return nil
}
954690
func (o *Observer) OnStartSpan(
	operationName string,
	options opentracing.StartSpanOptions,
) jaeger.SpanObserver {
	return NewSpanObserver(o.metricsByEndpoint, operationName, options)
}
186492
func StartSpan(
	parentCtx context.Context,
	opName string,
	tags opentracing.Tags,
	reference ...opentracing.SpanContext,
) context.Context {
	var ref opentracing.SpanContext
	if len(reference) > 0 {
		ref = reference[0]
	}
	span := opentracing.StartSpan(opName, opentracing.ChildOf(ref), tags)
	return opentracing.ContextWithSpan(parentCtx, span)
}
255485
func (mh *MethodHandler) Headers(headerNames ...string) *MethodHandler {
	if mh.headers == nil {
		mh.headers = make([]string, 0)
	}
	mh.headers = append(mh.headers, headerNames...)
	return mh
}
302004
func (r *InMemoryReporter) GetSpans() []opentracing.Span {
	r.lock.Lock()
	defer r.lock.Unlock()
	copied := make([]opentracing.Span, len(r.spans))
	copy(copied, r.spans)
	return copied
}
725517
func (s *SelectResults) start(pipeWriter *io.PipeWriter) {
	go func() {
		for {
			var prelude preludeInfo
			var headers = make(http.Header)
			var err error

			// Create CRC code
			crc := crc32.New(crc32.IEEETable)
			crcReader := io.TeeReader(s.resp.Body, crc)

			// Extract the prelude(12 bytes) into a struct to extract relevant information.
			prelude, err = processPrelude(crcReader, crc)
			if err != nil {
				pipeWriter.CloseWithError(err)
				closeResponse(s.resp)
				return
			}

			// Extract the headers(variable bytes) into a struct to extract relevant information
			if prelude.headerLen > 0 {
				if err = extractHeader(io.LimitReader(crcReader, int64(prelude.headerLen)), headers); err != nil {
					pipeWriter.CloseWithError(err)
					closeResponse(s.resp)
					return
				}
			}

			// Get the actual payload length so that the appropriate amount of
			// bytes can be read or parsed.
			payloadLen := prelude.PayloadLen()

			m := messageType(headers.Get("message-type"))

			switch m {
			case errorMsg:
				pipeWriter.CloseWithError(errors.New(headers.Get("error-code") + ":\"" + headers.Get("error-message") + "\""))
				closeResponse(s.resp)
				return
			case commonMsg:
				// Get content-type of the payload.
				c := contentType(headers.Get("content-type"))

				// Get event type of the payload.
				e := eventType(headers.Get("event-type"))

				// Handle all supported events.
				switch e {
				case endEvent:
					pipeWriter.Close()
					closeResponse(s.resp)
					return
				case recordsEvent:
					if _, err = io.Copy(pipeWriter, io.LimitReader(crcReader, payloadLen)); err != nil {
						pipeWriter.CloseWithError(err)
						closeResponse(s.resp)
						return
					}
				case progressEvent:
					switch c {
					case xmlContent:
						if err = xmlDecoder(io.LimitReader(crcReader, payloadLen), s.progress); err != nil {
							pipeWriter.CloseWithError(err)
							closeResponse(s.resp)
							return
						}
					default:
						pipeWriter.CloseWithError(fmt.Errorf("Unexpected content-type %s sent for event-type %s", c, progressEvent))
						closeResponse(s.resp)
						return
					}
				case statsEvent:
					switch c {
					case xmlContent:
						if err = xmlDecoder(io.LimitReader(crcReader, payloadLen), s.stats); err != nil {
							pipeWriter.CloseWithError(err)
							closeResponse(s.resp)
							return
						}
					default:
						pipeWriter.CloseWithError(fmt.Errorf("Unexpected content-type %s sent for event-type %s", c, statsEvent))
						closeResponse(s.resp)
						return
					}
				}
			}

			// Ensures that the full message's CRC is correct and
			// that the message is not corrupted
			if err := checkCRC(s.resp.Body, crc.Sum32()); err != nil {
				pipeWriter.CloseWithError(err)
				closeResponse(s.resp)
				return
			}

		}
	}()
}
800859
func (ns *NamedStopwatch) Start(name string) {
	if ns == nil {
		return // if we're not using stopwatches we just do nothing
	}
	ns.Lock()
	defer ns.Unlock()

	ns.start(name)
}
165114
func WithForwardedHeaders(fn func(header string) bool) Option {
	return func(p *Proxy) {
		p.headerForwarder = fn
	}
}
809156
func (s *Server) StartNetwork(w http.ResponseWriter, req *http.Request) {
	if err := s.network.StartAll(); err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	w.WriteHeader(http.StatusOK)
}
1099613
func (a *AdditionalSpanTags) AddSpans(ctx context.Context, spans []*trace.Span) error {
	for _, s := range spans {
		if s.Tags == nil {
			s.Tags = make(map[string]string, len(a.tags))
		}

		for k, v := range a.tags {
			s.Tags[k] = v
		}
	}
	return a.next.AddSpans(ctx, spans)
}
1236556
func (hct *httpClientTracer) startSpan(name string) *trace.Span {
	hct.mutex.Lock()
	defer hct.mutex.Unlock()
	newSpan, _ := trace.StartSpanFromContext(hct.ctx, name)
	hct.currentSpan.ClientFinish(hct.traceClient)

	hct.currentSpan = newSpan
	return newSpan
}
580747
func (s *Server) Start() {
	log.WithField("version", VERSION).Info("Starting server")

	// Set up the processors for spans:

	// Use the pre-allocated Workers slice to know how many to start.
	s.SpanWorker = NewSpanWorker(s.spanSinks, s.TraceClient, s.Statsd, s.SpanChan, s.TagsAsMap)

	go func() {
		log.Info("Starting Event worker")
		defer func() {
			ConsumePanic(s.Sentry, s.TraceClient, s.Hostname, recover())
		}()
		s.EventWorker.Work()
	}()

	log.WithField("n", s.SpanWorkerGoroutines).Info("Starting span workers")
	for i := 0; i < s.SpanWorkerGoroutines; i++ {
		go func() {
			defer func() {
				ConsumePanic(s.Sentry, s.TraceClient, s.Hostname, recover())
			}()
			s.SpanWorker.Work()
		}()
	}

	statsdPool := &sync.Pool{
		// We +1 this so we an "detect" when someone sends us too long of a metric!
		New: func() interface{} {
			return make([]byte, s.metricMaxLength+1)
		},
	}

	tracePool := &sync.Pool{
		New: func() interface{} {
			return make([]byte, s.traceMaxLengthBytes)
		},
	}

	for _, sink := range s.spanSinks {
		logrus.WithField("sink", sink.Name()).Info("Starting span sink")
		if err := sink.Start(s.TraceClient); err != nil {
			logrus.WithError(err).WithField("sink", sink).Panic("Error starting span sink")
		}
	}

	for _, sink := range s.metricSinks {
		logrus.WithField("sink", sink.Name()).Info("Starting metric sink")
		if err := sink.Start(s.TraceClient); err != nil {
			logrus.WithError(err).WithField("sink", sink).Fatal("Error starting metric sink")
		}
	}

	// Read Metrics Forever!
	concreteAddrs := make([]net.Addr, 0, len(s.StatsdListenAddrs))
	for _, addr := range s.StatsdListenAddrs {
		concreteAddrs = append(concreteAddrs, StartStatsd(s, addr, statsdPool))
	}
	s.StatsdListenAddrs = concreteAddrs

	// Read Traces Forever!
	if len(s.SSFListenAddrs) > 0 {
		concreteAddrs := make([]net.Addr, 0, len(s.StatsdListenAddrs))
		for _, addr := range s.SSFListenAddrs {
			concreteAddrs = append(concreteAddrs, StartSSF(s, addr, tracePool))
		}
		s.SSFListenAddrs = concreteAddrs
	} else {
		logrus.Info("Tracing sockets are not configured - not reading trace socket")
	}

	// Initialize a gRPC connection for forwarding
	if s.forwardUseGRPC {
		var err error
		s.grpcForwardConn, err = grpc.Dial(s.ForwardAddr, grpc.WithInsecure())
		if err != nil {
			log.WithError(err).WithFields(logrus.Fields{
				"forwardAddr": s.ForwardAddr,
			}).Fatal("Failed to initialize a gRPC connection for forwarding")
		}
	}

	// Flush every Interval forever!
	go func() {
		defer func() {
			ConsumePanic(s.Sentry, s.TraceClient, s.Hostname, recover())
		}()

		ctx, cancel := context.WithCancel(context.Background())
		go func() {
			// If the server is shutting down, cancel any in-flight flush:
			<-s.shutdown
			cancel()
		}()

		if s.synchronizeInterval {
			// We want to align our ticker to a multiple of its duration for
			// convenience of bucketing.
			<-time.After(CalculateTickDelay(s.interval, time.Now()))
		}

		// We aligned the ticker to our interval above. It's worth noting that just
		// because we aligned once we're not guaranteed to be perfect on each
		// subsequent tick. This code is small, however, and should service the
		// incoming tick signal fast enough that the amount we are "off" is
		// negligible.
		ticker := time.NewTicker(s.interval)
		for {
			select {
			case <-s.shutdown:
				// stop flushing on graceful shutdown
				ticker.Stop()
				return
			case triggered := <-ticker.C:
				ctx, cancel := context.WithDeadline(ctx, triggered.Add(s.interval))
				s.Flush(ctx)
				cancel()
			}
		}
	}()
}