observer/pkg/exporters/console_exporter/console_exporter.go

package console_exporter

import (
	"context"
	"fmt"
	"slices"

	"git.ma-al.com/maal-libraries/observer/pkg/attr"
	"git.ma-al.com/maal-libraries/observer/pkg/level"
	"go.opentelemetry.io/otel/attribute"
	"go.opentelemetry.io/otel/sdk/trace"
)

type TraceFormatter interface {
	FormatSpanStart(span trace.ReadOnlySpan, selectedAttrs []attribute.KeyValue, lvl level.SeverityLevel) (string, error)
	FormatSpanEnd(span trace.ReadOnlySpan, selectedAttrs []attribute.KeyValue, lvl level.SeverityLevel) (string, error)
	FormatEvent(event trace.Event, span trace.ReadOnlySpan, selectedAttr []attribute.KeyValue, lvl level.SeverityLevel) (string, error)
}

// Configuration for the exporter.
//
// Most of options are passed to the formatter.
type ProcessorOptions struct {
	// Try to parse filters from an environment variable with a name provided by this field.
	// Result will only by applied to unset options. NOT IMPLEMENTED!
	FilterFromEnvVar *string
	// Filter the output based on the [level.SeverityLevel].
	FilterOnLevel level.SeverityLevel
	// Fields that should be removed from the output.
	FilterOutFields []attribute.Key
	// Print only trace events instead of whole traces.
	EmitEventsOnly bool
	// Add trace id as an attribute.
	AddTraceId bool
	// Print output only when an error is found
	SkipNonErrors bool
	// Used only when `EmitEventsOnly` is set to true.
	TraceFormatter             *TraceFormatter
	SkipEmittingOnSpanStart    bool
	SkipEmittingOnSpanEnd      bool
	SkipAttributesOnSpanEnd    bool
	SkippAttributesOnSpanStart bool
}

type Processor struct {
	lvl                  level.SeverityLevel
	removedFields        []attribute.Key
	addTraceId           bool
	onlyErrs             bool
	onlyEvents           bool
	traceFormatter       TraceFormatter
	skipSpanStart        bool
	skipSpanEnd          bool
	skipAttrsOnSpanEnd   bool
	skipAttrsOnSpanStart bool
}

// NOTE: The configuration might change in future releases
func DefaultConsoleExportProcessor() trace.SpanProcessor {
	fmt := NewPrettyMultilineFormatter()
	return NewProcessor(ProcessorOptions{
		FilterFromEnvVar:           nil,
		FilterOutFields:            []attribute.Key{},
		EmitEventsOnly:             false,
		SkipEmittingOnSpanStart:    false,
		SkippAttributesOnSpanStart: true,
		AddTraceId:                 false,
		TraceFormatter:             &fmt,
	})
}

func NewProcessor(opts ProcessorOptions) trace.SpanProcessor {
	var formatter TraceFormatter
	var lvl level.SeverityLevel

	if opts.TraceFormatter != nil {
		formatter = *opts.TraceFormatter
	} else {
		formatter = TraceFormatter(nil)
	}
	if opts.FilterOnLevel != 0 {
		lvl = opts.FilterOnLevel
	} else {
		lvl = level.TRACE + 10
	}

	return &Processor{
		traceFormatter:       formatter,
		removedFields:        opts.FilterOutFields,
		addTraceId:           opts.AddTraceId,
		onlyEvents:           opts.EmitEventsOnly,
		onlyErrs:             opts.SkipNonErrors,
		skipSpanStart:        opts.SkipEmittingOnSpanStart,
		skipSpanEnd:          opts.SkipEmittingOnSpanEnd,
		skipAttrsOnSpanEnd:   opts.SkipAttributesOnSpanEnd,
		skipAttrsOnSpanStart: opts.SkippAttributesOnSpanStart,
		lvl:                  lvl,
	}
}

// Implements [trace.SpanProcessor]
func (e *Processor) OnStart(ctx context.Context, span trace.ReadWriteSpan) {
	if !e.skipSpanStart && !e.onlyEvents {
		attrs := span.Attributes()
		filteredAttrs := make([]attribute.KeyValue, 0)
		severityLvl := level.TRACE

		if !e.skipAttrsOnSpanStart {
			for i := range attrs {
				if !slices.Contains(e.removedFields, attrs[i].Key) {
					filteredAttrs = append(filteredAttrs, attrs[i])
				}
				if attrs[i].Key == attr.SeverityLevelKey {
					severityLvl = level.FromString(attrs[i].Value.AsString())
				}

			}
		}
		if e.addTraceId {
			filteredAttrs = append(filteredAttrs, attribute.String("trace_id", span.SpanContext().TraceID().String()))
		}
		if severityLvl <= e.lvl {
			line, err := e.traceFormatter.FormatSpanStart(span, filteredAttrs, severityLvl)
			if err != nil {
				fmt.Println("FAILED TO FORMAT SPAN START")
			} else {
				fmt.Printf("%s", line)
			}
		}
	}
	return
}

// Implements [trace.SpanProcessor]
func (e *Processor) OnEnd(span trace.ReadOnlySpan) {
	eventsString := ""
	spanEndString := ""

	for _, event := range span.Events() {
		attrs := event.Attributes
		filteredAttrs := make([]attribute.KeyValue, 0)
		severityLvl := level.TRACE
		for i := range attrs {
			if !slices.Contains(e.removedFields, attrs[i].Key) {
				filteredAttrs = append(filteredAttrs, attrs[i])
			}
			if attrs[i].Key == attr.SeverityLevelKey {
				severityLvl = level.FromString(attrs[i].Value.AsString())
			}
		}
		if e.addTraceId {
			filteredAttrs = append(filteredAttrs, attribute.String("trace_id", span.SpanContext().TraceID().String()))
		}
		if severityLvl <= e.lvl {
			eventString, err := e.traceFormatter.FormatEvent(event, span, filteredAttrs, severityLvl)
			if err != nil {
				fmt.Println("FAILED TO FORMAT TRACE EVENT")
			} else {
				eventsString += eventString
			}
		}
	}

	if !e.skipSpanEnd && !e.onlyEvents {
		attrs := span.Attributes()
		filteredAttrs := make([]attribute.KeyValue, len(attrs))
		severityLvl := level.TRACE

		if !e.skipAttrsOnSpanEnd {
			for i := range attrs {
				if !slices.Contains(e.removedFields, attrs[i].Key) {
					filteredAttrs = append(filteredAttrs, attrs[i])
				}
				if attrs[i].Key == attr.SeverityLevelKey {
					severityLvl = level.FromString(attrs[i].Value.AsString())

				}

			}
		}
		if e.addTraceId {
			filteredAttrs = append(filteredAttrs, attribute.String("trace_id", span.SpanContext().TraceID().String()))
		}
		if severityLvl <= e.lvl {
			spanString, err := e.traceFormatter.FormatSpanEnd(span, filteredAttrs, severityLvl)
			if err != nil {
				fmt.Println("FAILED TO FORMAT SPAN END")
			} else {
				spanEndString += spanString
			}
		}

	}

	if e.skipSpanStart {
		fmt.Printf("%s", spanEndString+eventsString)
	} else {
		fmt.Printf("%s", eventsString+spanEndString)
	}

	return
}

// Implements [trace.SpanProcessor]
func (e *Processor) ForceFlush(ctx context.Context) error {
	return nil
}

// Implements [trace.SpanExporter]
func (e *Processor) Shutdown(ctx context.Context) error {
	select {
	case <-ctx.Done():
		return ctx.Err()
	default:
	}
	return nil
}