144 lines
3.3 KiB
Go
144 lines
3.3 KiB
Go
package collector
|
|
|
|
import (
|
|
"container/ring"
|
|
"context"
|
|
"encoding/json"
|
|
"log"
|
|
"sync"
|
|
"time"
|
|
|
|
"git.tornberg.me/go-otel/internal/websocket"
|
|
|
|
"go.opentelemetry.io/collector/consumer"
|
|
"go.opentelemetry.io/collector/pdata/plog"
|
|
"go.opentelemetry.io/collector/pdata/pmetric"
|
|
"go.opentelemetry.io/collector/pdata/ptrace"
|
|
)
|
|
|
|
type Collector struct {
|
|
broadcaster websocket.Broadcaster
|
|
historySize int
|
|
history *ring.Ring
|
|
historyMutex sync.RWMutex
|
|
}
|
|
|
|
func New(broadcaster websocket.Broadcaster, historySize int) *Collector {
|
|
return &Collector{
|
|
broadcaster: broadcaster,
|
|
historySize: historySize,
|
|
history: ring.New(historySize),
|
|
}
|
|
}
|
|
|
|
func (c *Collector) SetBroadcaster(broadcaster websocket.Broadcaster) {
|
|
c.broadcaster = broadcaster
|
|
}
|
|
|
|
func (c *Collector) addToHistory(dataType websocket.DataType, data interface{}) {
|
|
c.historyMutex.Lock()
|
|
defer c.historyMutex.Unlock()
|
|
|
|
c.history.Value = websocket.HistoryEntry{
|
|
Type: string(dataType),
|
|
Timestamp: time.Now().UnixNano(),
|
|
Data: data,
|
|
}
|
|
c.history = c.history.Next()
|
|
}
|
|
|
|
func (c *Collector) GetHistory() []websocket.HistoryEntry {
|
|
c.historyMutex.RLock()
|
|
defer c.historyMutex.RUnlock()
|
|
|
|
var history []websocket.HistoryEntry
|
|
c.history.Do(func(v interface{}) {
|
|
if entry, ok := v.(websocket.HistoryEntry); ok && entry.Type != "" {
|
|
history = append(history, entry)
|
|
}
|
|
})
|
|
return history
|
|
}
|
|
|
|
func (c *Collector) ConsumeTraces(ctx context.Context, td ptrace.Traces) error {
|
|
marshaler := &ptrace.JSONMarshaler{}
|
|
buf, err := marshaler.MarshalTraces(td)
|
|
if err != nil {
|
|
log.Printf("failed to marshal traces: %v", err)
|
|
return err
|
|
}
|
|
|
|
// Parse to interface{} for history storage
|
|
var data interface{}
|
|
if err := json.Unmarshal(buf, &data); err != nil {
|
|
log.Printf("failed to unmarshal traces for history: %v", err)
|
|
} else {
|
|
c.addToHistory(websocket.DataTypeTraces, data)
|
|
}
|
|
|
|
entry := websocket.Entry{
|
|
Type: string(websocket.DataTypeTraces),
|
|
Data: data,
|
|
}
|
|
|
|
c.broadcaster.Broadcast(&entry)
|
|
return nil
|
|
}
|
|
|
|
func (c *Collector) ConsumeMetrics(ctx context.Context, md pmetric.Metrics) error {
|
|
marshaler := &pmetric.JSONMarshaler{}
|
|
buf, err := marshaler.MarshalMetrics(md)
|
|
if err != nil {
|
|
log.Printf("failed to marshal metrics: %v", err)
|
|
return err
|
|
}
|
|
|
|
// Parse to interface{} for history storage
|
|
var data interface{}
|
|
if err := json.Unmarshal(buf, &data); err != nil {
|
|
log.Printf("failed to unmarshal metrics for history: %v", err)
|
|
} else {
|
|
c.addToHistory(websocket.DataTypeMetrics, data)
|
|
}
|
|
|
|
entry := websocket.Entry{
|
|
Type: string(websocket.DataTypeMetrics),
|
|
Data: data,
|
|
}
|
|
|
|
c.broadcaster.Broadcast(&entry)
|
|
return nil
|
|
}
|
|
|
|
func (c *Collector) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
|
|
marshaler := &plog.JSONMarshaler{}
|
|
buf, err := marshaler.MarshalLogs(ld)
|
|
if err != nil {
|
|
log.Printf("failed to marshal logs: %v", err)
|
|
return err
|
|
}
|
|
|
|
// Parse to interface{} for history storage
|
|
var data interface{}
|
|
if err := json.Unmarshal(buf, &data); err != nil {
|
|
log.Printf("failed to unmarshal logs for history: %v", err)
|
|
} else {
|
|
c.addToHistory(websocket.DataTypeLogs, data)
|
|
}
|
|
|
|
entry := websocket.Entry{
|
|
Type: string(websocket.DataTypeLogs),
|
|
Data: data,
|
|
}
|
|
|
|
c.broadcaster.Broadcast(&entry)
|
|
|
|
//c.broadcaster.Broadcast(buf)
|
|
return nil
|
|
}
|
|
|
|
// Implement consumer interfaces
|
|
func (c *Collector) Capabilities() consumer.Capabilities {
|
|
return consumer.Capabilities{MutatesData: false}
|
|
}
|