some refinement
Build and Publish / Metadata (push) Has been cancelled
Build and Publish / BuildAndDeployAmd64 (push) Has been cancelled

This commit is contained in:
2026-04-07 12:44:55 +02:00
parent 89e1db8edf
commit 0ed1cb4355
14 changed files with 390 additions and 104 deletions
+12 -28
View File
@@ -47,7 +47,7 @@ func (c *Collector) SetBroadcaster(broadcaster websocket.Broadcaster) {
c.broadcaster = broadcaster
}
func (c *Collector) addToHistory(dataType websocket.DataType, data interface{}) {
func (c *Collector) addToHistory(dataType websocket.DataType, data json.RawMessage) {
entry := websocket.HistoryEntry{
Type: string(dataType),
Timestamp: time.Now().UnixNano(),
@@ -149,9 +149,8 @@ func (c *Collector) SearchHistory(ctx context.Context, query string, limit int)
if query == "" {
filtered = append(filtered, entry)
} else {
// Basic string filter for in-memory
data, _ := json.Marshal(entry.Data)
if strings.Contains(strings.ToLower(string(data)), lowerQuery) {
// Basic string filter for in-memory, no need to re-marshal RawMessage
if strings.Contains(strings.ToLower(string(entry.Data)), lowerQuery) {
filtered = append(filtered, entry)
}
}
@@ -180,17 +179,12 @@ func (c *Collector) ConsumeTraces(ctx context.Context, td ptrace.Traces) error {
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)
}
// Store and broadcast raw bytes as json.RawMessage to avoid unsafe interface{} usage
c.addToHistory(websocket.DataTypeTraces, buf)
entry := websocket.Entry{
Type: string(websocket.DataTypeTraces),
Data: data,
Data: buf,
}
c.broadcaster.Broadcast(&entry)
@@ -206,17 +200,12 @@ func (c *Collector) ConsumeMetrics(ctx context.Context, md pmetric.Metrics) erro
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)
}
// Store and broadcast raw bytes as json.RawMessage to avoid unsafe interface{} usage
c.addToHistory(websocket.DataTypeMetrics, buf)
entry := websocket.Entry{
Type: string(websocket.DataTypeMetrics),
Data: data,
Data: buf,
}
c.broadcaster.Broadcast(&entry)
@@ -232,17 +221,12 @@ func (c *Collector) ConsumeLogs(ctx context.Context, ld plog.Logs) error {
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)
}
// Store and broadcast raw bytes as json.RawMessage to avoid unsafe interface{} usage
c.addToHistory(websocket.DataTypeLogs, buf)
entry := websocket.Entry{
Type: string(websocket.DataTypeLogs),
Data: data,
Data: buf,
}
c.broadcaster.Broadcast(&entry)
+3 -2
View File
@@ -2,6 +2,7 @@ package collector
import (
"context"
"encoding/json"
"testing"
"time"
@@ -29,7 +30,7 @@ func TestCollectorWithStore(t *testing.T) {
store := &mockStore{}
coll := New(nil, 10, store)
coll.addToHistory(websocket.DataTypeLogs, map[string]interface{}{"msg": "test"})
coll.addToHistory(websocket.DataTypeLogs, json.RawMessage(`{"msg": "test"}`))
// Give it a moment since it's asynchronous
timeout := time.After(100 * time.Millisecond)
@@ -55,7 +56,7 @@ func TestCollectorHistoryLimit(t *testing.T) {
coll := New(nil, 10, nil)
for i := 0; i < 10; i++ {
coll.addToHistory(websocket.DataTypeLogs, map[string]interface{}{"msg": i})
coll.addToHistory(websocket.DataTypeLogs, json.RawMessage(`{"msg": "test"}`))
}
history := coll.GetHistory(10)
+2 -2
View File
@@ -21,12 +21,12 @@ func TestFileStore(t *testing.T) {
entry1 := websocket.HistoryEntry{
Type: "logs",
Timestamp: 100,
Data: map[string]interface{}{"msg": "hello world"},
Data: json.RawMessage(`{"msg":"hello world"}`),
}
entry2 := websocket.HistoryEntry{
Type: "traces",
Timestamp: 200,
Data: map[string]interface{}{"msg": "trace event"},
Data: json.RawMessage(`{"msg":"trace event"}`),
}
if err := fs.Append(ctx, entry1); err != nil {
+5 -5
View File
@@ -27,15 +27,15 @@ type HistoryProvider interface {
// HistoryEntry represents a single archived telemetry entry.
type HistoryEntry struct {
Type string `json:"type"`
Timestamp int64 `json:"timestamp"`
Data interface{} `json:"data"`
Type string `json:"type"`
Timestamp int64 `json:"timestamp"`
Data json.RawMessage `json:"data"`
}
// Entry represents a telemetry update sent to clients.
type Entry struct {
Type string `json:"type"`
Data interface{} `json:"data"`
Type string `json:"type"`
Data json.RawMessage `json:"data"`
}
// Broadcaster interface for sending telemetry updates to all connected clients.