trace.go

//
// Copyright 2022 SkyAPM org
//
// 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 go2sky

import (
	"context"

	"github.com/SkyAPM/go2sky/internal/idgen"
	"github.com/pkg/errors"

	"github.com/SkyAPM/go2sky/internal/tool"
	"github.com/SkyAPM/go2sky/propagation"
)

const (
	errParameter = tool.Error("parameter are nil")
)

// Tracer is go2sky tracer implementation.
type Tracer struct {
	service  string
	instance string
	reporter Reporter
	// 0 not init 1 init
	initFlag    int32
	sampler     Sampler
	correlation *CorrelationConfig
	cdsWatchers []AgentConfigChangeWatcher
}

// TracerOption allows for functional options to adjust behaviour
// of a Tracer to be created by NewTracer
type TracerOption func(t *Tracer)

// NewTracer return a new go2sky Tracer
func NewTracer(service string, opts ...TracerOption) (tracer *Tracer, err error) {
	// read the service in the environment variable
	service = serviceFormEnv(service)
	if service == "" {
		return nil, errParameter
	}

	// read the options in the environment variable
	envOps, err := traceOptionsFormEnv()
	if err != nil {
		return nil, err
	}
	opts = append(opts, envOps...)

	t := &Tracer{
		service:  service,
		initFlag: 0,
	}
	// default correlation config
	t.correlation = &CorrelationConfig{MaxKeyCount: 3, MaxValueSize: 128}
	for _, opt := range opts {
		opt(t)
	}

	if t.sampler == nil {
		t.sampler = NewDynamicSampler(1, t)
	}

	if t.reporter != nil {
		if t.instance == "" {
			id, err := idgen.UUID()
			if err != nil {
				return nil, err
			}
			t.instance = id + "@" + tool.IPV4()
		}
		t.reporter.Boot(t.service, t.instance, t.cdsWatchers)
		t.initFlag = 1
	}

	return t, nil
}

// CreateEntrySpan creates and starts an entry span for incoming request
func (t *Tracer) CreateEntrySpan(ctx context.Context, operationName string, extractor propagation.Extractor) (s Span, nCtx context.Context, err error) {
	if ctx == nil || operationName == "" || extractor == nil {
		return nil, nil, errParameter
	}
	if s, nCtx = t.createNoop(ctx); s != nil {
		return
	}
	var refSc = &propagation.SpanContext{}
	err = refSc.Decode(extractor)
	if err != nil {
		return
	}
	if !refSc.Valid {
		refSc = nil
	}
	s, nCtx, err = t.CreateLocalSpan(ctx, WithContext(refSc), WithSpanType(SpanTypeEntry), WithOperationName(operationName))
	return
}

// CreateLocalSpan creates and starts a span for local usage
func (t *Tracer) CreateLocalSpan(ctx context.Context, opts ...SpanOption) (s Span, c context.Context, err error) {
	if ctx == nil {
		return nil, nil, errParameter
	}
	if s, c = t.createNoop(ctx); s != nil {
		return
	}
	ds := newLocalSpan(t)
	for _, opt := range opts {
		opt(ds)
	}
	parentSpan, ok := ctx.Value(ctxKeyInstance).(segmentSpan)
	if !ok {
		parentSpan = nil
	}
	isForceSample := len(ds.Refs) > 0
	// Try to sample when it is not force sample
	if parentSpan == nil && !isForceSample {
		// Force sample
		sampled := t.sampler.IsSampled(ds.OperationName)
		if !sampled {
			// Filter by sample just return noop span
			s = &NoopSpan{}
			return s, context.WithValue(ctx, ctxKeyInstance, s), nil
		}
	}
	s, err = newSegmentSpan(ds, parentSpan)
	if err != nil {
		return nil, nil, err
	}
	return s, context.WithValue(ctx, ctxKeyInstance, s), nil
}

// CreateExitSpan creates and starts an exit span for client
func (t *Tracer) CreateExitSpan(ctx context.Context, operationName string, peer string, injector propagation.Injector) (s Span, err error) {
	s, _, err = t.CreateExitSpanWithContext(ctx, operationName, peer, injector)
	return
}

// CreateExitSpanWithContext creates and starts an exit span for client with context
func (t *Tracer) CreateExitSpanWithContext(ctx context.Context, operationName string, peer string,
	injector propagation.Injector) (s Span, nCtx context.Context, err error) {
	if ctx == nil || operationName == "" || peer == "" || injector == nil {
		return nil, nil, errParameter
	}
	if s, nCtx = t.createNoop(ctx); s != nil {
		return
	}
	s, nCtx, err = t.CreateLocalSpan(ctx, WithSpanType(SpanTypeExit), WithOperationName(operationName))
	if err != nil {
		return
	}
	noopSpan, ok := interface{}(s).(*NoopSpan)
	if ok {
		// Ignored, there is no need to inject SW8 in the request header
		return noopSpan, nCtx, nil
	}
	s.SetPeer(peer)
	spanContext := &propagation.SpanContext{}
	span, ok := s.(ReportedSpan)
	if !ok {
		return nil, nil, errors.New("span type is wrong")
	}

	firstSpan := span.Context().FirstSpan
	spanContext.Sample = 1
	spanContext.TraceID = span.Context().TraceID
	spanContext.ParentSegmentID = span.Context().SegmentID
	spanContext.ParentSpanID = span.Context().SpanID
	spanContext.ParentService = t.service
	spanContext.ParentServiceInstance = t.instance
	spanContext.ParentEndpoint = firstSpan.GetOperationName()
	spanContext.AddressUsedAtClient = peer
	spanContext.CorrelationContext = span.Context().CorrelationContext

	err = spanContext.Encode(injector)
	if err != nil {
		return nil, nil, err
	}
	return
}

func (t *Tracer) createNoop(ctx context.Context) (s Span, nCtx context.Context) {
	if ns, ok := ctx.Value(ctxKeyInstance).(*NoopSpan); ok {
		nCtx = ctx
		s = ns
		return
	}
	if t.initFlag == 0 {
		s = &NoopSpan{}
		nCtx = context.WithValue(ctx, ctxKeyInstance, s)
		return
	}
	return
}

//Reporter is a data transit specification
type Reporter interface {
	Boot(service string, serviceInstance string, cdsWatchers []AgentConfigChangeWatcher)
	Send(spans []ReportedSpan)
	SendLog(logData ReportedLogData)
	Close()
}