logotel/initialization;

logotel/go.mod

@@ -0,0 +1,12 @@
+module git.blauwelle.com/go/crate/logotel
+
+go 1.20
+
+require (
+	git.blauwelle.com/go/crate/log v0.3.0
+	git.blauwelle.com/go/crate/synchelper v0.1.0
+	go.opentelemetry.io/otel v1.13.0
+	go.opentelemetry.io/otel/trace v1.13.0
+)
+
+require git.blauwelle.com/go/crate/runtimehelper v0.1.0 // indirect

logotel/go.sum

@@ -0,0 +1,15 @@
+git.blauwelle.com/go/crate/log v0.3.0 h1:oLXMAShuPFQgHc5fNjWU3kcESaS8lpS88P2t5G2b2yA=
+git.blauwelle.com/go/crate/log v0.3.0/go.mod h1:NfiG7YKQCTnLIcn6fVkaa2qEu+DuYi1Kz783Sc/F3jI=
+git.blauwelle.com/go/crate/runtimehelper v0.1.0 h1:qNhtnt9YmHXNHKsGRbwD3AZ3pezpOwrbmX1o9Bz532I=
+git.blauwelle.com/go/crate/runtimehelper v0.1.0/go.mod h1:yVMA0GkO9AS7iuPmalHKeWyv9en0JWj25rY1vpTuHhk=
+git.blauwelle.com/go/crate/synchelper v0.1.0 h1:4yEXpshkklaws/57P94xN5bA3NmyyKGcZqYmzd6QIK4=
+git.blauwelle.com/go/crate/synchelper v0.1.0/go.mod h1:2JkfH+7sF0Q0wiIaDOqG42ZLO5JxpcMfSoyy7db4Y2g=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/google/go-cmp v0.5.9 h1:O2Tfq5qg4qc4AmwVlvv0oLiVAGB7enBSJ2x2DqQFi38=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
+go.opentelemetry.io/otel v1.13.0 h1:1ZAKnNQKwBBxFtww/GwxNUyTf0AxkZzrukO8MeXqe4Y=
+go.opentelemetry.io/otel v1.13.0/go.mod h1:FH3RtdZCzRkJYFTCsAKDy9l/XYjMdNv6QrkFFB8DvVg=
+go.opentelemetry.io/otel/trace v1.13.0 h1:CBgRZ6ntv+Amuj1jDsMhZtlAPT6gbyIRdaIzFhfBSdY=
+go.opentelemetry.io/otel/trace v1.13.0/go.mod h1:muCvmmO9KKpvuXSf3KKAXXB2ygNYHQ+ZfI5X08d3tds=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=

logotel/option.go

@@ -0,0 +1,43 @@
+package logotel
+
+import "git.blauwelle.com/go/crate/synchelper"
+
+func newConfig(opts ...Option) *config {
+	cfg := defaultConfig()
+	for _, opt := range opts {
+		opt.apply(cfg)
+	}
+	if !cfg.hasPool {
+		cfg.bufferPool = synchelper.NewBytesBufferPool(512, 4096)
+	}
+	return cfg
+}
+
+func defaultConfig() *config {
+	return &config{
+		hasPool: false,
+	}
+}
+
+// WithBufferPool 指定缓冲池
+func WithBufferPool(pool synchelper.BytesBufferPool) Option {
+	return optionFunc(func(cfg *config) {
+		cfg.bufferPool = pool
+		cfg.hasPool = true
+	})
+}
+
+// Option 配置 Processor
+type Option interface {
+	apply(cfg *config)
+}
+
+type config struct {
+	bufferPool synchelper.BytesBufferPool
+	hasPool    bool
+}
+type optionFunc func(cfg *config)
+
+func (fn optionFunc) apply(cfg *config) {
+	fn(cfg)
+}

logotel/processor.go

@@ -0,0 +1,90 @@
+// otellog 提供 git.blauwelle.com/go/crate/log 的 opentelemetry 处理功能
+
+package logotel
+
+import (
+	"fmt"
+	"strconv"
+
+	"git.blauwelle.com/go/crate/log/logsdk"
+	"git.blauwelle.com/go/crate/synchelper"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/codes"
+	semconv "go.opentelemetry.io/otel/semconv/v1.17.0"
+	"go.opentelemetry.io/otel/trace"
+)
+
+// New 创建 log/opentelemetry 处理器
+func New(opts ...Option) *Processor {
+	cfg := newConfig(opts...)
+	return &Processor{
+		bufferPool: cfg.bufferPool,
+	}
+}
+
+var _ logsdk.EntryProcessor = &Processor{}
+
+// Processor 用于把日志和 opentelemetry 对接
+type Processor struct {
+	bufferPool synchelper.BytesBufferPool
+}
+
+func (processor *Processor) Process(entry logsdk.ReadonlyEntry) {
+	span := trace.SpanFromContext(entry.Context)
+	if !span.IsRecording() {
+		return
+	}
+
+	attrs := make([]attribute.KeyValue, 0, len(entry.Fields)+6)
+	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("zz.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.Field) attribute.KeyValue {
+	switch value := field.Value.(type) {
+	case nil:
+		return attribute.String(field.Key, "<nil>")
+	case string:
+		return attribute.String(field.Key, value)
+	case int:
+		return attribute.Int(field.Key, value)
+	case int64:
+		return attribute.Int64(field.Key, value)
+	case float64:
+		return attribute.Float64(field.Key, value)
+	case bool:
+		return attribute.Bool(field.Key, value)
+	case error:
+		return attribute.String(field.Key, value.Error())
+	case fmt.Stringer:
+		return attribute.String(field.Key, value.String())
+	}
+
+	return attribute.String(field.Key, fmt.Sprint(field.Value))
+}