telegraf/plugins/inputs/zipkin/codec/jsonV1/jsonV1.go

package jsonV1

import (
	"encoding/json"
	"fmt"
	"strconv"
	"time"

	"github.com/influxdata/telegraf/plugins/inputs/zipkin/codec"
	"github.com/openzipkin/zipkin-go-opentracing/_thrift/gen-go/zipkincore"
)

// JSON decodes spans from  bodies `POST`ed to the spans endpoint
type JSON struct{}

// Decode unmarshals and validates the JSON body
func (j *JSON) Decode(octets []byte) ([]codec.Span, error) {
	var spans []span
	err := json.Unmarshal(octets, &spans)
	if err != nil {
		return nil, err
	}

	res := make([]codec.Span, len(spans))
	for i := range spans {
		if err := spans[i].Validate(); err != nil {
			return nil, err
		}
		res[i] = &spans[i]
	}
	return res, nil
}

type span struct {
	TraceID  string             `json:"traceId"`
	SpanName string             `json:"name"`
	ParentID string             `json:"parentId,omitempty"`
	ID       string             `json:"id"`
	Time     *int64             `json:"timestamp,omitempty"`
	Dur      *int64             `json:"duration,omitempty"`
	Debug    bool               `json:"debug,omitempty"`
	Anno     []annotation       `json:"annotations"`
	BAnno    []binaryAnnotation `json:"binaryAnnotations"`
}

func (s *span) Validate() error {
	var err error
	check := func(f func() (string, error)) {
		if err != nil {
			return
		}
		_, err = f()
	}

	check(s.Trace)
	check(s.SpanID)
	check(s.Parent)
	if err != nil {
		return err
	}

	_, err = s.BinaryAnnotations()
	return err
}

func (s *span) Trace() (string, error) {
	if s.TraceID == "" {
		return "", fmt.Errorf("Trace ID cannot be null")
	}
	return TraceIDFromString(s.TraceID)
}

func (s *span) SpanID() (string, error) {
	if s.ID == "" {
		return "", fmt.Errorf("Span ID cannot be null")
	}
	return IDFromString(s.ID)
}

func (s *span) Parent() (string, error) {
	if s.ParentID == "" {
		return "", nil
	}
	return IDFromString(s.ParentID)
}

func (s *span) Name() string {
	return s.SpanName
}

func (s *span) Annotations() []codec.Annotation {
	res := make([]codec.Annotation, len(s.Anno))
	for i := range s.Anno {
		res[i] = &s.Anno[i]
	}
	return res
}

func (s *span) BinaryAnnotations() ([]codec.BinaryAnnotation, error) {
	res := make([]codec.BinaryAnnotation, len(s.BAnno))
	for i, a := range s.BAnno {
		if a.Key() != "" && a.Value() == "" {
			return nil, fmt.Errorf("No value for key %s at binaryAnnotations[%d]", a.K, i)
		}
		if a.Value() != "" && a.Key() == "" {
			return nil, fmt.Errorf("No key at binaryAnnotations[%d]", i)
		}
		res[i] = &s.BAnno[i]
	}
	return res, nil
}

func (s *span) Timestamp() time.Time {
	if s.Time == nil {
		return time.Time{}
	}
	return codec.MicroToTime(*s.Time)
}

func (s *span) Duration() time.Duration {
	if s.Dur == nil {
		return 0
	}
	return time.Duration(*s.Dur) * time.Microsecond
}

type annotation struct {
	Endpoint *endpoint `json:"endpoint,omitempty"`
	Time     int64     `json:"timestamp"`
	Val      string    `json:"value,omitempty"`
}

func (a *annotation) Timestamp() time.Time {
	return codec.MicroToTime(a.Time)
}

func (a *annotation) Value() string {
	return a.Val
}

func (a *annotation) Host() codec.Endpoint {
	return a.Endpoint
}

type binaryAnnotation struct {
	K        string          `json:"key"`
	V        json.RawMessage `json:"value"`
	Type     string          `json:"type"`
	Endpoint *endpoint       `json:"endpoint,omitempty"`
}

func (b *binaryAnnotation) Key() string {
	return b.K
}

func (b *binaryAnnotation) Value() string {
	t, err := zipkincore.AnnotationTypeFromString(b.Type)
	// Assume this is a string if we cannot tell the type
	if err != nil {
		t = zipkincore.AnnotationType_STRING
	}

	switch t {
	case zipkincore.AnnotationType_BOOL:
		var v bool
		err := json.Unmarshal(b.V, &v)
		if err == nil {
			return strconv.FormatBool(v)
		}
	case zipkincore.AnnotationType_BYTES:
		return string(b.V)
	case zipkincore.AnnotationType_I16, zipkincore.AnnotationType_I32, zipkincore.AnnotationType_I64:
		var v int64
		err := json.Unmarshal(b.V, &v)
		if err == nil {
			return strconv.FormatInt(v, 10)
		}
	case zipkincore.AnnotationType_DOUBLE:
		var v float64
		err := json.Unmarshal(b.V, &v)
		if err == nil {
			return strconv.FormatFloat(v, 'f', -1, 64)
		}
	case zipkincore.AnnotationType_STRING:
		var v string
		err := json.Unmarshal(b.V, &v)
		if err == nil {
			return v
		}
	}

	return ""
}

func (b *binaryAnnotation) Host() codec.Endpoint {
	return b.Endpoint
}

type endpoint struct {
	ServiceName string `json:"serviceName"`
	Ipv4        string `json:"ipv4"`
	Ipv6        string `json:"ipv6,omitempty"`
	Port        int    `json:"port"`
}

func (e *endpoint) Host() string {
	if e.Port != 0 {
		return fmt.Sprintf("%s:%d", e.Ipv4, e.Port)
	}
	return e.Ipv4
}

func (e *endpoint) Name() string {
	return e.ServiceName
}

// TraceIDFromString creates a TraceID from a hexadecimal string
func TraceIDFromString(s string) (string, error) {
	var hi, lo uint64
	var err error
	if len(s) > 32 {
		return "", fmt.Errorf("TraceID cannot be longer than 32 hex characters: %s", s)
	} else if len(s) > 16 {
		hiLen := len(s) - 16
		if hi, err = strconv.ParseUint(s[0:hiLen], 16, 64); err != nil {
			return "", err
		}
		if lo, err = strconv.ParseUint(s[hiLen:], 16, 64); err != nil {
			return "", err
		}
	} else {
		if lo, err = strconv.ParseUint(s, 16, 64); err != nil {
			return "", err
		}
	}
	if hi == 0 {
		return fmt.Sprintf("%x", lo), nil
	}
	return fmt.Sprintf("%x%016x", hi, lo), nil
}

// IDFromString validates the ID and returns it in hexadecimal format.
func IDFromString(s string) (string, error) {
	if len(s) > 16 {
		return "", fmt.Errorf("ID cannot be longer than 16 hex characters: %s", s)
	}
	id, err := strconv.ParseUint(s, 16, 64)
	if err != nil {
		return "", err
	}
	return strconv.FormatUint(id, 16), nil
}