mirror of
https://github.com/SigNoz/signoz.git
synced 2026-10-30 13:00:42 +00:00
Compare commits
16 Commits
issue_579
...
ns/trace-s
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
fbe3b46a55 | ||
|
|
3211e880be | ||
|
|
0edfa93159 | ||
|
|
4a5311dc5c | ||
|
|
6159bc108e | ||
|
|
b2a0179931 | ||
|
|
3a97332520 | ||
|
|
06cab07a13 | ||
|
|
b347a0a74d | ||
|
|
7a7241fa30 | ||
|
|
90e13ea73c | ||
|
|
505ae859e5 | ||
|
|
3203809ca0 | ||
|
|
d4a9ea5c08 | ||
|
|
7aa8c7009c | ||
|
|
2f41a110d6 |
@@ -12131,6 +12131,8 @@ components:
|
||||
additionalProperties:
|
||||
type: string
|
||||
type: object
|
||||
scope:
|
||||
$ref: '#/components/schemas/SpantypesScope'
|
||||
spanId:
|
||||
type: string
|
||||
timestamp:
|
||||
@@ -12147,6 +12149,7 @@ components:
|
||||
- event
|
||||
- attributes
|
||||
- resource
|
||||
- scope
|
||||
type: object
|
||||
SpantypesGettableFlamegraphTrace:
|
||||
properties:
|
||||
@@ -12387,6 +12390,16 @@ components:
|
||||
nullable: true
|
||||
type: array
|
||||
type: object
|
||||
SpantypesScope:
|
||||
properties:
|
||||
attributes:
|
||||
additionalProperties: {}
|
||||
type: object
|
||||
name:
|
||||
type: string
|
||||
version:
|
||||
type: string
|
||||
type: object
|
||||
SpantypesSpanAggregation:
|
||||
properties:
|
||||
aggregation:
|
||||
@@ -12756,6 +12769,8 @@ components:
|
||||
type: object
|
||||
response_status_code:
|
||||
type: string
|
||||
scope:
|
||||
$ref: '#/components/schemas/SpantypesScope'
|
||||
span_id:
|
||||
type: string
|
||||
status_code:
|
||||
|
||||
@@ -17,7 +17,7 @@ Condition: `model`, `llm.input_messages` (OpenInference sets `llm.model_name` fr
|
||||
| `gen_ai.response.model`, last source for the request model | vLLM, which writes only the response model |
|
||||
| `ai.model.id`, `ai.model.provider`, `ai.response.model`, `ai.response.finishReason`, `ai.prompt.messages`, `ai.response.text`, `ai.response.toolCalls`, `ai.usage.promptTokens`, `ai.usage.completionTokens` | Vercel AI SDK v4, Mastra |
|
||||
| `ai.usage.inputTokens`, `ai.usage.outputTokens`, `ai.usage.cachedInputTokens`, `ai.usage.reasoningTokens` | Vercel AI SDK v5 |
|
||||
| `langfuse.observation.model.name`, `langfuse.observation.input`, `langfuse.observation.output`, `langfuse.observation.type`, `langfuse.session.id` | Langfuse SDK v3 (Python) and v4 (JS), LiteLLM `langfuse_otel` callback |
|
||||
| `langfuse.observation.model.name`, `langfuse.session.id` | Langfuse SDK v3 |
|
||||
| `session.id` | OpenInference, Langfuse, Claude Code |
|
||||
| `gen_ai.client.session_id` | opentelemetry-hooks runner (Cursor, Copilot, Claude Code hooks) |
|
||||
| `model`, `input_tokens`, `output_tokens`, `cache_read_tokens`, `cache_creation_tokens` | Claude Code `claude_code.llm_request` spans ([docs](https://code.claude.com/docs/en/monitoring-usage#span-attributes)) |
|
||||
|
||||
@@ -13264,6 +13264,23 @@ export type SpantypesFlamegraphSpanDTOAttributes = { [key: string]: unknown };
|
||||
|
||||
export type SpantypesFlamegraphSpanDTOResource = { [key: string]: string };
|
||||
|
||||
export type SpantypesScopeDTOAttributes = { [key: string]: unknown };
|
||||
|
||||
export interface SpantypesScopeDTO {
|
||||
/**
|
||||
* @type object
|
||||
*/
|
||||
attributes?: SpantypesScopeDTOAttributes;
|
||||
/**
|
||||
* @type string
|
||||
*/
|
||||
name?: string;
|
||||
/**
|
||||
* @type string
|
||||
*/
|
||||
version?: string;
|
||||
}
|
||||
|
||||
export interface SpantypesFlamegraphSpanDTO {
|
||||
/**
|
||||
* @type object
|
||||
@@ -13299,6 +13316,7 @@ export interface SpantypesFlamegraphSpanDTO {
|
||||
* @type object
|
||||
*/
|
||||
resource: SpantypesFlamegraphSpanDTOResource;
|
||||
scope: SpantypesScopeDTO;
|
||||
/**
|
||||
* @type string
|
||||
*/
|
||||
@@ -13844,6 +13862,7 @@ export interface SpantypesWaterfallSpanDTO {
|
||||
* @type string
|
||||
*/
|
||||
response_status_code?: string;
|
||||
scope?: SpantypesScopeDTO;
|
||||
/**
|
||||
* @type string
|
||||
*/
|
||||
|
||||
@@ -20,6 +20,7 @@ function makeSpan(
|
||||
event: [],
|
||||
resource: {},
|
||||
attributes: {},
|
||||
scope: {},
|
||||
...overrides,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -12,6 +12,7 @@ export const MOCK_SPAN: FlamegraphSpan = {
|
||||
event: [],
|
||||
resource: {},
|
||||
attributes: {},
|
||||
scope: {},
|
||||
};
|
||||
|
||||
/** Nested spans structure for findSpanById tests */
|
||||
|
||||
@@ -518,6 +518,7 @@ const flamegraphSpan = (span: BuiltSpan): SpantypesFlamegraphSpanDTO => ({
|
||||
level: span.level,
|
||||
attributes: span.template.attributes,
|
||||
resource: resourceOf(span),
|
||||
scope: {},
|
||||
event: [],
|
||||
});
|
||||
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
{
|
||||
"version": 5,
|
||||
"version": 4,
|
||||
"definition": {
|
||||
"name": "gen_ai.llm",
|
||||
"condition": {
|
||||
@@ -134,12 +134,6 @@
|
||||
"fieldContext": "attribute",
|
||||
"config": {
|
||||
"sources": [
|
||||
{
|
||||
"key": "langfuse.observation.type",
|
||||
"context": "attribute",
|
||||
"operation": "copy",
|
||||
"priority": 40
|
||||
},
|
||||
{
|
||||
"key": "openinference.span.kind",
|
||||
"context": "attribute",
|
||||
@@ -362,12 +356,6 @@
|
||||
"fieldContext": "attribute",
|
||||
"config": {
|
||||
"sources": [
|
||||
{
|
||||
"key": "langfuse.observation.input",
|
||||
"context": "attribute",
|
||||
"operation": "move",
|
||||
"priority": 40
|
||||
},
|
||||
{
|
||||
"key": "gen_ai.prompt",
|
||||
"context": "attribute",
|
||||
@@ -394,12 +382,6 @@
|
||||
"fieldContext": "attribute",
|
||||
"config": {
|
||||
"sources": [
|
||||
{
|
||||
"key": "langfuse.observation.output",
|
||||
"context": "attribute",
|
||||
"operation": "move",
|
||||
"priority": 40
|
||||
},
|
||||
{
|
||||
"key": "gen_ai.completion",
|
||||
"context": "attribute",
|
||||
|
||||
@@ -28,7 +28,7 @@ const colServiceName = `resource_string_service$$$$name` // $ gets escaped so $$
|
||||
// JSON holds the attributes.
|
||||
var storableSpanColumns = []string{
|
||||
"DISTINCT ON (span_id) timestamp", "duration_nano", "span_id", "has_error", "kind",
|
||||
colServiceName, "name", "attributes", "resources_string",
|
||||
colServiceName, "name", "attributes", "resources_string", "scope",
|
||||
"if(notEmpty(attributes), map(), attributes_string) AS attributes_string",
|
||||
"if(notEmpty(attributes), map(), attributes_number) AS attributes_number",
|
||||
"if(notEmpty(attributes), map(), attributes_bool) AS attributes_bool",
|
||||
@@ -382,6 +382,7 @@ func (s *traceStore) GetFlamegraphSpans(ctx context.Context, traceID string, sta
|
||||
"if(notEmpty(attributes), map(), any(attributes_number)) AS attributes_number",
|
||||
"if(notEmpty(attributes), map(), any(attributes_bool)) AS attributes_bool",
|
||||
"any(resources_string) AS resources_string",
|
||||
"any(scope) AS scope",
|
||||
)
|
||||
sb.From(fmt.Sprintf("%s.%s", spantypes.TraceDB, spantypes.TraceTable))
|
||||
conditions := []string{
|
||||
|
||||
@@ -154,7 +154,7 @@ func TestGetSpanCountByField(t *testing.T) {
|
||||
}
|
||||
|
||||
func TestGetTraceSpans(t *testing.T) {
|
||||
expectedSQL := "SELECT DISTINCT ON (span_id) timestamp, duration_nano, span_id, has_error, kind, resource_string_service$$name, name, attributes, resources_string, if(notEmpty(attributes), map(), attributes_string) AS attributes_string, if(notEmpty(attributes), map(), attributes_number) AS attributes_number, if(notEmpty(attributes), map(), attributes_bool) AS attributes_bool, events, status_message, status_code_string, kind_string, parent_span_id, flags, is_remote, trace_state, status_code, db_name, db_operation, http_method, http_url, http_host, external_http_method, external_http_url, response_status_code, links as references FROM signoz_traces.distributed_signoz_index_v3 WHERE trace_id = ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? ORDER BY timestamp ASC, name ASC"
|
||||
expectedSQL := "SELECT DISTINCT ON (span_id) timestamp, duration_nano, span_id, has_error, kind, resource_string_service$$name, name, attributes, resources_string, scope, if(notEmpty(attributes), map(), attributes_string) AS attributes_string, if(notEmpty(attributes), map(), attributes_number) AS attributes_number, if(notEmpty(attributes), map(), attributes_bool) AS attributes_bool, events, status_message, status_code_string, kind_string, parent_span_id, flags, is_remote, trace_state, status_code, db_name, db_operation, http_method, http_url, http_host, external_http_method, external_http_url, response_status_code, links as references FROM signoz_traces.distributed_signoz_index_v3 WHERE trace_id = ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? ORDER BY timestamp ASC, name ASC"
|
||||
|
||||
s := newTestStore(sqlmock.QueryMatcherEqual)
|
||||
s.Mock().ExpectSelect(expectedSQL).
|
||||
@@ -171,12 +171,12 @@ func TestGetFlamegraphSpans(t *testing.T) {
|
||||
}{
|
||||
{
|
||||
name: "NoSpanIDs_GeneratesBaseSQL",
|
||||
sql: "SELECT span_id, any(parent_span_id) AS parent_span_id, any(timestamp) AS timestamp, any(duration_nano) AS duration_nano, any(has_error) AS has_error, any(name) AS name, any(events) AS events, any(attributes) AS attributes, if(notEmpty(attributes), map(), any(attributes_string)) AS attributes_string, if(notEmpty(attributes), map(), any(attributes_number)) AS attributes_number, if(notEmpty(attributes), map(), any(attributes_bool)) AS attributes_bool, any(resources_string) AS resources_string FROM signoz_traces.distributed_signoz_index_v3 WHERE trace_id = ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? GROUP BY span_id ORDER BY timestamp ASC, name ASC",
|
||||
sql: "SELECT span_id, any(parent_span_id) AS parent_span_id, any(timestamp) AS timestamp, any(duration_nano) AS duration_nano, any(has_error) AS has_error, any(name) AS name, any(events) AS events, any(attributes) AS attributes, if(notEmpty(attributes), map(), any(attributes_string)) AS attributes_string, if(notEmpty(attributes), map(), any(attributes_number)) AS attributes_number, if(notEmpty(attributes), map(), any(attributes_bool)) AS attributes_bool, any(resources_string) AS resources_string, any(scope) AS scope FROM signoz_traces.distributed_signoz_index_v3 WHERE trace_id = ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? GROUP BY span_id ORDER BY timestamp ASC, name ASC",
|
||||
},
|
||||
{
|
||||
name: "WithSpanIDs_GeneratesInClauseSQL",
|
||||
spanIDs: []string{"span-1", "span-2"},
|
||||
sql: "SELECT span_id, any(parent_span_id) AS parent_span_id, any(timestamp) AS timestamp, any(duration_nano) AS duration_nano, any(has_error) AS has_error, any(name) AS name, any(events) AS events, any(attributes) AS attributes, if(notEmpty(attributes), map(), any(attributes_string)) AS attributes_string, if(notEmpty(attributes), map(), any(attributes_number)) AS attributes_number, if(notEmpty(attributes), map(), any(attributes_bool)) AS attributes_bool, any(resources_string) AS resources_string FROM signoz_traces.distributed_signoz_index_v3 WHERE trace_id = ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND span_id IN (?, ?) GROUP BY span_id ORDER BY timestamp ASC, name ASC",
|
||||
sql: "SELECT span_id, any(parent_span_id) AS parent_span_id, any(timestamp) AS timestamp, any(duration_nano) AS duration_nano, any(has_error) AS has_error, any(name) AS name, any(events) AS events, any(attributes) AS attributes, if(notEmpty(attributes), map(), any(attributes_string)) AS attributes_string, if(notEmpty(attributes), map(), any(attributes_number)) AS attributes_number, if(notEmpty(attributes), map(), any(attributes_bool)) AS attributes_bool, any(resources_string) AS resources_string, any(scope) AS scope FROM signoz_traces.distributed_signoz_index_v3 WHERE trace_id = ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? AND span_id IN (?, ?) GROUP BY span_id ORDER BY timestamp ASC, name ASC",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@@ -43,13 +43,15 @@ func stripKeyAlias(name string) string {
|
||||
// unwrapVariant decodes a scan envelope: the chcol.Variant the driver scans a Dynamic column (a JSON
|
||||
// path such as body_v2.level) into, and the JSONValue a whole JSON column scans into. The attributes
|
||||
// bag decodes flat so its dotted keys merge with the legacy attribute maps and a key stored at both a
|
||||
// scalar and an object path stays two keys; every other JSON column decodes nested.
|
||||
// scalar and an object path stays two keys; the scope decodes flat too, keeping its attributes dotted;
|
||||
// every other JSON column decodes nested.
|
||||
func unwrapVariant(name string, val any) any {
|
||||
switch v := val.(type) {
|
||||
case chcol.Variant:
|
||||
return v.Any()
|
||||
case telemetrystoretypes.JSONValue:
|
||||
if name == "attributes" {
|
||||
switch name {
|
||||
case "attributes", "scope":
|
||||
return v.Flatten()
|
||||
}
|
||||
return v.Nest()
|
||||
@@ -626,6 +628,9 @@ func mergeSpanAttributeColumns(data map[string]any) {
|
||||
if raw, ok := data["links"]; ok {
|
||||
data["links"] = spantypes.ParseLinks(raw)
|
||||
}
|
||||
if column, ok := data["scope"].(map[string]any); ok {
|
||||
data["scope"] = spantypes.NewScope(column)
|
||||
}
|
||||
}
|
||||
|
||||
// numericAsFloat converts numeric types to float64 efficiently.
|
||||
|
||||
@@ -95,6 +95,9 @@ func jsonColumn(paths map[string]any) telemetrystoretypes.JSONValue {
|
||||
func TestUnwrapVariant(t *testing.T) {
|
||||
jsonCol := jsonColumn(map[string]any{"level": "error", "attrs.code": int64(500)})
|
||||
nullCol := jsonColumn(map[string]any{"level": "error", "absent": chcol.Variant{}})
|
||||
scopeAttrs := chcol.NewJSON()
|
||||
scopeAttrs.SetValueAtPath("lib.lang", "go")
|
||||
scopeCol := jsonColumn(map[string]any{"name": "io.opentelemetry.http", "version": "1.2.0", "attributes": scopeAttrs})
|
||||
|
||||
testCases := []struct {
|
||||
name string
|
||||
@@ -108,6 +111,7 @@ func TestUnwrapVariant(t *testing.T) {
|
||||
{name: "JSONColumnNested", column: "body_v2", input: jsonCol, want: map[string]any{"level": "error", "attrs": map[string]any{"code": float64(500)}}},
|
||||
{name: "AttributesColumnFlat", column: "attributes", input: jsonCol, want: map[string]any{"level": "error", "attrs.code": float64(500)}},
|
||||
{name: "AttributesColumnDropsNullPaths", column: "attributes", input: nullCol, want: map[string]any{"level": "error"}},
|
||||
{name: "ScopeColumnFlat", column: "scope", input: scopeCol, want: map[string]any{"name": "io.opentelemetry.http", "version": "1.2.0", "attributes": map[string]any{"lib.lang": "go"}}},
|
||||
}
|
||||
for _, testCase := range testCases {
|
||||
t.Run(testCase.name, func(t *testing.T) {
|
||||
@@ -132,6 +136,32 @@ func TestMergeSpanAttributeColumns_EmptyEventsAndLinks(t *testing.T) {
|
||||
}
|
||||
}
|
||||
|
||||
func TestMergeSpanAttributeColumns_Scope(t *testing.T) {
|
||||
testCases := []struct {
|
||||
name string
|
||||
scope map[string]any
|
||||
want spantypes.Scope
|
||||
}{
|
||||
{
|
||||
name: "SetFields_Kept",
|
||||
scope: map[string]any{"name": "io.opentelemetry.http", "version": "1.2.0", "attributes": map[string]any{"lib.lang": "go"}},
|
||||
want: spantypes.Scope{"name": "io.opentelemetry.http", "version": "1.2.0", "attributes": map[string]any{"lib.lang": "go"}},
|
||||
},
|
||||
{
|
||||
name: "UnsetDeclaredPaths_Empty",
|
||||
scope: map[string]any{"name": "", "version": "", "attributes": map[string]any{}},
|
||||
want: spantypes.Scope{},
|
||||
},
|
||||
}
|
||||
for _, testCase := range testCases {
|
||||
t.Run(testCase.name, func(t *testing.T) {
|
||||
data := map[string]any{"scope": testCase.scope}
|
||||
mergeSpanAttributeColumns(data)
|
||||
assert.Equal(t, testCase.want, data["scope"])
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
// readAsRaw decodes the JSON column into a flat bag of dotted keys before this runs; mergeSpanAttributeColumns
|
||||
// copies that bag and overlays the legacy string/number/bool maps, which win on collision.
|
||||
func TestMergeSpanAttributeColumns_JSONColumn(t *testing.T) {
|
||||
|
||||
@@ -45,7 +45,7 @@ SELECT timestamp AS __SELECT_KEY_0_timestamp, trace_id AS __SELECT_KEY_1_trace_i
|
||||
http_url AS __SELECT_KEY_17_http_url, external_http_method AS __SELECT_KEY_18_external_http_method,
|
||||
http_method AS __SELECT_KEY_19_http_method, http_host AS __SELECT_KEY_20_http_host, db_name AS __SELECT_KEY_21_db_name,
|
||||
db_operation AS __SELECT_KEY_22_db_operation, has_error AS __SELECT_KEY_23_has_error, is_remote AS __SELECT_KEY_24_is_remote,
|
||||
attributes_string, attributes_number, attributes_bool, resources_string, attributes
|
||||
attributes_string, attributes_number, attributes_bool, resources_string, attributes, scope
|
||||
FROM signoz_traces.distributed_signoz_index_v3
|
||||
WHERE trace_id GLOBAL IN (SELECT trace_id FROM __trace_scope)
|
||||
AND (((mapContains(attributes_string, 'gen_ai.request.model')
|
||||
|
||||
@@ -676,7 +676,7 @@ SELECT timestamp AS __SELECT_KEY_0_timestamp, trace_id AS __SELECT_KEY_1_trace_i
|
||||
http_url AS __SELECT_KEY_17_http_url, external_http_method AS __SELECT_KEY_18_external_http_method,
|
||||
http_method AS __SELECT_KEY_19_http_method, http_host AS __SELECT_KEY_20_http_host, db_name AS __SELECT_KEY_21_db_name,
|
||||
db_operation AS __SELECT_KEY_22_db_operation, has_error AS __SELECT_KEY_23_has_error, is_remote AS __SELECT_KEY_24_is_remote,
|
||||
attributes_string, attributes_number, attributes_bool, resources_string, attributes
|
||||
attributes_string, attributes_number, attributes_bool, resources_string, attributes, scope
|
||||
FROM signoz_traces.distributed_signoz_index_v3
|
||||
WHERE (((mapContains(attributes_string, 'gen_ai.request.model')
|
||||
OR mapContains(attributes_string, 'gen_ai.tool.name')
|
||||
|
||||
@@ -581,7 +581,7 @@ func TestStatementBuilderListQuery(t *testing.T) {
|
||||
Limit: 10,
|
||||
},
|
||||
expected: qbtypes.Statement{
|
||||
Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint) SELECT timestamp AS `__SELECT_KEY_0_timestamp`, trace_id AS `__SELECT_KEY_1_trace_id`, span_id AS `__SELECT_KEY_2_span_id`, trace_state AS `__SELECT_KEY_3_trace_state`, parent_span_id AS `__SELECT_KEY_4_parent_span_id`, flags AS `__SELECT_KEY_5_flags`, name AS `__SELECT_KEY_6_name`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, duration_nano AS `__SELECT_KEY_9_duration_nano`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? ORDER BY multiIf(mapContains(attributes_string, 'user.id'), attributes_string['user.id'], NULL) desc LIMIT ?",
|
||||
Query: "WITH __resource_filter AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint) SELECT timestamp AS `__SELECT_KEY_0_timestamp`, trace_id AS `__SELECT_KEY_1_trace_id`, span_id AS `__SELECT_KEY_2_span_id`, trace_state AS `__SELECT_KEY_3_trace_state`, parent_span_id AS `__SELECT_KEY_4_parent_span_id`, flags AS `__SELECT_KEY_5_flags`, name AS `__SELECT_KEY_6_name`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, duration_nano AS `__SELECT_KEY_9_duration_nano`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes, scope FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? ORDER BY multiIf(mapContains(attributes_string, 'user.id'), attributes_string['user.id'], NULL) desc LIMIT ?",
|
||||
Args: []any{"redis-manual", "%service.name%", "%service.name\":\"redis-manual%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10},
|
||||
},
|
||||
expectedErr: nil,
|
||||
@@ -837,7 +837,7 @@ func TestStatementBuilderListQueryWithCorruptData(t *testing.T) {
|
||||
Limit: 10,
|
||||
},
|
||||
expected: qbtypes.Statement{
|
||||
Query: "SELECT timestamp AS `__SELECT_KEY_0_timestamp`, trace_id AS `__SELECT_KEY_1_trace_id`, span_id AS `__SELECT_KEY_2_span_id`, trace_state AS `__SELECT_KEY_3_trace_state`, parent_span_id AS `__SELECT_KEY_4_parent_span_id`, flags AS `__SELECT_KEY_5_flags`, name AS `__SELECT_KEY_6_name`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, duration_nano AS `__SELECT_KEY_9_duration_nano`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? LIMIT ?",
|
||||
Query: "SELECT timestamp AS `__SELECT_KEY_0_timestamp`, trace_id AS `__SELECT_KEY_1_trace_id`, span_id AS `__SELECT_KEY_2_span_id`, trace_state AS `__SELECT_KEY_3_trace_state`, parent_span_id AS `__SELECT_KEY_4_parent_span_id`, flags AS `__SELECT_KEY_5_flags`, name AS `__SELECT_KEY_6_name`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, duration_nano AS `__SELECT_KEY_9_duration_nano`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes, scope FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? LIMIT ?",
|
||||
Args: []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10},
|
||||
},
|
||||
expectedErr: nil,
|
||||
@@ -879,7 +879,7 @@ func TestStatementBuilderListQueryWithCorruptData(t *testing.T) {
|
||||
}},
|
||||
},
|
||||
expected: qbtypes.Statement{
|
||||
Query: "SELECT timestamp AS `__SELECT_KEY_0_timestamp`, trace_id AS `__SELECT_KEY_1_trace_id`, span_id AS `__SELECT_KEY_2_span_id`, trace_state AS `__SELECT_KEY_3_trace_state`, parent_span_id AS `__SELECT_KEY_4_parent_span_id`, flags AS `__SELECT_KEY_5_flags`, name AS `__SELECT_KEY_6_name`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, duration_nano AS `__SELECT_KEY_9_duration_nano`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? ORDER BY timestamp asc LIMIT ?",
|
||||
Query: "SELECT timestamp AS `__SELECT_KEY_0_timestamp`, trace_id AS `__SELECT_KEY_1_trace_id`, span_id AS `__SELECT_KEY_2_span_id`, trace_state AS `__SELECT_KEY_3_trace_state`, parent_span_id AS `__SELECT_KEY_4_parent_span_id`, flags AS `__SELECT_KEY_5_flags`, name AS `__SELECT_KEY_6_name`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, duration_nano AS `__SELECT_KEY_9_duration_nano`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes, scope FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? ORDER BY timestamp asc LIMIT ?",
|
||||
Args: []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10},
|
||||
},
|
||||
expectedErr: nil,
|
||||
@@ -906,7 +906,7 @@ func TestStatementBuilderListQueryWithCorruptData(t *testing.T) {
|
||||
Limit: 10,
|
||||
},
|
||||
expected: qbtypes.Statement{
|
||||
Query: "SELECT timestamp AS `__SELECT_KEY_0_timestamp`, trace_id AS `__SELECT_KEY_1_trace_id`, span_id AS `__SELECT_KEY_2_span_id`, trace_state AS `__SELECT_KEY_3_trace_state`, parent_span_id AS `__SELECT_KEY_4_parent_span_id`, flags AS `__SELECT_KEY_5_flags`, name AS `__SELECT_KEY_6_name`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, duration_nano AS `__SELECT_KEY_9_duration_nano`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes FROM signoz_traces.distributed_signoz_index_v3 WHERE (scope.version::String = ? AND scope.version::String <> '') AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? LIMIT ?",
|
||||
Query: "SELECT timestamp AS `__SELECT_KEY_0_timestamp`, trace_id AS `__SELECT_KEY_1_trace_id`, span_id AS `__SELECT_KEY_2_span_id`, trace_state AS `__SELECT_KEY_3_trace_state`, parent_span_id AS `__SELECT_KEY_4_parent_span_id`, flags AS `__SELECT_KEY_5_flags`, name AS `__SELECT_KEY_6_name`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, duration_nano AS `__SELECT_KEY_9_duration_nano`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes, scope FROM signoz_traces.distributed_signoz_index_v3 WHERE (scope.version::String = ? AND scope.version::String <> '') AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ? LIMIT ?",
|
||||
Args: []any{"1.0.0", "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10},
|
||||
},
|
||||
},
|
||||
|
||||
@@ -124,7 +124,7 @@ func TestTraceOperatorStatementBuilder(t *testing.T) {
|
||||
},
|
||||
},
|
||||
expected: qbtypes.Statement{
|
||||
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_INDIR_DESC_B AS (WITH RECURSIVE up AS (SELECT d.trace_id, d.span_id, d.parent_span_id, 0 AS depth FROM B AS d UNION ALL SELECT p.trace_id, p.span_id, p.parent_span_id, up.depth + 1 FROM all_spans AS p JOIN up ON p.trace_id = up.trace_id AND p.span_id = up.parent_span_id WHERE up.depth < 100) SELECT DISTINCT a.* FROM A AS a GLOBAL INNER JOIN (SELECT DISTINCT trace_id, span_id FROM up WHERE depth > 0 ) AS ancestors ON ancestors.trace_id = a.trace_id AND ancestors.span_id = a.span_id) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes FROM A_INDIR_DESC_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
|
||||
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_INDIR_DESC_B AS (WITH RECURSIVE up AS (SELECT d.trace_id, d.span_id, d.parent_span_id, 0 AS depth FROM B AS d UNION ALL SELECT p.trace_id, p.span_id, p.parent_span_id, up.depth + 1 FROM all_spans AS p JOIN up ON p.trace_id = up.trace_id AND p.span_id = up.parent_span_id WHERE up.depth < 100) SELECT DISTINCT a.* FROM A AS a GLOBAL INNER JOIN (SELECT DISTINCT trace_id, span_id FROM up WHERE depth > 0 ) AS ancestors ON ancestors.trace_id = a.trace_id AND ancestors.span_id = a.span_id) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes, scope FROM A_INDIR_DESC_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
|
||||
Args: []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "gateway", "%service.name%", "%service.name\":\"gateway%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "database", "%service.name%", "%service.name\":\"database%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 5},
|
||||
},
|
||||
expectedErr: nil,
|
||||
@@ -161,7 +161,7 @@ func TestTraceOperatorStatementBuilder(t *testing.T) {
|
||||
},
|
||||
},
|
||||
expected: qbtypes.Statement{
|
||||
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_AND_B AS (SELECT l.* FROM A AS l INNER JOIN B AS r ON l.trace_id = r.trace_id) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes FROM A_AND_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
|
||||
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_AND_B AS (SELECT l.* FROM A AS l INNER JOIN B AS r ON l.trace_id = r.trace_id) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes, scope FROM A_AND_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
|
||||
Args: []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "frontend", "%service.name%", "%service.name\":\"frontend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "backend", "%service.name%", "%service.name\":\"backend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 15},
|
||||
},
|
||||
expectedErr: nil,
|
||||
@@ -198,7 +198,7 @@ func TestTraceOperatorStatementBuilder(t *testing.T) {
|
||||
},
|
||||
},
|
||||
expected: qbtypes.Statement{
|
||||
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_OR_B AS (SELECT * FROM A UNION DISTINCT SELECT * FROM B) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes FROM A_OR_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
|
||||
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_OR_B AS (SELECT * FROM A UNION DISTINCT SELECT * FROM B) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes, scope FROM A_OR_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
|
||||
Args: []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "frontend", "%service.name%", "%service.name\":\"frontend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "backend", "%service.name%", "%service.name\":\"backend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 20},
|
||||
},
|
||||
expectedErr: nil,
|
||||
@@ -235,7 +235,7 @@ func TestTraceOperatorStatementBuilder(t *testing.T) {
|
||||
},
|
||||
},
|
||||
expected: qbtypes.Statement{
|
||||
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_not_B AS (SELECT l.* FROM A AS l WHERE l.trace_id GLOBAL NOT IN (SELECT DISTINCT trace_id FROM B)) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes FROM A_not_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
|
||||
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_not_B AS (SELECT l.* FROM A AS l WHERE l.trace_id GLOBAL NOT IN (SELECT DISTINCT trace_id FROM B)) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes, scope FROM A_not_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
|
||||
Args: []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "frontend", "%service.name%", "%service.name\":\"frontend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "backend", "%service.name%", "%service.name\":\"backend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10},
|
||||
},
|
||||
expectedErr: nil,
|
||||
@@ -400,7 +400,7 @@ func TestTraceOperatorStatementBuilder(t *testing.T) {
|
||||
},
|
||||
},
|
||||
expected: qbtypes.Statement{
|
||||
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_DIR_DESC_B AS (SELECT p.* FROM A AS p INNER JOIN B AS c ON p.trace_id = c.trace_id AND p.span_id = c.parent_span_id), __resource_filter_C AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), C AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_C) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_D AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), D AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_D) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), C_DIR_DESC_D AS (SELECT p.* FROM C AS p INNER JOIN D AS c ON p.trace_id = c.trace_id AND p.span_id = c.parent_span_id), A_DIR_DESC_B_AND_C_DIR_DESC_D AS (SELECT l.* FROM A_DIR_DESC_B AS l INNER JOIN C_DIR_DESC_D AS r ON l.trace_id = r.trace_id) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes FROM A_DIR_DESC_B_AND_C_DIR_DESC_D ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
|
||||
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_DIR_DESC_B AS (SELECT p.* FROM A AS p INNER JOIN B AS c ON p.trace_id = c.trace_id AND p.span_id = c.parent_span_id), __resource_filter_C AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), C AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_C) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_D AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), D AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_D) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), C_DIR_DESC_D AS (SELECT p.* FROM C AS p INNER JOIN D AS c ON p.trace_id = c.trace_id AND p.span_id = c.parent_span_id), A_DIR_DESC_B_AND_C_DIR_DESC_D AS (SELECT l.* FROM A_DIR_DESC_B AS l INNER JOIN C_DIR_DESC_D AS r ON l.trace_id = r.trace_id) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes, scope FROM A_DIR_DESC_B_AND_C_DIR_DESC_D ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
|
||||
Args: []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "frontend", "%service.name%", "%service.name\":\"frontend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "backend", "%service.name%", "%service.name\":\"backend%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "auth", "%service.name%", "%service.name\":\"auth%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "database", "%service.name%", "%service.name\":\"database%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 5},
|
||||
},
|
||||
expectedErr: nil,
|
||||
@@ -434,7 +434,7 @@ func TestTraceOperatorStatementBuilder(t *testing.T) {
|
||||
},
|
||||
},
|
||||
expected: qbtypes.Statement{
|
||||
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_INDIR_DESC_B AS (WITH RECURSIVE up AS (SELECT d.trace_id, d.span_id, d.parent_span_id, 0 AS depth FROM B AS d UNION ALL SELECT p.trace_id, p.span_id, p.parent_span_id, up.depth + 1 FROM all_spans AS p JOIN up ON p.trace_id = up.trace_id AND p.span_id = up.parent_span_id WHERE up.depth < 100) SELECT DISTINCT a.* FROM A AS a GLOBAL INNER JOIN (SELECT DISTINCT trace_id, span_id FROM up WHERE depth > 0 ) AS ancestors ON ancestors.trace_id = a.trace_id AND ancestors.span_id = a.span_id), __return_from_B AS (SELECT * FROM B WHERE trace_id IN (SELECT DISTINCT trace_id FROM A_INDIR_DESC_B)) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes FROM __return_from_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
|
||||
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_INDIR_DESC_B AS (WITH RECURSIVE up AS (SELECT d.trace_id, d.span_id, d.parent_span_id, 0 AS depth FROM B AS d UNION ALL SELECT p.trace_id, p.span_id, p.parent_span_id, up.depth + 1 FROM all_spans AS p JOIN up ON p.trace_id = up.trace_id AND p.span_id = up.parent_span_id WHERE up.depth < 100) SELECT DISTINCT a.* FROM A AS a GLOBAL INNER JOIN (SELECT DISTINCT trace_id, span_id FROM up WHERE depth > 0 ) AS ancestors ON ancestors.trace_id = a.trace_id AND ancestors.span_id = a.span_id), __return_from_B AS (SELECT * FROM B WHERE trace_id IN (SELECT DISTINCT trace_id FROM A_INDIR_DESC_B)) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes, scope FROM __return_from_B ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
|
||||
Args: []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "gateway", "%service.name%", "%service.name\":\"gateway%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "database", "%service.name%", "%service.name\":\"database%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10},
|
||||
},
|
||||
expectedErr: nil,
|
||||
@@ -476,7 +476,7 @@ func TestTraceOperatorStatementBuilder(t *testing.T) {
|
||||
},
|
||||
},
|
||||
expected: qbtypes.Statement{
|
||||
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_INDIR_DESC_B AS (WITH RECURSIVE up AS (SELECT d.trace_id, d.span_id, d.parent_span_id, 0 AS depth FROM B AS d UNION ALL SELECT p.trace_id, p.span_id, p.parent_span_id, up.depth + 1 FROM all_spans AS p JOIN up ON p.trace_id = up.trace_id AND p.span_id = up.parent_span_id WHERE up.depth < 100) SELECT DISTINCT a.* FROM A AS a GLOBAL INNER JOIN (SELECT DISTINCT trace_id, span_id FROM up WHERE depth > 0 ) AS ancestors ON ancestors.trace_id = a.trace_id AND ancestors.span_id = a.span_id), __resource_filter_C AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), C AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_C) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_INDIR_DESC_B_AND_C AS (SELECT l.* FROM A_INDIR_DESC_B AS l INNER JOIN C AS r ON l.trace_id = r.trace_id), __return_from_C AS (SELECT * FROM C WHERE trace_id IN (SELECT DISTINCT trace_id FROM A_INDIR_DESC_B_AND_C)) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes FROM __return_from_C ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
|
||||
Query: "WITH toDateTime64(1747947419000000000, 9) AS t_from, toDateTime64(1747983448000000000, 9) AS t_to, 1747945619 AS bucket_from, 1747983448 AS bucket_to, all_spans AS (SELECT *, resource_string_service$$name AS `service.name` FROM signoz_traces.distributed_signoz_index_v3 WHERE timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_A AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), A AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_A) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), __resource_filter_B AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), B AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_B) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_INDIR_DESC_B AS (WITH RECURSIVE up AS (SELECT d.trace_id, d.span_id, d.parent_span_id, 0 AS depth FROM B AS d UNION ALL SELECT p.trace_id, p.span_id, p.parent_span_id, up.depth + 1 FROM all_spans AS p JOIN up ON p.trace_id = up.trace_id AND p.span_id = up.parent_span_id WHERE up.depth < 100) SELECT DISTINCT a.* FROM A AS a GLOBAL INNER JOIN (SELECT DISTINCT trace_id, span_id FROM up WHERE depth > 0 ) AS ancestors ON ancestors.trace_id = a.trace_id AND ancestors.span_id = a.span_id), __resource_filter_C AS (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (simpleJSONExtractString(labels, 'service.name') = ? AND labels LIKE ? AND labels LIKE ?) AND seen_at_ts_bucket_start >= ? AND seen_at_ts_bucket_start <= ? GROUP BY fingerprint), C AS (SELECT * FROM signoz_traces.distributed_signoz_index_v3 WHERE resource_fingerprint GLOBAL IN (SELECT fingerprint FROM __resource_filter_C) AND timestamp >= ? AND timestamp < ? AND ts_bucket_start >= ? AND ts_bucket_start <= ?), A_INDIR_DESC_B_AND_C AS (SELECT l.* FROM A_INDIR_DESC_B AS l INNER JOIN C AS r ON l.trace_id = r.trace_id), __return_from_C AS (SELECT * FROM C WHERE trace_id IN (SELECT DISTINCT trace_id FROM A_INDIR_DESC_B_AND_C)) SELECT timestamp, trace_id, span_id, name, duration_nano, parent_span_id, trace_state AS `__SELECT_KEY_3_trace_state`, flags AS `__SELECT_KEY_5_flags`, kind AS `__SELECT_KEY_7_kind`, kind_string AS `__SELECT_KEY_8_kind_string`, status_code AS `__SELECT_KEY_10_status_code`, status_message AS `__SELECT_KEY_11_status_message`, status_code_string AS `__SELECT_KEY_12_status_code_string`, events AS `__SELECT_KEY_13_events`, links AS `__SELECT_KEY_14_links`, response_status_code AS `__SELECT_KEY_15_response_status_code`, external_http_url AS `__SELECT_KEY_16_external_http_url`, http_url AS `__SELECT_KEY_17_http_url`, external_http_method AS `__SELECT_KEY_18_external_http_method`, http_method AS `__SELECT_KEY_19_http_method`, http_host AS `__SELECT_KEY_20_http_host`, db_name AS `__SELECT_KEY_21_db_name`, db_operation AS `__SELECT_KEY_22_db_operation`, has_error AS `__SELECT_KEY_23_has_error`, is_remote AS `__SELECT_KEY_24_is_remote`, attributes_string, attributes_number, attributes_bool, resources_string, attributes, scope FROM __return_from_C ORDER BY timestamp DESC LIMIT ? SETTINGS distributed_product_mode='allow', max_memory_usage=10000000000",
|
||||
Args: []any{"1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "gateway", "%service.name%", "%service.name\":\"gateway%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "database", "%service.name%", "%service.name\":\"database%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), "auth", "%service.name%", "%service.name\":\"auth%", uint64(1747945619), uint64(1747983448), "1747947419000000000", "1747983448000000000", uint64(1747945619), uint64(1747983448), 10},
|
||||
},
|
||||
expectedErr: nil,
|
||||
|
||||
@@ -46,6 +46,7 @@ const (
|
||||
SpanAttributesColumn = "attributes"
|
||||
SpanAttributesPromotedColumn = "attributes_promoted"
|
||||
SpanResourcesStringColumn = "resources_string"
|
||||
SpanScopeColumn = "scope"
|
||||
)
|
||||
|
||||
var (
|
||||
@@ -455,6 +456,7 @@ var (
|
||||
SpanAttributesBoolColumn,
|
||||
SpanResourcesStringColumn,
|
||||
SpanAttributesColumn,
|
||||
SpanScopeColumn,
|
||||
}
|
||||
|
||||
DefaultFields = map[string]telemetrytypes.TelemetryFieldKey{
|
||||
|
||||
@@ -121,13 +121,10 @@ func (f *formatting) messages() {
|
||||
|
||||
// OpenRouter's answer object or an Anthropic response beside the OpenAI chat request it answers is one
|
||||
// format, not a mix: an Anthropic request without system and without Anthropic-only blocks reads as an
|
||||
// OpenAI chat request. A side kept as generic already warned, and a side read from OpenInference's
|
||||
// buckets beside the raw payload of the other is the documented fallback.
|
||||
// OpenAI chat request.
|
||||
func (f *formatting) warnMixedFormats(inputLabel, outputLabel string) {
|
||||
switch {
|
||||
case inputLabel == "" || outputLabel == "" || outputLabel == inputLabel:
|
||||
case inputLabel == FormatterGeneric || outputLabel == FormatterGeneric:
|
||||
case inputLabel == FormatterOpenInference || outputLabel == FormatterOpenInference:
|
||||
case inputLabel == FormatterOpenAIChat && (outputLabel == FormatterOpenRouter || outputLabel == FormatterAnthropic):
|
||||
default:
|
||||
f.warn("output formatted as %s, input as %s", outputLabel, inputLabel)
|
||||
@@ -186,16 +183,11 @@ func (f *formatting) side(value any, role Role) (OutputMessages, string) {
|
||||
return f.semconv(v), FormatterSemconv
|
||||
case isAnthropicMessages(v):
|
||||
return f.anthropicMessages(v), FormatterAnthropic
|
||||
case isResponsesItems(v):
|
||||
return foldAssistant(f.responsesItems(v)), FormatterOpenAIResponses
|
||||
case first["role"] != nil:
|
||||
return f.openAIChatMessages(v), FormatterOpenAIChat
|
||||
}
|
||||
}
|
||||
case map[string]any:
|
||||
if inner, ok := superJSON(v); ok {
|
||||
return f.side(decode(inner), role)
|
||||
}
|
||||
switch {
|
||||
case isLangChainRun(v):
|
||||
return f.langChainRun(v), FormatterLangChain
|
||||
@@ -222,9 +214,6 @@ func (f *formatting) side(value any, role Role) (OutputMessages, string) {
|
||||
if messages, ok := v["messages"].([]any); ok {
|
||||
return f.openAIChatMessages(messages), FormatterOpenAIChat
|
||||
}
|
||||
if (v["role"] != nil && hasKeys(v, "content")) || v["tool_calls"] != nil {
|
||||
return OutputMessages{f.openAIChatSingle(v)}, FormatterOpenAIChat
|
||||
}
|
||||
case string:
|
||||
if v == "" {
|
||||
return OutputMessages{}, FormatterText
|
||||
@@ -320,23 +309,6 @@ func foldAssistant(msgs OutputMessages) OutputMessages {
|
||||
return out
|
||||
}
|
||||
|
||||
// superJSON unwraps {json, meta?}, the envelope superjson, the transformer tRPC apps serialise with,
|
||||
// puts around a value; a Langfuse generation records it around the request when the app passes the
|
||||
// serialised form:
|
||||
// https://github.com/flightcontrolhq/superjson/blob/aaa65e36e8e31da740265a56b27c965ca1c7b754/README.md
|
||||
func superJSON(m map[string]any) (any, bool) {
|
||||
inner, ok := m["json"]
|
||||
if !ok {
|
||||
return nil, false
|
||||
}
|
||||
for key := range m {
|
||||
if key != "json" && key != "meta" {
|
||||
return nil, false
|
||||
}
|
||||
}
|
||||
return inner, true
|
||||
}
|
||||
|
||||
// decode parses a JSON-encoded attribute value, twice when an SDK encoded the string again.
|
||||
// Anything that is not JSON is returned as is.
|
||||
func decode(value any) any {
|
||||
|
||||
@@ -4,9 +4,8 @@ import (
|
||||
"strings"
|
||||
)
|
||||
|
||||
// Format: OpenAI Chat Completions, a request {messages}, a response {choices}, a bare message list, or a
|
||||
// single message.
|
||||
// Written by: bifrost, openrouter requests, the raw OpenAI calls OpenInference records, the Langfuse SDK.
|
||||
// Format: OpenAI Chat Completions, a request {messages}, a response {choices}, or a bare message list.
|
||||
// Written by: bifrost, openrouter requests, the raw OpenAI calls OpenInference records.
|
||||
// Handled: string and block content, tool_calls nested or flattened, tool messages, refusal,
|
||||
// reasoning_content, finish reasons.
|
||||
// Not yet: streaming delta chunks, legacy completions choices[].text, audio output.
|
||||
@@ -89,19 +88,6 @@ func (f *formatting) openAIChatMessage(m map[string]any) OutputMessage {
|
||||
return msg
|
||||
}
|
||||
|
||||
// openAIChatSingle converts a message recorded outside a list: the first choice's message the Langfuse
|
||||
// OpenAI wrapper writes as the output, or the additional_kwargs of a LangChain AIMessage, {tool_calls}
|
||||
// without a role, which the Langfuse LangChain handler writes for a tool-calling turn:
|
||||
// https://github.com/langfuse/langfuse-python/blob/18baf80450dd5f9c1f85916eb2d1b3c72d2052e1/langfuse/openai.py#L446
|
||||
// https://github.com/langfuse/langfuse-python/blob/18baf80450dd5f9c1f85916eb2d1b3c72d2052e1/langfuse/langchain/CallbackHandler.py#L1558
|
||||
func (f *formatting) openAIChatSingle(m map[string]any) OutputMessage {
|
||||
msg := f.openAIChatMessage(m)
|
||||
if msg.Role.IsZero() {
|
||||
msg.Role = RoleAssistant
|
||||
}
|
||||
return msg
|
||||
}
|
||||
|
||||
// toolCallPart reads {id, function: {name, arguments}}; a gateway may flatten it to {id, name, arguments|args}.
|
||||
func toolCallPart(c map[string]any) Part {
|
||||
call := ToolCallRequestPart{Type: PartTypeToolCall, ID: optionalString(c["id"]), Name: stringOf(c["name"])}
|
||||
|
||||
@@ -205,30 +205,6 @@ func TestFormat_OpenAIChat(t *testing.T) {
|
||||
]`,
|
||||
wantOutput: `[]`,
|
||||
},
|
||||
{
|
||||
// Langfuse SDK generation: the history as a list, the first choice's message as the output
|
||||
name: "SingleMessageOutput_AssistantMessage",
|
||||
input: `[{"role": "system", "content": "You are a support agent on a phone call."}, {"role": "user", "content": "Hello?"}]`,
|
||||
output: `{"role": "assistant", "content": "Hi, this is a courtesy call about your order. Do you have a minute?"}`,
|
||||
wantInput: `[{"role": "system", "parts": [{"type": "text", "content": "You are a support agent on a phone call."}]}, {"role": "user", "parts": [{"type": "text", "content": "Hello?"}]}]`,
|
||||
wantOutput: `[{"role": "assistant", "parts": [{"type": "text", "content": "Hi, this is a courtesy call about your order. Do you have a minute?"}]}]`,
|
||||
},
|
||||
{
|
||||
// a structured output that happens to carry a role key is not a message
|
||||
name: "ObjectWithRoleKeyButNoContent_KeptAsGeneric",
|
||||
output: `{"role": "admin", "permissions": ["read", "write"]}`,
|
||||
warnings: []string{"message format not recognised, kept as generic"},
|
||||
wantInput: `[]`,
|
||||
wantOutput: `[{"role": "", "parts": [{"type": "generic", "content": "{\"permissions\":[\"read\",\"write\"],\"role\":\"admin\"}"}]}]`,
|
||||
},
|
||||
{
|
||||
// Langfuse LangChain handler: the additional_kwargs of the AIMessage, without a role, as the output
|
||||
// of a tool-calling turn
|
||||
name: "RolelessToolCallsOutput_AssistantToolCallPart",
|
||||
output: `{"tool_calls": [{"id": "call_1", "type": "function", "function": {"name": "get_weather", "arguments": "{\"city\": \"Oslo\"}"}}], "refusal": null}`,
|
||||
wantInput: `[]`,
|
||||
wantOutput: `[{"role": "assistant", "parts": [{"type": "tool_call", "name": "get_weather", "id": "call_1", "arguments": {"city": "Oslo"}}]}]`,
|
||||
},
|
||||
{
|
||||
name: "ResponseWithScalarChoice_KeptAsGeneric",
|
||||
output: `{"choices":["oops",{"finish_reason":"stop","index":1,"message":{"role":"assistant","content":"hi"}}]}`,
|
||||
|
||||
@@ -2,15 +2,13 @@ package aiobservabilitytypes
|
||||
|
||||
import "strings"
|
||||
|
||||
// Format: OpenAI Responses, a request {input, instructions}, a response {output, status} or a bare list
|
||||
// of items:
|
||||
// Format: OpenAI Responses, a request {input, instructions} or a response {output, status} of items:
|
||||
// https://github.com/openai/openai-openapi/blob/aeb6a041dc56c011ca9f12127ec898f911100aa0/openapi.yaml#L47597
|
||||
// https://github.com/openai/openai-openapi/blob/aeb6a041dc56c011ca9f12127ec898f911100aa0/openapi.yaml#L69691
|
||||
// https://github.com/openai/openai-openapi/blob/aeb6a041dc56c011ca9f12127ec898f911100aa0/openapi.yaml#L53834
|
||||
// https://github.com/openai/openai-openapi/blob/aeb6a041dc56c011ca9f12127ec898f911100aa0/openapi.yaml#L57596
|
||||
// Written by: gateways recording raw Responses bodies, the raw calls OpenInference records for the
|
||||
// Responses API and the OpenAI Agents SDK when its flattened messages are absent, and LiteLLM's
|
||||
// langfuse_otel callback, the function_call items of a tool-calling turn as a bare list.
|
||||
// Written by: gateways recording raw Responses bodies, and the raw calls OpenInference records for the
|
||||
// Responses API and the OpenAI Agents SDK when its flattened messages are absent.
|
||||
// Handled: message items with input_text, output_text, refusal, input_image and input_file content;
|
||||
// function_call and custom_tool_call with their outputs; reasoning summaries; compaction; every other
|
||||
// *_call item as a server tool call and *_output item as its response; the status and incomplete_details
|
||||
@@ -34,19 +32,6 @@ func isResponsesRequest(m map[string]any) bool {
|
||||
return isItem
|
||||
}
|
||||
|
||||
// isResponsesItems looks for an item only the Responses API has, a tool call or its output, a reasoning
|
||||
// or compaction item, all without a role; a list of messages has none.
|
||||
func isResponsesItems(list []any) bool {
|
||||
for _, item := range list {
|
||||
m, _ := item.(map[string]any)
|
||||
typ := stringOf(m["type"])
|
||||
if m["role"] == nil && (typ == "reasoning" || typ == "compaction" || typ == "mcp_approval_response" || strings.HasSuffix(typ, "_call") || strings.HasSuffix(typ, "_output")) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func isResponsesResponse(m map[string]any) bool {
|
||||
if _, ok := m["output"].([]any); !ok {
|
||||
return false
|
||||
|
||||
@@ -73,23 +73,6 @@ func TestFormat_OpenAIResponses(t *testing.T) {
|
||||
wantInput: `[{"role": "system", "parts": [{"type": "text", "content": "Be brief."}]}, {"role": "user", "parts": [{"type": "text", "content": "Say hi"}]}]`,
|
||||
wantOutput: `[]`,
|
||||
},
|
||||
{
|
||||
// LiteLLM langfuse_otel: the function_call items of a tool-calling turn as a bare list
|
||||
name: "BareFunctionCallItems_OneAssistantMessage",
|
||||
output: `[{"id": "chatcmpl-1", "name": "set_stage", "call_id": "call_7", "type": "function_call", "arguments": {"fields": {"stage": "consent"}}}]`,
|
||||
wantInput: `[]`,
|
||||
wantOutput: `[{"role": "assistant", "parts": [{"type": "tool_call", "name": "set_stage", "id": "call_7", "arguments": {"fields": {"stage": "consent"}}}]}]`,
|
||||
},
|
||||
{
|
||||
name: "BareItemsWithMessage_ToolRound",
|
||||
input: `[{"type": "message", "role": "user", "content": [{"type": "input_text", "text": "Weather in Oslo?"}]}, {"type": "function_call", "call_id": "call_9", "name": "get_weather", "arguments": "{\"city\":\"Oslo\"}"}, {"type": "function_call_output", "call_id": "call_9", "output": "4C, cloudy"}]`,
|
||||
wantInput: `[
|
||||
{"role": "user", "parts": [{"type": "text", "content": "Weather in Oslo?"}]},
|
||||
{"role": "assistant", "parts": [{"type": "tool_call", "name": "get_weather", "id": "call_9", "arguments": {"city": "Oslo"}}]},
|
||||
{"role": "tool", "parts": [{"type": "tool_call_response", "id": "call_9", "response": "4C, cloudy"}]}
|
||||
]`,
|
||||
wantOutput: `[]`,
|
||||
},
|
||||
{
|
||||
// openinference capture, CreateEmbeddings span: an embeddings request also has input and model
|
||||
name: "EmbeddingsRequest_NotResponses_KeptAsGeneric",
|
||||
|
||||
@@ -4,9 +4,8 @@ import "github.com/SigNoz/signoz/pkg/valuer"
|
||||
|
||||
// Format: the flattened llm.input_messages.N.message.* and llm.output_messages.N.message.* keys, the
|
||||
// provider-neutral copy of the conversation the OpenInference instrumentations write on LLM spans. A side
|
||||
// without them, or whose buckets hold no message, falls back to the raw provider payload the mapper moved
|
||||
// onto gen_ai.*.messages; pipecat streams the reply into output.value alone, LiteLLM's OpenInference mode
|
||||
// leaves the output bucket empty on a tool-calling turn.
|
||||
// without them falls back to the raw provider payload the mapper moved onto gen_ai.*.messages; pipecat
|
||||
// streams the reply into output.value alone.
|
||||
// https://github.com/Arize-ai/openinference/blob/12f2cc035edfb00b60b656d977f07abfb7a69900/spec/semantic_conventions.md#L77
|
||||
// https://github.com/Arize-ai/openinference/blob/12f2cc035edfb00b60b656d977f07abfb7a69900/spec/tool_calling.md#L37
|
||||
// https://github.com/Arize-ai/openinference/blob/12f2cc035edfb00b60b656d977f07abfb7a69900/spec/multimodal_attributes.md#L35
|
||||
@@ -29,14 +28,15 @@ func (f *formatting) openInference() bool {
|
||||
return false
|
||||
}
|
||||
inputLabel, outputLabel := FormatterOpenInference, FormatterOpenInference
|
||||
if msgs := openInferenceMessages(input); len(msgs) > 0 {
|
||||
f.out.Input = inputMessages(msgs)
|
||||
if len(input) > 0 {
|
||||
f.out.Input = inputMessages(openInferenceMessages(input))
|
||||
} else if raw := attribute(f.attributes, GenAIInputMessages); raw != nil {
|
||||
var msgs OutputMessages
|
||||
msgs, inputLabel = f.side(decode(raw), RoleUser)
|
||||
f.out.Input = inputMessages(msgs)
|
||||
}
|
||||
if msgs := openInferenceMessages(output); len(msgs) > 0 {
|
||||
f.out.Output = msgs
|
||||
if len(output) > 0 {
|
||||
f.out.Output = openInferenceMessages(output)
|
||||
if fr := finishReason(stringOf(attribute(f.attributes, openInferenceFinishReason))); fr != nil {
|
||||
for i := range f.out.Output {
|
||||
f.out.Output[i].FinishReason = fr
|
||||
|
||||
@@ -160,30 +160,16 @@ func TestFormat_OpenInference(t *testing.T) {
|
||||
// which the mapper moves onto gen_ai.output.messages
|
||||
// https://github.com/Arize-ai/openinference/blob/12f2cc035edfb00b60b656d977f07abfb7a69900/python/instrumentation/openinference-instrumentation-pipecat/src/openinference/instrumentation/pipecat/_attributes.py#L430
|
||||
// https://github.com/Arize-ai/openinference/blob/12f2cc035edfb00b60b656d977f07abfb7a69900/python/instrumentation/openinference-instrumentation-pipecat/src/openinference/instrumentation/pipecat/_observer.py#L628
|
||||
name: "OutputKeysAbsent_RawOutputFormatted",
|
||||
name: "OutputKeysAbsent_RawOutputFormatted_Warns",
|
||||
attributes: map[string]any{
|
||||
"llm.input_messages.0.message.role": "user",
|
||||
"llm.input_messages.0.message.content": "hi",
|
||||
GenAIOutputMessages: "Hello there.",
|
||||
},
|
||||
warnings: []string{"bare assistant text, role assumed"},
|
||||
warnings: []string{"bare assistant text, role assumed", "output formatted as text, input as openinference"},
|
||||
wantInput: `[{"role": "user", "parts": [{"type": "text", "content": "hi"}]}]`,
|
||||
wantOutput: `[{"role": "assistant", "parts": [{"type": "text", "content": "Hello there."}]}]`,
|
||||
},
|
||||
{
|
||||
// LiteLLM's OpenInference mode leaves the output bucket empty on a tool-calling turn; the calls are
|
||||
// the Langfuse output the mapper moved onto gen_ai.output.messages
|
||||
name: "EmptyOutputBucket_RawOutputFormatted",
|
||||
attributes: map[string]any{
|
||||
"llm.input_messages.0.message.role": "user",
|
||||
"llm.input_messages.0.message.content": "Yes, go ahead.",
|
||||
"llm.output_messages.0.message.role": "assistant",
|
||||
"llm.output_messages.0.message.content": "",
|
||||
GenAIOutputMessages: `[{"id": "chatcmpl-2", "name": "set_stage", "call_id": "call_8", "type": "function_call", "arguments": {"fields": {"stage": "email"}}}]`,
|
||||
},
|
||||
wantInput: `[{"role": "user", "parts": [{"type": "text", "content": "Yes, go ahead."}]}]`,
|
||||
wantOutput: `[{"role": "assistant", "parts": [{"type": "tool_call", "name": "set_stage", "id": "call_8", "arguments": {"fields": {"stage": "email"}}}]}]`,
|
||||
},
|
||||
{
|
||||
// message.role left unset by some instrumentors: tool_call_id, tool_calls or neither decide it,
|
||||
// as the collector-contrib normalizer does
|
||||
|
||||
@@ -122,31 +122,6 @@ So, the ball costs 5 cents.`,
|
||||
wantInput: `[]`,
|
||||
wantOutput: `[]`,
|
||||
},
|
||||
{
|
||||
// Langfuse SDK generation: the app passes the request through superjson, the output is the parsed
|
||||
// structured output
|
||||
name: "SuperJSONEnvelope_Unwrapped_StructuredOutputGeneric",
|
||||
input: `{"json": {"model": "gpt-4o-mini", "messages": [{"role": "system", "content": "Extract the main article of the page as JSON."}, {"role": "user", "content": "# Council approves new bike lanes"}], "temperature": 0}}`,
|
||||
output: `{"hasArticle": true, "article": {"title": "Council approves new bike lanes"}}`,
|
||||
warnings: []string{"message format not recognised, kept as generic"},
|
||||
wantInput: `[{"role": "system", "parts": [{"type": "text", "content": "Extract the main article of the page as JSON."}]}, {"role": "user", "parts": [{"type": "text", "content": "# Council approves new bike lanes"}]}]`,
|
||||
wantOutput: `[{"role": "", "parts": [{"type": "generic", "content": "{\"article\":{\"title\":\"Council approves new bike lanes\"},\"hasArticle\":true}"}]}]`,
|
||||
},
|
||||
{
|
||||
name: "SuperJSONEnvelopeWithMeta_Unwrapped",
|
||||
input: `{"json": {"messages": [{"role": "user", "content": "What time is it in Oslo?"}]}, "meta": {"values": {"messages.0.at": ["Date"]}}}`,
|
||||
output: `{"json": {"choices": [{"finish_reason": "stop", "message": {"role": "assistant", "content": "It is 14:05 in Oslo."}}]}, "meta": {"values": {}}}`,
|
||||
wantInput: `[{"role": "user", "parts": [{"type": "text", "content": "What time is it in Oslo?"}]}]`,
|
||||
wantOutput: `[{"role": "assistant", "parts": [{"type": "text", "content": "It is 14:05 in Oslo."}], "finish_reason": "stop"}]`,
|
||||
},
|
||||
{
|
||||
// a json key beside others is a payload of its own, not the envelope
|
||||
name: "JSONKeyBesideOthers_NotUnwrapped_Generic",
|
||||
input: `{"json": {"messages": []}, "schema": "v1"}`,
|
||||
warnings: []string{"message format not recognised, kept as generic"},
|
||||
wantInput: `[{"role": "", "parts": [{"type": "generic", "content": "{\"json\":{\"messages\":[]},\"schema\":\"v1\"}"}]}]`,
|
||||
wantOutput: `[]`,
|
||||
},
|
||||
{
|
||||
name: "NoMessageAttributes_EmptyLists",
|
||||
attributes: map[string]any{GenAIRequestModel: "gpt-4o", GenAIToolName: "get_weather"},
|
||||
|
||||
@@ -17,6 +17,7 @@ type FlamegraphSpan struct {
|
||||
Events []Event `json:"event" required:"true" nullable:"false"`
|
||||
Attributes map[string]any `json:"attributes" required:"true" nullable:"false"`
|
||||
Resource map[string]string `json:"resource" required:"true" nullable:"false"`
|
||||
Scope Scope `json:"scope" required:"true" nullable:"false"`
|
||||
Children []*FlamegraphSpan `json:"-"` // internal tree use only
|
||||
}
|
||||
|
||||
@@ -66,11 +67,13 @@ func NewFlamegraphSpanFromStorable(s *StorableSpan, level int64, selectFields []
|
||||
Events: s.UnmarshalledEvents(),
|
||||
Attributes: make(map[string]any),
|
||||
Resource: make(map[string]string),
|
||||
Scope: Scope{},
|
||||
}
|
||||
if len(selectFields) == 0 {
|
||||
return span
|
||||
}
|
||||
var attributes map[string]any
|
||||
var scopeFields []string
|
||||
for _, field := range selectFields {
|
||||
switch field.FieldContext {
|
||||
case telemetrytypes.FieldContextResource:
|
||||
@@ -84,8 +87,13 @@ func NewFlamegraphSpanFromStorable(s *StorableSpan, level int64, selectFields []
|
||||
if v, ok := attributes[field.Name]; ok {
|
||||
span.Attributes[field.Name] = v
|
||||
}
|
||||
case telemetrytypes.FieldContextScope:
|
||||
scopeFields = append(scopeFields, field.Name)
|
||||
}
|
||||
}
|
||||
if len(scopeFields) > 0 {
|
||||
span.Scope = NewSelectedScope(s.Scope.Flatten(), scopeFields)
|
||||
}
|
||||
return span
|
||||
}
|
||||
|
||||
@@ -96,6 +104,7 @@ func NewMissingParentFlamegraphSpan(node *FlamegraphSpan) *FlamegraphSpan {
|
||||
Timestamp: node.Timestamp,
|
||||
DurationNano: node.DurationNano,
|
||||
Events: []Event{},
|
||||
Scope: Scope{},
|
||||
Children: []*FlamegraphSpan{node},
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,9 +1,11 @@
|
||||
package spantypes
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"testing"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
|
||||
)
|
||||
@@ -15,6 +17,7 @@ func TestNewFlamegraphSpanFromStorable_SelectFields(t *testing.T) {
|
||||
selectFields []telemetrytypes.TelemetryFieldKey
|
||||
wantAttrs map[string]any
|
||||
wantResource map[string]string
|
||||
wantScope Scope
|
||||
}{
|
||||
{
|
||||
name: "JSONColumn_ResolvesDottedPaths",
|
||||
@@ -26,6 +29,7 @@ func TestNewFlamegraphSpanFromStorable_SelectFields(t *testing.T) {
|
||||
},
|
||||
wantAttrs: map[string]any{"http.route": "/orders", "http.retry.count": float64(2)},
|
||||
wantResource: map[string]string{},
|
||||
wantScope: Scope{},
|
||||
},
|
||||
{
|
||||
name: "LegacyMaps_Resolve",
|
||||
@@ -37,6 +41,7 @@ func TestNewFlamegraphSpanFromStorable_SelectFields(t *testing.T) {
|
||||
},
|
||||
wantAttrs: map[string]any{"http.route": "/cart", "cache.hit": true},
|
||||
wantResource: map[string]string{},
|
||||
wantScope: Scope{},
|
||||
},
|
||||
{
|
||||
name: "ResourceFields_EmptyValueSkipped",
|
||||
@@ -47,6 +52,30 @@ func TestNewFlamegraphSpanFromStorable_SelectFields(t *testing.T) {
|
||||
},
|
||||
wantAttrs: map[string]any{},
|
||||
wantResource: map[string]string{"service.name": "checkout"},
|
||||
wantScope: Scope{},
|
||||
},
|
||||
{
|
||||
name: "ScopeFields_DeclaredPathsAndAttributes",
|
||||
span: StorableSpan{Scope: scopeColumn("io.opentelemetry.http", "1.2.0", map[string]any{"lib.lang": "go", "team": "payments"})},
|
||||
selectFields: []telemetrytypes.TelemetryFieldKey{
|
||||
{Name: "scope.name", FieldContext: telemetrytypes.FieldContextScope},
|
||||
{Name: "lib.lang", FieldContext: telemetrytypes.FieldContextScope},
|
||||
{Name: "missing", FieldContext: telemetrytypes.FieldContextScope},
|
||||
},
|
||||
wantAttrs: map[string]any{},
|
||||
wantResource: map[string]string{},
|
||||
wantScope: Scope{"name": "io.opentelemetry.http", "attributes": map[string]any{"lib.lang": "go"}},
|
||||
},
|
||||
{
|
||||
name: "ScopeFields_UnsetSelection_Empty",
|
||||
span: StorableSpan{Scope: scopeColumn("io.opentelemetry.grpc", "", map[string]any{"rpc.system": "grpc"})},
|
||||
selectFields: []telemetrytypes.TelemetryFieldKey{
|
||||
{Name: "scope.version", FieldContext: telemetrytypes.FieldContextScope},
|
||||
{Name: "absent", FieldContext: telemetrytypes.FieldContextScope},
|
||||
},
|
||||
wantAttrs: map[string]any{},
|
||||
wantResource: map[string]string{},
|
||||
wantScope: Scope{},
|
||||
},
|
||||
}
|
||||
for _, testCase := range testCases {
|
||||
@@ -54,6 +83,15 @@ func TestNewFlamegraphSpanFromStorable_SelectFields(t *testing.T) {
|
||||
span := NewFlamegraphSpanFromStorable(&testCase.span, 0, testCase.selectFields)
|
||||
assert.Equal(t, testCase.wantAttrs, span.Attributes)
|
||||
assert.Equal(t, testCase.wantResource, span.Resource)
|
||||
assert.Equal(t, testCase.wantScope, span.Scope)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func TestNewMissingParentFlamegraphSpan_EmptyScope(t *testing.T) {
|
||||
span := NewMissingParentFlamegraphSpan(&FlamegraphSpan{SpanID: "child", ParentSpanID: "parent"})
|
||||
|
||||
raw, err := json.Marshal(span)
|
||||
require.NoError(t, err)
|
||||
assert.Contains(t, string(raw), `"scope":{}`)
|
||||
}
|
||||
|
||||
72
pkg/types/spantypes/scope.go
Normal file
72
pkg/types/spantypes/scope.go
Normal file
@@ -0,0 +1,72 @@
|
||||
package spantypes
|
||||
|
||||
import (
|
||||
"strings"
|
||||
|
||||
"github.com/swaggest/jsonschema-go"
|
||||
)
|
||||
|
||||
const (
|
||||
scopeName = "name"
|
||||
scopeVersion = "version"
|
||||
scopeAttributes = "attributes"
|
||||
)
|
||||
|
||||
var _ jsonschema.Preparer = Scope{}
|
||||
|
||||
// Scope is a span's instrumentation scope holding only the fields that are set: name, version and
|
||||
// attributes, a flat bag of dotted keys like the span attributes.
|
||||
type Scope map[string]any
|
||||
|
||||
// NewScope builds a Scope from the flattened scope column, where an unset declared path reads back
|
||||
// as its zero value (name and version as empty strings).
|
||||
func NewScope(column map[string]any) Scope {
|
||||
scope := Scope{}
|
||||
if name, _ := column[scopeName].(string); name != "" {
|
||||
scope[scopeName] = name
|
||||
}
|
||||
if version, _ := column[scopeVersion].(string); version != "" {
|
||||
scope[scopeVersion] = version
|
||||
}
|
||||
if attributes, _ := column[scopeAttributes].(map[string]any); len(attributes) > 0 {
|
||||
scope[scopeAttributes] = attributes
|
||||
}
|
||||
return scope
|
||||
}
|
||||
|
||||
// NewSelectedScope builds a Scope with only the selected fields from the flattened scope column:
|
||||
// scope.name and scope.version select the declared paths, any other name a scope attribute.
|
||||
func NewSelectedScope(column map[string]any, names []string) Scope {
|
||||
attributes, _ := column[scopeAttributes].(map[string]any)
|
||||
selected := map[string]any{}
|
||||
selectedAttributes := map[string]any{}
|
||||
for _, name := range names {
|
||||
switch name {
|
||||
case "scope.name":
|
||||
selected[scopeName] = column[scopeName]
|
||||
case "scope.version":
|
||||
selected[scopeVersion] = column[scopeVersion]
|
||||
default:
|
||||
name = strings.TrimPrefix(name, "attribute.")
|
||||
if v, ok := attributes[name]; ok {
|
||||
selectedAttributes[name] = v
|
||||
}
|
||||
}
|
||||
}
|
||||
selected[scopeAttributes] = selectedAttributes
|
||||
return NewScope(selected)
|
||||
}
|
||||
|
||||
// PrepareJSONSchema types the map's fields. Scope is a map rather than a struct so unset fields are
|
||||
// left out instead of sent empty; omitempty would do that but decodes back with nil fields.
|
||||
func (Scope) PrepareJSONSchema(schema *jsonschema.Schema) error {
|
||||
attributes := jsonschema.Schema{}
|
||||
attributes.WithType(jsonschema.Object.Type())
|
||||
attributes.WithAdditionalProperties(jsonschema.SchemaOrBool{TypeObject: &jsonschema.Schema{}})
|
||||
|
||||
schema.AdditionalProperties = nil
|
||||
schema.WithPropertiesItem(scopeName, jsonschema.String.ToSchemaOrBool())
|
||||
schema.WithPropertiesItem(scopeVersion, jsonschema.String.ToSchemaOrBool())
|
||||
schema.WithPropertiesItem(scopeAttributes, attributes.ToSchemaOrBool())
|
||||
return nil
|
||||
}
|
||||
58
pkg/types/spantypes/scope_test.go
Normal file
58
pkg/types/spantypes/scope_test.go
Normal file
@@ -0,0 +1,58 @@
|
||||
package spantypes
|
||||
|
||||
import (
|
||||
"testing"
|
||||
|
||||
"github.com/ClickHouse/clickhouse-go/v2/lib/chcol"
|
||||
"github.com/stretchr/testify/assert"
|
||||
|
||||
"github.com/SigNoz/signoz/pkg/types/telemetrystoretypes"
|
||||
)
|
||||
|
||||
// scopeColumn mirrors the driver's scan of the scope column: declared String paths as plain strings and the
|
||||
// declared attributes path as a nested *chcol.JSON.
|
||||
func scopeColumn(name, version string, attributes map[string]any) telemetrystoretypes.JSONValue {
|
||||
j := chcol.NewJSON()
|
||||
j.SetValueAtPath("name", name)
|
||||
j.SetValueAtPath("version", version)
|
||||
attrs := chcol.NewJSON()
|
||||
for path, value := range attributes {
|
||||
attrs.SetValueAtPath(path, value)
|
||||
}
|
||||
j.SetValueAtPath("attributes", attrs)
|
||||
return telemetrystoretypes.NewJSONValue(*j)
|
||||
}
|
||||
|
||||
func TestNewScope(t *testing.T) {
|
||||
testCases := []struct {
|
||||
name string
|
||||
column telemetrystoretypes.JSONValue
|
||||
want Scope
|
||||
}{
|
||||
{
|
||||
name: "AttributesStayDotted",
|
||||
column: scopeColumn("io.opentelemetry.http", "1.2.0", map[string]any{"lib.lang": "go", "lib.lang.version": "1.22", "retry": int64(3)}),
|
||||
want: Scope{"name": "io.opentelemetry.http", "version": "1.2.0", "attributes": map[string]any{"lib.lang": "go", "lib.lang.version": "1.22", "retry": float64(3)}},
|
||||
},
|
||||
{
|
||||
name: "DropsNullAttributePaths_UnsetVersion",
|
||||
column: scopeColumn("checkout", "", map[string]any{"team": "payments", "absent": chcol.Variant{}}),
|
||||
want: Scope{"name": "checkout", "attributes": map[string]any{"team": "payments"}},
|
||||
},
|
||||
{
|
||||
name: "UnsetDeclaredPaths_Empty",
|
||||
column: scopeColumn("", "", nil),
|
||||
want: Scope{},
|
||||
},
|
||||
{
|
||||
name: "EmptyColumn",
|
||||
column: telemetrystoretypes.JSONValue{},
|
||||
want: Scope{},
|
||||
},
|
||||
}
|
||||
for _, testCase := range testCases {
|
||||
t.Run(testCase.name, func(t *testing.T) {
|
||||
assert.Equal(t, testCase.want, NewScope(testCase.column.Flatten()))
|
||||
})
|
||||
}
|
||||
}
|
||||
@@ -64,6 +64,7 @@ type WaterfallSpan struct {
|
||||
Name string `json:"name"`
|
||||
ParentSpanID string `json:"parent_span_id"`
|
||||
Resource map[string]string `json:"resource"`
|
||||
Scope Scope `json:"scope"`
|
||||
SpanID string `json:"span_id"`
|
||||
TimeUnix uint64 `json:"time_unix"`
|
||||
TraceID string `json:"trace_id"`
|
||||
@@ -113,6 +114,7 @@ type StorableSpan struct {
|
||||
AttributesBool map[string]bool `ch:"attributes_bool"`
|
||||
AttributesJSON telemetrystoretypes.JSONValue `ch:"attributes"`
|
||||
ResourcesString map[string]string `ch:"resources_string"`
|
||||
Scope telemetrystoretypes.JSONValue `ch:"scope"`
|
||||
Events []string `ch:"events"`
|
||||
StatusMessage string `ch:"status_message"`
|
||||
StatusCodeString string `ch:"status_code_string"`
|
||||
@@ -153,6 +155,7 @@ func (item *MinimalSpan) ToWaterfallSpan(traceID string) *WaterfallSpan {
|
||||
HasError: item.HasError,
|
||||
ServiceName: item.ServiceName,
|
||||
Resource: map[string]string{"service.name": item.ServiceName},
|
||||
Scope: Scope{},
|
||||
Children: make([]*WaterfallSpan, 0),
|
||||
Attributes: make(map[string]any),
|
||||
Events: make([]Event, 0),
|
||||
@@ -182,6 +185,7 @@ func NewMissingWaterfallSpan(spanID, traceID string, timeUnixNano, durationNano
|
||||
Children: make([]*WaterfallSpan, 0),
|
||||
Attributes: make(map[string]any),
|
||||
Resource: make(map[string]string),
|
||||
Scope: Scope{},
|
||||
}
|
||||
}
|
||||
|
||||
@@ -336,6 +340,7 @@ func (item *StorableSpan) ToWaterfallSpan(traceID string) *WaterfallSpan {
|
||||
Name: item.Name,
|
||||
ParentSpanID: item.ParentSpanID,
|
||||
Resource: resources,
|
||||
Scope: NewScope(item.Scope.Flatten()),
|
||||
ResponseStatusCode: item.ResponseStatusCode,
|
||||
SpanID: item.SpanID,
|
||||
StatusCode: item.StatusCode,
|
||||
|
||||
@@ -39,7 +39,7 @@ func (v JSONValue) Flatten() map[string]any {
|
||||
if isNull(value) {
|
||||
continue
|
||||
}
|
||||
flat[path] = value
|
||||
flat[path] = flatObject(value)
|
||||
}
|
||||
return unwrap(flat)
|
||||
}
|
||||
@@ -57,6 +57,15 @@ func (v JSONValue) MarshalJSON() ([]byte, error) {
|
||||
return json.Marshal(v.Flatten())
|
||||
}
|
||||
|
||||
// flatObject renders a declared JSON sub-path, which the driver scans as a nested chcol.JSON (such as
|
||||
// scope.attributes), as a flat attribute bag.
|
||||
func flatObject(value any) any {
|
||||
if object, ok := value.(*chcol.JSON); ok && object != nil {
|
||||
return JSONValue{column: *object}
|
||||
}
|
||||
return value
|
||||
}
|
||||
|
||||
func isNull(value any) bool {
|
||||
if value == nil {
|
||||
return true
|
||||
|
||||
1
tests/fixtures/traces.py
vendored
1
tests/fixtures/traces.py
vendored
@@ -50,6 +50,7 @@ ALL_SELECT_FIELDS = [
|
||||
# all contextual columns (merged in response layer)
|
||||
"attributes",
|
||||
"resource",
|
||||
"scope",
|
||||
]
|
||||
|
||||
|
||||
|
||||
@@ -545,7 +545,7 @@ def test_traces_list_response_shape_values(
|
||||
now = datetime.now(tz=UTC).replace(microsecond=0)
|
||||
trace_id = TraceIdGenerator.trace_id()
|
||||
|
||||
def span(marker: str, *, kind: TracesKind, status_code: TracesStatusCode = TracesStatusCode.STATUS_CODE_UNSET, status_message: str = "", flags: int = 0, attributes: dict[str, Any]) -> Traces:
|
||||
def span(marker: str, *, kind: TracesKind, status_code: TracesStatusCode = TracesStatusCode.STATUS_CODE_UNSET, status_message: str = "", flags: int = 0, attributes: dict[str, Any], scope: dict[str, Any] | None = None) -> Traces:
|
||||
return Traces(
|
||||
timestamp=now - timedelta(seconds=1),
|
||||
duration=timedelta(seconds=1),
|
||||
@@ -558,10 +558,11 @@ def test_traces_list_response_shape_values(
|
||||
flags=flags,
|
||||
resources={"service.name": "shape-service", **extra_resources},
|
||||
attributes={"marker": marker, **attributes, **extra_attrs},
|
||||
scope=scope or {},
|
||||
)
|
||||
|
||||
spans = [
|
||||
span("error", kind=TracesKind.SPAN_KIND_SERVER, status_code=TracesStatusCode.STATUS_CODE_ERROR, status_message="boom", attributes={"str_a": "x", "num_a": 7.5, "bool_a": True}),
|
||||
span("error", kind=TracesKind.SPAN_KIND_SERVER, status_code=TracesStatusCode.STATUS_CODE_ERROR, status_message="boom", attributes={"str_a": "x", "num_a": 7.5, "bool_a": True}, scope={"name": "io.signoz.shape", "version": "0.1.0", "attributes": {"lib.lang": "go"}}),
|
||||
span("client", kind=TracesKind.SPAN_KIND_CLIENT, attributes={"http.url": "https://api.example.com/v1/orders", "http.request.method": "GET"}),
|
||||
span("db", kind=TracesKind.SPAN_KIND_CLIENT, attributes={"db.name": "orders_db", "db.operation": "SELECT"}),
|
||||
span("host", kind=TracesKind.SPAN_KIND_PRODUCER, attributes={"server.address": "payments.internal:8080"}),
|
||||
@@ -622,6 +623,7 @@ def test_traces_list_response_shape_values(
|
||||
"is_remote": span_obj.is_remote,
|
||||
"attributes": {**span_obj.attribute_string, **span_obj.attributes_number, **span_obj.attributes_bool},
|
||||
"resource": span_obj.resources_string,
|
||||
"scope": {k: v for k, v in span_obj.scope_json.items() if v},
|
||||
}
|
||||
# `==` treats True/1 and 7.5/7.5 as equal regardless of type, so pin the
|
||||
# string/number/bool type preservation for the error span explicitly.
|
||||
@@ -1154,6 +1156,7 @@ def test_traces_list_span_scope(
|
||||
"is_remote": x[3].is_remote,
|
||||
"attributes": {**x[3].attribute_string, **x[3].attributes_number, **x[3].attributes_bool},
|
||||
"resource": x[3].resources_string,
|
||||
"scope": {k: v for k, v in x[3].scope_json.items() if v},
|
||||
},
|
||||
id="order_timestamp_full_shape",
|
||||
),
|
||||
@@ -1191,6 +1194,7 @@ def test_traces_list_span_scope(
|
||||
"is_remote": x[3].is_remote,
|
||||
"attributes": {**x[3].attribute_string, **x[3].attributes_number, **x[3].attributes_bool},
|
||||
"resource": x[3].resources_string,
|
||||
"scope": {k: v for k, v in x[3].scope_json.items() if v},
|
||||
},
|
||||
id="order_attribute_timestamp_full_shape",
|
||||
),
|
||||
|
||||
@@ -21,12 +21,13 @@ def test_waterfall_and_flamegraph_read_each_attribute_home(
|
||||
trace_id = TraceIdGenerator.trace_id()
|
||||
legacy_id, json_id, dual_id = (TraceIdGenerator.span_id() for _ in range(3))
|
||||
resources = {"service.name": "tracedetail-attribute-homes"}
|
||||
scope = {"name": "io.opentelemetry.http", "version": "1.2.0", "attributes": {"lib.lang": "go"}}
|
||||
|
||||
insert_traces(
|
||||
[
|
||||
Traces(timestamp=now - timedelta(seconds=10), duration=timedelta(seconds=9), trace_id=trace_id, span_id=legacy_id, name="GET /legacy", kind=TracesKind.SPAN_KIND_SERVER, resources=resources, attributes={"http.route": "/legacy", "http.retry.count": 1, "cache.hit": True}, attribute_write_mode="legacy_only"),
|
||||
Traces(timestamp=now - timedelta(seconds=8), trace_id=trace_id, span_id=json_id, parent_span_id=legacy_id, name="db call", resources=resources, attributes={"http.route": "/json", "db.function": "node_refresh", "db.function.arg_count": 2}, attribute_write_mode="json_only"),
|
||||
Traces(timestamp=now - timedelta(seconds=6), trace_id=trace_id, span_id=dual_id, parent_span_id=legacy_id, name="GET /dual", resources=resources, attributes={"http.route": "/dual"}),
|
||||
Traces(timestamp=now - timedelta(seconds=6), trace_id=trace_id, span_id=dual_id, parent_span_id=legacy_id, name="GET /dual", resources=resources, attributes={"http.route": "/dual"}, scope=scope),
|
||||
]
|
||||
)
|
||||
|
||||
@@ -40,11 +41,14 @@ def test_waterfall_and_flamegraph_read_each_attribute_home(
|
||||
timeout=10,
|
||||
)
|
||||
assert response.status_code == HTTPStatus.OK, response.text
|
||||
waterfall = {span["span_id"]: span["attributes"] for span in response.json()["data"]["spans"]}
|
||||
spans = {span["span_id"]: span for span in response.json()["data"]["spans"]}
|
||||
waterfall = {span_id: span["attributes"] for span_id, span in spans.items()}
|
||||
assert waterfall[legacy_id] == {"http.route": "/legacy", "http.retry.count": 1, "cache.hit": True}
|
||||
# a key stored at both a scalar and an object path stays two flat keys
|
||||
assert waterfall[json_id] == {"http.route": "/json", "db.function": "node_refresh", "db.function.arg_count": 2}
|
||||
assert waterfall[dual_id] == {"http.route": "/dual"}
|
||||
assert spans[dual_id]["scope"] == scope
|
||||
assert spans[legacy_id]["scope"] == {}
|
||||
|
||||
response = requests.post(
|
||||
signoz.self.host_configs["8080"].get(f"/api/v3/traces/{trace_id}/flamegraph"),
|
||||
@@ -54,6 +58,8 @@ def test_waterfall_and_flamegraph_read_each_attribute_home(
|
||||
{"name": "http.route", "fieldContext": "attribute"},
|
||||
{"name": "db.function.arg_count", "fieldContext": "attribute"},
|
||||
{"name": "service.name", "fieldContext": "resource"},
|
||||
{"name": "scope.name", "fieldContext": "scope"},
|
||||
{"name": "lib.lang", "fieldContext": "scope"},
|
||||
],
|
||||
},
|
||||
headers=headers,
|
||||
@@ -65,3 +71,5 @@ def test_waterfall_and_flamegraph_read_each_attribute_home(
|
||||
assert flamegraph[json_id]["attributes"] == {"http.route": "/json", "db.function.arg_count": 2}
|
||||
assert flamegraph[dual_id]["attributes"] == {"http.route": "/dual"}
|
||||
assert all(span["resource"] == resources for span in flamegraph.values())
|
||||
assert flamegraph[dual_id]["scope"] == {"name": "io.opentelemetry.http", "attributes": {"lib.lang": "go"}}
|
||||
assert flamegraph[json_id]["scope"] == {}
|
||||
|
||||
Reference in New Issue
Block a user