crate/logotel/processor.go

// otellog 提供 git.blauwelle.com/go/crate/log 的 opentelemetry 处理功能

package logotel

import (
	"context"
	"fmt"
	"strconv"

	"git.blauwelle.com/go/crate/log/logsdk"
	"git.blauwelle.com/go/crate/log/logsdk/logjson"
	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/codes"
	semconv "go.opentelemetry.io/otel/semconv/v1.17.0"
	"go.opentelemetry.io/otel/trace"
)

const (
	fieldPrefix = "log.field."
)

var (
	tracer = otel.Tracer(tracerName, trace.WithInstrumentationVersion("semver:"+version))
)

// New 创建 log/opentelemetry 处理器
func New(opts ...Option) *Processor {
	cfg := newConfig(opts...)
	return &Processor{
		bufferPool:  cfg.bytesBufferPool,
		defaultSpan: cfg.defaultSpan,
	}
}

var _ logsdk.EntryProcessor = &Processor{}

// Processor 用于把日志和 opentelemetry 对接
type Processor struct {
	bufferPool  logjson.BytesBufferPool
	defaultSpan bool
}

func (processor *Processor) Process(ctx context.Context, entry logsdk.ReadonlyEntry) {
	span := trace.SpanFromContext(ctx)
	if !span.IsRecording() {
		if processor.defaultSpan {
			name := entry.Message
			if name == "" {
				name = "default"
			}
			ctx, span = tracer.Start(ctx, name) //nolint:ineffassign,staticcheck,wastedassign
			defer span.End()
		}
	}
	if !span.IsRecording() {
		return
	}

	const entryInitSize = 6
	attrs := make([]attribute.KeyValue, 0, len(entry.Fields)+entryInitSize)
	attrs = append(attrs, attribute.String("log.severity", entry.Level.String()))
	attrs = append(attrs, attribute.String("log.message", entry.Message))
	if entry.Caller.IsValid() {
		attrs = append(attrs, semconv.CodeFunctionKey.String(entry.Caller.Function))
		attrs = append(attrs, semconv.CodeFilepathKey.String(entry.Caller.File))
		attrs = append(attrs, semconv.CodeLineNumberKey.Int(entry.Caller.Line))
	}
	if len(entry.Stack) > 0 {
		buf := processor.bufferPool.Get()
		for _, frame := range entry.Stack {
			buf.WriteString(frame.Function)
			buf.WriteByte('\n')
			buf.WriteByte('\t')
			buf.WriteString(frame.File)
			buf.WriteByte(':')
			buf.WriteString(strconv.Itoa(frame.Line))
			buf.WriteByte('\n')
		}
		processor.bufferPool.Put(buf)
		attrs = append(attrs, attribute.String("log.stack", buf.String()))
	}
	for _, field := range entry.Fields {
		attrs = append(attrs, fieldToKV(field))
	}
	span.AddEvent("log", trace.WithTimestamp(entry.Time), trace.WithAttributes(attrs...))
	if entry.Level <= logsdk.LevelError {
		span.SetStatus(codes.Error, entry.Message)
	}
}

func fieldToKV(field logsdk.KV) attribute.KeyValue {
	switch value := field.Value.(type) {
	case nil:
		return attribute.String(fieldPrefix+field.Key, "<nil>")
	case string:
		return attribute.String(fieldPrefix+field.Key, value)
	case int:
		return attribute.Int(fieldPrefix+field.Key, value)
	case int64:
		return attribute.Int64(fieldPrefix+field.Key, value)
	case float64:
		return attribute.Float64(fieldPrefix+field.Key, value)
	case bool:
		return attribute.Bool(fieldPrefix+field.Key, value)
	case error:
		return attribute.String(fieldPrefix+field.Key, value.Error())
	case fmt.Stringer:
		return attribute.String(fieldPrefix+field.Key, value.String())
	}

	return attribute.String(fieldPrefix+field.Key, fmt.Sprint(field.Value))
}