maal-libraries/observer
2
0
Fork 0
Code Issues 5 Pull Requests Packages Projects Releases Wiki Activity

sync

Browse Source
This commit is contained in:
Filip Góra
2024-04-30 12:41:05 +02:00
parent 7a6882c09a
commit c194967b33
9 changed files with 231 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
pkg/gelf_exporter/config.go
View File

@ -2,6 +2,7 @@ package gelfexporter
type config struct {
GelfUrl string
AppName string
}
// newConfig creates a validated Config configured with options.
@ -18,20 +19,6 @@ type Option interface {
apply(config) config
}
// WithWriter sets the export stream destination.
// func WithWriter(w io.Writer) Option {
// return writerOption{w}
// }
// type writerOption struct {
// W *gelf.Writer
// }
// func (o writerOption) apply(cfg config) config {
// cfg.Writer = o.W
// return cfg
// }
func WithGelfUrl(url string) Option {
return gelfUrlOption(url)
}
@ -43,26 +30,13 @@ func (o gelfUrlOption) apply(cfg config) config {
return cfg
}
// // WithPrettyPrint prettifies the emitted output.
// func WithPrettyPrint() Option {
// return prettyPrintOption(true)
// }
func WithAppName(url string) Option {
return appName(url)
}
// type prettyPrintOption bool
type appName string
// func (o prettyPrintOption) apply(cfg config) config {
// cfg.PrettyPrint = bool(o)
// return cfg
// }
// // WithoutTimestamps sets the export stream to not include timestamps.
// func WithoutTimestamps() Option {
// return timestampsOption(false)
// }
// type timestampsOption bool
// func (o timestampsOption) apply(cfg config) config {
// cfg.Timestamps = bool(o)
// return cfg
// }
func (o appName) apply(cfg config) config {
cfg.AppName = string(o)
return cfg
}

2
pkg/gelf_exporter/gelf.go
View File

@ -1,6 +1,7 @@
package gelfexporter
import (
"fmt"
"log"
"maal/tracer/pkg/level"
"time"
@ -30,6 +31,7 @@ func Log(writer *gelf.UDPWriter, msg GELFMessage) {
if err != nil {
log.Println(err)
}
fmt.Printf("msg: %v sent\n", msg)
}
}

84
pkg/gelf_exporter/trace.go
View File

@ -2,17 +2,15 @@ package gelfexporter
import (
"context"
"fmt"
"maal/tracer/pkg/level"
"sync"
"time"
"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 zeroTime time.Time
var _ trace.SpanExporter = &Exporter{}
// New creates an Exporter with the passed options.
@ -33,16 +31,13 @@ func New(options ...Option) (*Exporter, error) {
}
// Exporter is an implementation of trace.SpanSyncer that writes spans to stdout.
type Exporter struct {
timestamps bool
gelfWriter *gelf.UDPWriter
stoppedMu sync.RWMutex
stopped bool
appName string
stoppedMu sync.RWMutex
stopped bool
}
// ExportSpans writes spans in json format to stdout.
func (e *Exporter) ExportSpans(ctx context.Context, spans []trace.ReadOnlySpan) error {
if err := ctx.Err(); err != nil {
return err
@ -62,27 +57,70 @@ func (e *Exporter) ExportSpans(ctx context.Context, spans []trace.ReadOnlySpan)
for i := range stubs {
stub := &stubs[i]
// Remove timestamps
if !e.timestamps {
stub.StartTime = zeroTime
stub.EndTime = zeroTime
for j := range stub.Events {
ev := &stub.Events[j]
ev.Time = zeroTime
}
var attributes = make(map[string]interface{})
for _, attr := range stub.Attributes {
attributes[string(attr.Key)] = GetByType(attr.Value)
}
fmt.Printf("stub: %v\n", stub)
attributes["from"] = "test"
Log(e.gelfWriter, GELFMessage{
Host: "test",
ShortMessage: "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()

32
pkg/tracer/event.go Normal file
View File

@ -0,0 +1,32 @@
package tracer
import (
"encoding/json"
"maal/tracer/pkg/level"
"go.opentelemetry.io/otel/attribute"
"go.opentelemetry.io/otel/codes"
"go.opentelemetry.io/otel/trace"
)
func LongMessage(message string) attribute.KeyValue {
return attribute.KeyValue{Key: "long_message_", Value: attribute.StringValue(message)}
}
func Level(level level.SyslogLevel) attribute.KeyValue {
return attribute.KeyValue{Key: "level_", Value: attribute.Int64Value(int64(level))}
}
func JsonAttr(key string, jsonEl map[string]interface{}) attribute.KeyValue {
jsonStr, _ := json.Marshal(jsonEl)
return attribute.KeyValue{Key: attribute.Key(key), Value: attribute.StringValue(string(jsonStr))}
}
func RecordError(span trace.Span, err error) error {
span.SetStatus(codes.Error, err.Error())
span.RecordError(err)
return nil
}

48
pkg/tracer/middleware.go
View File

@ -2,9 +2,11 @@ package tracer
import (
"context"
"fmt"
"log"
gelfexporter "maal/tracer/pkg/gelf_exporter"
"os"
"runtime"
"github.com/gofiber/fiber/v2"
fiberOpentelemetry "github.com/psmarcin/fiber-opentelemetry/pkg/fiber-otel"
@ -111,13 +113,15 @@ func NewTracer(config Config) func(*fiber.Ctx) error {
otlpExporter := otlptracehttp.NewUnstarted(otlptracehttp.WithEndpointURL(config.JaegerUrl))
gelfExporter, err := gelfexporter.New(gelfexporter.WithGelfUrl(config.GelfUrl))
gelfExporter, err := gelfexporter.New(
gelfexporter.WithGelfUrl(config.GelfUrl),
gelfexporter.WithAppName("salego"),
)
if err != nil {
l.Fatal(err)
}
tracerProviders = append(tracerProviders, trc.WithBatcher(otlpExporter))
tracerProviders = append(tracerProviders, trc.WithBatcher(gelfExporter))
tracerProviders = append(tracerProviders, trc.WithResource(newResource(config)))
TP = *trc.NewTracerProvider(tracerProviders...)
@ -154,7 +158,45 @@ func Handler(fc *fiber.Ctx) (context.Context, trace.Span) {
simpleCtx, span := fiberOpentelemetry.Tracer.Start(fc.UserContext(), fc.OriginalURL())
fc.SetUserContext(simpleCtx)
span.SetAttributes(attribute.String("service.layer", "handler"))
_, file, line, _ := runtime.Caller(1)
span.SetAttributes(
attribute.String("service.layer", "handler"),
attribute.String("file", file),
attribute.String("line", fmt.Sprintf("%d", line)),
)
return simpleCtx, span
}
func Service(c context.Context, serviceName string, spanName string) (context.Context, trace.Span) {
simpleCtx, span := fiberOpentelemetry.Tracer.Start(c, spanName)
var attribs []attribute.KeyValue
_, file, line, _ := runtime.Caller(1)
attribs = append(
attribs,
attribute.String("service.name", serviceName),
attribute.String("service.layer", "service"),
attribute.String("file", file),
attribute.String("line", fmt.Sprintf("%d", line)),
)
span.SetAttributes(attribs...)
return simpleCtx, span
}
func Repository(c context.Context, repositoryName string, spanName string) (context.Context, trace.Span) {
ctx2, span := fiberOpentelemetry.Tracer.Start(c, spanName)
var attribs []attribute.KeyValue
_, file, line, _ := runtime.Caller(1)
attribs = append(attribs,
attribute.String("service.repository", repositoryName),
attribute.String("file", file),
attribute.String("line", fmt.Sprintf("%d", line)),
)
span.SetAttributes(attribs...)
return ctx2, span
}