1package telemetry
2
3import (
4 "context"
5 "sync"
6
7 "go.opentelemetry.io/otel/attribute"
8 otellog "go.opentelemetry.io/otel/log"
9 sdklog "go.opentelemetry.io/otel/sdk/log"
10 "go.opentelemetry.io/otel/sdk/resource"
11 semconv "go.opentelemetry.io/otel/semconv/v1.26.0"
12
13 "github.com/bnema/gordon/internal/boundaries/out"
14 "github.com/bnema/gordon/internal/domain"
15)
16
17// Attribute keys shared by every exported record.
18const (
19 attrGordonApp = "gordon.app"
20 attrGordonService = "gordon.service"
21 attrLogType = "log.type"
22 attrLogStream = "log.iostream"
23)
24
25var _ out.LogExporter = (*LogExporter)(nil)
26
27// LogExporter implements out.LogExporter over OTLP.
28//
29// OTLP carries service identity on the resource, not on the record, so
30// each source (Gordon or one app service) gets its own lazily created
31// LoggerProvider and resource. All providers share one exporter; the
32// batch processors of every provider serialize through it.
33type LogExporter struct {
34 exporter sdklog.Exporter
35 shared *serialExporter
36 base *resource.Resource
37
38 mu sync.Mutex
39 providers map[domain.LogSource]*sdklog.LoggerProvider
40 closed bool
41}
42
43// NewLogExporter wraps exporter. base carries host/OS/version attributes
44// merged into every source resource; it may be nil.
45func NewLogExporter(exporter sdklog.Exporter, base *resource.Resource) *LogExporter {
46 shared := &serialExporter{Exporter: exporter}
47 return &LogExporter{
48 exporter: exporter,
49 shared: shared,
50 base: base,
51 providers: map[domain.LogSource]*sdklog.LoggerProvider{},
52 }
53}
54
55// Export implements out.LogExporter. It never blocks on the network:
56// the batch processor queues records and drops the oldest on overload.
57func (e *LogExporter) Export(ctx context.Context, record domain.LogRecord) {
58 provider := e.provider(record.Source)
59 if provider == nil {
60 return
61 }
62 var r otellog.Record
63 r.SetTimestamp(record.Time)
64 r.SetObservedTimestamp(record.Time)
65 r.SetBody(attribute.StringValue(record.Body))
66 if sev, text := severity(record.Severity); sev != otellog.SeverityUndefined {
67 r.SetSeverity(sev)
68 r.SetSeverityText(text)
69 }
70 attrs := make([]attribute.KeyValue, 0, len(record.Attributes)+2)
71 if record.Type != "" {
72 attrs = append(attrs, attribute.String(attrLogType, record.Type))
73 }
74 if record.Stream != "" {
75 attrs = append(attrs, attribute.String(attrLogStream, record.Stream))
76 }
77 for key, value := range record.Attributes {
78 attrs = append(attrs, attribute.String(key, value))
79 }
80 r.AddAttributes(attrs...)
81 provider.Logger("gordon").Emit(ctx, r)
82}
83
84// Shutdown flushes every source provider, then closes the exporter.
85func (e *LogExporter) Shutdown(ctx context.Context) error {
86 e.mu.Lock()
87 e.closed = true
88 providers := e.providers
89 e.providers = map[domain.LogSource]*sdklog.LoggerProvider{}
90 e.mu.Unlock()
91
92 var firstErr error
93 for _, provider := range providers {
94 if err := provider.Shutdown(ctx); err != nil && firstErr == nil {
95 firstErr = err
96 }
97 }
98 if err := e.exporter.Shutdown(ctx); err != nil && firstErr == nil {
99 firstErr = err
100 }
101 return firstErr
102}
103
104func (e *LogExporter) provider(source domain.LogSource) *sdklog.LoggerProvider {
105 e.mu.Lock()
106 defer e.mu.Unlock()
107 if e.closed {
108 return nil
109 }
110 if provider, ok := e.providers[source]; ok {
111 return provider
112 }
113 // sourceResource is schemaless, so Merge cannot fail on schema
114 // conflicts; its attributes win over base (service.name).
115 res, err := resource.Merge(e.base, sourceResource(source))
116 if err != nil {
117 res = sourceResource(source)
118 }
119 provider := sdklog.NewLoggerProvider(
120 sdklog.WithProcessor(sdklog.NewBatchProcessor(e.shared)),
121 sdklog.WithResource(res),
122 )
123 e.providers[source] = provider
124 return provider
125}
126
127// sourceResource builds the resource identity of one source:
128// service.name "<app>.<service>" plus namespace and Gordon attributes.
129func sourceResource(source domain.LogSource) *resource.Resource {
130 attrs := []attribute.KeyValue{semconv.ServiceName(source.ServiceName())}
131 if !source.IsGordon() {
132 attrs = append(attrs,
133 semconv.ServiceNamespace(source.App),
134 attribute.String(attrGordonApp, source.App),
135 attribute.String(attrGordonService, source.Service),
136 )
137 }
138 return resource.NewSchemaless(attrs...)
139}
140
141func severity(level domain.LogSeverity) (otellog.Severity, string) {
142 switch level {
143 case domain.LogSeverityInfo:
144 return otellog.SeverityInfo, "INFO"
145 case domain.LogSeverityWarn:
146 return otellog.SeverityWarn, "WARN"
147 case domain.LogSeverityError:
148 return otellog.SeverityError, "ERROR"
149 default:
150 return otellog.SeverityUndefined, ""
151 }
152}
153
154// serialExporter serializes Export calls: the SDK forbids concurrent
155// Export on one exporter, and each source provider runs its own batcher.
156// Shutdown is ignored so one source provider shutting down cannot close
157// the exporter shared by the others; the owner shuts it down once.
158type serialExporter struct {
159 sdklog.Exporter
160 mu sync.Mutex
161}
162
163func (s *serialExporter) Export(ctx context.Context, records []sdklog.Record) error {
164 s.mu.Lock()
165 defer s.mu.Unlock()
166 return s.Exporter.Export(ctx, records)
167}
168
169func (s *serialExporter) Shutdown(context.Context) error { return nil }