Compare commits

...

27 Commits

Author SHA1 Message Date
nikhilmantri0902
9203602abd chore: rename 2026-04-28 17:30:11 +05:30
Nikhil Mantri
c10d278ec4 Merge branch 'infraM/v2_nodes_list_api' into infraM/v2_namespaces_list_api 2026-04-28 17:12:27 +05:30
nikhilmantri0902
8b0e8f666e chore: v2 nodes api 2026-04-28 17:02:19 +05:30
nikhilmantri0902
1d89b03f10 chore: merged main and resolved conflicts 2026-04-28 16:38:57 +05:30
Nikhil Mantri
ffd493617f feat(infra-monitoring): v2 pods list API (#10833)
* chore: baseline setup

* chore: endpoint detail update

* chore: added logic for hosts v3 api

* fix: bug fix

* chore: disk usage

* chore: added validate function

* chore: added some unit tests

* chore: return status as a string

* chore: yarn generate api

* chore: removed isSendingK8sAgentsMetricsCode

* chore: moved funcs

* chore: added validation on order by

* chore: added pods list logic

* chore: updated openapi yml

* chore: updated spec

* chore: pods api meta start time

* chore: nil pointer check

* chore: nil pointer dereference fix in req.Filter

* chore: added temporalities of metrics

* chore: added pods metrics temporality

* chore: unified composite key function

* chore: code improvements

* chore: added pods list api updates

* chore: hostStatusNone added for clarity that this field can be left empty as well in payload

* chore: yarn generate api

* chore: return errors from getMetadata and lint fix

* chore: return errors from getMetadata and lint fix

* chore: added hostName logic

* chore: modified getMetadata query

* chore: add type for response and files rearrange

* chore: warnings added passing from queryResponse warning to host lists response struct

* chore: added better metrics existence check

* chore: added a TODO remark

* chore: added required metrics check

* chore: distributed samples table to local table change for get metadata

* chore: frontend fix

* chore: endpoint correction

* chore: endpoint modification openapi

* chore: escape backtick to prevent sql injection

* chore: rearrage

* chore: improvements

* chore: validate order by to validate function

* chore: improved description

* chore: added TODOs and made filterByStatus a part of filter struct

* chore: ignore empty string hosts in get active hosts

* feat(infra-monitoring): v2 hosts list - return counts of active & inactive hosts for custom group by attributes (#10956)

* chore: add functionality for showing active and inactive counts in custom group by

* chore: bug fix

* chore: added subquery for active and total count

* chore: ignore empty string hosts in get active hosts

* fix: sinceUnixMilli for determining active hosts compute once per request

* chore: refactor code

* chore: rename HostsList -> ListHosts

* chore: rearrangement

* chore: inframonitoring types renaming

* chore: added types package

* chore: file structure further breakdown for clarity

* chore: comments correction

* chore: removed temporalities

* chore: pods code restructuring

* chore: comments resolve

* chore: added json tag required: true

* chore: removed pod metric temporalities

* chore: removed internal server error

* chore: added status unauthorized

* chore: remove a defensive nil map check, the function ensure non-nil map when err nil

* chore: cleanup and rename

* chore: make sort stable in case of tiebreaker by comparing composite group by keys

* chore: regen api client for inframonitoring

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* chore: added required tags

* chore: added support for pod phase unknown

* chore: removed pods - order by phase

* chore: improved api description to document -1 as no data in numeric fields

* fix: rebase fixes

* feat(infra-monitoring): v2 pods list apis - phase counts when custom grouping (#11088)

* chore: added phase counts feature

* chore: added queries for pod phase counts in custom group by

* chore: added unknown phase count

* fix: isPodUIDInGroupBy in buildPodRecords

* chore: 3 cte --> 2 cte

* chore: pod phase with local table of time series as counts

* chore: comment correction

* chore: corrected comment

* chore: value column for samples table added

* chore: removed query G for phase counts

* chore: rename variable

* chore: added PodPhaseNum constants to types

* chore: updated comment

* chore: formatted file

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
Co-authored-by: Ashwin Bhatkal <ashwin96@gmail.com>
2026-04-28 10:19:26 +00:00
Pandey
3f95d35ad5 chore: add @therealpandey with @srikanthcvv (#11126)
#### Temporary

- add @therealpandey with @srikanthcvv
2026-04-28 10:01:39 +00:00
Piyush Singariya
9e5678d6b3 CI: JSON QB E2E Integration Suite (#10863)
* feat: enable JSON Path index

* fix: contextual path index usage

* test: fix unit tests

* feat: align negative operators to include other logs

* fix: primitive conditions working

* fix: indexed tests passing

* fix: array type filtering from dynamic arrays

* fix: unit tests

* fix: remove not used paths from testdata

* fix: unit tests

* fix: comment

* fix: indexed unit tests

* fix: array json element comparison

* feat: change filtering of dynamic arrays

* fix: dynamic array tests

* fix: stringified integer value input

* fix: better review for test file

* fix: negative operator check

* ci: lint changes

* chore: import tests from older pr

* fix: better tests

* fix: logs.py

* fix: body tests ready

* fix: better validations

* fix: dynamically change insert stmt for body_v2 availability

* test: with higher migrator version

* fix: type ambiguity

* fix: test

* test: updated validation

* fix: tons of changes

* chore: remove redundent comparison

* ci: tests fixed

* fix: upgraded collector version

* fix: qbtoexpr tests

* fix: go sum

* chore: upgrade collector version v0.144.3-rc.4

* fix: tests

* ci: test fix

* revert: remove db binaries

* test: selectField tests added

* fix: added safeguards in plan generation

* fix: name changed to field_map

* chore: changes based on review

* fix: changes based on review

* fix: remove unused promoted fixture

* test: integration test fix attempt 1

* fix: json access plan remval of AvailableTypes

* fix: invalid index usage on terminal condition

* test: added indexed tests separately

* test: select order by tests added

* fix: update jsontypeexporter

* fix: branches should tell missing array types

* fix: comment removed

* ci: test with updated collector

* fix: issue with FuzzyMatching and API failing

* test: select orderby works

* fix: int64 mapping

* fix: replacing JSONIndex with TelemetryFieldKeyIndex

* chore: update collector to stable release

* chore: update migrator version to stable release

* chore: error fix and unused code

* chore: go mod tidy

* chore: some changes in test to improve quality

* test: enhanced tests to work with backtick required key

* chore: comment removal

* fix: change based on review

* fix: change based on review

* fix: openapi ci

* ci: fix go tests

* fix: index needs body prefix

* fix: tests

* ci: polluted test fix

* fix: shift test files

* refactor(telemetrytypes): replace JSONDataTypeIndex with TelemetryFieldKeySkipIndex

* revert: mcp related changes

* feat: check query log as fixture

* fix: reuse querier fixtures

* chore: rename jsontypeexporter.py

* chore: py-fmt

* fix: minor fix

* fix: py-lint

* fix: changes based on review

* chore: fmt
2026-04-28 08:49:44 +00:00
nikhilmantri0902
7d702763cc chore: namespaces code 2026-04-28 13:44:21 +05:30
nikhilmantri0902
3a61a78986 chore: merged base branch 2026-04-28 12:48:20 +05:30
Abhi kumar
941016f12c Feat/crosshair series highlight (#11013)
* chore: added changes for crosshair sync for tooltip

* chore: minor cleanup

* chore: updated the core structure

* chore: updated the types

* chore: minor cleanup

* feat: added changes for sereis highlighting on crosshair sync

* chore: pr review fixes

* chore: handled other cases of groupby

* chore: minor changes

* feat: tooltip sync across panels (#11114)

* feat: added changes for syncing tooltip

* fix: fixed other tooltips closing when clicked on top of root tooltip

* fix: highlighting first series

* chore: removed y-axis sync for tooltip mode

* chore: minor fix

* chore: fmt fix
2026-04-28 06:52:51 +00:00
nikhilmantri0902
401701e036 chore: metadata fix 2026-04-27 19:23:11 +05:30
nikhilmantri0902
3bec0df0ad chore: nodes list v2 full blown 2026-04-27 16:17:39 +05:30
nikhilmantri0902
520e92049c Merge branch 'feat/v2_pods_list_api_phase_counts' of github.com:SigNoz/signoz into infraM/v2_nodes_list_api 2026-04-27 15:43:09 +05:30
nikhilmantri0902
eff29aefba chore: added PodPhaseNum constants to types 2026-04-27 13:59:08 +05:30
nikhilmantri0902
ca73453c9e chore: rename variable 2026-04-27 13:53:17 +05:30
nikhilmantri0902
1d836d674f chore: removed query G for phase counts 2026-04-27 13:46:03 +05:30
nikhilmantri0902
83724b0cde chore: value column for samples table added 2026-04-27 13:09:46 +05:30
nikhilmantri0902
3050e37ec7 chore: corrected comment 2026-04-27 12:43:17 +05:30
nikhilmantri0902
55b2215025 chore: comment correction 2026-04-24 17:14:56 +05:30
nikhilmantri0902
e90378e618 Merge branch 'infraM/v2_pods_list_api' into feat/v2_pods_list_api_phase_counts 2026-04-24 16:47:15 +05:30
nikhilmantri0902
4592b78f48 chore: pod phase with local table of time series as counts 2026-04-24 15:27:38 +05:30
nikhilmantri0902
d81c99feae chore: 3 cte --> 2 cte 2026-04-24 14:58:26 +05:30
nikhilmantri0902
65a456ff9e fix: isPodUIDInGroupBy in buildPodRecords 2026-04-24 14:36:43 +05:30
nikhilmantri0902
264577b673 chore: added unknown phase count 2026-04-24 13:29:09 +05:30
nikhilmantri0902
c78c9a42db chore: merged base 2026-04-24 13:03:21 +05:30
nikhilmantri0902
b7d4f18aae chore: added queries for pod phase counts in custom group by 2026-04-23 18:01:26 +05:30
nikhilmantri0902
9ad2ec428a chore: added phase counts feature 2026-04-23 16:50:39 +05:30
49 changed files with 4902 additions and 128 deletions

56
.github/CODEOWNERS vendored
View File

@@ -52,49 +52,49 @@ go.mod @therealpandey
# Querier Owners
/pkg/querier/ @srikanthccv
/pkg/variables/ @srikanthccv
/pkg/types/querybuildertypes/ @srikanthccv
/pkg/types/telemetrytypes/ @srikanthccv
/pkg/querybuilder/ @srikanthccv
/pkg/telemetrylogs/ @srikanthccv
/pkg/telemetrymetadata/ @srikanthccv
/pkg/telemetrymetrics/ @srikanthccv
/pkg/telemetrytraces/ @srikanthccv
/pkg/querier/ @srikanthccv @therealpandey
/pkg/variables/ @srikanthccv @therealpandey
/pkg/types/querybuildertypes/ @srikanthccv @therealpandey
/pkg/types/telemetrytypes/ @srikanthccv @therealpandey
/pkg/querybuilder/ @srikanthccv @therealpandey
/pkg/telemetrylogs/ @srikanthccv @therealpandey
/pkg/telemetrymetadata/ @srikanthccv @therealpandey
/pkg/telemetrymetrics/ @srikanthccv @therealpandey
/pkg/telemetrytraces/ @srikanthccv @therealpandey
# Metrics
/pkg/types/metrictypes/ @srikanthccv
/pkg/types/metricsexplorertypes/ @srikanthccv
/pkg/modules/metricsexplorer/ @srikanthccv
/pkg/prometheus/ @srikanthccv
/pkg/types/metrictypes/ @srikanthccv @therealpandey
/pkg/types/metricsexplorertypes/ @srikanthccv @therealpandey
/pkg/modules/metricsexplorer/ @srikanthccv @therealpandey
/pkg/prometheus/ @srikanthccv @therealpandey
# APM
/pkg/types/servicetypes/ @srikanthccv
/pkg/types/apdextypes/ @srikanthccv
/pkg/modules/apdex/ @srikanthccv
/pkg/modules/services/ @srikanthccv
/pkg/types/servicetypes/ @srikanthccv @therealpandey
/pkg/types/apdextypes/ @srikanthccv @therealpandey
/pkg/modules/apdex/ @srikanthccv @therealpandey
/pkg/modules/services/ @srikanthccv @therealpandey
# Dashboard
/pkg/types/dashboardtypes/ @srikanthccv
/pkg/modules/dashboard/ @srikanthccv
/pkg/types/dashboardtypes/ @srikanthccv @therealpandey
/pkg/modules/dashboard/ @srikanthccv @therealpandey
# Rule/Alertmanager
/pkg/types/ruletypes/ @srikanthccv
/pkg/types/alertmanagertypes @srikanthccv
/pkg/alertmanager/ @srikanthccv
/pkg/ruler/ @srikanthccv
/pkg/modules/rulestatehistory/ @srikanthccv
/pkg/types/rulestatehistorytypes/ @srikanthccv
/pkg/types/ruletypes/ @srikanthccv @therealpandey
/pkg/types/alertmanagertypes @srikanthccv @therealpandey
/pkg/alertmanager/ @srikanthccv @therealpandey
/pkg/ruler/ @srikanthccv @therealpandey
/pkg/modules/rulestatehistory/ @srikanthccv @therealpandey
/pkg/types/rulestatehistorytypes/ @srikanthccv @therealpandey
# Correlation-adjacent
/pkg/contextlinks/ @srikanthccv
/pkg/types/parsertypes/ @srikanthccv
/pkg/queryparser/ @srikanthccv
/pkg/contextlinks/ @srikanthccv @therealpandey
/pkg/types/parsertypes/ @srikanthccv @therealpandey
/pkg/queryparser/ @srikanthccv @therealpandey
# AuthN / AuthZ Owners

View File

@@ -51,6 +51,7 @@ jobs:
- role
- rootuser
- serviceaccount
- querier_json_body
- ttl
sqlstore-provider:
- postgres
@@ -61,7 +62,7 @@ jobs:
- 25.5.6
- 25.12.5
schema-migrator-version:
- v0.142.0
- v0.144.3
postgres-version:
- 15
if: |

View File

@@ -2474,6 +2474,132 @@ components:
- requiredMetricsCheck
- endTimeBeforeRetention
type: object
InframonitoringtypesNamespaceRecord:
properties:
failedPodCount:
type: integer
meta:
additionalProperties: {}
nullable: true
type: object
namespaceCPU:
format: double
type: number
namespaceMemory:
format: double
type: number
namespaceName:
type: string
pendingPodCount:
type: integer
runningPodCount:
type: integer
succeededPodCount:
type: integer
unknownPodCount:
type: integer
required:
- namespaceName
- namespaceCPU
- namespaceMemory
- pendingPodCount
- runningPodCount
- succeededPodCount
- failedPodCount
- unknownPodCount
- meta
type: object
InframonitoringtypesNamespaces:
properties:
endTimeBeforeRetention:
type: boolean
records:
items:
$ref: '#/components/schemas/InframonitoringtypesNamespaceRecord'
nullable: true
type: array
requiredMetricsCheck:
$ref: '#/components/schemas/InframonitoringtypesRequiredMetricsCheck'
total:
type: integer
type:
$ref: '#/components/schemas/InframonitoringtypesResponseType'
warning:
$ref: '#/components/schemas/Querybuildertypesv5QueryWarnData'
required:
- type
- records
- total
- requiredMetricsCheck
- endTimeBeforeRetention
type: object
InframonitoringtypesNodeCondition:
enum:
- ready
- not_ready
- ""
type: string
InframonitoringtypesNodeRecord:
properties:
condition:
$ref: '#/components/schemas/InframonitoringtypesNodeCondition'
meta:
additionalProperties: {}
nullable: true
type: object
nodeCPU:
format: double
type: number
nodeCPUAllocatable:
format: double
type: number
nodeMemory:
format: double
type: number
nodeMemoryAllocatable:
format: double
type: number
nodeName:
type: string
notReadyNodesCount:
type: integer
readyNodesCount:
type: integer
required:
- nodeName
- condition
- readyNodesCount
- notReadyNodesCount
- nodeCPU
- nodeCPUAllocatable
- nodeMemory
- nodeMemoryAllocatable
- meta
type: object
InframonitoringtypesNodes:
properties:
endTimeBeforeRetention:
type: boolean
records:
items:
$ref: '#/components/schemas/InframonitoringtypesNodeRecord'
nullable: true
type: array
requiredMetricsCheck:
$ref: '#/components/schemas/InframonitoringtypesRequiredMetricsCheck'
total:
type: integer
type:
$ref: '#/components/schemas/InframonitoringtypesResponseType'
warning:
$ref: '#/components/schemas/Querybuildertypesv5QueryWarnData'
required:
- type
- records
- total
- requiredMetricsCheck
- endTimeBeforeRetention
type: object
InframonitoringtypesPodPhase:
enum:
- pending
@@ -2591,6 +2717,58 @@ components:
- end
- limit
type: object
InframonitoringtypesPostableNamespaces:
properties:
end:
format: int64
type: integer
filter:
$ref: '#/components/schemas/Querybuildertypesv5Filter'
groupBy:
items:
$ref: '#/components/schemas/Querybuildertypesv5GroupByKey'
nullable: true
type: array
limit:
type: integer
offset:
type: integer
orderBy:
$ref: '#/components/schemas/Querybuildertypesv5OrderBy'
start:
format: int64
type: integer
required:
- start
- end
- limit
type: object
InframonitoringtypesPostableNodes:
properties:
end:
format: int64
type: integer
filter:
$ref: '#/components/schemas/Querybuildertypesv5Filter'
groupBy:
items:
$ref: '#/components/schemas/Querybuildertypesv5GroupByKey'
nullable: true
type: array
limit:
type: integer
offset:
type: integer
orderBy:
$ref: '#/components/schemas/Querybuildertypesv5OrderBy'
start:
format: int64
type: integer
required:
- start
- end
- limit
type: object
InframonitoringtypesPostablePods:
properties:
end:
@@ -11059,6 +11237,145 @@ paths:
summary: List Hosts for Infra Monitoring
tags:
- inframonitoring
/api/v2/infra_monitoring/namespaces:
post:
deprecated: false
description: 'Returns a paginated list of Kubernetes namespaces with key aggregated
pod metrics: CPU usage and memory working set (summed across pods in the group),
plus per-group pod counts bucketed by each pod''s latest k8s.pod.phase value
in the window (pendingPodCount, runningPodCount, succeededPodCount, failedPodCount,
unknownPodCount). Each namespace includes metadata attributes (k8s.namespace.name,
k8s.cluster.name). The response type is ''list'' for the default k8s.namespace.name
grouping or ''grouped_list'' for custom groupBy keys; in both modes every
row aggregates pods in the group. Supports filtering via a filter expression,
custom groupBy, ordering by cpu / memory, and pagination via offset/limit.
Also reports missing required metrics and whether the requested time range
falls before the data retention boundary. Numeric metric fields (namespaceCPU,
namespaceMemory) return -1 as a sentinel when no data is available for that
field.'
operationId: ListNamespaces
requestBody:
content:
application/json:
schema:
$ref: '#/components/schemas/InframonitoringtypesPostableNamespaces'
responses:
"200":
content:
application/json:
schema:
properties:
data:
$ref: '#/components/schemas/InframonitoringtypesNamespaces'
status:
type: string
required:
- status
- data
type: object
description: OK
"400":
content:
application/json:
schema:
$ref: '#/components/schemas/RenderErrorResponse'
description: Bad Request
"401":
content:
application/json:
schema:
$ref: '#/components/schemas/RenderErrorResponse'
description: Unauthorized
"403":
content:
application/json:
schema:
$ref: '#/components/schemas/RenderErrorResponse'
description: Forbidden
"500":
content:
application/json:
schema:
$ref: '#/components/schemas/RenderErrorResponse'
description: Internal Server Error
security:
- api_key:
- VIEWER
- tokenizer:
- VIEWER
summary: List Namespaces for Infra Monitoring
tags:
- inframonitoring
/api/v2/infra_monitoring/nodes:
post:
deprecated: false
description: 'Returns a paginated list of Kubernetes nodes with key metrics:
CPU usage, CPU allocatable, memory working set, memory allocatable, and per-group
readyNodesCount / notReadyNodesCount derived from each node''s latest k8s.node.condition_ready
value in the window. Each node includes metadata attributes (k8s.node.uid,
k8s.cluster.name). The response type is ''list'' for the default k8s.node.name
grouping (each row is one node with its current condition string: ready /
not_ready / '''') or ''grouped_list'' for custom groupBy keys (each row aggregates
nodes in the group with readyNodesCount and notReadyNodesCount; condition
stays empty). Supports filtering via a filter expression, custom groupBy,
ordering by cpu / cpu_allocatable / memory / memory_allocatable, and pagination
via offset/limit. Also reports missing required metrics and whether the requested
time range falls before the data retention boundary. Numeric metric fields
(nodeCPU, nodeCPUAllocatable, nodeMemory, nodeMemoryAllocatable) return -1
as a sentinel when no data is available for that field.'
operationId: ListNodes
requestBody:
content:
application/json:
schema:
$ref: '#/components/schemas/InframonitoringtypesPostableNodes'
responses:
"200":
content:
application/json:
schema:
properties:
data:
$ref: '#/components/schemas/InframonitoringtypesNodes'
status:
type: string
required:
- status
- data
type: object
description: OK
"400":
content:
application/json:
schema:
$ref: '#/components/schemas/RenderErrorResponse'
description: Bad Request
"401":
content:
application/json:
schema:
$ref: '#/components/schemas/RenderErrorResponse'
description: Unauthorized
"403":
content:
application/json:
schema:
$ref: '#/components/schemas/RenderErrorResponse'
description: Forbidden
"500":
content:
application/json:
schema:
$ref: '#/components/schemas/RenderErrorResponse'
description: Internal Server Error
security:
- api_key:
- VIEWER
- tokenizer:
- VIEWER
summary: List Nodes for Infra Monitoring
tags:
- inframonitoring
/api/v2/infra_monitoring/pods:
post:
deprecated: false

View File

@@ -13,8 +13,12 @@ import type {
import type {
InframonitoringtypesPostableHostsDTO,
InframonitoringtypesPostableNamespacesDTO,
InframonitoringtypesPostableNodesDTO,
InframonitoringtypesPostablePodsDTO,
ListHosts200,
ListNamespaces200,
ListNodes200,
ListPods200,
RenderErrorResponseDTO,
} from '../sigNoz.schemas';
@@ -106,6 +110,174 @@ export const useListHosts = <
return useMutation(mutationOptions);
};
/**
* Returns a paginated list of Kubernetes namespaces with key aggregated pod metrics: CPU usage and memory working set (summed across pods in the group), plus per-group pod counts bucketed by each pod's latest k8s.pod.phase value in the window (pendingPodCount, runningPodCount, succeededPodCount, failedPodCount, unknownPodCount). Each namespace includes metadata attributes (k8s.namespace.name, k8s.cluster.name). The response type is 'list' for the default k8s.namespace.name grouping or 'grouped_list' for custom groupBy keys; in both modes every row aggregates pods in the group. Supports filtering via a filter expression, custom groupBy, ordering by cpu / memory, and pagination via offset/limit. Also reports missing required metrics and whether the requested time range falls before the data retention boundary. Numeric metric fields (namespaceCPU, namespaceMemory) return -1 as a sentinel when no data is available for that field.
* @summary List Namespaces for Infra Monitoring
*/
export const listNamespaces = (
inframonitoringtypesPostableNamespacesDTO: BodyType<InframonitoringtypesPostableNamespacesDTO>,
signal?: AbortSignal,
) => {
return GeneratedAPIInstance<ListNamespaces200>({
url: `/api/v2/infra_monitoring/namespaces`,
method: 'POST',
headers: { 'Content-Type': 'application/json' },
data: inframonitoringtypesPostableNamespacesDTO,
signal,
});
};
export const getListNamespacesMutationOptions = <
TError = ErrorType<RenderErrorResponseDTO>,
TContext = unknown,
>(options?: {
mutation?: UseMutationOptions<
Awaited<ReturnType<typeof listNamespaces>>,
TError,
{ data: BodyType<InframonitoringtypesPostableNamespacesDTO> },
TContext
>;
}): UseMutationOptions<
Awaited<ReturnType<typeof listNamespaces>>,
TError,
{ data: BodyType<InframonitoringtypesPostableNamespacesDTO> },
TContext
> => {
const mutationKey = ['listNamespaces'];
const { mutation: mutationOptions } = options
? options.mutation &&
'mutationKey' in options.mutation &&
options.mutation.mutationKey
? options
: { ...options, mutation: { ...options.mutation, mutationKey } }
: { mutation: { mutationKey } };
const mutationFn: MutationFunction<
Awaited<ReturnType<typeof listNamespaces>>,
{ data: BodyType<InframonitoringtypesPostableNamespacesDTO> }
> = (props) => {
const { data } = props ?? {};
return listNamespaces(data);
};
return { mutationFn, ...mutationOptions };
};
export type ListNamespacesMutationResult = NonNullable<
Awaited<ReturnType<typeof listNamespaces>>
>;
export type ListNamespacesMutationBody =
BodyType<InframonitoringtypesPostableNamespacesDTO>;
export type ListNamespacesMutationError = ErrorType<RenderErrorResponseDTO>;
/**
* @summary List Namespaces for Infra Monitoring
*/
export const useListNamespaces = <
TError = ErrorType<RenderErrorResponseDTO>,
TContext = unknown,
>(options?: {
mutation?: UseMutationOptions<
Awaited<ReturnType<typeof listNamespaces>>,
TError,
{ data: BodyType<InframonitoringtypesPostableNamespacesDTO> },
TContext
>;
}): UseMutationResult<
Awaited<ReturnType<typeof listNamespaces>>,
TError,
{ data: BodyType<InframonitoringtypesPostableNamespacesDTO> },
TContext
> => {
const mutationOptions = getListNamespacesMutationOptions(options);
return useMutation(mutationOptions);
};
/**
* Returns a paginated list of Kubernetes nodes with key metrics: CPU usage, CPU allocatable, memory working set, memory allocatable, and per-group readyNodesCount / notReadyNodesCount derived from each node's latest k8s.node.condition_ready value in the window. Each node includes metadata attributes (k8s.node.uid, k8s.cluster.name). The response type is 'list' for the default k8s.node.name grouping (each row is one node with its current condition string: ready / not_ready / '') or 'grouped_list' for custom groupBy keys (each row aggregates nodes in the group with readyNodesCount and notReadyNodesCount; condition stays empty). Supports filtering via a filter expression, custom groupBy, ordering by cpu / cpu_allocatable / memory / memory_allocatable, and pagination via offset/limit. Also reports missing required metrics and whether the requested time range falls before the data retention boundary. Numeric metric fields (nodeCPU, nodeCPUAllocatable, nodeMemory, nodeMemoryAllocatable) return -1 as a sentinel when no data is available for that field.
* @summary List Nodes for Infra Monitoring
*/
export const listNodes = (
inframonitoringtypesPostableNodesDTO: BodyType<InframonitoringtypesPostableNodesDTO>,
signal?: AbortSignal,
) => {
return GeneratedAPIInstance<ListNodes200>({
url: `/api/v2/infra_monitoring/nodes`,
method: 'POST',
headers: { 'Content-Type': 'application/json' },
data: inframonitoringtypesPostableNodesDTO,
signal,
});
};
export const getListNodesMutationOptions = <
TError = ErrorType<RenderErrorResponseDTO>,
TContext = unknown,
>(options?: {
mutation?: UseMutationOptions<
Awaited<ReturnType<typeof listNodes>>,
TError,
{ data: BodyType<InframonitoringtypesPostableNodesDTO> },
TContext
>;
}): UseMutationOptions<
Awaited<ReturnType<typeof listNodes>>,
TError,
{ data: BodyType<InframonitoringtypesPostableNodesDTO> },
TContext
> => {
const mutationKey = ['listNodes'];
const { mutation: mutationOptions } = options
? options.mutation &&
'mutationKey' in options.mutation &&
options.mutation.mutationKey
? options
: { ...options, mutation: { ...options.mutation, mutationKey } }
: { mutation: { mutationKey } };
const mutationFn: MutationFunction<
Awaited<ReturnType<typeof listNodes>>,
{ data: BodyType<InframonitoringtypesPostableNodesDTO> }
> = (props) => {
const { data } = props ?? {};
return listNodes(data);
};
return { mutationFn, ...mutationOptions };
};
export type ListNodesMutationResult = NonNullable<
Awaited<ReturnType<typeof listNodes>>
>;
export type ListNodesMutationBody =
BodyType<InframonitoringtypesPostableNodesDTO>;
export type ListNodesMutationError = ErrorType<RenderErrorResponseDTO>;
/**
* @summary List Nodes for Infra Monitoring
*/
export const useListNodes = <
TError = ErrorType<RenderErrorResponseDTO>,
TContext = unknown,
>(options?: {
mutation?: UseMutationOptions<
Awaited<ReturnType<typeof listNodes>>,
TError,
{ data: BodyType<InframonitoringtypesPostableNodesDTO> },
TContext
>;
}): UseMutationResult<
Awaited<ReturnType<typeof listNodes>>,
TError,
{ data: BodyType<InframonitoringtypesPostableNodesDTO> },
TContext
> => {
const mutationOptions = getListNodesMutationOptions(options);
return useMutation(mutationOptions);
};
/**
* Returns a paginated list of Kubernetes pods with key metrics: CPU usage, CPU request/limit utilization, memory working set, memory request/limit utilization, current pod phase (pending/running/succeeded/failed/unknown), and pod age (ms since start time). Each pod includes metadata attributes (namespace, node, workload owner such as deployment/statefulset/daemonset/job/cronjob, cluster). Supports filtering via a filter expression, custom groupBy to aggregate pods by any attribute, ordering by any of the six metrics (cpu, cpu_request, cpu_limit, memory, memory_request, memory_limit), and pagination via offset/limit. The response type is 'list' for the default k8s.pod.uid grouping (each row is one pod with its current phase) or 'grouped_list' for custom groupBy keys (each row aggregates pods in the group with per-phase counts: pendingPodCount, runningPodCount, succeededPodCount, failedPodCount, unknownPodCount derived from each pod's latest phase in the window). Also reports missing required metrics and whether the requested time range falls before the data retention boundary. Numeric metric fields (podCPU, podCPURequest, podCPULimit, podMemory, podMemoryRequest, podMemoryLimit, podAge) return -1 as a sentinel when no data is available for that field.
* @summary List Pods for Infra Monitoring

View File

@@ -3243,6 +3243,146 @@ export interface InframonitoringtypesHostsDTO {
warning?: Querybuildertypesv5QueryWarnDataDTO;
}
/**
* @nullable
*/
export type InframonitoringtypesNamespaceRecordDTOMeta = {
[key: string]: unknown;
} | null;
export interface InframonitoringtypesNamespaceRecordDTO {
/**
* @type integer
*/
failedPodCount: number;
/**
* @type object
* @nullable true
*/
meta: InframonitoringtypesNamespaceRecordDTOMeta;
/**
* @type number
* @format double
*/
namespaceCPU: number;
/**
* @type number
* @format double
*/
namespaceMemory: number;
/**
* @type string
*/
namespaceName: string;
/**
* @type integer
*/
pendingPodCount: number;
/**
* @type integer
*/
runningPodCount: number;
/**
* @type integer
*/
succeededPodCount: number;
/**
* @type integer
*/
unknownPodCount: number;
}
export interface InframonitoringtypesNamespacesDTO {
/**
* @type boolean
*/
endTimeBeforeRetention: boolean;
/**
* @type array
* @nullable true
*/
records: InframonitoringtypesNamespaceRecordDTO[] | null;
requiredMetricsCheck: InframonitoringtypesRequiredMetricsCheckDTO;
/**
* @type integer
*/
total: number;
type: InframonitoringtypesResponseTypeDTO;
warning?: Querybuildertypesv5QueryWarnDataDTO;
}
export enum InframonitoringtypesNodeConditionDTO {
ready = 'ready',
not_ready = 'not_ready',
'' = '',
}
/**
* @nullable
*/
export type InframonitoringtypesNodeRecordDTOMeta = {
[key: string]: unknown;
} | null;
export interface InframonitoringtypesNodeRecordDTO {
condition: InframonitoringtypesNodeConditionDTO;
/**
* @type object
* @nullable true
*/
meta: InframonitoringtypesNodeRecordDTOMeta;
/**
* @type number
* @format double
*/
nodeCPU: number;
/**
* @type number
* @format double
*/
nodeCPUAllocatable: number;
/**
* @type number
* @format double
*/
nodeMemory: number;
/**
* @type number
* @format double
*/
nodeMemoryAllocatable: number;
/**
* @type string
*/
nodeName: string;
/**
* @type integer
*/
notReadyNodesCount: number;
/**
* @type integer
*/
readyNodesCount: number;
}
export interface InframonitoringtypesNodesDTO {
/**
* @type boolean
*/
endTimeBeforeRetention: boolean;
/**
* @type array
* @nullable true
*/
records: InframonitoringtypesNodeRecordDTO[] | null;
requiredMetricsCheck: InframonitoringtypesRequiredMetricsCheckDTO;
/**
* @type integer
*/
total: number;
type: InframonitoringtypesResponseTypeDTO;
warning?: Querybuildertypesv5QueryWarnDataDTO;
}
export enum InframonitoringtypesPodPhaseDTO {
pending = 'pending',
running = 'running',
@@ -3373,6 +3513,62 @@ export interface InframonitoringtypesPostableHostsDTO {
start: number;
}
export interface InframonitoringtypesPostableNamespacesDTO {
/**
* @type integer
* @format int64
*/
end: number;
filter?: Querybuildertypesv5FilterDTO;
/**
* @type array
* @nullable true
*/
groupBy?: Querybuildertypesv5GroupByKeyDTO[] | null;
/**
* @type integer
*/
limit: number;
/**
* @type integer
*/
offset?: number;
orderBy?: Querybuildertypesv5OrderByDTO;
/**
* @type integer
* @format int64
*/
start: number;
}
export interface InframonitoringtypesPostableNodesDTO {
/**
* @type integer
* @format int64
*/
end: number;
filter?: Querybuildertypesv5FilterDTO;
/**
* @type array
* @nullable true
*/
groupBy?: Querybuildertypesv5GroupByKeyDTO[] | null;
/**
* @type integer
*/
limit: number;
/**
* @type integer
*/
offset?: number;
orderBy?: Querybuildertypesv5OrderByDTO;
/**
* @type integer
* @format int64
*/
start: number;
}
export interface InframonitoringtypesPostablePodsDTO {
/**
* @type integer
@@ -7480,6 +7676,22 @@ export type ListHosts200 = {
status: string;
};
export type ListNamespaces200 = {
data: InframonitoringtypesNamespacesDTO;
/**
* @type string
*/
status: string;
};
export type ListNodes200 = {
data: InframonitoringtypesNodesDTO;
/**
* @type string
*/
status: string;
};
export type ListPods200 = {
data: InframonitoringtypesPodsDTO;
/**

View File

@@ -33,6 +33,7 @@ export default function ChartWrapper({
children,
layoutChildren,
yAxisUnit,
groupBy,
customTooltip,
pinnedTooltipElement,
'data-testid': testId,
@@ -68,8 +69,9 @@ export default function ChartWrapper({
const syncMetadata = useMemo(
() => ({
yAxisUnit,
groupBy,
}),
[yAxisUnit],
[yAxisUnit, groupBy],
);
return (

View File

@@ -6,6 +6,7 @@ import {
DashboardCursorSync,
TooltipClickData,
} from 'lib/uPlotV2/plugins/TooltipPlugin/types';
import { BaseAutocompleteData } from 'types/api/queryBuilder/queryAutocompleteResponse';
interface BaseChartProps {
width: number;
@@ -38,6 +39,7 @@ interface UPlotBasedChartProps {
interface UPlotChartDataProps {
yAxisUnit?: string;
decimalPrecision?: PrecisionOption;
groupBy?: BaseAutocompleteData[];
}
export interface TimeSeriesChartProps

View File

@@ -113,6 +113,10 @@ function BarPanel(props: PanelWrapperProps): JSX.Element {
uPlotRef.current = plot;
}, []);
const groupBy = useMemo(() => {
return widget.query.builder.queryData[0].groupBy;
}, [widget.query]);
return (
<div className="panel-container" ref={graphRef}>
{containerDimensions.width > 0 && containerDimensions.height > 0 && (
@@ -128,6 +132,7 @@ function BarPanel(props: PanelWrapperProps): JSX.Element {
width={containerDimensions.width}
height={containerDimensions.height}
layoutChildren={layoutChildren}
groupBy={groupBy}
isStackedBarChart={widget.stackedBarChart ?? false}
yAxisUnit={widget.yAxisUnit}
decimalPrecision={widget.decimalPrecision}

View File

@@ -105,6 +105,7 @@ export function prepareBarPanelConfig({
colorMapping: widget.customLegendColors ?? {},
isDarkMode,
stepInterval: currentStepInterval,
metric: series.metric,
});
});

View File

@@ -104,6 +104,10 @@ function TimeSeriesPanel(props: PanelWrapperProps): JSX.Element {
widget.decimalPrecision,
]);
const groupBy = useMemo(() => {
return widget.query.builder.queryData[0].groupBy;
}, [widget.query]);
return (
<div className="panel-container" ref={graphRef}>
{containerDimensions.width > 0 && containerDimensions.height > 0 && (
@@ -117,6 +121,7 @@ function TimeSeriesPanel(props: PanelWrapperProps): JSX.Element {
yAxisUnit={widget.yAxisUnit}
decimalPrecision={widget.decimalPrecision}
data={chartData as uPlot.AlignedData}
groupBy={groupBy}
width={containerDimensions.width}
height={containerDimensions.height}
layoutChildren={layoutChildren}

View File

@@ -131,6 +131,7 @@ export const prepareUPlotConfig = ({
pointSize: 5,
fillMode: widget.fillMode || FillMode.None,
isDarkMode,
metric: series.metric,
});
});

View File

@@ -115,7 +115,9 @@ export const useGetQueryRange: UseGetQueryRange = (
const updatedQuery = updateBarStepInterval(
requestData.query,
requestData.start ? requestData.start * 1e3 : Number.parseInt(start, 10) * 1e3,
requestData.start
? requestData.start * 1e3
: Number.parseInt(start, 10) * 1e3,
requestData.end ? requestData.end * 1e3 : Number.parseInt(end, 10) * 1e3,
);

View File

@@ -16,6 +16,7 @@ export default function BarChartTooltip(props: BarTooltipProps): JSX.Element {
yAxisUnit: props.yAxisUnit ?? '',
decimalPrecision: props.decimalPrecision,
isStackedBarChart: props.isStackedBarChart,
syncedSeriesIndexes: props.syncedSeriesIndexes,
}),
[
props.uPlotInstance,
@@ -24,6 +25,7 @@ export default function BarChartTooltip(props: BarTooltipProps): JSX.Element {
props.yAxisUnit,
props.decimalPrecision,
props.isStackedBarChart,
props.syncedSeriesIndexes,
],
);

View File

@@ -17,6 +17,7 @@ export default function HistogramTooltip(
uPlotInstance: props.uPlotInstance,
yAxisUnit: props.yAxisUnit ?? '',
decimalPrecision: props.decimalPrecision,
syncedSeriesIndexes: props.syncedSeriesIndexes,
}),
[
props.uPlotInstance,
@@ -24,6 +25,7 @@ export default function HistogramTooltip(
props.dataIndexes,
props.yAxisUnit,
props.decimalPrecision,
props.syncedSeriesIndexes,
],
);

View File

@@ -17,6 +17,7 @@ export default function TimeSeriesTooltip(
uPlotInstance: props.uPlotInstance,
yAxisUnit: props.yAxisUnit ?? '',
decimalPrecision: props.decimalPrecision,
syncedSeriesIndexes: props.syncedSeriesIndexes,
}),
[
props.uPlotInstance,
@@ -24,6 +25,7 @@ export default function TimeSeriesTooltip(
props.dataIndexes,
props.yAxisUnit,
props.decimalPrecision,
props.syncedSeriesIndexes,
],
);

View File

@@ -62,6 +62,7 @@ export function buildTooltipContent({
yAxisUnit,
decimalPrecision,
isStackedBarChart,
syncedSeriesIndexes,
}: {
data: AlignedData;
series: Series[];
@@ -71,18 +72,34 @@ export function buildTooltipContent({
yAxisUnit: string;
decimalPrecision?: PrecisionOption;
isStackedBarChart?: boolean;
syncedSeriesIndexes?: number[] | null;
}): TooltipContentItem[] {
const items: TooltipContentItem[] = [];
const allowedIndexes =
syncedSeriesIndexes != null ? new Set(syncedSeriesIndexes) : null;
for (let seriesIndex = 1; seriesIndex < series.length; seriesIndex += 1) {
const seriesItem = series[seriesIndex];
if (!seriesItem?.show) {
continue;
}
if (allowedIndexes != null && !allowedIndexes.has(seriesIndex)) {
continue;
}
const dataIndex = dataIndexes[seriesIndex];
// Skip series with no data at the current cursor position
const isSync = allowedIndexes != null;
if (dataIndex === null) {
if (isSync) {
items.push({
label: String(seriesItem.label ?? ''),
value: 0,
tooltipValue: 'No Data',
color: resolveSeriesColor(seriesItem.stroke, uPlotInstance, seriesIndex),
isActive: false,
});
}
continue;
}
@@ -102,6 +119,14 @@ export function buildTooltipContent({
color: resolveSeriesColor(seriesItem.stroke, uPlotInstance, seriesIndex),
isActive: seriesIndex === activeSeriesIndex,
});
} else if (isSync) {
items.push({
label: String(seriesItem.label ?? ''),
value: 0,
tooltipValue: 'No Data',
color: resolveSeriesColor(seriesItem.stroke, uPlotInstance, seriesIndex),
isActive: false,
});
}
}

View File

@@ -58,6 +58,9 @@ export interface TooltipRenderArgs {
isPinned: boolean;
dismiss: () => void;
viaSync: boolean;
/** In Tooltip sync mode, limits which series are rendered in the receiver tooltip.
* null = no filtering; [] = no matches (tooltip hidden upstream); [...] = allowed indexes */
syncedSeriesIndexes?: number[] | null;
}
export interface BaseTooltipProps {

View File

@@ -9,6 +9,7 @@ import {
BarAlignment,
ConfigBuilder,
DrawStyle,
ExtendedSeries,
FillMode,
LineInterpolation,
LineStyle,
@@ -27,7 +28,10 @@ let builders: PathBuilders | null = null;
const DEFAULT_LINE_WIDTH = 2;
export const POINT_SIZE_FACTOR = 2.5;
export class UPlotSeriesBuilder extends ConfigBuilder<SeriesProps, Series> {
export class UPlotSeriesBuilder extends ConfigBuilder<
SeriesProps,
ExtendedSeries
> {
constructor(props: SeriesProps) {
super(props);
const pathBuilders = uPlot.paths;
@@ -205,8 +209,8 @@ export class UPlotSeriesBuilder extends ConfigBuilder<SeriesProps, Series> {
);
}
getConfig(): Series {
const { scaleKey, label, spanGaps, show = true } = this.props;
getConfig(): ExtendedSeries {
const { scaleKey, label, spanGaps, show = true, metric } = this.props;
const resolvedLineColor = this.getLineColor();
@@ -233,6 +237,7 @@ export class UPlotSeriesBuilder extends ConfigBuilder<SeriesProps, Series> {
...lineConfig,
...pathConfig,
points: Object.keys(pointsConfig).length > 0 ? pointsConfig : undefined,
metric,
};
}
}

View File

@@ -171,6 +171,10 @@ export enum FillMode {
None = 'none',
}
export type ExtendedSeries = Series & {
metric?: { [key: string]: string };
};
export interface SeriesProps extends LineConfig, PointsConfig, BarConfig {
scaleKey: string;
label?: string;
@@ -194,6 +198,7 @@ export interface SeriesProps extends LineConfig, PointsConfig, BarConfig {
fillMode?: FillMode;
isDarkMode?: boolean;
stepInterval?: number;
metric?: { [key: string]: string };
}
export interface LegendItem {

View File

@@ -4,7 +4,7 @@ import cx from 'classnames';
import uPlot from 'uplot';
import logEvent from 'api/common/logEvent';
import { syncCursorRegistry } from './syncCursorRegistry';
import { createSyncDisplayHook } from './syncDisplayHook';
import {
createInitialControllerState,
createSetCursorHandler,
@@ -107,32 +107,20 @@ export default function TooltipPlugin({
// Enable uPlot's built-in cursor sync when requested so that
// crosshair / tooltip can follow the dashboard-wide cursor.
let removeSyncDisplayHook: (() => void) | null = null;
if (syncMode !== DashboardCursorSync.None && config.scales[0]?.props.time) {
config.setCursor({
sync: { key: syncKey, scales: ['x', 'y'] },
sync: {
key: syncKey,
scales:
syncMode === DashboardCursorSync.Crosshair ? ['x', 'y'] : ['x', null],
},
});
// Show the horizontal crosshair only when the receiving panel shares
// the same y-axis unit as the source panel. When this panel is the
// source (cursor.event != null) the line is always shown and this
// panel's metadata is written to the registry so receivers can read it.
config.addHook('setCursor', (u: uPlot): void => {
const yCursorEl = u.root.querySelector<HTMLElement>('.u-cursor-y');
if (!yCursorEl) {
return;
}
if (u.cursor.event != null) {
// This panel is the source — publish metadata and always show line.
syncCursorRegistry.setMetadata(syncKey, syncMetadata);
yCursorEl.style.display = '';
} else {
// This panel is receiving sync — show only if units match.
const sourceMeta = syncCursorRegistry.getMetadata(syncKey);
yCursorEl.style.display =
sourceMeta?.yAxisUnit === syncMetadata?.yAxisUnit ? '' : 'none';
}
});
removeSyncDisplayHook = config.addHook(
'setCursor',
createSyncDisplayHook(syncKey, syncMetadata, controller),
);
}
// Dismiss the tooltip when the user clicks / presses a key
@@ -140,7 +128,12 @@ export default function TooltipPlugin({
const onOutsideInteraction = (event: Event): void => {
const target = event.target as Node;
if (!containerRef.current?.contains(target)) {
dismissTooltip();
// Don't dismiss if the click landed inside any other pinned tooltip.
const isInsideAnyPinnedTooltip =
(target as Element).closest?.('[data-pinned="true"]') != null;
if (!isInsideAnyPinnedTooltip) {
dismissTooltip();
}
}
};
@@ -206,6 +199,16 @@ export default function TooltipPlugin({
if (!controller.hoverActive || !plot) {
return null;
}
// In Tooltip sync mode, suppress the receiver tooltip entirely when
// no receiver series match the source panel's focused series.
if (
syncTooltipWithDashboard &&
controller.cursorDrivenBySync &&
Array.isArray(controller.syncedSeriesIndexes) &&
controller.syncedSeriesIndexes.length === 0
) {
return null;
}
return renderRef.current({
uPlotInstance: plot,
dataIndexes: controller.seriesIndexes,
@@ -213,6 +216,7 @@ export default function TooltipPlugin({
isPinned: controller.pinned,
dismiss: dismissTooltip,
viaSync: controller.cursorDrivenBySync,
syncedSeriesIndexes: controller.syncedSeriesIndexes,
});
}
@@ -443,6 +447,7 @@ export default function TooltipPlugin({
removeSetSeriesHook();
removeSetLegendHook();
removeSetCursorHook();
removeSyncDisplayHook?.();
if (overClickHandler) {
const plot = getPlot(controller);
plot?.over.removeEventListener('click', overClickHandler);
@@ -505,7 +510,7 @@ export default function TooltipPlugin({
isHovering,
contents,
]);
const isTooltipVisible = isHovering || tooltipBody != null;
const isTooltipVisible = tooltipBody != null;
if (!hasPlot) {
return null;

View File

@@ -9,9 +9,13 @@ import type { TooltipSyncMetadata } from './types';
*
* Receivers use this to make decisions such as:
* - Whether to show the horizontal crosshair line (matching yAxisUnit)
* - Future: what to render inside the tooltip (matching groupBy, etc.)
* - Which series to highlight when panels share the same groupBy
*/
const metadataBySyncKey = new Map<string, TooltipSyncMetadata | undefined>();
const activeSeriesMetricBySyncKey = new Map<
string,
Record<string, string> | null
>();
export const syncCursorRegistry = {
setMetadata(syncKey: string, metadata: TooltipSyncMetadata | undefined): void {
@@ -21,4 +25,15 @@ export const syncCursorRegistry = {
getMetadata(syncKey: string): TooltipSyncMetadata | undefined {
return metadataBySyncKey.get(syncKey);
},
setActiveSeriesMetric(
syncKey: string,
metric: Record<string, string> | null,
): void {
activeSeriesMetricBySyncKey.set(syncKey, metric);
},
getActiveSeriesMetric(syncKey: string): Record<string, string> | null {
return activeSeriesMetricBySyncKey.get(syncKey) ?? null;
},
};

View File

@@ -0,0 +1,185 @@
import uPlot from 'uplot';
import type { ExtendedSeries } from '../../config/types';
import { syncCursorRegistry } from './syncCursorRegistry';
import type { TooltipControllerState, TooltipSyncMetadata } from './types';
/**
* Returns the dimension keys present in both groupBy arrays.
* An empty result means no overlap — series highlighting should not run.
*
* exact [A, B] vs [A, B] → [A, B] one match
* subset [A] vs [A, B] → [A] multiple receiver series may match
* superset [A, B] vs [A] → [A] one receiver series matches
* partial [A, B] vs [B, C] → [B]
*/
function getCommonGroupByKeys(
a: TooltipSyncMetadata['groupBy'],
b: TooltipSyncMetadata['groupBy'],
): string[] {
if (
!Array.isArray(a) ||
a.length === 0 ||
!Array.isArray(b) ||
b.length === 0
) {
return [];
}
const bKeys = new Set(b.map((g) => g.key));
return a.filter((g) => bKeys.has(g.key)).map((g) => g.key);
}
/**
* Returns the 1-based indexes of every series whose metric matches
* sourceMetric on all commonKeys.
*/
function findMatchingSeriesIndexes(
series: uPlot.Series[],
sourceMetric: Record<string, string>,
commonKeys: string[],
): number[] {
return series.reduce<number[]>((acc, s, i) => {
if (i === 0) {
return acc;
}
const metric = (s as ExtendedSeries).metric;
if (
metric != null &&
commonKeys.every((key) => metric[key] === sourceMetric[key])
) {
acc.push(i);
}
return acc;
}, []);
}
function applySourceSync({
uPlotInstance,
syncKey,
syncMetadata,
focusedSeriesIndex,
}: {
uPlotInstance: uPlot;
syncKey: string;
syncMetadata: TooltipSyncMetadata | undefined;
focusedSeriesIndex: number | null;
}): void {
syncCursorRegistry.setMetadata(syncKey, syncMetadata);
const focusedSeries =
focusedSeriesIndex != null
? (uPlotInstance.series[focusedSeriesIndex] as ExtendedSeries)
: null;
syncCursorRegistry.setActiveSeriesMetric(
syncKey,
focusedSeries?.metric ?? null,
);
}
/**
* Returns:
* null no groupBy filtering configured or cursor off-chart (no-op for tooltip)
* [] groupBy configured but no receiver series match the source (hide synced tooltip)
* number[] 1-based indexes of matching receiver series (show only these)
*/
function applyReceiverSync({
uPlotInstance,
yCrosshairEl,
syncKey,
syncMetadata,
sourceMetadata,
commonKeys,
}: {
uPlotInstance: uPlot;
yCrosshairEl: HTMLElement;
syncKey: string;
syncMetadata: TooltipSyncMetadata | undefined;
sourceMetadata: TooltipSyncMetadata | undefined;
commonKeys: string[];
}): number[] | null {
yCrosshairEl.style.display =
sourceMetadata?.yAxisUnit === syncMetadata?.yAxisUnit ? '' : 'none';
if (commonKeys.length === 0) {
return null;
}
if ((uPlotInstance.cursor.left ?? -1) < 0) {
uPlotInstance.setSeries(null, { focus: false });
return null;
}
const sourceSeriesMetric = syncCursorRegistry.getActiveSeriesMetric(syncKey);
if (sourceSeriesMetric == null) {
uPlotInstance.setSeries(null, { focus: false });
return [];
}
const matchingIdxs = findMatchingSeriesIndexes(
uPlotInstance.series,
sourceSeriesMetric,
commonKeys,
);
if (matchingIdxs.length === 0) {
uPlotInstance.setSeries(null, { focus: false });
return [];
}
uPlotInstance.setSeries(matchingIdxs[0], { focus: true });
return matchingIdxs;
}
export function createSyncDisplayHook(
syncKey: string,
syncMetadata: TooltipSyncMetadata | undefined,
controller: TooltipControllerState,
): (u: uPlot) => void {
// Cached once — avoids a DOM query on every cursor move.
let yCrosshairEl: HTMLElement | null = null;
// groupBy on both panels is stable (set at config time). Recompute the
// intersection only when the source panel's groupBy reference changes.
let lastSourceGroupBy: TooltipSyncMetadata['groupBy'];
let cachedCommonKeys: string[] = [];
return (u: uPlot): void => {
yCrosshairEl ??= u.root.querySelector<HTMLElement>('.u-cursor-y');
if (!yCrosshairEl) {
return;
}
if (u.cursor.event != null) {
controller.syncedSeriesIndexes = null;
applySourceSync({
uPlotInstance: u,
syncKey,
syncMetadata,
focusedSeriesIndex: controller.focusedSeriesIndex,
});
yCrosshairEl.style.display = '';
return;
}
// Read metadata once and pass it down — avoids a second registry lookup
// inside applyReceiverSync.
const sourceMetadata = syncCursorRegistry.getMetadata(syncKey);
if (sourceMetadata?.groupBy !== lastSourceGroupBy) {
lastSourceGroupBy = sourceMetadata?.groupBy;
cachedCommonKeys = getCommonGroupByKeys(
sourceMetadata?.groupBy,
syncMetadata?.groupBy,
);
}
controller.syncedSeriesIndexes = applyReceiverSync({
uPlotInstance: u,
yCrosshairEl,
syncKey,
syncMetadata,
sourceMetadata,
commonKeys: cachedCommonKeys,
});
};
}

View File

@@ -27,6 +27,7 @@ export function createInitialControllerState(): TooltipControllerState {
verticalOffset: 0,
seriesIndexes: [],
focusedSeriesIndex: null,
syncedSeriesIndexes: null,
cursorDrivenBySync: false,
plotWithinViewport: false,
windowWidth: window.innerWidth - WINDOW_OFFSET,
@@ -184,7 +185,7 @@ export function createSetLegendHandler(
return;
}
const newSeriesIndexes = plot.cursor.idxs.slice();
const newSeriesIndexes = [...plot.cursor.idxs];
const isAnySeriesActive = newSeriesIndexes.some((v, i) => i > 0 && v != null);
const previousCursorDrivenBySync = controller.cursorDrivenBySync;

View File

@@ -4,6 +4,7 @@ import type {
ReactNode,
RefObject,
} from 'react';
import { BaseAutocompleteData } from 'types/api/queryBuilder/queryAutocompleteResponse';
import type uPlot from 'uplot';
import type { TooltipRenderArgs } from '../../components/types';
@@ -39,6 +40,7 @@ export interface TooltipLayoutInfo {
export interface TooltipSyncMetadata {
yAxisUnit?: string;
groupBy?: BaseAutocompleteData[];
}
export interface TooltipPluginProps {
@@ -95,6 +97,11 @@ export interface TooltipControllerState {
verticalOffset: number;
seriesIndexes: Array<number | null>;
focusedSeriesIndex: number | null;
/** Receiver-side series filtering for Tooltip sync mode.
* null = no filtering (source panel or no groupBy configured)
* [] = no matching series found → hide the synced tooltip
* [...] = only these 1-based series indexes should appear in the synced tooltip */
syncedSeriesIndexes: number[] | null;
cursorDrivenBySync: boolean;
plotWithinViewport: boolean;
windowWidth: number;

View File

@@ -210,7 +210,7 @@ describe('TooltipPlugin', () => {
expect(container.parentElement).toBe(document.body);
const fullscreenRoot = document.createElement('div');
document.body.appendChild(fullscreenRoot);
document.body.append(fullscreenRoot);
act(() => {
mockedFullscreenElement = fullscreenRoot;
@@ -252,7 +252,7 @@ describe('TooltipPlugin', () => {
renderAndActivateHover(config, undefined, { canPinTooltip: true });
const container = screen.getByTestId('tooltip-plugin-container');
expect(container.getAttribute('data-pinned') === 'true').toBe(false);
expect(container.dataset.pinned === 'true').toBe(false);
act(() => {
document.body.dispatchEvent(
@@ -266,7 +266,7 @@ describe('TooltipPlugin', () => {
return waitFor(() => {
const updated = screen.getByTestId('tooltip-plugin-container');
expect(updated).toBeInTheDocument();
expect(updated.getAttribute('data-pinned') === 'true').toBe(true);
expect(updated.dataset.pinned === 'true').toBe(true);
});
});
@@ -340,7 +340,7 @@ describe('TooltipPlugin', () => {
// Wait until the tooltip is actually pinned.
await waitFor(() => {
const container = screen.getByTestId('tooltip-plugin-container');
expect(container.getAttribute('data-pinned') === 'true').toBe(true);
expect(container.dataset.pinned === 'true').toBe(true);
});
const button = await screen.findByRole('button', { name: 'Dismiss' });
@@ -353,7 +353,7 @@ describe('TooltipPlugin', () => {
expect(container).toBeInTheDocument();
expect(container.getAttribute('aria-hidden')).toBe('true');
expect(container.getAttribute('data-pinned') === 'true').toBe(false);
expect(container.dataset.pinned === 'true').toBe(false);
expect(container.textContent).toBe('');
});
});
@@ -391,9 +391,7 @@ describe('TooltipPlugin', () => {
});
expect(
screen
.getByTestId('tooltip-plugin-container')
.getAttribute('data-pinned') === 'true',
screen.getByTestId('tooltip-plugin-container').dataset.pinned === 'true',
).toBe(true);
// Simulate data update should dismiss the pinned tooltip.
@@ -405,7 +403,7 @@ describe('TooltipPlugin', () => {
const container = screen.getByTestId('tooltip-plugin-container');
expect(container).toBeInTheDocument();
expect(container.getAttribute('aria-hidden')).toBe('true');
expect(container.getAttribute('data-pinned') === 'true').toBe(false);
expect(container.dataset.pinned === 'true').toBe(false);
jest.useRealTimers();
});
@@ -443,9 +441,7 @@ describe('TooltipPlugin', () => {
});
expect(
screen
.getByTestId('tooltip-plugin-container')
.getAttribute('data-pinned') === 'true',
screen.getByTestId('tooltip-plugin-container').dataset.pinned === 'true',
).toBe(true);
// Click outside the tooltip container.
@@ -459,7 +455,7 @@ describe('TooltipPlugin', () => {
expect(container).toBeInTheDocument();
expect(container.getAttribute('aria-hidden')).toBe('true');
expect(container.getAttribute('data-pinned') === 'true').toBe(false);
expect(container.dataset.pinned === 'true').toBe(false);
});
jest.useRealTimers();
@@ -498,9 +494,7 @@ describe('TooltipPlugin', () => {
});
expect(
screen
.getByTestId('tooltip-plugin-container')
.getAttribute('data-pinned') === 'true',
screen.getByTestId('tooltip-plugin-container').dataset.pinned === 'true',
).toBe(true);
// Press Escape to release.
@@ -515,7 +509,7 @@ describe('TooltipPlugin', () => {
const container = screen.getByTestId('tooltip-plugin-container');
expect(container).toBeInTheDocument();
expect(container.getAttribute('aria-hidden')).toBe('true');
expect(container.getAttribute('data-pinned') === 'true').toBe(false);
expect(container.dataset.pinned === 'true').toBe(false);
});
jest.useRealTimers();
@@ -541,9 +535,7 @@ describe('TooltipPlugin', () => {
await waitFor(() => {
expect(
screen
.getByTestId('tooltip-plugin-container')
.getAttribute('data-pinned') === 'true',
screen.getByTestId('tooltip-plugin-container').dataset.pinned === 'true',
).toBe(true);
});
@@ -560,7 +552,7 @@ describe('TooltipPlugin', () => {
await waitFor(() => {
const container = screen.getByTestId('tooltip-plugin-container');
expect(container.getAttribute('data-pinned') === 'true').toBe(false);
expect(container.dataset.pinned === 'true').toBe(false);
});
jest.useRealTimers();
@@ -580,7 +572,7 @@ describe('TooltipPlugin', () => {
const container = screen.getByTestId('tooltip-plugin-container');
// Tooltip should still be hovering (visible), not dismissed.
expect(container.getAttribute('aria-hidden')).toBe('false');
expect(container.getAttribute('data-pinned') === 'true').toBe(false);
expect(container.dataset.pinned === 'true').toBe(false);
});
it('does not unpin on arbitrary keys that are not Escape or the pin key', async () => {
@@ -603,9 +595,7 @@ describe('TooltipPlugin', () => {
await waitFor(() => {
expect(
screen
.getByTestId('tooltip-plugin-container')
.getAttribute('data-pinned') === 'true',
screen.getByTestId('tooltip-plugin-container').dataset.pinned === 'true',
).toBe(true);
});
@@ -619,9 +609,7 @@ describe('TooltipPlugin', () => {
await waitFor(() => {
expect(
screen
.getByTestId('tooltip-plugin-container')
.getAttribute('data-pinned') === 'true',
screen.getByTestId('tooltip-plugin-container').dataset.pinned === 'true',
).toBe(true);
});
@@ -665,7 +653,7 @@ describe('TooltipPlugin', () => {
});
const container = screen.getByTestId('tooltip-plugin-container');
expect(container.getAttribute('data-pinned') === 'true').toBe(false);
expect(container.dataset.pinned === 'true').toBe(false);
});
it('does not pin when hover is not active', () => {
@@ -699,7 +687,7 @@ describe('TooltipPlugin', () => {
// The container exists once the plot is initialised, but it should
// be hidden and not pinned since hover was never activated.
const container = screen.getByTestId('tooltip-plugin-container');
expect(container.getAttribute('data-pinned') === 'true').toBe(false);
expect(container.dataset.pinned === 'true').toBe(false);
expect(container.getAttribute('aria-hidden')).toBe('true');
});
@@ -723,9 +711,7 @@ describe('TooltipPlugin', () => {
await waitFor(() => {
expect(
screen
.getByTestId('tooltip-plugin-container')
.getAttribute('data-pinned') === 'true',
screen.getByTestId('tooltip-plugin-container').dataset.pinned === 'true',
).toBe(false);
});
@@ -738,9 +724,7 @@ describe('TooltipPlugin', () => {
await waitFor(() => {
expect(
screen
.getByTestId('tooltip-plugin-container')
.getAttribute('data-pinned') === 'true',
screen.getByTestId('tooltip-plugin-container').dataset.pinned === 'true',
).toBe(true);
});
});
@@ -796,7 +780,7 @@ describe('TooltipPlugin', () => {
// ---- Cursor sync ------------------------------------------------------------
describe('cursor sync', () => {
it('enables uPlot cursor sync for time-based scales when mode is Tooltip', () => {
it('enables uPlot cursor sync on x-axis only when mode is Tooltip', () => {
const config = createConfigMock();
const setCursorSpy = jest.spyOn(config, 'setCursor');
config.addScale({ scaleKey: 'x', time: true });
@@ -810,6 +794,25 @@ describe('TooltipPlugin', () => {
}),
);
expect(setCursorSpy).toHaveBeenCalledWith({
sync: { key: 'dashboard-sync', scales: ['x', null] },
});
});
it('enables uPlot cursor sync on both axes when mode is Crosshair', () => {
const config = createConfigMock();
const setCursorSpy = jest.spyOn(config, 'setCursor');
config.addScale({ scaleKey: 'x', time: true });
render(
React.createElement(TooltipPlugin, {
config,
render: () => null,
syncMode: DashboardCursorSync.Crosshair,
syncKey: 'dashboard-sync',
}),
);
expect(setCursorSpy).toHaveBeenCalledWith({
sync: { key: 'dashboard-sync', scales: ['x', 'y'] },
});

View File

@@ -48,5 +48,43 @@ func (provider *provider) addInfraMonitoringRoutes(router *mux.Router) error {
return err
}
if err := router.Handle("/api/v2/infra_monitoring/nodes", handler.New(
provider.authZ.ViewAccess(provider.infraMonitoringHandler.ListNodes),
handler.OpenAPIDef{
ID: "ListNodes",
Tags: []string{"inframonitoring"},
Summary: "List Nodes for Infra Monitoring",
Description: "Returns a paginated list of Kubernetes nodes with key metrics: CPU usage, CPU allocatable, memory working set, memory allocatable, and per-group readyNodesCount / notReadyNodesCount derived from each node's latest k8s.node.condition_ready value in the window. Each node includes metadata attributes (k8s.node.uid, k8s.cluster.name). The response type is 'list' for the default k8s.node.name grouping (each row is one node with its current condition string: ready / not_ready / '') or 'grouped_list' for custom groupBy keys (each row aggregates nodes in the group with readyNodesCount and notReadyNodesCount; condition stays empty). Supports filtering via a filter expression, custom groupBy, ordering by cpu / cpu_allocatable / memory / memory_allocatable, and pagination via offset/limit. Also reports missing required metrics and whether the requested time range falls before the data retention boundary. Numeric metric fields (nodeCPU, nodeCPUAllocatable, nodeMemory, nodeMemoryAllocatable) return -1 as a sentinel when no data is available for that field.",
Request: new(inframonitoringtypes.PostableNodes),
RequestContentType: "application/json",
Response: new(inframonitoringtypes.Nodes),
ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{http.StatusBadRequest, http.StatusUnauthorized},
Deprecated: false,
SecuritySchemes: newSecuritySchemes(types.RoleViewer),
})).Methods(http.MethodPost).GetError(); err != nil {
return err
}
if err := router.Handle("/api/v2/infra_monitoring/namespaces", handler.New(
provider.authZ.ViewAccess(provider.infraMonitoringHandler.ListNamespaces),
handler.OpenAPIDef{
ID: "ListNamespaces",
Tags: []string{"inframonitoring"},
Summary: "List Namespaces for Infra Monitoring",
Description: "Returns a paginated list of Kubernetes namespaces with key aggregated pod metrics: CPU usage and memory working set (summed across pods in the group), plus per-group pod counts bucketed by each pod's latest k8s.pod.phase value in the window (pendingPodCount, runningPodCount, succeededPodCount, failedPodCount, unknownPodCount). Each namespace includes metadata attributes (k8s.namespace.name, k8s.cluster.name). The response type is 'list' for the default k8s.namespace.name grouping or 'grouped_list' for custom groupBy keys; in both modes every row aggregates pods in the group. Supports filtering via a filter expression, custom groupBy, ordering by cpu / memory, and pagination via offset/limit. Also reports missing required metrics and whether the requested time range falls before the data retention boundary. Numeric metric fields (namespaceCPU, namespaceMemory) return -1 as a sentinel when no data is available for that field.",
Request: new(inframonitoringtypes.PostableNamespaces),
RequestContentType: "application/json",
Response: new(inframonitoringtypes.Namespaces),
ResponseContentType: "application/json",
SuccessStatusCode: http.StatusOK,
ErrorStatusCodes: []int{http.StatusBadRequest, http.StatusUnauthorized},
Deprecated: false,
SecuritySchemes: newSecuritySchemes(types.RoleViewer),
})).Methods(http.MethodPost).GetError(); err != nil {
return err
}
return nil
}

View File

@@ -69,3 +69,51 @@ func (h *handler) ListPods(rw http.ResponseWriter, req *http.Request) {
render.Success(rw, http.StatusOK, result)
}
func (h *handler) ListNodes(rw http.ResponseWriter, req *http.Request) {
claims, err := authtypes.ClaimsFromContext(req.Context())
if err != nil {
render.Error(rw, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
var parsedReq inframonitoringtypes.PostableNodes
if err := binding.JSON.BindBody(req.Body, &parsedReq); err != nil {
render.Error(rw, err)
return
}
result, err := h.module.ListNodes(req.Context(), orgID, &parsedReq)
if err != nil {
render.Error(rw, err)
return
}
render.Success(rw, http.StatusOK, result)
}
func (h *handler) ListNamespaces(rw http.ResponseWriter, req *http.Request) {
claims, err := authtypes.ClaimsFromContext(req.Context())
if err != nil {
render.Error(rw, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
var parsedReq inframonitoringtypes.PostableNamespaces
if err := binding.JSON.BindBody(req.Body, &parsedReq); err != nil {
render.Error(rw, err)
return
}
result, err := h.module.ListNamespaces(req.Context(), orgID, &parsedReq)
if err != nil {
render.Error(rw, err)
return
}
render.Success(rw, http.StatusOK, result)
}

View File

@@ -23,3 +23,9 @@ type podPhaseCounts struct {
Failed int
Unknown int
}
// nodeConditionCounts holds per-group node counts bucketed by latest condition_ready in window.
type nodeConditionCounts struct {
Ready int
NotReady int
}

View File

@@ -242,3 +242,175 @@ func (m *module) ListPods(ctx context.Context, orgID valuer.UUID, req *inframoni
return resp, nil
}
func (m *module) ListNodes(ctx context.Context, orgID valuer.UUID, req *inframonitoringtypes.PostableNodes) (*inframonitoringtypes.Nodes, error) {
if err := req.Validate(); err != nil {
return nil, err
}
resp := &inframonitoringtypes.Nodes{}
if req.OrderBy == nil {
req.OrderBy = &qbtypes.OrderBy{
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: inframonitoringtypes.NodesOrderByCPU,
},
},
Direction: qbtypes.OrderDirectionDesc,
}
}
if len(req.GroupBy) == 0 {
req.GroupBy = []qbtypes.GroupByKey{nodeNameGroupByKey}
resp.Type = inframonitoringtypes.ResponseTypeList
} else {
resp.Type = inframonitoringtypes.ResponseTypeGroupedList
}
missingMetrics, minFirstReportedUnixMilli, err := m.getMetricsExistenceAndEarliestTime(ctx, nodesTableMetricNamesList)
if err != nil {
return nil, err
}
if len(missingMetrics) > 0 {
resp.RequiredMetricsCheck = inframonitoringtypes.RequiredMetricsCheck{MissingMetrics: missingMetrics}
resp.Records = []inframonitoringtypes.NodeRecord{}
resp.Total = 0
return resp, nil
}
if req.End < int64(minFirstReportedUnixMilli) {
resp.EndTimeBeforeRetention = true
resp.Records = []inframonitoringtypes.NodeRecord{}
resp.Total = 0
return resp, nil
}
resp.RequiredMetricsCheck = inframonitoringtypes.RequiredMetricsCheck{MissingMetrics: []string{}}
metadataMap, err := m.getNodesTableMetadata(ctx, req)
if err != nil {
return nil, err
}
resp.Total = len(metadataMap)
pageGroups, err := m.getTopNodeGroups(ctx, orgID, req, metadataMap)
if err != nil {
return nil, err
}
if len(pageGroups) == 0 {
resp.Records = []inframonitoringtypes.NodeRecord{}
return resp, nil
}
filterExpr := ""
if req.Filter != nil {
filterExpr = req.Filter.Expression
}
fullQueryReq := buildFullQueryRequest(req.Start, req.End, filterExpr, req.GroupBy, pageGroups, m.newNodesTableListQuery())
queryResp, err := m.querier.QueryRange(ctx, orgID, fullQueryReq)
if err != nil {
return nil, err
}
conditionCounts, err := m.getPerGroupNodeConditionCounts(ctx, req, pageGroups)
if err != nil {
return nil, err
}
isNodeNameInGroupBy := isKeyInGroupByAttrs(req.GroupBy, nodeNameAttrKey)
resp.Records = buildNodeRecords(isNodeNameInGroupBy, queryResp, pageGroups, req.GroupBy, metadataMap, conditionCounts)
resp.Warning = queryResp.Warning
return resp, nil
}
func (m *module) ListNamespaces(ctx context.Context, orgID valuer.UUID, req *inframonitoringtypes.PostableNamespaces) (*inframonitoringtypes.Namespaces, error) {
if err := req.Validate(); err != nil {
return nil, err
}
resp := &inframonitoringtypes.Namespaces{}
if req.OrderBy == nil {
req.OrderBy = &qbtypes.OrderBy{
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: inframonitoringtypes.NamespacesOrderByCPU,
},
},
Direction: qbtypes.OrderDirectionDesc,
}
}
if len(req.GroupBy) == 0 {
req.GroupBy = []qbtypes.GroupByKey{namespaceNameGroupByKey}
resp.Type = inframonitoringtypes.ResponseTypeList
} else {
resp.Type = inframonitoringtypes.ResponseTypeGroupedList
}
missingMetrics, minFirstReportedUnixMilli, err := m.getMetricsExistenceAndEarliestTime(ctx, namespacesTableMetricNamesList)
if err != nil {
return nil, err
}
if len(missingMetrics) > 0 {
resp.RequiredMetricsCheck = inframonitoringtypes.RequiredMetricsCheck{MissingMetrics: missingMetrics}
resp.Records = []inframonitoringtypes.NamespaceRecord{}
resp.Total = 0
return resp, nil
}
if req.End < int64(minFirstReportedUnixMilli) {
resp.EndTimeBeforeRetention = true
resp.Records = []inframonitoringtypes.NamespaceRecord{}
resp.Total = 0
return resp, nil
}
resp.RequiredMetricsCheck = inframonitoringtypes.RequiredMetricsCheck{MissingMetrics: []string{}}
metadataMap, err := m.getNamespacesTableMetadata(ctx, req)
if err != nil {
return nil, err
}
resp.Total = len(metadataMap)
pageGroups, err := m.getTopNamespaceGroups(ctx, orgID, req, metadataMap)
if err != nil {
return nil, err
}
if len(pageGroups) == 0 {
resp.Records = []inframonitoringtypes.NamespaceRecord{}
return resp, nil
}
filterExpr := ""
if req.Filter != nil {
filterExpr = req.Filter.Expression
}
fullQueryReq := buildFullQueryRequest(req.Start, req.End, filterExpr, req.GroupBy, pageGroups, m.newNamespacesTableListQuery())
queryResp, err := m.querier.QueryRange(ctx, orgID, fullQueryReq)
if err != nil {
return nil, err
}
// Reuse the pods phase-counts CTE function via a temp struct — it reads only
// Start/End/Filter/GroupBy from PostablePods.
phaseCounts, err := m.getPerGroupPodPhaseCounts(ctx, &inframonitoringtypes.PostablePods{
Start: req.Start,
End: req.End,
Filter: req.Filter,
GroupBy: req.GroupBy,
}, pageGroups)
if err != nil {
return nil, err
}
resp.Records = buildNamespaceRecords(queryResp, pageGroups, req.GroupBy, metadataMap, phaseCounts)
resp.Warning = queryResp.Warning
return resp, nil
}

View File

@@ -0,0 +1,121 @@
package implinframonitoring
import (
"context"
"slices"
"github.com/SigNoz/signoz/pkg/types/inframonitoringtypes"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/valuer"
)
// buildNamespaceRecords assembles the page records. Pod phase counts come from
// phaseCounts in both modes; every row is a group of pods, so there's no
// per-row "current phase" concept (unlike pods/nodes list mode).
func buildNamespaceRecords(
resp *qbtypes.QueryRangeResponse,
pageGroups []map[string]string,
groupBy []qbtypes.GroupByKey,
metadataMap map[string]map[string]string,
phaseCounts map[string]podPhaseCounts,
) []inframonitoringtypes.NamespaceRecord {
metricsMap := parseFullQueryResponse(resp, groupBy)
records := make([]inframonitoringtypes.NamespaceRecord, 0, len(pageGroups))
for _, labels := range pageGroups {
compositeKey := compositeKeyFromLabels(labels, groupBy)
namespaceName := labels[namespaceNameAttrKey]
record := inframonitoringtypes.NamespaceRecord{ // initialize with default values
NamespaceName: namespaceName,
NamespaceCPU: -1,
NamespaceMemory: -1,
Meta: map[string]any{},
}
if metrics, ok := metricsMap[compositeKey]; ok {
if v, exists := metrics["A"]; exists {
record.NamespaceCPU = v
}
if v, exists := metrics["D"]; exists {
record.NamespaceMemory = v
}
}
if phaseCountsForGroup, ok := phaseCounts[compositeKey]; ok {
record.PendingPodCount = phaseCountsForGroup.Pending
record.RunningPodCount = phaseCountsForGroup.Running
record.SucceededPodCount = phaseCountsForGroup.Succeeded
record.FailedPodCount = phaseCountsForGroup.Failed
record.UnknownPodCount = phaseCountsForGroup.Unknown
}
if attrs, ok := metadataMap[compositeKey]; ok {
for k, v := range attrs {
record.Meta[k] = v
}
}
records = append(records, record)
}
return records
}
func (m *module) getTopNamespaceGroups(
ctx context.Context,
orgID valuer.UUID,
req *inframonitoringtypes.PostableNamespaces,
metadataMap map[string]map[string]string,
) ([]map[string]string, error) {
orderByKey := req.OrderBy.Key.Name
queryNamesForOrderBy := orderByToNamespacesQueryNames[orderByKey]
rankingQueryName := queryNamesForOrderBy[len(queryNamesForOrderBy)-1]
topReq := &qbtypes.QueryRangeRequest{
Start: uint64(req.Start),
End: uint64(req.End),
RequestType: qbtypes.RequestTypeScalar,
CompositeQuery: qbtypes.CompositeQuery{
Queries: make([]qbtypes.QueryEnvelope, 0, len(queryNamesForOrderBy)),
},
}
for _, envelope := range m.newNamespacesTableListQuery().CompositeQuery.Queries {
if !slices.Contains(queryNamesForOrderBy, envelope.GetQueryName()) {
continue
}
copied := envelope
if copied.Type == qbtypes.QueryTypeBuilder {
existingExpr := ""
if f := copied.GetFilter(); f != nil {
existingExpr = f.Expression
}
reqFilterExpr := ""
if req.Filter != nil {
reqFilterExpr = req.Filter.Expression
}
merged := mergeFilterExpressions(existingExpr, reqFilterExpr)
copied.SetFilter(&qbtypes.Filter{Expression: merged})
copied.SetGroupBy(req.GroupBy)
}
topReq.CompositeQuery.Queries = append(topReq.CompositeQuery.Queries, copied)
}
resp, err := m.querier.QueryRange(ctx, orgID, topReq)
if err != nil {
return nil, err
}
allMetricGroups := parseAndSortGroups(resp, rankingQueryName, req.GroupBy, req.OrderBy.Direction)
return paginateWithBackfill(allMetricGroups, metadataMap, req.GroupBy, req.Offset, req.Limit), nil
}
func (m *module) getNamespacesTableMetadata(ctx context.Context, req *inframonitoringtypes.PostableNamespaces) (map[string]map[string]string, error) {
var nonGroupByAttrs []string
for _, key := range namespaceAttrKeysForMetadata {
if !isKeyInGroupByAttrs(req.GroupBy, key) {
nonGroupByAttrs = append(nonGroupByAttrs, key)
}
}
return m.getMetadata(ctx, namespacesTableMetricNamesList, req.GroupBy, nonGroupByAttrs, req.Filter, req.Start, req.End)
}

View File

@@ -0,0 +1,92 @@
package implinframonitoring
import (
"github.com/SigNoz/signoz/pkg/types/inframonitoringtypes"
"github.com/SigNoz/signoz/pkg/types/metrictypes"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
)
const (
namespaceNameAttrKey = "k8s.namespace.name"
)
var namespaceNameGroupByKey = qbtypes.GroupByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: namespaceNameAttrKey,
FieldContext: telemetrytypes.FieldContextResource,
FieldDataType: telemetrytypes.FieldDataTypeString,
},
}
// namespacesTableMetricNamesList drives the existence/retention check.
// Includes k8s.pod.phase so the response short-circuits cleanly when a
// cluster doesn't ship the metric — even though phase isn't part of the
// QB composite query (it's queried separately via getPerGroupPodPhaseCounts).
var namespacesTableMetricNamesList = []string{
"k8s.pod.cpu.usage",
"k8s.pod.memory.working_set",
"k8s.pod.phase",
}
var namespaceAttrKeysForMetadata = []string{
"k8s.namespace.name",
"k8s.cluster.name",
}
var orderByToNamespacesQueryNames = map[string][]string{
inframonitoringtypes.NamespacesOrderByCPU: {"A"},
inframonitoringtypes.NamespacesOrderByMemory: {"D"},
}
// newNamespacesTableListQuery builds the composite QB v5 request for the namespaces list.
// Pod phase counts are derived separately via getPerGroupPodPhaseCounts (works for both
// list and grouped_list modes), so no phase query is included here.
// Query letters A and D are kept aligned with the v1 implementation.
func (m *module) newNamespacesTableListQuery() *qbtypes.QueryRangeRequest {
queries := []qbtypes.QueryEnvelope{
// Query A: CPU usage — sum of pod CPU within the group.
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
Name: "A",
Signal: telemetrytypes.SignalMetrics,
Aggregations: []qbtypes.MetricAggregation{
{
MetricName: "k8s.pod.cpu.usage",
TimeAggregation: metrictypes.TimeAggregationAvg,
SpaceAggregation: metrictypes.SpaceAggregationSum,
ReduceTo: qbtypes.ReduceToAvg,
},
},
GroupBy: []qbtypes.GroupByKey{namespaceNameGroupByKey},
Disabled: false,
},
},
// Query D: Memory working set — sum of pod memory within the group.
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
Name: "D",
Signal: telemetrytypes.SignalMetrics,
Aggregations: []qbtypes.MetricAggregation{
{
MetricName: "k8s.pod.memory.working_set",
TimeAggregation: metrictypes.TimeAggregationAvg,
SpaceAggregation: metrictypes.SpaceAggregationSum,
ReduceTo: qbtypes.ReduceToAvg,
},
},
GroupBy: []qbtypes.GroupByKey{namespaceNameGroupByKey},
Disabled: false,
},
},
}
return &qbtypes.QueryRangeRequest{
RequestType: qbtypes.RequestTypeScalar,
CompositeQuery: qbtypes.CompositeQuery{
Queries: queries,
},
}
}

View File

@@ -0,0 +1,299 @@
package implinframonitoring
import (
"context"
"fmt"
"slices"
"strings"
"github.com/SigNoz/signoz/pkg/querybuilder"
"github.com/SigNoz/signoz/pkg/telemetrymetrics"
"github.com/SigNoz/signoz/pkg/types/inframonitoringtypes"
"github.com/SigNoz/signoz/pkg/types/metrictypes"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/valuer"
"github.com/huandu/go-sqlbuilder"
)
// buildNodeRecords assembles the page records. Condition counts come from
// conditionCounts in both modes. In list mode (isNodeNameInGroupBy=true) each
// group is one node, so exactly one count is 1; Condition is derived from
// which one. In grouped_list mode Condition stays NodeConditionNone.
func buildNodeRecords(
isNodeNameInGroupBy bool,
resp *qbtypes.QueryRangeResponse,
pageGroups []map[string]string,
groupBy []qbtypes.GroupByKey,
metadataMap map[string]map[string]string,
conditionCounts map[string]nodeConditionCounts,
) []inframonitoringtypes.NodeRecord {
metricsMap := parseFullQueryResponse(resp, groupBy)
records := make([]inframonitoringtypes.NodeRecord, 0, len(pageGroups))
for _, labels := range pageGroups {
compositeKey := compositeKeyFromLabels(labels, groupBy)
nodeName := labels[nodeNameAttrKey]
record := inframonitoringtypes.NodeRecord{ // initialize with default values
NodeName: nodeName,
Condition: inframonitoringtypes.NodeConditionNone,
NodeCPU: -1,
NodeCPUAllocatable: -1,
NodeMemory: -1,
NodeMemoryAllocatable: -1,
Meta: map[string]any{},
}
if metrics, ok := metricsMap[compositeKey]; ok {
if v, exists := metrics["A"]; exists {
record.NodeCPU = v
}
if v, exists := metrics["B"]; exists {
record.NodeCPUAllocatable = v
}
if v, exists := metrics["C"]; exists {
record.NodeMemory = v
}
if v, exists := metrics["D"]; exists {
record.NodeMemoryAllocatable = v
}
}
if conditionCountsForGroup, ok := conditionCounts[compositeKey]; ok {
record.ReadyNodesCount = conditionCountsForGroup.Ready
record.NotReadyNodesCount = conditionCountsForGroup.NotReady
// In list mode each group is one node; the count==1 bucket identifies the condition.
if isNodeNameInGroupBy {
switch {
case conditionCountsForGroup.Ready == 1:
record.Condition = inframonitoringtypes.NodeConditionReady
case conditionCountsForGroup.NotReady == 1:
record.Condition = inframonitoringtypes.NodeConditionNotReady
}
}
}
if attrs, ok := metadataMap[compositeKey]; ok {
for k, v := range attrs {
record.Meta[k] = v
}
}
records = append(records, record)
}
return records
}
func (m *module) getTopNodeGroups(
ctx context.Context,
orgID valuer.UUID,
req *inframonitoringtypes.PostableNodes,
metadataMap map[string]map[string]string,
) ([]map[string]string, error) {
orderByKey := req.OrderBy.Key.Name
queryNamesForOrderBy := orderByToNodesQueryNames[orderByKey]
rankingQueryName := queryNamesForOrderBy[len(queryNamesForOrderBy)-1]
topReq := &qbtypes.QueryRangeRequest{
Start: uint64(req.Start),
End: uint64(req.End),
RequestType: qbtypes.RequestTypeScalar,
CompositeQuery: qbtypes.CompositeQuery{
Queries: make([]qbtypes.QueryEnvelope, 0, len(queryNamesForOrderBy)),
},
}
for _, envelope := range m.newNodesTableListQuery().CompositeQuery.Queries {
if !slices.Contains(queryNamesForOrderBy, envelope.GetQueryName()) {
continue
}
copied := envelope
if copied.Type == qbtypes.QueryTypeBuilder {
existingExpr := ""
if f := copied.GetFilter(); f != nil {
existingExpr = f.Expression
}
reqFilterExpr := ""
if req.Filter != nil {
reqFilterExpr = req.Filter.Expression
}
merged := mergeFilterExpressions(existingExpr, reqFilterExpr)
copied.SetFilter(&qbtypes.Filter{Expression: merged})
copied.SetGroupBy(req.GroupBy)
}
topReq.CompositeQuery.Queries = append(topReq.CompositeQuery.Queries, copied)
}
resp, err := m.querier.QueryRange(ctx, orgID, topReq)
if err != nil {
return nil, err
}
allMetricGroups := parseAndSortGroups(resp, rankingQueryName, req.GroupBy, req.OrderBy.Direction)
return paginateWithBackfill(allMetricGroups, metadataMap, req.GroupBy, req.Offset, req.Limit), nil
}
func (m *module) getNodesTableMetadata(ctx context.Context, req *inframonitoringtypes.PostableNodes) (map[string]map[string]string, error) {
var nonGroupByAttrs []string
for _, key := range nodeAttrKeysForMetadata {
if !isKeyInGroupByAttrs(req.GroupBy, key) {
nonGroupByAttrs = append(nonGroupByAttrs, key)
}
}
return m.getMetadata(ctx, nodesTableMetricNamesList, req.GroupBy, nonGroupByAttrs, req.Filter, req.Start, req.End)
}
// getPerGroupNodeConditionCounts computes per-group node counts bucketed by each
// node's latest condition_ready value (0 / 1) in the requested window.
// Pipeline:
//
// timeSeriesFPs: fp ↔ (node_name, groupBy cols) from the time_series table.
// User filter + page-groups filter applied here.
// latestConditionPerNode: INNER JOIN samples × timeSeriesFPs, collapsed to
// the latest condition value per node via argMax(value, unix_milli).
// countNodesPerCondition: per-group uniqExactIf into ready/not_ready buckets.
//
// Groups absent from the result map have implicit zero counts (caller default).
func (m *module) getPerGroupNodeConditionCounts(
ctx context.Context,
req *inframonitoringtypes.PostableNodes,
pageGroups []map[string]string,
) (map[string]nodeConditionCounts, error) {
if len(pageGroups) == 0 || len(req.GroupBy) == 0 {
return map[string]nodeConditionCounts{}, nil
}
// Merged filter expression (user filter + page-groups IN clauses).
reqFilterExpr := ""
if req.Filter != nil {
reqFilterExpr = req.Filter.Expression
}
pageGroupsFilterExpr := buildPageGroupsFilterExpr(pageGroups)
filterExpr := mergeFilterExpressions(reqFilterExpr, pageGroupsFilterExpr)
// Resolve tables. Same convention as pods.
adjustedStart, adjustedEnd, _, localTimeSeriesTable := telemetrymetrics.WhichTSTableToUse(
uint64(req.Start), uint64(req.End), nil,
)
samplesTable := telemetrymetrics.WhichSamplesTableToUse(
uint64(req.Start), uint64(req.End),
metrictypes.UnspecifiedType, metrictypes.TimeAggregationUnspecified, nil,
)
valueCol := telemetrymetrics.ValueColumnForSamplesTable(samplesTable)
// ----- timeSeriesFPs -----
timeSeriesFPs := sqlbuilder.NewSelectBuilder()
timeSeriesFPsSelectCols := []string{
"fingerprint",
fmt.Sprintf("JSONExtractString(labels, %s) AS node_name", timeSeriesFPs.Var(nodeNameAttrKey)),
}
for _, key := range req.GroupBy {
timeSeriesFPsSelectCols = append(timeSeriesFPsSelectCols,
fmt.Sprintf("JSONExtractString(labels, %s) AS %s", timeSeriesFPs.Var(key.Name), quoteIdentifier(key.Name)),
)
}
timeSeriesFPs.Select(timeSeriesFPsSelectCols...)
timeSeriesFPs.From(fmt.Sprintf("%s.%s", telemetrymetrics.DBName, localTimeSeriesTable))
timeSeriesFPs.Where(
timeSeriesFPs.E("metric_name", nodeConditionMetricName),
timeSeriesFPs.GE("unix_milli", adjustedStart),
timeSeriesFPs.L("unix_milli", adjustedEnd),
)
if filterExpr != "" {
filterClause, err := m.buildFilterClause(ctx, &qbtypes.Filter{Expression: filterExpr}, req.Start, req.End)
if err != nil {
return nil, err
}
if filterClause != nil {
timeSeriesFPs.AddWhereClause(filterClause)
}
}
timeSeriesFPsGroupBy := []string{"fingerprint", "node_name"}
for _, key := range req.GroupBy {
timeSeriesFPsGroupBy = append(timeSeriesFPsGroupBy, quoteIdentifier(key.Name))
}
timeSeriesFPs.GroupBy(timeSeriesFPsGroupBy...)
timeSeriesFPsSQL, timeSeriesFPsArgs := timeSeriesFPs.BuildWithFlavor(sqlbuilder.ClickHouse)
// ----- latestConditionPerNode -----
latestConditionPerNode := sqlbuilder.NewSelectBuilder()
latestConditionPerNodeSelectCols := []string{"tsfp.node_name AS node_name"}
latestConditionPerNodeGroupBy := []string{"node_name"}
for _, key := range req.GroupBy {
col := quoteIdentifier(key.Name)
latestConditionPerNodeSelectCols = append(latestConditionPerNodeSelectCols, fmt.Sprintf("tsfp.%s AS %s", col, col))
latestConditionPerNodeGroupBy = append(latestConditionPerNodeGroupBy, col)
}
latestConditionPerNodeSelectCols = append(latestConditionPerNodeSelectCols,
fmt.Sprintf("argMax(samples.%s, samples.unix_milli) AS condition_value", valueCol),
)
latestConditionPerNode.Select(latestConditionPerNodeSelectCols...)
latestConditionPerNode.From(fmt.Sprintf(
"%s.%s AS samples INNER JOIN time_series_fps AS tsfp ON samples.fingerprint = tsfp.fingerprint",
telemetrymetrics.DBName, samplesTable,
))
latestConditionPerNode.Where(
latestConditionPerNode.E("samples.metric_name", nodeConditionMetricName),
latestConditionPerNode.GE("samples.unix_milli", req.Start),
latestConditionPerNode.L("samples.unix_milli", req.End),
"tsfp.node_name != ''",
)
latestConditionPerNode.GroupBy(latestConditionPerNodeGroupBy...)
latestConditionPerNodeSQL, latestConditionPerNodeArgs := latestConditionPerNode.BuildWithFlavor(sqlbuilder.ClickHouse)
// ----- countNodesPerCondition (outer SELECT) -----
countNodesPerConditionSelectCols := make([]string, 0, len(req.GroupBy)+2)
countNodesPerConditionGroupBy := make([]string, 0, len(req.GroupBy))
for _, key := range req.GroupBy {
col := quoteIdentifier(key.Name)
countNodesPerConditionSelectCols = append(countNodesPerConditionSelectCols, col)
countNodesPerConditionGroupBy = append(countNodesPerConditionGroupBy, col)
}
countNodesPerConditionSelectCols = append(countNodesPerConditionSelectCols,
fmt.Sprintf("uniqExactIf(node_name, condition_value = %d) AS ready_count", inframonitoringtypes.NodeConditionNumReady),
fmt.Sprintf("uniqExactIf(node_name, condition_value = %d) AS not_ready_count", inframonitoringtypes.NodeConditionNumNotReady),
)
countNodesPerConditionSQL := fmt.Sprintf(
"SELECT %s FROM latest_condition_per_node GROUP BY %s",
strings.Join(countNodesPerConditionSelectCols, ", "),
strings.Join(countNodesPerConditionGroupBy, ", "),
)
// Combine CTEs + outer.
cteFragments := []string{
fmt.Sprintf("time_series_fps AS (%s)", timeSeriesFPsSQL),
fmt.Sprintf("latest_condition_per_node AS (%s)", latestConditionPerNodeSQL),
}
finalSQL := querybuilder.CombineCTEs(cteFragments) + countNodesPerConditionSQL
finalArgs := querybuilder.PrependArgs([][]any{timeSeriesFPsArgs, latestConditionPerNodeArgs}, nil)
rows, err := m.telemetryStore.ClickhouseDB().Query(ctx, finalSQL, finalArgs...)
if err != nil {
return nil, err
}
defer rows.Close()
result := make(map[string]nodeConditionCounts)
for rows.Next() {
groupVals := make([]string, len(req.GroupBy))
scanPtrs := make([]any, 0, len(req.GroupBy)+2)
for i := range groupVals {
scanPtrs = append(scanPtrs, &groupVals[i])
}
var ready, notReady uint64
scanPtrs = append(scanPtrs, &ready, &notReady)
if err := rows.Scan(scanPtrs...); err != nil {
return nil, err
}
result[compositeKeyFromList(groupVals)] = nodeConditionCounts{
Ready: int(ready),
NotReady: int(notReady),
}
}
if err := rows.Err(); err != nil {
return nil, err
}
return result, nil
}

View File

@@ -0,0 +1,134 @@
package implinframonitoring
import (
"github.com/SigNoz/signoz/pkg/types/inframonitoringtypes"
"github.com/SigNoz/signoz/pkg/types/metrictypes"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
)
const (
nodeNameAttrKey = "k8s.node.name"
nodeConditionMetricName = "k8s.node.condition_ready"
)
var nodeNameGroupByKey = qbtypes.GroupByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: nodeNameAttrKey,
FieldContext: telemetrytypes.FieldContextResource,
FieldDataType: telemetrytypes.FieldDataTypeString,
},
}
// nodesTableMetricNamesList drives the existence/retention check.
// Includes condition_ready so the response short-circuits cleanly when a
// cluster doesn't ship the metric — even though condition_ready isn't part
// of the QB composite query (it's queried separately via getPerGroupNodeConditionCounts).
var nodesTableMetricNamesList = []string{
"k8s.node.cpu.usage",
"k8s.node.allocatable_cpu",
"k8s.node.memory.working_set",
"k8s.node.allocatable_memory",
"k8s.node.condition_ready",
}
var nodeAttrKeysForMetadata = []string{
"k8s.node.uid",
"k8s.cluster.name",
}
var orderByToNodesQueryNames = map[string][]string{
inframonitoringtypes.NodesOrderByCPU: {"A"},
inframonitoringtypes.NodesOrderByCPUAllocatable: {"B"},
inframonitoringtypes.NodesOrderByMemory: {"C"},
inframonitoringtypes.NodesOrderByMemoryAllocatable: {"D"},
}
// newNodesTableListQuery builds the composite QB v5 request for the nodes list.
// Node condition is derived separately via getPerGroupNodeConditionCounts (works
// for both list and grouped_list modes), so no condition query is included here.
func (m *module) newNodesTableListQuery() *qbtypes.QueryRangeRequest {
queries := []qbtypes.QueryEnvelope{
// Query A: CPU usage
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
Name: "A",
Signal: telemetrytypes.SignalMetrics,
Aggregations: []qbtypes.MetricAggregation{
{
MetricName: "k8s.node.cpu.usage",
TimeAggregation: metrictypes.TimeAggregationAvg,
SpaceAggregation: metrictypes.SpaceAggregationSum,
ReduceTo: qbtypes.ReduceToAvg,
},
},
GroupBy: []qbtypes.GroupByKey{nodeNameGroupByKey},
Disabled: false,
},
},
// Query B: CPU allocatable.
// TimeAggregationLatest is the closest v5 equivalent of v1's AnyLast;
// allocatable values change rarely so divergence in practice is negligible.
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
Name: "B",
Signal: telemetrytypes.SignalMetrics,
Aggregations: []qbtypes.MetricAggregation{
{
MetricName: "k8s.node.allocatable_cpu",
TimeAggregation: metrictypes.TimeAggregationLatest,
SpaceAggregation: metrictypes.SpaceAggregationSum,
ReduceTo: qbtypes.ReduceToAvg,
},
},
GroupBy: []qbtypes.GroupByKey{nodeNameGroupByKey},
Disabled: false,
},
},
// Query C: Memory working set
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
Name: "C",
Signal: telemetrytypes.SignalMetrics,
Aggregations: []qbtypes.MetricAggregation{
{
MetricName: "k8s.node.memory.working_set",
TimeAggregation: metrictypes.TimeAggregationAvg,
SpaceAggregation: metrictypes.SpaceAggregationSum,
ReduceTo: qbtypes.ReduceToAvg,
},
},
GroupBy: []qbtypes.GroupByKey{nodeNameGroupByKey},
Disabled: false,
},
},
// Query D: Memory allocatable. Same Latest caveat as Query B.
{
Type: qbtypes.QueryTypeBuilder,
Spec: qbtypes.QueryBuilderQuery[qbtypes.MetricAggregation]{
Name: "D",
Signal: telemetrytypes.SignalMetrics,
Aggregations: []qbtypes.MetricAggregation{
{
MetricName: "k8s.node.allocatable_memory",
TimeAggregation: metrictypes.TimeAggregationLatest,
SpaceAggregation: metrictypes.SpaceAggregationSum,
ReduceTo: qbtypes.ReduceToAvg,
},
},
GroupBy: []qbtypes.GroupByKey{nodeNameGroupByKey},
Disabled: false,
},
},
}
return &qbtypes.QueryRangeRequest{
RequestType: qbtypes.RequestTypeScalar,
CompositeQuery: qbtypes.CompositeQuery{
Queries: queries,
},
}
}

View File

@@ -11,9 +11,13 @@ import (
type Handler interface {
ListHosts(http.ResponseWriter, *http.Request)
ListPods(http.ResponseWriter, *http.Request)
ListNodes(http.ResponseWriter, *http.Request)
ListNamespaces(http.ResponseWriter, *http.Request)
}
type Module interface {
ListHosts(ctx context.Context, orgID valuer.UUID, req *inframonitoringtypes.PostableHosts) (*inframonitoringtypes.Hosts, error)
ListPods(ctx context.Context, orgID valuer.UUID, req *inframonitoringtypes.PostablePods) (*inframonitoringtypes.Pods, error)
ListNodes(ctx context.Context, orgID valuer.UUID, req *inframonitoringtypes.PostableNodes) (*inframonitoringtypes.Nodes, error)
ListNamespaces(ctx context.Context, orgID valuer.UUID, req *inframonitoringtypes.PostableNamespaces) (*inframonitoringtypes.Namespaces, error)
}

View File

@@ -0,0 +1,103 @@
package inframonitoringtypes
import (
"encoding/json"
"slices"
"github.com/SigNoz/signoz/pkg/errors"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
)
type Namespaces struct {
Type ResponseType `json:"type" required:"true"`
Records []NamespaceRecord `json:"records" required:"true"`
Total int `json:"total" required:"true"`
RequiredMetricsCheck RequiredMetricsCheck `json:"requiredMetricsCheck" required:"true"`
EndTimeBeforeRetention bool `json:"endTimeBeforeRetention" required:"true"`
Warning *qbtypes.QueryWarnData `json:"warning,omitempty"`
}
type NamespaceRecord struct {
NamespaceName string `json:"namespaceName" required:"true"`
NamespaceCPU float64 `json:"namespaceCPU" required:"true"`
NamespaceMemory float64 `json:"namespaceMemory" required:"true"`
PendingPodCount int `json:"pendingPodCount" required:"true"`
RunningPodCount int `json:"runningPodCount" required:"true"`
SucceededPodCount int `json:"succeededPodCount" required:"true"`
FailedPodCount int `json:"failedPodCount" required:"true"`
UnknownPodCount int `json:"unknownPodCount" required:"true"`
Meta map[string]interface{} `json:"meta" required:"true"`
}
// PostableNamespaces is the request body for the v2 namespaces list API.
type PostableNamespaces struct {
Start int64 `json:"start" required:"true"`
End int64 `json:"end" required:"true"`
Filter *qbtypes.Filter `json:"filter"`
GroupBy []qbtypes.GroupByKey `json:"groupBy"`
OrderBy *qbtypes.OrderBy `json:"orderBy"`
Offset int `json:"offset"`
Limit int `json:"limit" required:"true"`
}
// Validate ensures PostableNamespaces contains acceptable values.
func (req *PostableNamespaces) Validate() error {
if req == nil {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "request is nil")
}
if req.Start <= 0 {
return errors.NewInvalidInputf(
errors.CodeInvalidInput,
"invalid start time %d: start must be greater than 0",
req.Start,
)
}
if req.End <= 0 {
return errors.NewInvalidInputf(
errors.CodeInvalidInput,
"invalid end time %d: end must be greater than 0",
req.End,
)
}
if req.Start >= req.End {
return errors.NewInvalidInputf(
errors.CodeInvalidInput,
"invalid time range: start (%d) must be less than end (%d)",
req.Start,
req.End,
)
}
if req.Limit < 1 || req.Limit > 5000 {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "limit must be between 1 and 5000")
}
if req.Offset < 0 {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "offset cannot be negative")
}
if req.OrderBy != nil {
if !slices.Contains(NamespacesValidOrderByKeys, req.OrderBy.Key.Name) {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid order by key: %s", req.OrderBy.Key.Name)
}
if req.OrderBy.Direction != qbtypes.OrderDirectionAsc && req.OrderBy.Direction != qbtypes.OrderDirectionDesc {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid order by direction: %s", req.OrderBy.Direction)
}
}
return nil
}
// UnmarshalJSON validates input immediately after decoding.
func (req *PostableNamespaces) UnmarshalJSON(data []byte) error {
type raw PostableNamespaces
var decoded raw
if err := json.Unmarshal(data, &decoded); err != nil {
return err
}
*req = PostableNamespaces(decoded)
return req.Validate()
}

View File

@@ -0,0 +1,11 @@
package inframonitoringtypes
const (
NamespacesOrderByCPU = "cpu"
NamespacesOrderByMemory = "memory"
)
var NamespacesValidOrderByKeys = []string{
NamespacesOrderByCPU,
NamespacesOrderByMemory,
}

View File

@@ -0,0 +1,237 @@
package inframonitoringtypes
import (
"testing"
"github.com/SigNoz/signoz/pkg/errors"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
"github.com/stretchr/testify/require"
)
func TestPostableNamespaces_Validate(t *testing.T) {
tests := []struct {
name string
req *PostableNamespaces
wantErr bool
}{
{
name: "valid request",
req: &PostableNamespaces{
Start: 1000,
End: 2000,
Limit: 100,
Offset: 0,
},
wantErr: false,
},
{
name: "nil request",
req: nil,
wantErr: true,
},
{
name: "start time zero",
req: &PostableNamespaces{
Start: 0,
End: 2000,
Limit: 100,
Offset: 0,
},
wantErr: true,
},
{
name: "start time negative",
req: &PostableNamespaces{
Start: -1000,
End: 2000,
Limit: 100,
Offset: 0,
},
wantErr: true,
},
{
name: "end time zero",
req: &PostableNamespaces{
Start: 1000,
End: 0,
Limit: 100,
Offset: 0,
},
wantErr: true,
},
{
name: "start time greater than end time",
req: &PostableNamespaces{
Start: 2000,
End: 1000,
Limit: 100,
Offset: 0,
},
wantErr: true,
},
{
name: "start time equal to end time",
req: &PostableNamespaces{
Start: 1000,
End: 1000,
Limit: 100,
Offset: 0,
},
wantErr: true,
},
{
name: "limit zero",
req: &PostableNamespaces{
Start: 1000,
End: 2000,
Limit: 0,
Offset: 0,
},
wantErr: true,
},
{
name: "limit negative",
req: &PostableNamespaces{
Start: 1000,
End: 2000,
Limit: -10,
Offset: 0,
},
wantErr: true,
},
{
name: "limit exceeds max",
req: &PostableNamespaces{
Start: 1000,
End: 2000,
Limit: 5001,
Offset: 0,
},
wantErr: true,
},
{
name: "offset negative",
req: &PostableNamespaces{
Start: 1000,
End: 2000,
Limit: 100,
Offset: -5,
},
wantErr: true,
},
{
name: "orderBy nil is valid",
req: &PostableNamespaces{
Start: 1000,
End: 2000,
Limit: 100,
Offset: 0,
},
wantErr: false,
},
{
name: "orderBy with valid key cpu and direction asc",
req: &PostableNamespaces{
Start: 1000,
End: 2000,
Limit: 100,
Offset: 0,
OrderBy: &qbtypes.OrderBy{
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: NamespacesOrderByCPU,
},
},
Direction: qbtypes.OrderDirectionAsc,
},
},
wantErr: false,
},
{
name: "orderBy with valid key memory and direction desc",
req: &PostableNamespaces{
Start: 1000,
End: 2000,
Limit: 100,
Offset: 0,
OrderBy: &qbtypes.OrderBy{
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: NamespacesOrderByMemory,
},
},
Direction: qbtypes.OrderDirectionDesc,
},
},
wantErr: false,
},
{
name: "orderBy with pod_phase key is rejected",
req: &PostableNamespaces{
Start: 1000,
End: 2000,
Limit: 100,
Offset: 0,
OrderBy: &qbtypes.OrderBy{
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: "pod_phase",
},
},
Direction: qbtypes.OrderDirectionDesc,
},
},
wantErr: true,
},
{
name: "orderBy with invalid key",
req: &PostableNamespaces{
Start: 1000,
End: 2000,
Limit: 100,
Offset: 0,
OrderBy: &qbtypes.OrderBy{
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: "unknown",
},
},
Direction: qbtypes.OrderDirectionDesc,
},
},
wantErr: true,
},
{
name: "orderBy with valid key but invalid direction",
req: &PostableNamespaces{
Start: 1000,
End: 2000,
Limit: 100,
Offset: 0,
OrderBy: &qbtypes.OrderBy{
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: NamespacesOrderByMemory,
},
},
Direction: qbtypes.OrderDirection{String: valuer.NewString("invalid")},
},
},
wantErr: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
err := tt.req.Validate()
if tt.wantErr {
require.Error(t, err)
require.True(t, errors.Ast(err, errors.TypeInvalidInput), "expected error to be of type InvalidInput")
} else {
require.NoError(t, err)
}
})
}
}

View File

@@ -0,0 +1,103 @@
package inframonitoringtypes
import (
"encoding/json"
"slices"
"github.com/SigNoz/signoz/pkg/errors"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
)
type Nodes struct {
Type ResponseType `json:"type" required:"true"`
Records []NodeRecord `json:"records" required:"true"`
Total int `json:"total" required:"true"`
RequiredMetricsCheck RequiredMetricsCheck `json:"requiredMetricsCheck" required:"true"`
EndTimeBeforeRetention bool `json:"endTimeBeforeRetention" required:"true"`
Warning *qbtypes.QueryWarnData `json:"warning,omitempty"`
}
type NodeRecord struct {
NodeName string `json:"nodeName" required:"true"`
Condition NodeCondition `json:"condition" required:"true"`
ReadyNodesCount int `json:"readyNodesCount" required:"true"`
NotReadyNodesCount int `json:"notReadyNodesCount" required:"true"`
NodeCPU float64 `json:"nodeCPU" required:"true"`
NodeCPUAllocatable float64 `json:"nodeCPUAllocatable" required:"true"`
NodeMemory float64 `json:"nodeMemory" required:"true"`
NodeMemoryAllocatable float64 `json:"nodeMemoryAllocatable" required:"true"`
Meta map[string]interface{} `json:"meta" required:"true"`
}
// PostableNodes is the request body for the v2 nodes list API.
type PostableNodes struct {
Start int64 `json:"start" required:"true"`
End int64 `json:"end" required:"true"`
Filter *qbtypes.Filter `json:"filter"`
GroupBy []qbtypes.GroupByKey `json:"groupBy"`
OrderBy *qbtypes.OrderBy `json:"orderBy"`
Offset int `json:"offset"`
Limit int `json:"limit" required:"true"`
}
// Validate ensures PostableNodes contains acceptable values.
func (req *PostableNodes) Validate() error {
if req == nil {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "request is nil")
}
if req.Start <= 0 {
return errors.NewInvalidInputf(
errors.CodeInvalidInput,
"invalid start time %d: start must be greater than 0",
req.Start,
)
}
if req.End <= 0 {
return errors.NewInvalidInputf(
errors.CodeInvalidInput,
"invalid end time %d: end must be greater than 0",
req.End,
)
}
if req.Start >= req.End {
return errors.NewInvalidInputf(
errors.CodeInvalidInput,
"invalid time range: start (%d) must be less than end (%d)",
req.Start,
req.End,
)
}
if req.Limit < 1 || req.Limit > 5000 {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "limit must be between 1 and 5000")
}
if req.Offset < 0 {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "offset cannot be negative")
}
if req.OrderBy != nil {
if !slices.Contains(NodesValidOrderByKeys, req.OrderBy.Key.Name) {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid order by key: %s", req.OrderBy.Key.Name)
}
if req.OrderBy.Direction != qbtypes.OrderDirectionAsc && req.OrderBy.Direction != qbtypes.OrderDirectionDesc {
return errors.NewInvalidInputf(errors.CodeInvalidInput, "invalid order by direction: %s", req.OrderBy.Direction)
}
}
return nil
}
// UnmarshalJSON validates input immediately after decoding.
func (req *PostableNodes) UnmarshalJSON(data []byte) error {
type raw PostableNodes
var decoded raw
if err := json.Unmarshal(data, &decoded); err != nil {
return err
}
*req = PostableNodes(decoded)
return req.Validate()
}

View File

@@ -0,0 +1,42 @@
package inframonitoringtypes
import "github.com/SigNoz/signoz/pkg/valuer"
type NodeCondition struct {
valuer.String
}
var (
NodeConditionReady = NodeCondition{valuer.NewString("ready")}
NodeConditionNotReady = NodeCondition{valuer.NewString("not_ready")}
NodeConditionNone = NodeCondition{valuer.NewString("")}
)
func (NodeCondition) Enum() []any {
return []any{
NodeConditionReady,
NodeConditionNotReady,
NodeConditionNone,
}
}
// Numeric values emitted by the k8s.node.condition_ready metric
// (source: OTel kubeletstats receiver).
const (
NodeConditionNumReady = 1
NodeConditionNumNotReady = 0
)
const (
NodesOrderByCPU = "cpu"
NodesOrderByCPUAllocatable = "cpu_allocatable"
NodesOrderByMemory = "memory"
NodesOrderByMemoryAllocatable = "memory_allocatable"
)
var NodesValidOrderByKeys = []string{
NodesOrderByCPU,
NodesOrderByCPUAllocatable,
NodesOrderByMemory,
NodesOrderByMemoryAllocatable,
}

View File

@@ -0,0 +1,255 @@
package inframonitoringtypes
import (
"testing"
"github.com/SigNoz/signoz/pkg/errors"
qbtypes "github.com/SigNoz/signoz/pkg/types/querybuildertypes/querybuildertypesv5"
"github.com/SigNoz/signoz/pkg/types/telemetrytypes"
"github.com/SigNoz/signoz/pkg/valuer"
"github.com/stretchr/testify/require"
)
func TestPostableNodes_Validate(t *testing.T) {
tests := []struct {
name string
req *PostableNodes
wantErr bool
}{
{
name: "valid request",
req: &PostableNodes{
Start: 1000,
End: 2000,
Limit: 100,
Offset: 0,
},
wantErr: false,
},
{
name: "nil request",
req: nil,
wantErr: true,
},
{
name: "start time zero",
req: &PostableNodes{
Start: 0,
End: 2000,
Limit: 100,
Offset: 0,
},
wantErr: true,
},
{
name: "start time negative",
req: &PostableNodes{
Start: -1000,
End: 2000,
Limit: 100,
Offset: 0,
},
wantErr: true,
},
{
name: "end time zero",
req: &PostableNodes{
Start: 1000,
End: 0,
Limit: 100,
Offset: 0,
},
wantErr: true,
},
{
name: "start time greater than end time",
req: &PostableNodes{
Start: 2000,
End: 1000,
Limit: 100,
Offset: 0,
},
wantErr: true,
},
{
name: "start time equal to end time",
req: &PostableNodes{
Start: 1000,
End: 1000,
Limit: 100,
Offset: 0,
},
wantErr: true,
},
{
name: "limit zero",
req: &PostableNodes{
Start: 1000,
End: 2000,
Limit: 0,
Offset: 0,
},
wantErr: true,
},
{
name: "limit negative",
req: &PostableNodes{
Start: 1000,
End: 2000,
Limit: -10,
Offset: 0,
},
wantErr: true,
},
{
name: "limit exceeds max",
req: &PostableNodes{
Start: 1000,
End: 2000,
Limit: 5001,
Offset: 0,
},
wantErr: true,
},
{
name: "offset negative",
req: &PostableNodes{
Start: 1000,
End: 2000,
Limit: 100,
Offset: -5,
},
wantErr: true,
},
{
name: "orderBy nil is valid",
req: &PostableNodes{
Start: 1000,
End: 2000,
Limit: 100,
Offset: 0,
},
wantErr: false,
},
{
name: "orderBy with valid key cpu and direction asc",
req: &PostableNodes{
Start: 1000,
End: 2000,
Limit: 100,
Offset: 0,
OrderBy: &qbtypes.OrderBy{
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: NodesOrderByCPU,
},
},
Direction: qbtypes.OrderDirectionAsc,
},
},
wantErr: false,
},
{
name: "orderBy with valid key cpu_allocatable and direction desc",
req: &PostableNodes{
Start: 1000,
End: 2000,
Limit: 100,
Offset: 0,
OrderBy: &qbtypes.OrderBy{
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: NodesOrderByCPUAllocatable,
},
},
Direction: qbtypes.OrderDirectionDesc,
},
},
wantErr: false,
},
{
name: "orderBy with valid key memory_allocatable and direction asc",
req: &PostableNodes{
Start: 1000,
End: 2000,
Limit: 100,
Offset: 0,
OrderBy: &qbtypes.OrderBy{
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: NodesOrderByMemoryAllocatable,
},
},
Direction: qbtypes.OrderDirectionAsc,
},
},
wantErr: false,
},
{
name: "orderBy with condition key is rejected",
req: &PostableNodes{
Start: 1000,
End: 2000,
Limit: 100,
Offset: 0,
OrderBy: &qbtypes.OrderBy{
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: "condition",
},
},
Direction: qbtypes.OrderDirectionDesc,
},
},
wantErr: true,
},
{
name: "orderBy with invalid key",
req: &PostableNodes{
Start: 1000,
End: 2000,
Limit: 100,
Offset: 0,
OrderBy: &qbtypes.OrderBy{
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: "unknown",
},
},
Direction: qbtypes.OrderDirectionDesc,
},
},
wantErr: true,
},
{
name: "orderBy with valid key but invalid direction",
req: &PostableNodes{
Start: 1000,
End: 2000,
Limit: 100,
Offset: 0,
OrderBy: &qbtypes.OrderBy{
Key: qbtypes.OrderByKey{
TelemetryFieldKey: telemetrytypes.TelemetryFieldKey{
Name: NodesOrderByMemory,
},
},
Direction: qbtypes.OrderDirection{String: valuer.NewString("invalid")},
},
},
wantErr: true,
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
err := tt.req.Validate()
if tt.wantErr {
require.Error(t, err)
require.True(t, errors.Ast(err, errors.TypeInvalidInput), "expected error to be of type InvalidInput")
} else {
require.NoError(t, err)
}
})
}
}

View File

@@ -23,6 +23,7 @@ pytest_plugins = [
"fixtures.notification_channel",
"fixtures.alerts",
"fixtures.cloudintegrations",
"fixtures.jsontypes",
"fixtures.seeder",
]
@@ -79,6 +80,6 @@ def pytest_addoption(parser: pytest.Parser):
parser.addoption(
"--schema-migrator-version",
action="store",
default="v0.144.2",
default="v0.144.3",
help="schema migrator version",
)

View File

@@ -1,5 +1,6 @@
import os
from collections.abc import Generator
from collections.abc import Callable, Generator
from datetime import datetime
from typing import Any
import clickhouse_connect
@@ -262,3 +263,76 @@ def clickhouse(
delete=delete,
restore=restore,
)
@pytest.fixture(name="check_query_log")
def check_query_log(
signoz: types.SigNoz,
) -> Callable[..., None]:
"""
Returns a callable that flushes system.query_log and asserts that at
least one recent SELECT satisfies check_fn.
Args:
after_ts: Only consider queries logged after this timestamp.
case_name: Label used in assertion failure messages.
check_fn: Predicate run against each candidate query string.
tables: Filter to queries that touched all of these tables, as
'db.table' strings (uses hasAll(tables, [...])).
must_contain: Substrings that must appear in the query text (AND-ed).
must_not_contain: Substrings that must not appear in the query text (AND-ed).
limit: How many most-recent queries to examine (default 10).
Usage:
before = datetime.now(tz=timezone.utc)
# ... trigger the query under test ...
check_query_log(
before, "my.case",
lambda q: "assumeNotNull" in q,
tables=["signoz_logs.distributed_logs_v2"],
)
"""
def _check(
after_ts: datetime,
case_name: str,
check_fn: Callable[[str], bool],
*,
tables: list[str] | None = None,
must_contain: list[str] | None = None,
must_not_contain: list[str] | None = None,
limit: int = 10,
) -> None:
conn = signoz.telemetrystore.conn
conn.command("SYSTEM FLUSH LOGS")
# Use millisecond precision to avoid timestamp collisions between
# adjacent test cases (second-level precision causes bleed-through).
params: dict = {"after_ms": int(after_ts.timestamp() * 1000)}
conditions = [
"type = 'QueryFinish'",
"query_kind = 'Select'",
"toUnixTimestamp64Milli(event_time_microseconds) >= %(after_ms)s",
]
if tables:
params["tables"] = tables
conditions.append("hasAll(tables, %(tables)s)")
for i, pattern in enumerate(must_contain or []):
key = f"mc_{i}"
params[key] = pattern
conditions.append(f"position(query, %({key})s) > 0")
for i, pattern in enumerate(must_not_contain or []):
key = f"mnc_{i}"
params[key] = pattern
conditions.append(f"position(query, %({key})s) = 0")
where = " AND ".join(conditions)
result = conn.query(
f"SELECT query FROM system.query_log WHERE {where} ORDER BY event_time_microseconds DESC LIMIT {limit}",
parameters=params,
)
queries = [row[0] for row in result.result_rows]
assert queries, f"No matching SELECT in system.query_log for case '{case_name}'"
assert all(check_fn(q) for q in queries), f"query_log check failed for case '{case_name}'.\n" + "Queries:\n" + "\n---\n".join(queries)
return _check

441
tests/fixtures/jsontypes.py vendored Normal file
View File

@@ -0,0 +1,441 @@
"""
Simpler version of metadataexporter for exporting jsontypes for test fixtures.
This exports JSON type metadata to the path_types table by parsing JSON bodies
and extracting all paths with their types, similar to how the real metadataexporter works.
"""
import datetime
import json
from abc import ABC
from collections.abc import Callable, Generator
from http import HTTPStatus
from typing import (
Any,
)
import numpy as np
import pytest
import requests
from fixtures import types
class JSONPathType(ABC):
"""Represents a JSON path with its type information"""
field_name: str
field_data_type: str
last_seen: np.uint64
signal: str = "logs"
field_context: str = "body"
def __init__(
self,
field_name: str,
field_data_type: str,
last_seen: datetime.datetime | None = None,
) -> None:
self.field_name = field_name
self.field_data_type = field_data_type
self.signal = "logs"
self.field_context = "body"
if last_seen is None:
last_seen = datetime.datetime.now()
self.last_seen = np.uint64(int(last_seen.timestamp() * 1e9))
def np_arr(self) -> np.array:
"""Return path type data as numpy array for database insertion"""
return np.array([self.signal, self.field_context, self.field_name, self.field_data_type, self.last_seen])
# Constants matching metadataexporter
ARRAY_SEPARATOR = "[]." # Used in paths like "education[].name"
ARRAY_SUFFIX = "[]" # Used when traversing into array element objects
def _infer_array_type_from_type_strings(types: list[str]) -> str | None:
"""
Infer array type from a list of pre-classified type strings.
Matches metadataexporter's inferArrayMask logic.
Internal type strings are: "JSON", "String", "Bool", "Float64", "Int64"
SuperTyping rules (matching Go inferArrayMask):
- JSON alone → []json
- JSON + any primitive → []dynamic
- String alone → []string; String + other → []dynamic
- Float64 wins over Int64 and Bool
- Int64 wins over Bool
- Bool alone → []bool
"""
if len(types) == 0:
return None
unique = set(types)
has_json = "JSON" in unique
# hasPrimitive mirrors Go: (hasJSON && len(unique) > 1) || (!hasJSON && len(unique) > 0)
has_primitive = (has_json and len(unique) > 1) or (not has_json and len(unique) > 0)
if has_json:
if not has_primitive:
return "[]json"
return "[]dynamic"
# ---- Primitive Type Resolution (Float > Int > Bool) ----
if "String" in unique:
if len(unique) > 1:
return "[]dynamic"
return "[]string"
if "Float64" in unique:
return "[]float64"
if "Int64" in unique:
return "[]int64"
if "Bool" in unique:
return "[]bool"
return "[]dynamic"
def _infer_array_type(elements: list[Any]) -> str | None:
"""
Infer array type from raw Python list elements.
Classifies each element then delegates to _infer_array_type_from_type_strings.
"""
if len(elements) == 0:
return None
types = []
for elem in elements:
if elem is None:
continue
if isinstance(elem, dict):
types.append("JSON")
elif isinstance(elem, str):
types.append("String")
elif isinstance(elem, bool): # must be before int (bool is subclass of int)
types.append("Bool")
elif isinstance(elem, float):
types.append("Float64")
elif isinstance(elem, int):
types.append("Int64")
return _infer_array_type_from_type_strings(types)
def _python_type_to_clickhouse_type(value: Any) -> str:
"""
Convert Python type to ClickHouse JSON type string.
Maps Python types to ClickHouse JSON data types.
Matches metadataexporter's mapPCommonValueTypeToDataType.
"""
if isinstance(value, bool):
return "bool"
elif isinstance(value, int):
return "int64"
elif isinstance(value, float):
return "float64"
elif isinstance(value, str):
return "string"
elif isinstance(value, list):
# Use the sophisticated array type inference
array_type = _infer_array_type(value)
return array_type if array_type else "[]dynamic"
elif isinstance(value, dict):
return "json"
else:
return "string" # Default fallback
def _extract_json_paths(
obj: Any,
current_path: str = "",
path_types: dict[str, set[str]] | None = None,
level: int = 0,
) -> dict[str, set[str]]:
"""
Recursively extract all paths and their types from a JSON object.
Matches metadataexporter's analyzePValue logic.
Args:
obj: The JSON object to traverse
current_path: Current path being built (e.g., "user.name")
path_types: Dictionary mapping paths to sets of types found
level: Current nesting level (for depth limiting)
Returns:
Dictionary mapping paths to sets of type strings
"""
if path_types is None:
path_types = {}
if obj is None:
# Skip null values — matches Go walkNode which errors on ValueTypeEmpty
return path_types
if isinstance(obj, dict):
# For objects, recurse into keys without recording the object itself as a type.
# Matches Go walkMap which recurses without calling ta.record on the map node.
for key, value in obj.items():
# Build the path for this key
if current_path:
new_path = f"{current_path}.{key}"
else:
new_path = key
# Recurse into the value
_extract_json_paths(value, new_path, path_types, level + 1)
elif isinstance(obj, list):
# Skip empty arrays
if len(obj) == 0:
return path_types
# Collect types from array elements (matching Go: types := make([]pcommon.ValueType, 0, s.Len()))
types = []
for item in obj:
if isinstance(item, dict):
# When traversing into array element objects, use ArraySuffix ([])
# This matches: prefix+ArraySuffix in the Go code
# Example: if current_path is "education", we use "education[]" to traverse into objects
array_prefix = current_path + ARRAY_SUFFIX if current_path else ""
for key, value in item.items():
if array_prefix:
# Use array separator: education[].name
array_path = f"{array_prefix}.{key}"
else:
array_path = key
# Recurse without increasing level (matching Go behavior)
_extract_json_paths(value, array_path, path_types, level)
types.append("JSON")
elif isinstance(item, list):
# Arrays inside arrays are not supported - skip the whole path
# Matching Go: e.logger.Error("arrays inside arrays are not supported!", ...); return nil
return path_types
elif isinstance(item, str):
types.append("String")
elif isinstance(item, bool):
types.append("Bool")
elif isinstance(item, float):
types.append("Float64")
elif isinstance(item, int):
types.append("Int64")
# Infer array type from collected types (matching Go: if mask := inferArrayMask(types); mask != 0)
if len(types) > 0:
array_type = _infer_array_type_from_type_strings(types)
if array_type and current_path:
if current_path not in path_types:
path_types[current_path] = set()
path_types[current_path].add(array_type)
# Primitive value (string, number, bool)
elif current_path:
if current_path not in path_types:
path_types[current_path] = set()
obj_type = _python_type_to_clickhouse_type(obj)
path_types[current_path].add(obj_type)
return path_types
def _parse_json_bodies_and_extract_paths(
json_bodies: list[str],
timestamp: datetime.datetime | None = None,
) -> list[JSONPathType]:
"""
Parse JSON bodies and extract all paths with their types.
This mimics the behavior of metadataexporter.
Args:
json_bodies: List of JSON body strings to parse
timestamp: Timestamp to use for last_seen (defaults to now)
Returns:
List of JSONPathType objects with all discovered paths and types
"""
if timestamp is None:
timestamp = datetime.datetime.now()
# Aggregate all paths and their types across all JSON bodies
all_path_types: dict[str, set[str]] = {}
for json_body in json_bodies:
try:
parsed = json.loads(json_body)
_extract_json_paths(parsed, "", all_path_types, level=0)
except (json.JSONDecodeError, TypeError):
# Skip invalid JSON
continue
# Convert to list of JSONPathType objects
# Each path can have multiple types, so we create one JSONPathType per type
path_type_objects: list[JSONPathType] = []
for path, types_set in all_path_types.items():
for type_str in types_set:
path_type_objects.append(JSONPathType(field_name=path, field_data_type=type_str, last_seen=timestamp))
return path_type_objects
@pytest.fixture(name="export_json_types", scope="function")
def export_json_types(
clickhouse: types.TestContainerClickhouse,
) -> Generator[Callable[[list[JSONPathType] | list[str] | list[Any]], None], Any]:
"""
Fixture for exporting JSON type metadata to the path_types table.
This is a simpler version of metadataexporter for test fixtures.
The function can accept:
1. List of JSONPathType objects (manual specification)
2. List of JSON body strings (auto-extract paths)
3. List of Logs objects (extract from body_json field)
Usage examples:
# Manual specification
export_json_types([
JSONPathType(field_name="user.name", field_data_type="string"),
JSONPathType(field_name="user.age", field_data_type="int64"),
])
# Auto-extract from JSON strings
export_json_types([
'{"user": {"name": "alice", "age": 25}}',
'{"user": {"name": "bob", "age": 30}}',
])
# Auto-extract from Logs objects
export_json_types(logs_list)
"""
def _export_json_types(
data: list[JSONPathType] | list[str] | list[Any], # List[Logs] but avoiding circular import
) -> None:
"""
Export JSON type metadata to signoz_metadata.distributed_field_keys table.
This table stores signal, context, path, and type information for body JSON fields.
"""
path_types: list[JSONPathType] = []
if len(data) == 0:
return
# Determine input type and convert to JSONPathType list
first_item = data[0]
if isinstance(first_item, JSONPathType):
# Already JSONPathType objects
path_types = data # type: ignore
elif isinstance(first_item, str):
# List of JSON strings - parse and extract paths
path_types = _parse_json_bodies_and_extract_paths(data) # type: ignore
else:
# Assume it's a list of Logs objects - extract body_v2
json_bodies: list[str] = []
for log in data: # type: ignore
# Try to get body_v2 attribute
if hasattr(log, "body_v2") and log.body_v2:
json_bodies.append(log.body_v2)
elif hasattr(log, "body") and log.body:
# Fallback to body if body_v2 not available
try:
# Try to parse as JSON
json.loads(log.body)
json_bodies.append(log.body)
except (json.JSONDecodeError, TypeError):
pass
if json_bodies:
path_types = _parse_json_bodies_and_extract_paths(json_bodies)
if len(path_types) == 0:
return
clickhouse.conn.insert(
database="signoz_metadata",
table="distributed_field_keys",
data=[path_type.np_arr() for path_type in path_types],
column_names=[
"signal",
"field_context",
"field_name",
"field_data_type",
"last_seen",
],
)
yield _export_json_types
# Cleanup - truncate the local table after tests (following pattern from logs fixture)
clickhouse.conn.query(f"TRUNCATE TABLE signoz_metadata.field_keys ON CLUSTER '{clickhouse.env['SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_CLUSTER']}' SYNC")
@pytest.fixture(name="create_json_index", scope="function")
def create_json_index(
signoz: types.SigNoz,
) -> Generator[Callable[[str, list[dict[str, Any]]], None]]:
"""
Create ClickHouse data-skipping indexes on body_v2 JSON sub-columns via
POST /api/v1/logs/promote_paths.
**Must be called BEFORE insert_logs** so that newly inserted data parts are
covered by the index and the QB uses the indexed condition path.
Each entry in `paths` follows the PromotePath API shape:
{
"path": "body.user.name", # must start with "body."
"indexes": [
{
"fieldDataType": "string", # string | int64 | float64
"type": "ngrambf_v1(3, 256, 2, 0)", # or "minmax", "tokenbf_v1(...)"
"granularity": 1,
}
],
}
Teardown drops every index created during the test by querying
system.data_skipping_indices for matching expressions.
Example::
def test_foo(signoz, get_token, insert_logs, export_json_types, create_json_body_index):
token = get_token(...)
export_json_types(logs_list)
create_json_body_index(token, [
{"path": "body.user.name",
"indexes": [{"fieldDataType": "string", "type": "ngrambf_v1(3, 256, 2, 0)", "granularity": 1}]},
{"path": "body.user.age",
"indexes": [{"fieldDataType": "int64", "type": "minmax", "granularity": 1}]},
])
insert_logs(logs_list) # data inserted after index exists — index is built automatically
"""
created_paths: list[str] = []
def _create_json_body_index(token: str, paths: list[dict[str, Any]]) -> None:
response = requests.post(
signoz.self.host_configs["8080"].get("/api/v1/logs/promote_paths"),
headers={"authorization": f"Bearer {token}"},
json=paths,
timeout=30,
)
assert response.status_code == HTTPStatus.CREATED, f"Failed to create JSON body indexes: {response.status_code} {response.text}"
for path in paths:
# The API strips the "body." prefix before storing — mirror that here
# so our cleanup query uses the bare path (e.g. "user.name").
raw = path["path"].removeprefix("body.")
if raw not in created_paths:
created_paths.append(raw)
yield _create_json_body_index
if not created_paths:
return
cluster = signoz.telemetrystore.env["SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_CLUSTER"]
for path in created_paths:
result = signoz.telemetrystore.conn.query(f"SELECT name FROM system.data_skipping_indices WHERE database = 'signoz_logs' AND table = 'logs_v2' AND expr LIKE '%{path}%'")
for (index_name,) in result.result_rows:
signoz.telemetrystore.conn.query(f"ALTER TABLE signoz_logs.logs_v2 ON CLUSTER '{cluster}' DROP INDEX IF EXISTS `{index_name}`")

View File

@@ -121,6 +121,8 @@ class Logs(ABC):
resources: dict[str, Any] = {},
attributes: dict[str, Any] = {},
body: str = "default body",
body_v2: str | None = None,
body_promoted: str | None = None,
severity_text: str = "INFO",
trace_id: str = "",
span_id: str = "",
@@ -166,6 +168,33 @@ class Logs(ABC):
# Set body
self.body = body
# Set body_v2 - if body is JSON, parse and stringify it, otherwise use empty string
# ClickHouse accepts String input for JSON column
if body_v2 is not None:
self.body_v2 = body_v2
else:
# Try to parse body as JSON; if successful use it directly,
# otherwise wrap as {"message": body} matching the normalize operator behavior.
try:
json.loads(body)
self.body_v2 = body
except (json.JSONDecodeError, TypeError):
self.body_v2 = json.dumps({"message": body})
# Set body_promoted - must be valid JSON
# Tests will explicitly pass promoted column's content, but we validate it
if body_promoted is not None:
# Validate that it's valid JSON
try:
json.loads(body_promoted)
self.body_promoted = body_promoted
except (json.JSONDecodeError, TypeError):
# If invalid, default to empty JSON object
self.body_promoted = "{}"
else:
# Default to empty JSON object (valid JSON)
self.body_promoted = "{}"
# Process resources and attributes
self.resources_string = {k: str(v) for k, v in resources.items()}
self.resource_json = {} if resource_write_mode == "legacy_only" else dict(self.resources_string)
@@ -309,6 +338,8 @@ class Logs(ABC):
self.severity_text,
self.severity_number,
self.body,
self.body_v2,
self.body_promoted,
self.attributes_string,
self.attributes_number,
self.attributes_bool,
@@ -436,31 +467,47 @@ def insert_logs_to_clickhouse(conn, logs: list[Logs]) -> None:
data=[resource_key.np_arr() for resource_key in resource_keys],
)
all_column_names = [
"ts_bucket_start",
"resource_fingerprint",
"timestamp",
"observed_timestamp",
"id",
"trace_id",
"span_id",
"trace_flags",
"severity_text",
"severity_number",
"body",
"body_v2",
"body_promoted",
"attributes_string",
"attributes_number",
"attributes_bool",
"resources_string",
"scope_name",
"scope_version",
"scope_string",
"resource",
]
result = conn.query("SELECT count() FROM system.columns WHERE database = 'signoz_logs' AND table = 'logs_v2' AND name = 'body_v2'")
has_json_body = result.result_rows[0][0] > 0
if has_json_body:
column_names = all_column_names
data = [log.np_arr() for log in logs]
else:
json_body_cols = {"body_v2", "body_promoted"}
keep_indices = [i for i, c in enumerate(all_column_names) if c not in json_body_cols]
column_names = [all_column_names[i] for i in keep_indices]
data = [log.np_arr()[keep_indices] for log in logs]
conn.insert(
database="signoz_logs",
table="distributed_logs_v2",
data=[log.np_arr() for log in logs],
column_names=[
"ts_bucket_start",
"resource_fingerprint",
"timestamp",
"observed_timestamp",
"id",
"trace_id",
"span_id",
"trace_flags",
"severity_text",
"severity_number",
"body",
"attributes_string",
"attributes_number",
"attributes_bool",
"resources_string",
"scope_name",
"scope_version",
"scope_string",
"resource",
],
data=data,
column_names=column_names,
)

View File

@@ -8,27 +8,32 @@ from fixtures.logger import setup_logger
logger = setup_logger(__name__)
@pytest.fixture(name="migrator", scope="package")
def migrator(
def create_migrator(
network: Network,
clickhouse: types.TestContainerClickhouse,
request: pytest.FixtureRequest,
pytestconfig: pytest.Config,
cache_key: str = "migrator",
env_overrides: dict | None = None,
) -> types.Operation:
"""
Package-scoped fixture for running schema migrations.
Factory function for running schema migrations.
Accepts optional env_overrides to customize the migrator environment.
"""
def create() -> None:
version = request.config.getoption("--schema-migrator-version")
client = docker.from_env()
environment = dict(env_overrides) if env_overrides else {}
container = client.containers.run(
image=f"signoz/signoz-schema-migrator:{version}",
command=f"sync --replication=true --cluster-name=cluster --up= --dsn={clickhouse.env['SIGNOZ_TELEMETRYSTORE_CLICKHOUSE_DSN']}",
detach=True,
auto_remove=False,
network=network.id,
environment=environment,
)
result = container.wait()
@@ -47,6 +52,7 @@ def migrator(
detach=True,
auto_remove=False,
network=network.id,
environment=environment,
)
result = container.wait()
@@ -59,7 +65,7 @@ def migrator(
container.remove()
return types.Operation(name="migrator")
return types.Operation(name=cache_key)
def delete(_: types.Operation) -> None:
pass
@@ -70,9 +76,27 @@ def migrator(
return reuse.wrap(
request,
pytestconfig,
"migrator",
cache_key,
lambda: types.Operation(name=""),
create,
delete,
restore,
)
@pytest.fixture(name="migrator", scope="package")
def migrator(
network: Network,
clickhouse: types.TestContainerClickhouse,
request: pytest.FixtureRequest,
pytestconfig: pytest.Config,
) -> types.Operation:
"""
Package-scoped fixture for running schema migrations.
"""
return create_migrator(
network=network,
clickhouse=clickhouse,
request=request,
pytestconfig=pytestconfig,
)

View File

@@ -500,6 +500,14 @@ def get_scalar_columns(response_json: dict) -> list[dict]:
return results[0].get("columns", [])
def get_rows(response: requests.Response) -> list[dict[str, Any]]:
assert response.json()["status"] == "success"
results = response.json()["data"]["data"]["results"]
assert len(results) == 1
# The server returns rows:null (not []) when there are 0 matching logs.
return results[0].get("rows") or []
def get_column_data_from_response(response_json: dict, column_name: str) -> list[Any]:
results = response_json.get("data", {}).get("data", {}).get("results", [])
if not results:

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,67 @@
import pytest
from testcontainers.core.container import Network
from fixtures import types
from fixtures.migrator import create_migrator
from fixtures.signoz import create_signoz
UNSUPPORTED_CLICKHOUSE_VERSIONS = {"25.5.6"}
def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None:
version = config.getoption("--clickhouse-version")
if version in UNSUPPORTED_CLICKHOUSE_VERSIONS:
skip = pytest.mark.skip(reason=f"JSON body QB tests require ClickHouse > {version}")
for item in items:
item.add_marker(skip)
@pytest.fixture(name="migrator", scope="package")
def migrator_json(
network: Network,
clickhouse: types.TestContainerClickhouse,
request: pytest.FixtureRequest,
pytestconfig: pytest.Config,
) -> types.Operation:
"""
Package-scoped migrator with ENABLE_LOGS_MIGRATIONS_V2=1.
"""
return create_migrator(
network=network,
clickhouse=clickhouse,
request=request,
pytestconfig=pytestconfig,
cache_key="migrator-json-body",
env_overrides={
"ENABLE_LOGS_MIGRATIONS_V2": "1",
},
)
@pytest.fixture(name="signoz", scope="package")
def signoz_json_body(
network: Network,
migrator: types.Operation, # pylint: disable=unused-argument
zeus: types.TestContainerDocker,
gateway: types.TestContainerDocker,
sqlstore: types.TestContainerSQL,
clickhouse: types.TestContainerClickhouse,
request: pytest.FixtureRequest,
pytestconfig: pytest.Config,
) -> types.SigNoz:
"""
Package-scoped fixture for SigNoz with BODY_JSON_QUERY_ENABLED=true.
"""
return create_signoz(
network=network,
zeus=zeus,
gateway=gateway,
sqlstore=sqlstore,
clickhouse=clickhouse,
request=request,
pytestconfig=pytestconfig,
cache_key="signoz-json-body",
env_overrides={
"BODY_JSON_QUERY_ENABLED": "true",
},
)