package storage import ( "context" "fmt" "time" "git.tornberg.me/go-otel/internal/uaclass" ) // EndpointStats reads raw trace spans because the production entries table is // the source of truth and may not have the optional normalized_events table. // The query intentionally returns both span and trace counts: a reverse proxy // can legitimately produce nested spans for one request, as with nginx // ingress -> upstream instrumentation. func (s *ClickHouseStore) EndpointStats(ctx context.Context, host, service string, hours, limit int) ([]EndpointStat, error) { if hours <= 0 { hours = 72 } if limit <= 0 || limit > 500 { limit = 100 } where := "host != ''" rawWhere := "type = 'traces' AND timestamp >= ?" args := []interface{}{time.Now().UTC().Add(-time.Duration(hours) * time.Hour)} if host != "" { where += " AND host = ?" args = append(args, host) rawWhere += " AND positionCaseInsensitive(data, ?) > 0" args = append(args, host) } if service != "" { where += " AND service = ?" args = append(args, service) } query := fmt.Sprintf(` SELECT service, host, method, path, if(route_raw = '/' AND path != '/', '', route_raw) AS route, status_code, user_agent, origin_host, referer_host, fetch_site, fetch_mode, count() AS span_count, uniqExact(trace_id) AS trace_count, min(timestamp) AS first_seen, max(timestamp) AS last_seen FROM ( WITH JSONExtractArrayRaw(span, 'attributes') AS attrs SELECT timestamp, JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') = 'service.name', JSONExtractArrayRaw(JSONExtractRaw(resource_span, 'resource'), 'attributes')), 'value', 'stringValue') AS service, JSONExtractString(span, 'traceId') AS trace_id, upper(coalesce(nullIf(JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') IN ('http.request.method', 'http.method'), attrs), 'value', 'stringValue'), ''), '')) AS method, replaceRegexpOne(coalesce(nullIf(JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') = 'url.path', attrs), 'value', 'stringValue'), ''), JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') = 'http.target', attrs), 'value', 'stringValue')), '[?#].*$', '') AS path, JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') = 'http.route', attrs), 'value', 'stringValue') AS route_raw, lower(replaceRegexpOne(coalesce(nullIf(JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') IN ('http.host', 'server.address', 'net.host.name'), attrs), 'value', 'stringValue'), ''), ''), ':\\d+$', '')) AS host, toUInt16OrZero(coalesce(nullIf(JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') IN ('http.response.status_code', 'http.status_code'), attrs), 'value', 'intValue'), ''), JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') IN ('http.response.status_code', 'http.status_code'), attrs), 'value', 'stringValue'))) AS status_code, JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') IN ('http.user_agent', 'user_agent.original'), attrs), 'value', 'stringValue') AS user_agent, replaceRegexpOne(replaceRegexpOne(JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') IN ('http.request.header.origin', 'http.origin', 'origin'), attrs), 'value', 'stringValue'), '^https?://', ''), '/.*$', '') AS origin_host, replaceRegexpOne(replaceRegexpOne(JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') IN ('http.request.header.referer', 'http.referer'), attrs), 'value', 'stringValue'), '^https?://', ''), '/.*$', '') AS referer_host, JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') = 'http.request.header.sec_fetch_site', attrs), 'value', 'stringValue') AS fetch_site, JSONExtractString(arrayFirst(x -> JSONExtractString(x, 'key') = 'http.request.header.sec_fetch_mode', attrs), 'value', 'stringValue') AS fetch_mode FROM ( SELECT timestamp, span FROM %s.entries ARRAY JOIN JSONExtractArrayRaw(data, 'resourceSpans') AS resource_span ARRAY JOIN JSONExtractArrayRaw(resource_span, 'scopeSpans') AS scope_span ARRAY JOIN JSONExtractArrayRaw(scope_span, 'spans') AS span WHERE %s ) ) AS spans WHERE %s GROUP BY service, host, method, path, route_raw, status_code, user_agent, origin_host, referer_host, fetch_site, fetch_mode ORDER BY trace_count DESC, span_count DESC, last_seen DESC LIMIT ?`, s.db, rawWhere, where) args = append(args, limit) rows, err := s.conn.Query(ctx, query, args...) if err != nil { return nil, fmt.Errorf("clickhouse endpoint stats: %w", err) } defer rows.Close() var out []EndpointStat for rows.Next() { var stat EndpointStat if err := rows.Scan( &stat.Service, &stat.Host, &stat.Method, &stat.Path, &stat.Route, &stat.StatusCode, &stat.UserAgent, &stat.OriginHost, &stat.RefererHost, &stat.FetchSite, &stat.FetchMode, &stat.SpanCount, &stat.TraceCount, &stat.FirstSeen, &stat.LastSeen, ); err != nil { return nil, err } if stat.UserAgent != "" { stat.UAClass = uaclass.Of(stat.UserAgent) } if stat.Route == "/" && stat.Path != "/" { stat.Route = "" } out = append(out, stat) } return out, rows.Err() }