Skip to content

Commit c754072

Browse files
AnnatarHeAnnatarHe
authored
Merge pull request #148 from shelltime/refactor/ccotel-remove-session-flat-structure
refactor(otel): remove session concept, embed resource attrs in metrics/events
2 parents d24211d + 38a5570 commit c754072

3 files changed

Lines changed: 246 additions & 75 deletions

File tree

‎daemon/ccotel_processor.go‎

Lines changed: 175 additions & 43 deletions
Original file line numberDiff line numberDiff line change
@@ -5,7 +5,7 @@ import (
55
"encoding/json"
66
"log/slog"
77
"os"
8-
"time"
8+
"strconv"
99

1010
"github.com/google/uuid"
1111
"github.com/malamtime/cli/model"
@@ -54,14 +54,15 @@ func (p *CCOtelProcessor) ProcessMetrics(ctx context.Context, req *collmetricsv1
5454
continue
5555
}
5656

57-
session := extractSessionFromResource(resource)
57+
// Extract resource attributes once for all metrics in this resource
58+
resourceAttrs := extractResourceAttributes(resource)
5859
project := p.detectProject(resource)
5960

6061
var metrics []model.CCOtelMetric
6162

6263
for _, sm := range rm.GetScopeMetrics() {
6364
for _, m := range sm.GetMetrics() {
64-
parsedMetrics := p.parseMetric(m)
65+
parsedMetrics := p.parseMetric(m, resourceAttrs)
6566
metrics = append(metrics, parsedMetrics...)
6667
}
6768
}
@@ -70,11 +71,10 @@ func (p *CCOtelProcessor) ProcessMetrics(ctx context.Context, req *collmetricsv1
7071
continue
7172
}
7273

73-
// Build and send request immediately
74+
// Build and send request immediately - flat structure without session
7475
ccReq := &model.CCOtelRequest{
7576
Host: p.hostname,
7677
Project: project,
77-
Session: session,
7878
Metrics: metrics,
7979
}
8080

@@ -103,14 +103,15 @@ func (p *CCOtelProcessor) ProcessLogs(ctx context.Context, req *collogsv1.Export
103103
continue
104104
}
105105

106-
session := extractSessionFromResource(resource)
106+
// Extract resource attributes once for all events in this resource
107+
resourceAttrs := extractResourceAttributes(resource)
107108
project := p.detectProject(resource)
108109

109110
var events []model.CCOtelEvent
110111

111112
for _, sl := range rl.GetScopeLogs() {
112113
for _, lr := range sl.GetLogRecords() {
113-
event := p.parseLogRecord(lr)
114+
event := p.parseLogRecord(lr, resourceAttrs)
114115
if event != nil {
115116
events = append(events, *event)
116117
}
@@ -121,11 +122,10 @@ func (p *CCOtelProcessor) ProcessLogs(ctx context.Context, req *collogsv1.Export
121122
continue
122123
}
123124

124-
// Build and send request immediately
125+
// Build and send request immediately - flat structure without session
125126
ccReq := &model.CCOtelRequest{
126127
Host: p.hostname,
127128
Project: project,
128-
Session: session,
129129
Events: events,
130130
}
131131

@@ -155,50 +155,105 @@ func isClaudeCodeResource(resource *resourcev1.Resource) bool {
155155
return false
156156
}
157157

158-
// extractSessionFromResource extracts session info from resource attributes
159-
func extractSessionFromResource(resource *resourcev1.Resource) *model.CCOtelSession {
160-
session := &model.CCOtelSession{
161-
StartedAt: time.Now().Unix(),
162-
}
158+
// extractResourceAttributes extracts resource-level attributes from OTEL resource
159+
// Returns a struct that can be used to populate metrics and events
160+
func extractResourceAttributes(resource *resourcev1.Resource) *model.CCOtelResourceAttributes {
161+
attrs := &model.CCOtelResourceAttributes{}
163162

164163
if resource == nil {
165-
return session
164+
return attrs
166165
}
167166

168167
for _, attr := range resource.GetAttributes() {
169168
key := attr.GetKey()
170169
value := attr.GetValue()
171170

172171
switch key {
172+
// Standard resource attributes
173173
case "session.id":
174-
session.SessionID = value.GetStringValue()
174+
attrs.SessionID = value.GetStringValue()
175175
case "app.version":
176-
session.AppVersion = value.GetStringValue()
176+
attrs.AppVersion = value.GetStringValue()
177177
case "organization.id":
178-
session.OrganizationID = value.GetStringValue()
178+
attrs.OrganizationID = value.GetStringValue()
179179
case "user.account_uuid":
180-
session.UserAccountUUID = value.GetStringValue()
180+
attrs.UserAccountUUID = value.GetStringValue()
181181
case "terminal.type":
182-
session.TerminalType = value.GetStringValue()
182+
attrs.TerminalType = value.GetStringValue()
183183
case "service.version":
184-
session.ServiceVersion = value.GetStringValue()
184+
attrs.ServiceVersion = value.GetStringValue()
185185
case "os.type":
186-
session.OSType = value.GetStringValue()
186+
attrs.OSType = value.GetStringValue()
187187
case "os.version":
188-
session.OSVersion = value.GetStringValue()
188+
attrs.OSVersion = value.GetStringValue()
189189
case "host.arch":
190-
session.HostArch = value.GetStringValue()
190+
attrs.HostArch = value.GetStringValue()
191191
case "wsl.version":
192-
session.WSLVersion = value.GetStringValue()
192+
attrs.WSLVersion = value.GetStringValue()
193+
// Additional identifiers
194+
case "user.id":
195+
attrs.UserID = value.GetStringValue()
196+
case "user.email":
197+
attrs.UserEmail = value.GetStringValue()
198+
// Custom resource attributes (from OTEL_RESOURCE_ATTRIBUTES)
199+
case "user.name":
200+
attrs.UserName = value.GetStringValue()
201+
case "machine.name":
202+
attrs.MachineName = value.GetStringValue()
203+
case "team.id":
204+
attrs.TeamID = value.GetStringValue()
205+
case "pwd":
206+
attrs.Pwd = value.GetStringValue()
193207
}
194208
}
195209

196-
// Generate session ID if not present
197-
if session.SessionID == "" {
198-
session.SessionID = uuid.New().String()
199-
}
210+
return attrs
211+
}
200212

201-
return session
213+
// applyResourceAttributesToMetric copies resource attributes into a metric
214+
func applyResourceAttributesToMetric(metric *model.CCOtelMetric, attrs *model.CCOtelResourceAttributes) {
215+
// Standard resource attributes
216+
metric.SessionID = attrs.SessionID
217+
metric.UserAccountUUID = attrs.UserAccountUUID
218+
metric.OrganizationID = attrs.OrganizationID
219+
metric.TerminalType = attrs.TerminalType
220+
metric.AppVersion = attrs.AppVersion
221+
metric.OSType = attrs.OSType
222+
metric.OSVersion = attrs.OSVersion
223+
metric.HostArch = attrs.HostArch
224+
225+
// Additional identifiers
226+
metric.UserID = attrs.UserID
227+
metric.UserEmail = attrs.UserEmail
228+
229+
// Custom resource attributes
230+
metric.UserName = attrs.UserName
231+
metric.MachineName = attrs.MachineName
232+
metric.TeamID = attrs.TeamID
233+
metric.Pwd = attrs.Pwd
234+
}
235+
236+
// applyResourceAttributesToEvent copies resource attributes into an event
237+
func applyResourceAttributesToEvent(event *model.CCOtelEvent, attrs *model.CCOtelResourceAttributes) {
238+
// Standard resource attributes
239+
event.SessionID = attrs.SessionID
240+
event.UserAccountUUID = attrs.UserAccountUUID
241+
event.OrganizationID = attrs.OrganizationID
242+
event.TerminalType = attrs.TerminalType
243+
event.AppVersion = attrs.AppVersion
244+
event.OSType = attrs.OSType
245+
event.OSVersion = attrs.OSVersion
246+
event.HostArch = attrs.HostArch
247+
248+
// Additional identifiers
249+
event.UserID = attrs.UserID
250+
event.UserEmail = attrs.UserEmail
251+
252+
// Custom resource attributes
253+
event.UserName = attrs.UserName
254+
event.MachineName = attrs.MachineName
255+
event.TeamID = attrs.TeamID
256+
event.Pwd = attrs.Pwd
202257
}
203258

204259
// detectProject extracts project from resource attributes or environment
@@ -224,7 +279,7 @@ func (p *CCOtelProcessor) detectProject(resource *resourcev1.Resource) string {
224279
}
225280

226281
// parseMetric parses an OTEL metric into CCOtelMetric(s)
227-
func (p *CCOtelProcessor) parseMetric(m *metricsv1.Metric) []model.CCOtelMetric {
282+
func (p *CCOtelProcessor) parseMetric(m *metricsv1.Metric, resourceAttrs *model.CCOtelResourceAttributes) []model.CCOtelMetric {
228283
var metrics []model.CCOtelMetric
229284

230285
name := m.GetName()
@@ -243,7 +298,9 @@ func (p *CCOtelProcessor) parseMetric(m *metricsv1.Metric) []model.CCOtelMetric
243298
Timestamp: int64(dp.GetTimeUnixNano() / 1e9), // Convert to seconds
244299
Value: getDataPointValue(dp),
245300
}
246-
// Extract attributes
301+
// Apply resource attributes first
302+
applyResourceAttributesToMetric(&metric, resourceAttrs)
303+
// Then extract data point attributes (can override resource attrs)
247304
for _, attr := range dp.GetAttributes() {
248305
applyMetricAttribute(&metric, attr, metricType)
249306
}
@@ -257,6 +314,9 @@ func (p *CCOtelProcessor) parseMetric(m *metricsv1.Metric) []model.CCOtelMetric
257314
Timestamp: int64(dp.GetTimeUnixNano() / 1e9),
258315
Value: getDataPointValue(dp),
259316
}
317+
// Apply resource attributes first
318+
applyResourceAttributesToMetric(&metric, resourceAttrs)
319+
// Then extract data point attributes (can override resource attrs)
260320
for _, attr := range dp.GetAttributes() {
261321
applyMetricAttribute(&metric, attr, metricType)
262322
}
@@ -268,34 +328,39 @@ func (p *CCOtelProcessor) parseMetric(m *metricsv1.Metric) []model.CCOtelMetric
268328
}
269329

270330
// parseLogRecord parses an OTEL log record into a CCOtelEvent
271-
func (p *CCOtelProcessor) parseLogRecord(lr *logsv1.LogRecord) *model.CCOtelEvent {
331+
func (p *CCOtelProcessor) parseLogRecord(lr *logsv1.LogRecord, resourceAttrs *model.CCOtelResourceAttributes) *model.CCOtelEvent {
272332
event := &model.CCOtelEvent{
273333
EventID: uuid.New().String(),
274334
Timestamp: int64(lr.GetTimeUnixNano() / 1e9), // Convert to seconds
275335
}
276336

277-
// Extract event type and other attributes
337+
// Apply resource attributes first
338+
applyResourceAttributesToEvent(event, resourceAttrs)
339+
340+
// Extract event type and other attributes from log record
278341
for _, attr := range lr.GetAttributes() {
279342
key := attr.GetKey()
280343
value := attr.GetValue()
281344

282345
switch key {
283346
case "event.name":
284347
event.EventType = mapEventName(value.GetStringValue())
348+
case "event.timestamp":
349+
event.EventTimestamp = value.GetStringValue()
285350
case "model":
286351
event.Model = value.GetStringValue()
287352
case "cost_usd":
288-
event.CostUSD = value.GetDoubleValue()
353+
event.CostUSD = getFloatFromValue(value)
289354
case "duration_ms":
290-
event.DurationMs = int(value.GetIntValue())
355+
event.DurationMs = getIntFromValue(value)
291356
case "input_tokens":
292-
event.InputTokens = int(value.GetIntValue())
357+
event.InputTokens = getIntFromValue(value)
293358
case "output_tokens":
294-
event.OutputTokens = int(value.GetIntValue())
359+
event.OutputTokens = getIntFromValue(value)
295360
case "cache_read_tokens":
296-
event.CacheReadTokens = int(value.GetIntValue())
361+
event.CacheReadTokens = getIntFromValue(value)
297362
case "cache_creation_tokens":
298-
event.CacheCreationTokens = int(value.GetIntValue())
363+
event.CacheCreationTokens = getIntFromValue(value)
299364
case "tool_name":
300365
event.ToolName = value.GetStringValue()
301366
case "success":
@@ -307,7 +372,7 @@ func (p *CCOtelProcessor) parseLogRecord(lr *logsv1.LogRecord) *model.CCOtelEven
307372
case "error":
308373
event.Error = value.GetStringValue()
309374
case "prompt_length":
310-
event.PromptLength = int(value.GetIntValue())
375+
event.PromptLength = getIntFromValue(value)
311376
case "prompt":
312377
event.Prompt = value.GetStringValue()
313378
case "tool_parameters":
@@ -321,11 +386,26 @@ func (p *CCOtelProcessor) parseLogRecord(lr *logsv1.LogRecord) *model.CCOtelEven
321386
}
322387
}
323388
case "status_code":
324-
event.StatusCode = int(value.GetIntValue())
389+
event.StatusCode = getIntFromValue(value)
325390
case "attempt":
326-
event.Attempt = int(value.GetIntValue())
391+
event.Attempt = getIntFromValue(value)
327392
case "language":
328393
event.Language = value.GetStringValue()
394+
// Log record level attributes that override resource attrs
395+
case "user.id":
396+
event.UserID = value.GetStringValue()
397+
case "user.email":
398+
event.UserEmail = value.GetStringValue()
399+
case "session.id":
400+
event.SessionID = value.GetStringValue()
401+
case "app.version":
402+
event.AppVersion = value.GetStringValue()
403+
case "organization.id":
404+
event.OrganizationID = value.GetStringValue()
405+
case "user.account_uuid":
406+
event.UserAccountUUID = value.GetStringValue()
407+
case "terminal.type":
408+
event.TerminalType = value.GetStringValue()
329409
}
330410
}
331411

@@ -391,6 +471,36 @@ func getDataPointValue(dp *metricsv1.NumberDataPoint) float64 {
391471
}
392472
}
393473

474+
// getIntFromValue extracts an int from an OTEL value, handling both int and string formats
475+
func getIntFromValue(value *commonv1.AnyValue) int {
476+
// First try to get as int
477+
if intVal := value.GetIntValue(); intVal != 0 {
478+
return int(intVal)
479+
}
480+
// Try to parse from string (Claude Code sends some values as strings)
481+
if strVal := value.GetStringValue(); strVal != "" {
482+
if parsed, err := strconv.Atoi(strVal); err == nil {
483+
return parsed
484+
}
485+
}
486+
return 0
487+
}
488+
489+
// getFloatFromValue extracts a float64 from an OTEL value, handling both double and string formats
490+
func getFloatFromValue(value *commonv1.AnyValue) float64 {
491+
// First try to get as double
492+
if doubleVal := value.GetDoubleValue(); doubleVal != 0 {
493+
return doubleVal
494+
}
495+
// Try to parse from string (Claude Code sends some values as strings)
496+
if strVal := value.GetStringValue(); strVal != "" {
497+
if parsed, err := strconv.ParseFloat(strVal, 64); err == nil {
498+
return parsed
499+
}
500+
}
501+
return 0
502+
}
503+
394504
// applyMetricAttribute applies an attribute to a metric
395505
func applyMetricAttribute(metric *model.CCOtelMetric, attr *commonv1.KeyValue, metricType string) {
396506
key := attr.GetKey()
@@ -411,5 +521,27 @@ func applyMetricAttribute(metric *model.CCOtelMetric, attr *commonv1.KeyValue, m
411521
metric.Decision = value.GetStringValue()
412522
case "language":
413523
metric.Language = value.GetStringValue()
524+
// Resource attributes at data point level - apply them (override if already set from resource)
525+
case "session.id":
526+
metric.SessionID = value.GetStringValue()
527+
case "user.account_uuid":
528+
metric.UserAccountUUID = value.GetStringValue()
529+
case "organization.id":
530+
metric.OrganizationID = value.GetStringValue()
531+
case "terminal.type":
532+
metric.TerminalType = value.GetStringValue()
533+
case "app.version":
534+
metric.AppVersion = value.GetStringValue()
535+
case "os.type":
536+
metric.OSType = value.GetStringValue()
537+
case "os.version":
538+
metric.OSVersion = value.GetStringValue()
539+
case "host.arch":
540+
metric.HostArch = value.GetStringValue()
541+
// Additional identifiers at data point level
542+
case "user.id":
543+
metric.UserID = value.GetStringValue()
544+
case "user.email":
545+
metric.UserEmail = value.GetStringValue()
414546
}
415547
}

0 commit comments

Comments
 (0)