100 lines
5.0 KiB
Go
100 lines
5.0 KiB
Go
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()
|
|
}
|