chore: function refactor

pkg/modules/inframonitoring/implinframonitoring/module.go

@@ -231,7 +231,7 @@ func (m *module) ListPods(ctx context.Context, orgID valuer.UUID, req *inframoni
 		return nil, err
 	}
 
-	phaseCounts, err := m.getPerGroupPodPhaseCounts(ctx, req, pageGroups)
+	phaseCounts, err := m.getPerGroupPodPhaseCounts(ctx, req.Start, req.End, req.Filter, req.GroupBy, pageGroups)
 	if err != nil {
 		return nil, err
 	}
@@ -314,19 +314,12 @@ func (m *module) ListNodes(ctx context.Context, orgID valuer.UUID, req *inframon
 		return nil, err
 	}
 
-	nodeConditionCounts, err := m.getPerGroupNodeConditionCounts(ctx, req, pageGroups)
+	nodeConditionCounts, err := m.getPerGroupNodeConditionCounts(ctx, req.Start, req.End, req.Filter, req.GroupBy, pageGroups)
 	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.
-	podPhaseCounts, err := m.getPerGroupPodPhaseCounts(ctx, &inframonitoringtypes.PostablePods{
-		Start:   req.Start,
-		End:     req.End,
-		Filter:  req.Filter,
-		GroupBy: req.GroupBy,
-	}, pageGroups)
+	podPhaseCounts, err := m.getPerGroupPodPhaseCounts(ctx, req.Start, req.End, req.Filter, req.GroupBy, pageGroups)
 	if err != nil {
 		return nil, err
 	}
@@ -409,14 +402,7 @@ func (m *module) ListNamespaces(ctx context.Context, orgID valuer.UUID, req *inf
 		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)
+	phaseCounts, err := m.getPerGroupPodPhaseCounts(ctx, req.Start, req.End, req.Filter, req.GroupBy, pageGroups)
 	if err != nil {
 		return nil, err
 	}
@@ -498,27 +484,14 @@ func (m *module) ListClusters(ctx context.Context, orgID valuer.UUID, req *infra
 		return nil, err
 	}
 
-	// Reuse the nodes condition-counts CTE function via a temp struct — it reads only
-	// Start/End/Filter/GroupBy from PostableNodes. With default groupBy
-	// [k8s.cluster.name], counts are bucketed per cluster; with a custom groupBy,
-	// they aggregate across clusters in that group.
-	nodeConditionCountsMap, err := m.getPerGroupNodeConditionCounts(ctx, &inframonitoringtypes.PostableNodes{
-		Start:   req.Start,
-		End:     req.End,
-		Filter:  req.Filter,
-		GroupBy: req.GroupBy,
-	}, pageGroups)
+	// With default groupBy [k8s.cluster.name], counts are bucketed per cluster;
+	// with a custom groupBy, they aggregate across clusters in that group.
+	nodeConditionCountsMap, err := m.getPerGroupNodeConditionCounts(ctx, req.Start, req.End, req.Filter, req.GroupBy, pageGroups)
 	if err != nil {
 		return nil, err
 	}
 
-	// Same pattern for pod phase counts via PostablePods shim.
-	podPhaseCountsMap, err := m.getPerGroupPodPhaseCounts(ctx, &inframonitoringtypes.PostablePods{
-		Start:   req.Start,
-		End:     req.End,
-		Filter:  req.Filter,
-		GroupBy: req.GroupBy,
-	}, pageGroups)
+	podPhaseCountsMap, err := m.getPerGroupPodPhaseCounts(ctx, req.Start, req.End, req.Filter, req.GroupBy, pageGroups)
 	if err != nil {
 		return nil, err
 	}
@@ -689,14 +662,7 @@ func (m *module) ListDeployments(ctx context.Context, orgID valuer.UUID, req *in
 		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)
+	phaseCounts, err := m.getPerGroupPodPhaseCounts(ctx, req.Start, req.End, req.Filter, req.GroupBy, pageGroups)
 	if err != nil {
 		return nil, err
 	}
@@ -784,16 +750,9 @@ func (m *module) ListStatefulSets(ctx context.Context, orgID valuer.UUID, req *i
 		return nil, err
 	}
 
-	// Reuse the pods phase-counts CTE function via a temp struct — it reads only
-	// Start/End/Filter/GroupBy from PostablePods. Pods owned by a StatefulSet carry
-	// k8s.statefulset.name as a resource attribute, so default-groupBy gives
-	// per-statefulset phase counts automatically.
-	phaseCounts, err := m.getPerGroupPodPhaseCounts(ctx, &inframonitoringtypes.PostablePods{
-		Start:   req.Start,
-		End:     req.End,
-		Filter:  req.Filter,
-		GroupBy: req.GroupBy,
-	}, pageGroups)
+	// Pods owned by a StatefulSet carry k8s.statefulset.name as a resource attribute,
+	// so default-groupBy gives per-statefulset phase counts automatically.
+	phaseCounts, err := m.getPerGroupPodPhaseCounts(ctx, req.Start, req.End, req.Filter, req.GroupBy, pageGroups)
 	if err != nil {
 		return nil, err
 	}
@@ -881,16 +840,9 @@ func (m *module) ListJobs(ctx context.Context, orgID valuer.UUID, req *inframoni
 		return nil, err
 	}
 
-	// Reuse the pods phase-counts CTE function via a temp struct — it reads only
-	// Start/End/Filter/GroupBy from PostablePods. Pods owned by a Job carry
-	// k8s.job.name as a resource attribute, so default-groupBy gives
-	// per-job phase counts automatically.
-	phaseCounts, err := m.getPerGroupPodPhaseCounts(ctx, &inframonitoringtypes.PostablePods{
-		Start:   req.Start,
-		End:     req.End,
-		Filter:  req.Filter,
-		GroupBy: req.GroupBy,
-	}, pageGroups)
+	// Pods owned by a Job carry k8s.job.name as a resource attribute, so default-groupBy
+	// gives per-job phase counts automatically.
+	phaseCounts, err := m.getPerGroupPodPhaseCounts(ctx, req.Start, req.End, req.Filter, req.GroupBy, pageGroups)
 	if err != nil {
 		return nil, err
 	}
@@ -978,16 +930,9 @@ func (m *module) ListDaemonSets(ctx context.Context, orgID valuer.UUID, req *inf
 		return nil, err
 	}
 
-	// Reuse the pods phase-counts CTE function via a temp struct — it reads only
-	// Start/End/Filter/GroupBy from PostablePods. Pods owned by a DaemonSet carry
-	// k8s.daemonset.name as a resource attribute, so default-groupBy gives
-	// per-daemonset phase counts automatically.
-	phaseCounts, err := m.getPerGroupPodPhaseCounts(ctx, &inframonitoringtypes.PostablePods{
-		Start:   req.Start,
-		End:     req.End,
-		Filter:  req.Filter,
-		GroupBy: req.GroupBy,
-	}, pageGroups)
+	// Pods owned by a DaemonSet carry k8s.daemonset.name as a resource attribute,
+	// so default-groupBy gives per-daemonset phase counts automatically.
+	phaseCounts, err := m.getPerGroupPodPhaseCounts(ctx, req.Start, req.End, req.Filter, req.GroupBy, pageGroups)
 	if err != nil {
 		return nil, err
 	}

pkg/modules/inframonitoring/implinframonitoring/nodes.go

@@ -170,27 +170,29 @@ func (m *module) getNodesTableMetadata(ctx context.Context, req *inframonitoring
 // Groups absent from the result map have implicit zero counts (caller default).
 func (m *module) getPerGroupNodeConditionCounts(
 	ctx context.Context,
-	req *inframonitoringtypes.PostableNodes,
+	start, end int64,
+	filter *qbtypes.Filter,
+	groupBy []qbtypes.GroupByKey,
 	pageGroups []map[string]string,
 ) (map[string]nodeConditionCounts, error) {
-	if len(pageGroups) == 0 || len(req.GroupBy) == 0 {
+	if len(pageGroups) == 0 || len(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
+	// Merge user filter with page-groups IN clauses.
+	userFilterExpr := ""
+	if filter != nil {
+		userFilterExpr = filter.Expression
 	}
 	pageGroupsFilterExpr := buildPageGroupsFilterExpr(pageGroups)
-	filterExpr := mergeFilterExpressions(reqFilterExpr, pageGroupsFilterExpr)
+	mergedFilterExpr := mergeFilterExpressions(userFilterExpr, pageGroupsFilterExpr)
 
 	// Resolve tables. Same convention as pods.
 	adjustedStart, adjustedEnd, _, localTimeSeriesTable := telemetrymetrics.WhichTSTableToUse(
-		uint64(req.Start), uint64(req.End), nil,
+		uint64(start), uint64(end), nil,
 	)
 	samplesTable := telemetrymetrics.WhichSamplesTableToUse(
-		uint64(req.Start), uint64(req.End),
+		uint64(start), uint64(end),
 		metrictypes.UnspecifiedType, metrictypes.TimeAggregationUnspecified, nil,
 	)
 	valueCol := telemetrymetrics.ValueColumnForSamplesTable(samplesTable)
@@ -201,7 +203,7 @@ func (m *module) getPerGroupNodeConditionCounts(
 		"fingerprint",
 		fmt.Sprintf("JSONExtractString(labels, %s) AS node_name", timeSeriesFPs.Var(nodeNameAttrKey)),
 	}
-	for _, key := range req.GroupBy {
+	for _, key := range groupBy {
 		timeSeriesFPsSelectCols = append(timeSeriesFPsSelectCols,
 			fmt.Sprintf("JSONExtractString(labels, %s) AS %s", timeSeriesFPs.Var(key.Name), quoteIdentifier(key.Name)),
 		)
@@ -213,8 +215,8 @@ func (m *module) getPerGroupNodeConditionCounts(
 		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 mergedFilterExpr != "" {
+		filterClause, err := m.buildFilterClause(ctx, &qbtypes.Filter{Expression: mergedFilterExpr}, start, end)
 		if err != nil {
 			return nil, err
 		}
@@ -223,7 +225,7 @@ func (m *module) getPerGroupNodeConditionCounts(
 		}
 	}
 	timeSeriesFPsGroupBy := []string{"fingerprint", "node_name"}
-	for _, key := range req.GroupBy {
+	for _, key := range groupBy {
 		timeSeriesFPsGroupBy = append(timeSeriesFPsGroupBy, quoteIdentifier(key.Name))
 	}
 	timeSeriesFPs.GroupBy(timeSeriesFPsGroupBy...)
@@ -233,7 +235,7 @@ func (m *module) getPerGroupNodeConditionCounts(
 	latestConditionPerNode := sqlbuilder.NewSelectBuilder()
 	latestConditionPerNodeSelectCols := []string{"tsfp.node_name AS node_name"}
 	latestConditionPerNodeGroupBy := []string{"node_name"}
-	for _, key := range req.GroupBy {
+	for _, key := range groupBy {
 		col := quoteIdentifier(key.Name)
 		latestConditionPerNodeSelectCols = append(latestConditionPerNodeSelectCols, fmt.Sprintf("tsfp.%s AS %s", col, col))
 		latestConditionPerNodeGroupBy = append(latestConditionPerNodeGroupBy, col)
@@ -248,17 +250,17 @@ func (m *module) getPerGroupNodeConditionCounts(
 	))
 	latestConditionPerNode.Where(
 		latestConditionPerNode.E("samples.metric_name", nodeConditionMetricName),
-		latestConditionPerNode.GE("samples.unix_milli", req.Start),
-		latestConditionPerNode.L("samples.unix_milli", req.End),
+		latestConditionPerNode.GE("samples.unix_milli", start),
+		latestConditionPerNode.L("samples.unix_milli", 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 {
+	countNodesPerConditionSelectCols := make([]string, 0, len(groupBy)+2)
+	countNodesPerConditionGroupBy := make([]string, 0, len(groupBy))
+	for _, key := range groupBy {
 		col := quoteIdentifier(key.Name)
 		countNodesPerConditionSelectCols = append(countNodesPerConditionSelectCols, col)
 		countNodesPerConditionGroupBy = append(countNodesPerConditionGroupBy, col)
@@ -289,8 +291,8 @@ func (m *module) getPerGroupNodeConditionCounts(
 
 	result := make(map[string]nodeConditionCounts)
 	for rows.Next() {
-		groupVals := make([]string, len(req.GroupBy))
-		scanPtrs := make([]any, 0, len(req.GroupBy)+2)
+		groupVals := make([]string, len(groupBy))
+		scanPtrs := make([]any, 0, len(groupBy)+2)
 		for i := range groupVals {
 			scanPtrs = append(scanPtrs, &groupVals[i])
 		}

pkg/modules/inframonitoring/implinframonitoring/pods.go

@@ -189,27 +189,29 @@ func (m *module) getPodsTableMetadata(ctx context.Context, req *inframonitoringt
 // Groups absent from the result map have implicit zero counts (caller default).
 func (m *module) getPerGroupPodPhaseCounts(
 	ctx context.Context,
-	req *inframonitoringtypes.PostablePods,
+	start, end int64,
+	filter *qbtypes.Filter,
+	groupBy []qbtypes.GroupByKey,
 	pageGroups []map[string]string,
 ) (map[string]podPhaseCounts, error) {
-	if len(pageGroups) == 0 || len(req.GroupBy) == 0 {
+	if len(pageGroups) == 0 || len(groupBy) == 0 {
 		return map[string]podPhaseCounts{}, nil
 	}
 
-	// Merged filter expression (user filter + page-groups IN clauses).
-	reqFilterExpr := ""
-	if req.Filter != nil {
-		reqFilterExpr = req.Filter.Expression
+	// Merge user filter with page-groups IN clauses.
+	userFilterExpr := ""
+	if filter != nil {
+		userFilterExpr = filter.Expression
 	}
 	pageGroupsFilterExpr := buildPageGroupsFilterExpr(pageGroups)
-	filterExpr := mergeFilterExpressions(reqFilterExpr, pageGroupsFilterExpr)
+	mergedFilterExpr := mergeFilterExpressions(userFilterExpr, pageGroupsFilterExpr)
 
 	// Resolve tables. Same convention as hosts (distributed names from helpers).
 	adjustedStart, adjustedEnd, _, localTimeSeriesTable := telemetrymetrics.WhichTSTableToUse(
-		uint64(req.Start), uint64(req.End), nil,
+		uint64(start), uint64(end), nil,
 	)
 	samplesTable := telemetrymetrics.WhichSamplesTableToUse(
-		uint64(req.Start), uint64(req.End),
+		uint64(start), uint64(end),
 		metrictypes.UnspecifiedType, metrictypes.TimeAggregationUnspecified, nil,
 	)
 	valueCol := telemetrymetrics.ValueColumnForSamplesTable(samplesTable)
@@ -220,7 +222,7 @@ func (m *module) getPerGroupPodPhaseCounts(
 		"fingerprint",
 		fmt.Sprintf("JSONExtractString(labels, %s) AS pod_uid", timeSeriesFPs.Var(podUIDAttrKey)),
 	}
-	for _, key := range req.GroupBy {
+	for _, key := range groupBy {
 		timeSeriesFPsSelectCols = append(timeSeriesFPsSelectCols,
 			fmt.Sprintf("JSONExtractString(labels, %s) AS %s", timeSeriesFPs.Var(key.Name), quoteIdentifier(key.Name)),
 		)
@@ -232,8 +234,8 @@ func (m *module) getPerGroupPodPhaseCounts(
 		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 mergedFilterExpr != "" {
+		filterClause, err := m.buildFilterClause(ctx, &qbtypes.Filter{Expression: mergedFilterExpr}, start, end)
 		if err != nil {
 			return nil, err
 		}
@@ -242,7 +244,7 @@ func (m *module) getPerGroupPodPhaseCounts(
 		}
 	}
 	timeSeriesFPsGroupBy := []string{"fingerprint", "pod_uid"}
-	for _, key := range req.GroupBy {
+	for _, key := range groupBy {
 		timeSeriesFPsGroupBy = append(timeSeriesFPsGroupBy, quoteIdentifier(key.Name))
 	}
 	timeSeriesFPs.GroupBy(timeSeriesFPsGroupBy...)
@@ -251,7 +253,7 @@ func (m *module) getPerGroupPodPhaseCounts(
 	latestPhasePerPod := sqlbuilder.NewSelectBuilder()
 	latestPhasePerPodSelectCols := []string{"tsfp.pod_uid AS pod_uid"}
 	latestPhasePerPodGroupBy := []string{"pod_uid"}
-	for _, key := range req.GroupBy {
+	for _, key := range groupBy {
 		col := quoteIdentifier(key.Name)
 		latestPhasePerPodSelectCols = append(latestPhasePerPodSelectCols, fmt.Sprintf("tsfp.%s AS %s", col, col))
 		latestPhasePerPodGroupBy = append(latestPhasePerPodGroupBy, col)
@@ -266,17 +268,17 @@ func (m *module) getPerGroupPodPhaseCounts(
 	))
 	latestPhasePerPod.Where(
 		latestPhasePerPod.E("samples.metric_name", podPhaseMetricName),
-		latestPhasePerPod.GE("samples.unix_milli", req.Start),
-		latestPhasePerPod.L("samples.unix_milli", req.End),
+		latestPhasePerPod.GE("samples.unix_milli", start),
+		latestPhasePerPod.L("samples.unix_milli", end),
 		"tsfp.pod_uid != ''",
 	)
 	latestPhasePerPod.GroupBy(latestPhasePerPodGroupBy...)
 	latestPhasePerPodSQL, latestPhasePerPodArgs := latestPhasePerPod.BuildWithFlavor(sqlbuilder.ClickHouse)
 
 	// ----- countPodsPerPhase (outer SELECT) -----
-	countPodsPerPhaseSelectCols := make([]string, 0, len(req.GroupBy)+5)
-	countPodsPerPhaseGroupBy := make([]string, 0, len(req.GroupBy))
-	for _, key := range req.GroupBy {
+	countPodsPerPhaseSelectCols := make([]string, 0, len(groupBy)+5)
+	countPodsPerPhaseGroupBy := make([]string, 0, len(groupBy))
+	for _, key := range groupBy {
 		col := quoteIdentifier(key.Name)
 		countPodsPerPhaseSelectCols = append(countPodsPerPhaseSelectCols, col)
 		countPodsPerPhaseGroupBy = append(countPodsPerPhaseGroupBy, col)
@@ -310,8 +312,8 @@ func (m *module) getPerGroupPodPhaseCounts(
 
 	result := make(map[string]podPhaseCounts)
 	for rows.Next() {
-		groupVals := make([]string, len(req.GroupBy))
-		scanPtrs := make([]any, 0, len(req.GroupBy)+5)
+		groupVals := make([]string, len(groupBy))
+		scanPtrs := make([]any, 0, len(groupBy)+5)
 		for i := range groupVals {
 			scanPtrs = append(scanPtrs, &groupVals[i])
 		}