diff --git a/pkg/telemetrymeter/stmt_builder_test.go b/pkg/telemetrymeter/stmt_builder_test.go index aa4d833157d9..3121cf15dc03 100644 --- a/pkg/telemetrymeter/stmt_builder_test.go +++ b/pkg/telemetrymeter/stmt_builder_test.go @@ -22,77 +22,77 @@ func TestStatementBuilder(t *testing.T) { expected qbtypes.Statement expectedErr error }{ - // { - // name: "test_cumulative_rate_sum", - // requestType: qbtypes.RequestTypeTimeSeries, - // query: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{ - // Signal: telemetrytypes.SignalMetrics, - // StepInterval: qbtypes.Step{Duration: 30 * time.Second}, - // Aggregations: []qbtypes.MetricAggregation{ - // { - // MetricName: "signoz_calls_total", - // Type: metrictypes.SumType, - // Temporality: metrictypes.Cumulative, - // TimeAggregation: metrictypes.TimeAggregationRate, - // SpaceAggregation: metrictypes.SpaceAggregationSum, - // }, - // }, - // Filter: &qbtypes.Filter{ - // Expression: "service.name = 'cartservice'", - // }, - // Limit: 10, - // GroupBy: []qbtypes.GroupByKey{ - // { - // TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{ - // Name: "service.name", - // }, - // }, - // }, - // }, - // expected: qbtypes.Statement{ - // Query: "WITH __temporal_aggregation_cte AS (SELECT ts, `service.name`, If((per_series_value - lagInFrame(per_series_value, 1, 0) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1, toDateTime(fromUnixTimestamp64Milli(1747947360000))) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1, 0) OVER rate_window) / (ts - lagInFrame(ts, 1, toDateTime(fromUnixTimestamp64Milli(1747947360000))) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `service.name`, max(value) AS per_series_value FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'service.name') AS `service.name` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? AND JSONExtractString(labels, 'service.name') = ? GROUP BY fingerprint, `service.name`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts, `service.name` ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, `service.name`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts, `service.name`) SELECT * FROM __spatial_aggregation_cte", - // Args: []any{"signoz_calls_total", uint64(1747936800000), uint64(1747983420000), "cumulative", false, "cartservice", "signoz_calls_total", uint64(1747947360000), uint64(1747983420000), 0}, - // }, - // expectedErr: nil, - // }, - // { - // name: "test_delta_rate_sum", - // requestType: qbtypes.RequestTypeTimeSeries, - // query: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{ - // Signal: telemetrytypes.SignalMetrics, - // StepInterval: qbtypes.Step{Duration: 30 * time.Second}, - // Aggregations: []qbtypes.MetricAggregation{ - // { - // MetricName: "signoz_calls_total", - // Type: metrictypes.SumType, - // Temporality: metrictypes.Delta, - // TimeAggregation: metrictypes.TimeAggregationRate, - // SpaceAggregation: metrictypes.SpaceAggregationSum, - // }, - // }, - // Filter: &qbtypes.Filter{ - // Expression: "service.name = 'cartservice'", - // }, - // Limit: 10, - // GroupBy: []qbtypes.GroupByKey{ - // { - // TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{ - // Name: "service.name", - // }, - // }, - // }, - // }, - // expected: qbtypes.Statement{ - // Query: "WITH __spatial_aggregation_cte AS (SELECT toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `service.name`, sum(value)/30 AS value FROM signoz_meter.distributed_samples AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'service.name') AS `service.name` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? AND JSONExtractString(labels, 'service.name') = ? GROUP BY fingerprint, `service.name`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY ts, `service.name`) SELECT * FROM __spatial_aggregation_cte", - // Args: []any{"signoz_calls_total", uint64(1747936800000), uint64(1747983420000), "delta", false, "cartservice", "signoz_calls_total", uint64(1747947390000), uint64(1747983420000)}, - // }, - // expectedErr: nil, - // }, + { + name: "test_cumulative_rate_sum", + requestType: qbtypes.RequestTypeTimeSeries, + query: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{ + Signal: telemetrytypes.SignalMeter, + StepInterval: qbtypes.Step{Duration: 24 * time.Hour}, + Aggregations: []qbtypes.MetricAggregation{ + { + MetricName: "signoz_calls_total", + Type: metrictypes.SumType, + Temporality: metrictypes.Cumulative, + TimeAggregation: metrictypes.TimeAggregationRate, + SpaceAggregation: metrictypes.SpaceAggregationSum, + }, + }, + Filter: &qbtypes.Filter{ + Expression: "service.name = 'cartservice'", + }, + Limit: 10, + GroupBy: []qbtypes.GroupByKey{ + { + TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{ + Name: "service.name", + }, + }, + }, + }, + expected: qbtypes.Statement{ + Query: "WITH __temporal_aggregation_cte AS (SELECT ts, `service.name`, If((per_series_value - lagInFrame(per_series_value, 1, 0) OVER rate_window) < 0, per_series_value / (ts - lagInFrame(ts, 1, toDateTime(fromUnixTimestamp64Milli(1747785600000))) OVER rate_window), (per_series_value - lagInFrame(per_series_value, 1, 0) OVER rate_window) / (ts - lagInFrame(ts, 1, toDateTime(fromUnixTimestamp64Milli(1747785600000))) OVER rate_window)) AS per_series_value FROM (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(86400)) AS ts, JSONExtractString(labels, 'service.name') AS `service.name`, max(max) AS per_series_value FROM signoz_meter.distributed_samples AS points WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? AND JSONExtractString(labels, 'service.name') = ? AND LOWER(temporality) LIKE LOWER(?) GROUP BY fingerprint, ts, `service.name` ORDER BY fingerprint, ts) WINDOW rate_window AS (PARTITION BY fingerprint ORDER BY fingerprint, ts)), __spatial_aggregation_cte AS (SELECT ts, `service.name`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts, `service.name`) SELECT * FROM __spatial_aggregation_cte", + Args: []any{"signoz_calls_total", uint64(1747785600000), uint64(1747983420000), "cartservice", "cumulative", 0}, + }, + expectedErr: nil, + }, + { + name: "test_delta_rate_sum", + requestType: qbtypes.RequestTypeTimeSeries, + query: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{ + Signal: telemetrytypes.SignalMeter, + StepInterval: qbtypes.Step{Duration: 24 * time.Hour}, + Aggregations: []qbtypes.MetricAggregation{ + { + MetricName: "signoz_calls_total", + Type: metrictypes.SumType, + Temporality: metrictypes.Delta, + TimeAggregation: metrictypes.TimeAggregationRate, + SpaceAggregation: metrictypes.SpaceAggregationSum, + }, + }, + Filter: &qbtypes.Filter{ + Expression: "service.name = 'cartservice'", + }, + Limit: 10, + GroupBy: []qbtypes.GroupByKey{ + { + TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{ + Name: "service.name", + }, + }, + }, + }, + expected: qbtypes.Statement{ + Query: "WITH __spatial_aggregation_cte AS (SELECT toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(86400)) AS ts, JSONExtractString(labels, 'service.name') AS `service.name`, sum(sum)/86400 AS value FROM signoz_meter.distributed_samples AS points WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? AND JSONExtractString(labels, 'service.name') = ? AND LOWER(temporality) LIKE LOWER(?) GROUP BY ts, `service.name`) SELECT * FROM __spatial_aggregation_cte", + Args: []any{"signoz_calls_total", uint64(1747872000000), uint64(1747983420000), "cartservice", "delta"}, + }, + expectedErr: nil, + }, { name: "test_delta_rate_avg", requestType: qbtypes.RequestTypeTimeSeries, query: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{ - Signal: telemetrytypes.SignalMetrics, + Signal: telemetrytypes.SignalMeter, StepInterval: qbtypes.Step{Duration: 24 * time.Hour}, Aggregations: []qbtypes.MetricAggregation{ { @@ -116,44 +116,44 @@ func TestStatementBuilder(t *testing.T) { }, }, expected: qbtypes.Statement{ - Query: "WITH __spatial_aggregation_cte AS (SELECT toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `service.name`, sum(value)/30 AS value FROM signoz_meter.distributed_samples AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'service.name') AS `service.name` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? AND JSONExtractString(labels, 'service.name') = ? GROUP BY fingerprint, `service.name`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY ts, `service.name`) SELECT * FROM __spatial_aggregation_cte", - Args: []any{"signoz_calls_total", uint64(1747936800000), uint64(1747983420000), "delta", false, "cartservice", "signoz_calls_total", uint64(1747947390000), uint64(1747983420000)}, + Query: "WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(86400)) AS ts, JSONExtractString(labels, 'service.name') AS `service.name`, sum(sum)/86400 AS per_series_value FROM signoz_meter.distributed_samples AS points WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? AND JSONExtractString(labels, 'service.name') = ? AND LOWER(temporality) LIKE LOWER(?) GROUP BY fingerprint, ts, `service.name` ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, `service.name`, avg(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts, `service.name`) SELECT * FROM __spatial_aggregation_cte", + Args: []any{"signoz_calls_total", uint64(1747872000000), uint64(1747983420000), "cartservice", "delta", 0}, + }, + expectedErr: nil, + }, + { + name: "test_gauge_avg_sum", + requestType: qbtypes.RequestTypeTimeSeries, + query: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{ + Signal: telemetrytypes.SignalMeter, + StepInterval: qbtypes.Step{Duration: 24 * time.Hour}, + Aggregations: []qbtypes.MetricAggregation{ + { + MetricName: "system.memory.usage", + Type: metrictypes.GaugeType, + Temporality: metrictypes.Unspecified, + TimeAggregation: metrictypes.TimeAggregationAvg, + SpaceAggregation: metrictypes.SpaceAggregationSum, + }, + }, + Filter: &qbtypes.Filter{ + Expression: "host.name = 'big-data-node-1'", + }, + Limit: 10, + GroupBy: []qbtypes.GroupByKey{ + { + TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{ + Name: "host.name", + }, + }, + }, + }, + expected: qbtypes.Statement{ + Query: "WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(86400)) AS ts, JSONExtractString(labels, 'host.name') AS `host.name`, sum(sum) / sum(count) AS per_series_value FROM signoz_meter.distributed_samples AS points WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? AND JSONExtractString(labels, 'host.name') = ? AND LOWER(temporality) LIKE LOWER(?) GROUP BY fingerprint, ts, `host.name` ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, `host.name`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts, `host.name`) SELECT * FROM __spatial_aggregation_cte", + Args: []any{"system.memory.usage", uint64(1747872000000), uint64(1747983420000), "big-data-node-1", "unspecified", 0}, }, expectedErr: nil, }, - // { - // name: "test_gauge_avg_sum", - // requestType: qbtypes.RequestTypeTimeSeries, - // query: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{ - // Signal: telemetrytypes.SignalMetrics, - // StepInterval: qbtypes.Step{Duration: 30 * time.Second}, - // Aggregations: []qbtypes.MetricAggregation{ - // { - // MetricName: "system.memory.usage", - // Type: metrictypes.GaugeType, - // Temporality: metrictypes.Unspecified, - // TimeAggregation: metrictypes.TimeAggregationAvg, - // SpaceAggregation: metrictypes.SpaceAggregationSum, - // }, - // }, - // Filter: &qbtypes.Filter{ - // Expression: "host.name = 'big-data-node-1'", - // }, - // Limit: 10, - // GroupBy: []qbtypes.GroupByKey{ - // { - // TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{ - // Name: "host.name", - // }, - // }, - // }, - // }, - // expected: qbtypes.Statement{ - // Query: "WITH __temporal_aggregation_cte AS (SELECT fingerprint, toStartOfInterval(toDateTime(intDiv(unix_milli, 1000)), toIntervalSecond(30)) AS ts, `host.name`, avg(value) AS per_series_value FROM signoz_metrics.distributed_samples_v4 AS points INNER JOIN (SELECT fingerprint, JSONExtractString(labels, 'host.name') AS `host.name` FROM signoz_metrics.time_series_v4_6hrs WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli <= ? AND LOWER(temporality) LIKE LOWER(?) AND __normalized = ? AND JSONExtractString(labels, 'host.name') = ? GROUP BY fingerprint, `host.name`) AS filtered_time_series ON points.fingerprint = filtered_time_series.fingerprint WHERE metric_name IN (?) AND unix_milli >= ? AND unix_milli < ? GROUP BY fingerprint, ts, `host.name` ORDER BY fingerprint, ts), __spatial_aggregation_cte AS (SELECT ts, `host.name`, sum(per_series_value) AS value FROM __temporal_aggregation_cte WHERE isNaN(per_series_value) = ? GROUP BY ts, `host.name`) SELECT * FROM __spatial_aggregation_cte", - // Args: []any{"system.memory.usage", uint64(1747936800000), uint64(1747983420000), "unspecified", false, "big-data-node-1", "system.memory.usage", uint64(1747947390000), uint64(1747983420000), 0}, - // }, - // expectedErr: nil, - // }, } fm := telemetrymetrics.NewFieldMapper()