observer/pkg/gelf_exporter/trace.go
2024-04-30 14:42:05 +02:00

143 lines
2.9 KiB
Go

package gelfexporter
import (
"context"
"sync"
"git.ma-al.com/gora_filip/observer/pkg/level"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/sdk/trace"
"go.opentelemetry.io/otel/sdk/trace/tracetest"
"gopkg.in/Graylog2/go-gelf.v2/gelf"
)
var _ trace.SpanExporter = &Exporter{}
// New creates an Exporter with the passed options.
func New(options ...Option) (*Exporter, error) {
cfg, err := newConfig(options...)
if err != nil {
return nil, err
}
writer, err := gelf.NewUDPWriter(cfg.GelfUrl)
if err != nil {
return nil, err
}
return &Exporter{
gelfWriter: writer,
}, nil
}
type Exporter struct {
gelfWriter *gelf.UDPWriter
appName string
stoppedMu sync.RWMutex
stopped bool
}
func (e *Exporter) ExportSpans(ctx context.Context, spans []trace.ReadOnlySpan) error {
if err := ctx.Err(); err != nil {
return err
}
e.stoppedMu.RLock()
stopped := e.stopped
e.stoppedMu.RUnlock()
if stopped {
return nil
}
if len(spans) == 0 {
return nil
}
stubs := tracetest.SpanStubsFromReadOnlySpans(spans)
for i := range stubs {
stub := &stubs[i]
var attributes = make(map[string]interface{})
for _, attr := range stub.Attributes {
attributes[string(attr.Key)] = GetByType(attr.Value)
}
attributes["from"] = "test"
for i := range stub.Events {
event := stub.Events[i]
var gelf GELFMessage = GELFMessage{
Host: "salego",
ShortMessage: event.Name,
Timestamp: stub.StartTime,
Level: level.DEBUG,
ExtraFields: attributes,
}
for _, attr := range event.Attributes {
if attr.Key == "long_message_" {
gelf.LongMessage = attr.Value.AsString()
continue
}
if attr.Key == "level_" {
gelf.Level = level.SyslogLevel(attr.Value.AsInt64())
continue
}
attributes[string(attr.Key)] = GetByType(attr.Value)
}
Log(e.gelfWriter, gelf)
}
}
return nil
}
func GetByType(val attribute.Value) interface{} {
switch val.Type() {
case attribute.INVALID:
return "invalid value"
case attribute.BOOL:
return val.AsBool()
case attribute.INT64:
return val.AsInt64()
case attribute.FLOAT64:
return val.AsFloat64()
case attribute.STRING:
return val.AsString()
case attribute.BOOLSLICE:
return val.AsBoolSlice()
case attribute.INT64SLICE:
return val.AsInt64Slice()
case attribute.FLOAT64SLICE:
return val.AsFloat64Slice()
case attribute.STRINGSLICE:
return val.AsStringSlice()
}
return "invalid value"
}
// Shutdown is called to stop the exporter, it performs no action.
func (e *Exporter) Shutdown(ctx context.Context) error {
e.stoppedMu.Lock()
e.stopped = true
e.stoppedMu.Unlock()
return nil
}
// // MarshalLog is the marshaling function used by the logging system to represent this Exporter.
// func (e *Exporter) MarshalLog() interface{} {
// return struct {
// Type string
// WithTimestamps bool
// }{
// Type: "stdout",
// WithTimestamps: e.timestamps,
// }
// }