Compare commits

..

1 Commits

Author SHA1 Message Date
srikanthccv
a077ec03f4 feat(prometheus): add the Prometheus query API under a /prometheus prefix
New GET|POST /prometheus/api/v1/query_range and /prometheus/api/v1/query
endpoints (pkg/prometheus/promapi), following the Prometheus HTTP API:
float-unix or RFC3339 times, float-seconds or duration-string durations,
the {status, data, errorType, error, warnings, infos} envelope with
Prometheus' status codes, and the 11,000-point cap. Wired through
signoz.Handlers like the other domain handlers. Range queries serve
through the RangeExecutor capability when the provider has it, so a
clickhousev2-serving deployment transpiles through these endpoints too.

The promapiconformance suite replays the frozen promqltest corpus against
these endpoints with clickhousev2 as the serving provider - the two paths
nothing else exercises. Instant cases go through /query with a real time
parameter. Its ledger holds the seven known Kahan-class divergences of
transpiled coarse-step serving.

Purely additive: the existing GET /api/v1/query_range and /api/v1/query
handlers are untouched.

Assisted-by: Claude Fable 5
2026-08-13 01:00:55 +05:30
41 changed files with 8770 additions and 62 deletions

View File

@@ -58,6 +58,7 @@ jobs:
- querierai
- rawexportdata
- promqlconformance
- promapiconformance
- querierauthz
- role
- rootuser

View File

@@ -183,6 +183,7 @@ func (s *Server) createPublicServer(apiHandler *api.APIHandler, web web.Web) (*h
apiHandler.RegisterLogsRoutes(r, am)
apiHandler.RegisterIntegrationRoutes(r, am)
apiHandler.RegisterQueryRangeV3Routes(r, am)
apiHandler.RegisterInfraMetricsRoutes(r, am)
apiHandler.RegisterQueryRangeV4Routes(r, am)
apiHandler.RegisterWebSocketPaths(r, am)
apiHandler.RegisterMessagingQueuesRoutes(r, am)

View File

@@ -0,0 +1,82 @@
import axios from 'api';
import { ErrorResponseHandler } from 'api/ErrorResponseHandler';
import { AxiosError } from 'axios';
import { ErrorResponse, SuccessResponse } from 'types/api';
import { BaseAutocompleteData } from 'types/api/queryBuilder/queryAutocompleteResponse';
import { TagFilter } from 'types/api/queryBuilder/queryBuilderData';
export interface HostListPayload {
filters: TagFilter;
groupBy: BaseAutocompleteData[];
offset?: number;
limit?: number;
orderBy?: {
columnName: string;
order: 'asc' | 'desc';
} | null;
start?: number;
end?: number;
}
export interface TimeSeriesValue {
timestamp: number;
value: string;
}
export interface TimeSeries {
labels: Record<string, string>;
labelsArray: Array<Record<string, string>>;
values: TimeSeriesValue[];
}
export interface HostData {
hostName: string;
active: boolean;
os: string;
/** Present when the list API returns grouped rows or extra resource attributes. */
meta?: Record<string, string>;
cpu: number;
cpuTimeSeries: TimeSeries;
memory: number;
memoryTimeSeries: TimeSeries;
wait: number;
waitTimeSeries: TimeSeries;
load15: number;
load15TimeSeries: TimeSeries;
}
export interface HostListResponse {
status: string;
data: {
type: string;
records: HostData[];
groups: null;
total: number;
sentAnyHostMetricsData: boolean;
isSendingK8SAgentMetrics: boolean;
endTimeBeforeRetention: boolean;
};
}
export const getHostLists = async (
props: HostListPayload,
signal?: AbortSignal,
headers?: Record<string, string>,
): Promise<SuccessResponse<HostListResponse> | ErrorResponse> => {
try {
const response = await axios.post('/hosts/list', props, {
signal,
headers,
});
return {
statusCode: 200,
error: null,
message: 'Success',
payload: response.data,
params: props,
};
} catch (error) {
return ErrorResponseHandler(error as AxiosError);
}
};

View File

@@ -37,7 +37,8 @@ var daemonSetsTableMetricNamesList = []string{
"k8s.daemonset.misscheduled_nodes",
}
// Carried forward from v1 daemonSetAttrsToEnrich (removed).
// Carried forward from v1 daemonSetAttrsToEnrich
// (pkg/query-service/app/inframetrics/daemonsets.go:29-33).
var daemonSetAttrKeysForMetadata = []string{
"k8s.daemonset.name",
"k8s.namespace.name",

View File

@@ -35,7 +35,8 @@ var deploymentsTableMetricNamesList = []string{
"k8s.deployment.available",
}
// Carried forward from v1 deploymentAttrsToEnrich (removed).
// Carried forward from v1 deploymentAttrsToEnrich
// (pkg/query-service/app/inframetrics/deployments.go:29-33).
var deploymentAttrKeysForMetadata = []string{
"k8s.deployment.name",
"k8s.namespace.name",

View File

@@ -37,7 +37,8 @@ var jobsTableMetricNamesList = []string{
"k8s.job.desired_successful_pods",
}
// Carried forward from v1 jobAttrsToEnrich (removed).
// Carried forward from v1 jobAttrsToEnrich
// (pkg/query-service/app/inframetrics/jobs.go:31-35).
var jobAttrKeysForMetadata = []string{
"k8s.job.name",
"k8s.namespace.name",

View File

@@ -35,7 +35,8 @@ var statefulSetsTableMetricNamesList = []string{
"k8s.statefulset.current_pods",
}
// Carried forward from v1 statefulSetAttrsToEnrich (removed).
// Carried forward from v1 statefulSetAttrsToEnrich
// (pkg/query-service/app/inframetrics/statefulsets.go:29-33).
var statefulSetAttrKeysForMetadata = []string{
"k8s.statefulset.name",
"k8s.namespace.name",

View File

@@ -26,7 +26,8 @@ var volumesTableMetricNamesList = []string{
"k8s.volume.inodes.used",
}
// Carried forward from v1 volumeAttrsToEnrich (removed).
// Carried forward from v1 volumeAttrsToEnrich
// (pkg/query-service/app/inframetrics/pvcs.go:23-31).
var volumeAttrKeysForMetadata = []string{
"k8s.persistentvolumeclaim.name",
"k8s.pod.uid",

View File

@@ -0,0 +1,9 @@
package prometheus
import "net/http"
type Handler interface {
Query(http.ResponseWriter, *http.Request)
QueryRange(http.ResponseWriter, *http.Request)
}

View File

@@ -0,0 +1,259 @@
// Package promapi serves the Prometheus HTTP query API over a
// prometheus.Prometheus provider: /query and /query_range in the shape of
// Prometheus' /api/v1 endpoints (https://prometheus.io/docs/prometheus/latest/querying/api/),
// intended to be mounted under a distinguishing prefix (/prometheus/api/v1)
// so PromQL-only endpoints are separate from the SigNoz query APIs. The
// request and response contracts follow Prometheus: form-encoded GET/POST
// params, {"status":"success","data":{resultType,result}} on success and
// {"status":"error","errorType","error"} with Prometheus' status codes on
// failure — so Prometheus-compatible clients can point at the prefix.
package promapi
import (
"context"
"encoding/json"
"log/slog"
"math"
"net/http"
"strconv"
"time"
promModel "github.com/prometheus/common/model"
"github.com/prometheus/prometheus/promql"
"github.com/prometheus/prometheus/promql/parser"
"github.com/prometheus/prometheus/util/stats"
"github.com/SigNoz/signoz/pkg/errors"
"github.com/SigNoz/signoz/pkg/prometheus"
)
type handler struct {
logger *slog.Logger
prom prometheus.Prometheus
}
func NewHandler(logger *slog.Logger, prom prometheus.Prometheus) prometheus.Handler {
return &handler{logger: logger, prom: prom}
}
type errorType string
const (
errBadData errorType = "bad_data"
errExec errorType = "execution"
errCanceled errorType = "canceled"
errTimeout errorType = "timeout"
errInternal errorType = "internal"
)
type queryData struct {
ResultType parser.ValueType `json:"resultType"`
Result parser.Value `json:"result"`
Stats stats.QueryStats `json:"stats,omitempty"`
}
type response struct {
Status string `json:"status"`
Data *queryData `json:"data,omitempty"`
ErrorType errorType `json:"errorType,omitempty"`
Error string `json:"error,omitempty"`
Warnings []string `json:"warnings,omitempty"`
Infos []string `json:"infos,omitempty"`
}
// QueryRange evaluates an expression over a grid: query, start, end, step,
// and optional timeout/stats params, all in Prometheus' formats.
func (h *handler) QueryRange(w http.ResponseWriter, r *http.Request) {
start, err := parseTime(r.FormValue("start"))
if err != nil {
h.respondError(r.Context(), w, errBadData, err)
return
}
end, err := parseTime(r.FormValue("end"))
if err != nil {
h.respondError(r.Context(), w, errBadData, err)
return
}
if end.Before(start) {
h.respondError(r.Context(), w, errBadData, errors.NewInvalidInputf(errors.CodeInvalidInput, "end timestamp must not be before start time"))
return
}
step, err := parseDuration(r.FormValue("step"))
if err != nil {
h.respondError(r.Context(), w, errBadData, err)
return
}
if step <= 0 {
h.respondError(r.Context(), w, errBadData, errors.NewInvalidInputf(errors.CodeInvalidInput, "zero or negative query resolution step widths are not accepted. Try a positive integer"))
return
}
// The engine materializes every point of every series; an unbounded
// grid is an unbounded allocation. 11,000 points covers 60s resolution
// for a week or 1h resolution for a year.
if end.Sub(start)/step > 11000 {
h.respondError(r.Context(), w, errBadData, errors.NewInvalidInputf(errors.CodeInvalidInput, "exceeded maximum resolution of 11,000 points per timeseries. Try decreasing the query resolution (?step=XX)"))
return
}
ctx, cancel, err := h.contextWithTimeout(r)
if err != nil {
h.respondError(r.Context(), w, errBadData, err)
return
}
defer cancel()
if h.tryRangeExecutor(ctx, w, r, start, end, step) {
return
}
qry, err := h.prom.Engine().NewRangeQuery(ctx, h.prom.Storage(), nil, r.FormValue("query"), start, end, step)
if err != nil {
h.respondError(r.Context(), w, errBadData, err)
return
}
h.exec(ctx, w, r, qry)
}
// tryRangeExecutor serves the query the way a RangeExecutor provider is
// designed to serve: evaluated inside the datastore when the shape allows.
// It reports whether the response was written.
func (h *handler) tryRangeExecutor(ctx context.Context, w http.ResponseWriter, r *http.Request, start, end time.Time, step time.Duration) bool {
re, ok := h.prom.(prometheus.RangeExecutor)
if !ok {
return false
}
matrix, served, err := re.TryExecuteRange(ctx, r.FormValue("query"), start, end, step)
if err != nil {
h.respondError(ctx, w, errExec, err)
return true
}
if !served {
return false
}
h.respond(ctx, w, &queryData{ResultType: matrix.Type(), Result: matrix}, nil, nil)
return true
}
// Query evaluates an expression at a single instant: query and optional
// time/timeout/stats params. A missing time evaluates at the server's now,
// as in Prometheus.
func (h *handler) Query(w http.ResponseWriter, r *http.Request) {
ts := time.Now()
if t := r.FormValue("time"); t != "" {
var err error
ts, err = parseTime(t)
if err != nil {
h.respondError(r.Context(), w, errBadData, err)
return
}
}
ctx, cancel, err := h.contextWithTimeout(r)
if err != nil {
h.respondError(r.Context(), w, errBadData, err)
return
}
defer cancel()
qry, err := h.prom.Engine().NewInstantQuery(ctx, h.prom.Storage(), nil, r.FormValue("query"), ts)
if err != nil {
h.respondError(r.Context(), w, errBadData, err)
return
}
h.exec(ctx, w, r, qry)
}
func (h *handler) exec(ctx context.Context, w http.ResponseWriter, r *http.Request, qry promql.Query) {
defer qry.Close()
res := qry.Exec(ctx)
if res.Err != nil {
h.logger.ErrorContext(ctx, "error evaluating promql query", errors.Attr(res.Err))
switch res.Err.(type) {
case promql.ErrQueryCanceled:
h.respondError(ctx, w, errCanceled, res.Err)
case promql.ErrQueryTimeout:
h.respondError(ctx, w, errTimeout, res.Err)
case promql.ErrStorage:
h.respondError(ctx, w, errInternal, res.Err)
default:
h.respondError(ctx, w, errExec, res.Err)
}
return
}
data := &queryData{ResultType: res.Value.Type(), Result: res.Value}
if r.FormValue("stats") != "" {
data.Stats = stats.NewQueryStats(qry.Stats())
}
warnings, infos := res.Warnings.AsStrings(r.FormValue("query"), 10, 10)
h.respond(ctx, w, data, warnings, infos)
}
func (h *handler) contextWithTimeout(r *http.Request) (context.Context, context.CancelFunc, error) {
ctx := r.Context()
if to := r.FormValue("timeout"); to != "" {
timeout, err := parseDuration(to)
if err != nil {
return nil, nil, err
}
ctx, cancel := context.WithTimeout(ctx, timeout)
return ctx, cancel, nil
}
ctx, cancel := context.WithCancel(ctx)
return ctx, cancel, nil
}
func (h *handler) respond(ctx context.Context, w http.ResponseWriter, data *queryData, warnings, infos []string) {
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
if err := json.NewEncoder(w).Encode(&response{Status: "success", Data: data, Warnings: warnings, Infos: infos}); err != nil {
h.logger.ErrorContext(ctx, "error writing prometheus api response", errors.Attr(err))
}
}
// respondError follows Prometheus' status-code mapping: bad_data 400,
// execution 422, canceled/timeout 503, internal 500.
func (h *handler) respondError(ctx context.Context, w http.ResponseWriter, typ errorType, err error) {
code := http.StatusInternalServerError
switch typ {
case errBadData:
code = http.StatusBadRequest
case errExec:
code = http.StatusUnprocessableEntity
case errCanceled, errTimeout:
code = http.StatusServiceUnavailable
}
w.Header().Set("Content-Type", "application/json")
w.WriteHeader(code)
if encErr := json.NewEncoder(w).Encode(&response{Status: "error", ErrorType: typ, Error: err.Error()}); encErr != nil {
h.logger.ErrorContext(ctx, "error writing prometheus api error response", errors.Attr(encErr))
}
}
// parseTime accepts Prometheus' time formats: float unix seconds or RFC3339.
func parseTime(s string) (time.Time, error) {
if t, err := strconv.ParseFloat(s, 64); err == nil {
sec, ns := math.Modf(t)
return time.Unix(int64(sec), int64(ns*float64(time.Second))), nil
}
if t, err := time.Parse(time.RFC3339Nano, s); err == nil {
return t, nil
}
return time.Time{}, errors.NewInvalidInputf(errors.CodeInvalidInput, "cannot parse %q to a valid timestamp", s)
}
// parseDuration accepts Prometheus' duration formats: float seconds or a
// duration string like 5m.
func parseDuration(s string) (time.Duration, error) {
if d, err := strconv.ParseFloat(s, 64); err == nil {
ts := d * float64(time.Second)
if ts > float64(math.MaxInt64) || ts < float64(math.MinInt64) {
return 0, errors.NewInvalidInputf(errors.CodeInvalidInput, "cannot parse %q to a valid duration. It overflows int64", s)
}
return time.Duration(ts), nil
}
if d, err := promModel.ParseDuration(s); err == nil {
return time.Duration(d), nil
}
return 0, errors.NewInvalidInputf(errors.CodeInvalidInput, "cannot parse %q to a valid duration", s)
}

View File

@@ -3334,6 +3334,60 @@ func (r *ClickHouseReader) GetMetricMetadata(ctx context.Context, orgID valuer.U
}, nil
}
// GetCountOfThings returns the count of things in the query
// This is a generic function that can be used to check if any data exists for a given query
func (r *ClickHouseReader) GetCountOfThings(ctx context.Context, query string) (uint64, error) {
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "clickhouse-reader",
instrumentationtypes.CodeFunctionName: "GetCountOfThings",
})
var count uint64
err := r.db.QueryRow(ctx, query).Scan(&count)
if err != nil {
return 0, err
}
return count, nil
}
func (r *ClickHouseReader) GetActiveHostsFromMetricMetadata(ctx context.Context, metricNames []string, hostNameAttr string, sinceUnixMilli int64) (map[string]bool, error) {
activeHosts := map[string]bool{}
query := fmt.Sprintf(
`SELECT DISTINCT attr_string_value
FROM %s.%s
WHERE metric_name IN @metricNames
AND attr_name = @attrName
AND last_reported_unix_milli >= @sinceUnixMilli`,
signozMetricDBName,
constants.SIGNOZ_METADATA_TABLENAME,
)
rows, err := r.db.Query(ctx, query,
clickhouse.Named("metricNames", metricNames),
clickhouse.Named("attrName", hostNameAttr),
clickhouse.Named("sinceUnixMilli", sinceUnixMilli),
)
if err != nil {
return nil, errorsV2.WrapInternalf(err, errorsV2.CodeInternal, "error querying active hosts")
}
defer rows.Close()
for rows.Next() {
var hostName string
if err := rows.Scan(&hostName); err != nil {
return nil, errorsV2.WrapInternalf(err, errorsV2.CodeInternal, "error scanning active host row")
}
if hostName != "" {
activeHosts[hostName] = true
}
}
if err := rows.Err(); err != nil {
return nil, errorsV2.WrapInternalf(err, errorsV2.CodeInternal, "error iterating active host rows")
}
return activeHosts, nil
}
func (r *ClickHouseReader) GetLatestReceivedMetric(
ctx context.Context, orgID valuer.UUID, metricNames []string, labelValues map[string]string,
) (*model.MetricStatus, *model.ApiError) {
@@ -4166,6 +4220,33 @@ func (r *ClickHouseReader) GetListResultV3(ctx context.Context, query string) ([
}
// GetHostMetricsExistenceAndEarliestTime returns (count, minFirstReportedUnixMilli, error) for the given host metric names
// from distributed_metadata. When count is 0, minFirstReportedUnixMilli is 0.
func (r *ClickHouseReader) GetMetricsExistenceAndEarliestTime(ctx context.Context, metricNames []string) (uint64, uint64, error) {
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.TelemetrySignal: telemetrytypes.SignalMetrics.StringValue(),
instrumentationtypes.CodeNamespace: "clickhouse-reader",
instrumentationtypes.CodeFunctionName: "GetMetricsExistenceAndEarliestTime",
})
if len(metricNames) == 0 {
return 0, 0, nil
}
query := fmt.Sprintf(
`SELECT count(*) AS cnt, min(first_reported_unix_milli) AS min_first_reported
FROM %s.%s
WHERE metric_name IN @metric_names`,
constants.SIGNOZ_METRIC_DBNAME, constants.SIGNOZ_METADATA_TABLENAME)
var count, minFirstReported uint64
err := r.db.QueryRow(ctx, query, clickhouse.Named("metric_names", metricNames)).Scan(&count, &minFirstReported)
if err != nil {
r.logger.Error("error getting host metrics existence and earliest time", errorsV2.Attr(err))
return 0, 0, err
}
return count, minFirstReported, nil
}
func getPersonalisedError(err error) error {
if err == nil {
return nil

View File

@@ -44,6 +44,7 @@ import (
"github.com/SigNoz/signoz/pkg/contextlinks"
traceFunnelsModule "github.com/SigNoz/signoz/pkg/modules/tracefunnel"
"github.com/SigNoz/signoz/pkg/query-service/agentConf"
"github.com/SigNoz/signoz/pkg/query-service/app/inframetrics"
queues2 "github.com/SigNoz/signoz/pkg/query-service/app/integrations/messagingQueues/queues"
"github.com/SigNoz/signoz/pkg/query-service/app/logs"
logsv3 "github.com/SigNoz/signoz/pkg/query-service/app/logs/v3"
@@ -119,6 +120,20 @@ type APIHandler struct {
// Websocket connection upgrader
Upgrader *websocket.Upgrader
hostsRepo *inframetrics.HostsRepo
processesRepo *inframetrics.ProcessesRepo
podsRepo *inframetrics.PodsRepo
nodesRepo *inframetrics.NodesRepo
namespacesRepo *inframetrics.NamespacesRepo
clustersRepo *inframetrics.ClustersRepo
// workloads
deploymentsRepo *inframetrics.DeploymentsRepo
daemonsetsRepo *inframetrics.DaemonSetsRepo
statefulsetsRepo *inframetrics.StatefulSetsRepo
jobsRepo *inframetrics.JobsRepo
pvcsRepo *inframetrics.PvcsRepo
LicensingAPI licensing.API
QueryParserAPI *queryparser.API
@@ -165,6 +180,17 @@ func NewAPIHandler(opts APIHandlerOpts, config signoz.Config) (*APIHandler, erro
querier := querier.NewQuerier(querierOpts)
querierv2 := querierV2.NewQuerier(querierOptsV2)
hostsRepo := inframetrics.NewHostsRepo(opts.Reader, querierv2)
processesRepo := inframetrics.NewProcessesRepo(opts.Reader, querierv2)
podsRepo := inframetrics.NewPodsRepo(opts.Reader, querierv2)
nodesRepo := inframetrics.NewNodesRepo(opts.Reader, querierv2)
namespacesRepo := inframetrics.NewNamespacesRepo(opts.Reader, querierv2)
clustersRepo := inframetrics.NewClustersRepo(opts.Reader, querierv2)
deploymentsRepo := inframetrics.NewDeploymentsRepo(opts.Reader, querierv2)
daemonsetsRepo := inframetrics.NewDaemonSetsRepo(opts.Reader, querierv2)
statefulsetsRepo := inframetrics.NewStatefulSetsRepo(opts.Reader, querierv2)
jobsRepo := inframetrics.NewJobsRepo(opts.Reader, querierv2)
pvcsRepo := inframetrics.NewPvcsRepo(opts.Reader, querierv2)
//quickFilterModule := quickfilter.NewAPI(opts.QuickFilterModule)
aH := &APIHandler{
@@ -176,6 +202,17 @@ func NewAPIHandler(opts APIHandlerOpts, config signoz.Config) (*APIHandler, erro
LogsParsingPipelineController: opts.LogsParsingPipelineController,
querier: querier,
querierV2: querierv2,
hostsRepo: hostsRepo,
processesRepo: processesRepo,
podsRepo: podsRepo,
nodesRepo: nodesRepo,
namespacesRepo: namespacesRepo,
clustersRepo: clustersRepo,
deploymentsRepo: deploymentsRepo,
daemonsetsRepo: daemonsetsRepo,
statefulsetsRepo: statefulsetsRepo,
jobsRepo: jobsRepo,
pvcsRepo: pvcsRepo,
LicensingAPI: opts.LicensingAPI,
Signoz: opts.Signoz,
QueryParserAPI: opts.QueryParserAPI,
@@ -367,6 +404,66 @@ func (aH *APIHandler) RegisterQueryRangeV3Routes(router *mux.Router, am *middlew
subRouter.HandleFunc("/logs/livetail", am.ViewAccess(aH.Signoz.Handlers.QuerierHandler.QueryRawStream)).Methods(http.MethodGet)
}
func (aH *APIHandler) RegisterInfraMetricsRoutes(router *mux.Router, am *middleware.AuthZ) {
hostsSubRouter := router.PathPrefix("/api/v1/hosts").Subrouter()
hostsSubRouter.HandleFunc("/attribute_keys", am.ViewAccess(aH.getHostAttributeKeys)).Methods(http.MethodGet)
hostsSubRouter.HandleFunc("/attribute_values", am.ViewAccess(aH.getHostAttributeValues)).Methods(http.MethodGet)
hostsSubRouter.HandleFunc("/list", am.ViewAccess(aH.getHostList)).Methods(http.MethodPost)
processesSubRouter := router.PathPrefix("/api/v1/processes").Subrouter()
processesSubRouter.HandleFunc("/attribute_keys", am.ViewAccess(aH.getProcessAttributeKeys)).Methods(http.MethodGet)
processesSubRouter.HandleFunc("/attribute_values", am.ViewAccess(aH.getProcessAttributeValues)).Methods(http.MethodGet)
processesSubRouter.HandleFunc("/list", am.ViewAccess(aH.getProcessList)).Methods(http.MethodPost)
podsSubRouter := router.PathPrefix("/api/v1/pods").Subrouter()
podsSubRouter.HandleFunc("/attribute_keys", am.ViewAccess(aH.getPodAttributeKeys)).Methods(http.MethodGet)
podsSubRouter.HandleFunc("/attribute_values", am.ViewAccess(aH.getPodAttributeValues)).Methods(http.MethodGet)
podsSubRouter.HandleFunc("/list", am.ViewAccess(aH.getPodList)).Methods(http.MethodPost)
pvcsSubRouter := router.PathPrefix("/api/v1/pvcs").Subrouter()
pvcsSubRouter.HandleFunc("/attribute_keys", am.ViewAccess(aH.getPvcAttributeKeys)).Methods(http.MethodGet)
pvcsSubRouter.HandleFunc("/attribute_values", am.ViewAccess(aH.getPvcAttributeValues)).Methods(http.MethodGet)
pvcsSubRouter.HandleFunc("/list", am.ViewAccess(aH.getPvcList)).Methods(http.MethodPost)
nodesSubRouter := router.PathPrefix("/api/v1/nodes").Subrouter()
nodesSubRouter.HandleFunc("/attribute_keys", am.ViewAccess(aH.getNodeAttributeKeys)).Methods(http.MethodGet)
nodesSubRouter.HandleFunc("/attribute_values", am.ViewAccess(aH.getNodeAttributeValues)).Methods(http.MethodGet)
nodesSubRouter.HandleFunc("/list", am.ViewAccess(aH.getNodeList)).Methods(http.MethodPost)
namespacesSubRouter := router.PathPrefix("/api/v1/namespaces").Subrouter()
namespacesSubRouter.HandleFunc("/attribute_keys", am.ViewAccess(aH.getNamespaceAttributeKeys)).Methods(http.MethodGet)
namespacesSubRouter.HandleFunc("/attribute_values", am.ViewAccess(aH.getNamespaceAttributeValues)).Methods(http.MethodGet)
namespacesSubRouter.HandleFunc("/list", am.ViewAccess(aH.getNamespaceList)).Methods(http.MethodPost)
clustersSubRouter := router.PathPrefix("/api/v1/clusters").Subrouter()
clustersSubRouter.HandleFunc("/attribute_keys", am.ViewAccess(aH.getClusterAttributeKeys)).Methods(http.MethodGet)
clustersSubRouter.HandleFunc("/attribute_values", am.ViewAccess(aH.getClusterAttributeValues)).Methods(http.MethodGet)
clustersSubRouter.HandleFunc("/list", am.ViewAccess(aH.getClusterList)).Methods(http.MethodPost)
deploymentsSubRouter := router.PathPrefix("/api/v1/deployments").Subrouter()
deploymentsSubRouter.HandleFunc("/attribute_keys", am.ViewAccess(aH.getDeploymentAttributeKeys)).Methods(http.MethodGet)
deploymentsSubRouter.HandleFunc("/attribute_values", am.ViewAccess(aH.getDeploymentAttributeValues)).Methods(http.MethodGet)
deploymentsSubRouter.HandleFunc("/list", am.ViewAccess(aH.getDeploymentList)).Methods(http.MethodPost)
daemonsetsSubRouter := router.PathPrefix("/api/v1/daemonsets").Subrouter()
daemonsetsSubRouter.HandleFunc("/attribute_keys", am.ViewAccess(aH.getDaemonSetAttributeKeys)).Methods(http.MethodGet)
daemonsetsSubRouter.HandleFunc("/attribute_values", am.ViewAccess(aH.getDaemonSetAttributeValues)).Methods(http.MethodGet)
daemonsetsSubRouter.HandleFunc("/list", am.ViewAccess(aH.getDaemonSetList)).Methods(http.MethodPost)
statefulsetsSubRouter := router.PathPrefix("/api/v1/statefulsets").Subrouter()
statefulsetsSubRouter.HandleFunc("/attribute_keys", am.ViewAccess(aH.getStatefulSetAttributeKeys)).Methods(http.MethodGet)
statefulsetsSubRouter.HandleFunc("/attribute_values", am.ViewAccess(aH.getStatefulSetAttributeValues)).Methods(http.MethodGet)
statefulsetsSubRouter.HandleFunc("/list", am.ViewAccess(aH.getStatefulSetList)).Methods(http.MethodPost)
jobsSubRouter := router.PathPrefix("/api/v1/jobs").Subrouter()
jobsSubRouter.HandleFunc("/attribute_keys", am.ViewAccess(aH.getJobAttributeKeys)).Methods(http.MethodGet)
jobsSubRouter.HandleFunc("/attribute_values", am.ViewAccess(aH.getJobAttributeValues)).Methods(http.MethodGet)
jobsSubRouter.HandleFunc("/list", am.ViewAccess(aH.getJobList)).Methods(http.MethodPost)
infraOnboardingSubRouter := router.PathPrefix("/api/v1/infra_onboarding").Subrouter()
infraOnboardingSubRouter.HandleFunc("/k8s/status", am.ViewAccess(aH.getK8sInfraOnboardingStatus)).Methods(http.MethodGet)
}
func (aH *APIHandler) RegisterWebSocketPaths(router *mux.Router, am *middleware.AuthZ) {
subRouter := router.PathPrefix("/ws").Subrouter()
subRouter.HandleFunc("/query_progress", am.ViewAccess(aH.GetQueryProgressUpdates)).Methods(http.MethodGet)
@@ -387,6 +484,9 @@ func (aH *APIHandler) Respond(w http.ResponseWriter, data interface{}) {
func (aH *APIHandler) RegisterRoutes(router *mux.Router, am *middleware.AuthZ) {
router.HandleFunc("/api/v1/query_range", am.ViewAccess(aH.queryRangeMetrics)).Methods(http.MethodGet)
router.HandleFunc("/api/v1/query", am.ViewAccess(aH.queryMetrics)).Methods(http.MethodGet)
router.HandleFunc("/prometheus/api/v1/query_range", am.ViewAccess(aH.Signoz.Handlers.PrometheusHandler.QueryRange)).Methods(http.MethodGet, http.MethodPost)
router.HandleFunc("/prometheus/api/v1/query", am.ViewAccess(aH.Signoz.Handlers.PrometheusHandler.Query)).Methods(http.MethodGet, http.MethodPost)
router.HandleFunc("/api/v1/rules", am.ViewAccess(aH.listRules)).Methods(http.MethodGet)
router.HandleFunc("/api/v1/rules/{id}", am.ViewAccess(aH.getRule)).Methods(http.MethodGet)
router.HandleFunc("/api/v1/rules", am.EditAccess(aH.createRule)).Methods(http.MethodPost)

View File

@@ -0,0 +1,908 @@
package app
import (
"encoding/json"
"net/http"
"github.com/SigNoz/signoz/pkg/http/render"
"github.com/SigNoz/signoz/pkg/query-service/model"
"github.com/SigNoz/signoz/pkg/types/authtypes"
"github.com/SigNoz/signoz/pkg/valuer"
)
func (aH *APIHandler) getHostAttributeKeys(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
req, err := parseFilterAttributeKeyRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
// get attribute keys
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
keys, err := aH.hostsRepo.GetHostAttributeKeys(ctx, orgID, *req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
// write response
aH.Respond(w, keys)
}
func (aH *APIHandler) getHostAttributeValues(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
// parse request
req, err := parseFilterAttributeValueRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
// get attribute values
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
values, err := aH.hostsRepo.GetHostAttributeValues(ctx, orgID, *req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
// write response
aH.Respond(w, values)
}
func (aH *APIHandler) getHostList(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
req := model.HostListRequest{}
// parse request
err = json.NewDecoder(r.Body).Decode(&req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
// get host list
hostList, err := aH.hostsRepo.GetHostList(ctx, orgID, req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
// write response
aH.Respond(w, hostList)
}
func (aH *APIHandler) getProcessAttributeKeys(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
req, err := parseFilterAttributeKeyRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
keys, err := aH.processesRepo.GetProcessAttributeKeys(ctx, orgID, *req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, keys)
}
func (aH *APIHandler) getProcessAttributeValues(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
req, err := parseFilterAttributeValueRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
values, err := aH.processesRepo.GetProcessAttributeValues(ctx, orgID, *req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, values)
}
func (aH *APIHandler) getProcessList(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
req := model.ProcessListRequest{}
err = json.NewDecoder(r.Body).Decode(&req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
hostList, err := aH.processesRepo.GetProcessList(ctx, orgID, req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, hostList)
}
func (aH *APIHandler) getPodAttributeKeys(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
req, err := parseFilterAttributeKeyRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
keys, err := aH.podsRepo.GetPodAttributeKeys(ctx, orgID, *req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, keys)
}
func (aH *APIHandler) getPodAttributeValues(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
req, err := parseFilterAttributeValueRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
values, err := aH.podsRepo.GetPodAttributeValues(ctx, orgID, *req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, values)
}
func (aH *APIHandler) getPodList(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
req := model.PodListRequest{}
err = json.NewDecoder(r.Body).Decode(&req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
podList, err := aH.podsRepo.GetPodList(ctx, orgID, req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, podList)
}
func (aH *APIHandler) getNodeAttributeKeys(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
req, err := parseFilterAttributeKeyRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
keys, err := aH.nodesRepo.GetNodeAttributeKeys(ctx, orgID, *req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, keys)
}
func (aH *APIHandler) getNodeAttributeValues(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
req, err := parseFilterAttributeValueRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
values, err := aH.nodesRepo.GetNodeAttributeValues(ctx, orgID, *req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, values)
}
func (aH *APIHandler) getNodeList(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
req := model.NodeListRequest{}
err = json.NewDecoder(r.Body).Decode(&req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
nodeList, err := aH.nodesRepo.GetNodeList(ctx, orgID, req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, nodeList)
}
func (aH *APIHandler) getNamespaceAttributeKeys(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
req, err := parseFilterAttributeKeyRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
keys, err := aH.namespacesRepo.GetNamespaceAttributeKeys(ctx, *req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, keys)
}
func (aH *APIHandler) getNamespaceAttributeValues(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
req, err := parseFilterAttributeValueRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
values, err := aH.namespacesRepo.GetNamespaceAttributeValues(ctx, orgID, *req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, values)
}
func (aH *APIHandler) getNamespaceList(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
req := model.NamespaceListRequest{}
err = json.NewDecoder(r.Body).Decode(&req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
namespaceList, err := aH.namespacesRepo.GetNamespaceList(ctx, orgID, req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, namespaceList)
}
func (aH *APIHandler) getClusterAttributeKeys(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
req, err := parseFilterAttributeKeyRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
keys, err := aH.clustersRepo.GetClusterAttributeKeys(ctx, orgID, *req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, keys)
}
func (aH *APIHandler) getClusterAttributeValues(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
req, err := parseFilterAttributeValueRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
values, err := aH.clustersRepo.GetClusterAttributeValues(ctx, orgID, *req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, values)
}
func (aH *APIHandler) getClusterList(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
req := model.ClusterListRequest{}
err = json.NewDecoder(r.Body).Decode(&req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
clusterList, err := aH.clustersRepo.GetClusterList(ctx, orgID, req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, clusterList)
}
func (aH *APIHandler) getDeploymentAttributeKeys(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
req, err := parseFilterAttributeKeyRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
keys, err := aH.deploymentsRepo.GetDeploymentAttributeKeys(ctx, orgID, *req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, keys)
}
func (aH *APIHandler) getDeploymentAttributeValues(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
req, err := parseFilterAttributeValueRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
values, err := aH.deploymentsRepo.GetDeploymentAttributeValues(ctx, orgID, *req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, values)
}
func (aH *APIHandler) getDeploymentList(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
req := model.DeploymentListRequest{}
err = json.NewDecoder(r.Body).Decode(&req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
deploymentList, err := aH.deploymentsRepo.GetDeploymentList(ctx, orgID, req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, deploymentList)
}
func (aH *APIHandler) getDaemonSetAttributeKeys(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
req, err := parseFilterAttributeKeyRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
keys, err := aH.daemonsetsRepo.GetDaemonSetAttributeKeys(ctx, orgID, *req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, keys)
}
func (aH *APIHandler) getDaemonSetAttributeValues(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
req, err := parseFilterAttributeValueRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
values, err := aH.daemonsetsRepo.GetDaemonSetAttributeValues(ctx, orgID, *req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, values)
}
func (aH *APIHandler) getDaemonSetList(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
req := model.DaemonSetListRequest{}
err = json.NewDecoder(r.Body).Decode(&req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
daemonSetList, err := aH.daemonsetsRepo.GetDaemonSetList(ctx, orgID, req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, daemonSetList)
}
func (aH *APIHandler) getStatefulSetAttributeKeys(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
req, err := parseFilterAttributeKeyRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
keys, err := aH.statefulsetsRepo.GetStatefulSetAttributeKeys(ctx, orgID, *req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, keys)
}
func (aH *APIHandler) getStatefulSetAttributeValues(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
req, err := parseFilterAttributeValueRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
values, err := aH.statefulsetsRepo.GetStatefulSetAttributeValues(ctx, orgID, *req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, values)
}
func (aH *APIHandler) getStatefulSetList(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
req := model.StatefulSetListRequest{}
err = json.NewDecoder(r.Body).Decode(&req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
statefulSetList, err := aH.statefulsetsRepo.GetStatefulSetList(ctx, orgID, req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, statefulSetList)
}
func (aH *APIHandler) getJobAttributeKeys(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
req, err := parseFilterAttributeKeyRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
keys, err := aH.jobsRepo.GetJobAttributeKeys(ctx, orgID, *req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, keys)
}
func (aH *APIHandler) getJobAttributeValues(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
req, err := parseFilterAttributeValueRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
values, err := aH.jobsRepo.GetJobAttributeValues(ctx, orgID, *req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, values)
}
func (aH *APIHandler) getJobList(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
req := model.JobListRequest{}
err = json.NewDecoder(r.Body).Decode(&req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
jobList, err := aH.jobsRepo.GetJobList(ctx, orgID, req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, jobList)
}
func (aH *APIHandler) getPvcList(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID, err := valuer.NewUUID(claims.OrgID)
if err != nil {
render.Error(w, err)
return
}
req := model.VolumeListRequest{}
err = json.NewDecoder(r.Body).Decode(&req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
pvcList, err := aH.pvcsRepo.GetPvcList(ctx, orgID, req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, pvcList)
}
func (aH *APIHandler) getPvcAttributeKeys(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
req, err := parseFilterAttributeKeyRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
keys, err := aH.pvcsRepo.GetPvcAttributeKeys(ctx, orgID, *req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, keys)
}
func (aH *APIHandler) getPvcAttributeValues(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
req, err := parseFilterAttributeValueRequest(r)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
claims, err := authtypes.ClaimsFromContext(r.Context())
if err != nil {
render.Error(w, err)
return
}
orgID := valuer.MustNewUUID(claims.OrgID)
values, err := aH.pvcsRepo.GetPvcAttributeValues(ctx, orgID, *req)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
aH.Respond(w, values)
}
func (aH *APIHandler) getK8sInfraOnboardingStatus(w http.ResponseWriter, r *http.Request) {
ctx := r.Context()
status := model.OnboardingStatus{}
didSendPodMetrics, err := aH.podsRepo.DidSendPodMetrics(ctx)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
if !didSendPodMetrics {
aH.Respond(w, status)
return
}
didSendClusterMetrics, err := aH.podsRepo.DidSendClusterMetrics(ctx)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
didSendNodeMetrics, err := aH.nodesRepo.DidSendNodeMetrics(ctx)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
didSendOptionalPodMetrics, err := aH.podsRepo.IsSendingOptionalPodMetrics(ctx)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
requiredMetadata, err := aH.podsRepo.SendingRequiredMetadata(ctx)
if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
return
}
status.DidSendPodMetrics = didSendPodMetrics
status.DidSendClusterMetrics = didSendClusterMetrics
status.DidSendNodeMetrics = didSendNodeMetrics
status.IsSendingOptionalPodMetrics = didSendOptionalPodMetrics
status.IsSendingRequiredMetadata = requiredMetadata
aH.Respond(w, status)
}

View File

@@ -0,0 +1,356 @@
package inframetrics
import (
"context"
"math"
"sort"
"github.com/SigNoz/signoz/pkg/query-service/app/metrics/v4/helpers"
"github.com/SigNoz/signoz/pkg/query-service/common"
"github.com/SigNoz/signoz/pkg/query-service/interfaces"
"github.com/SigNoz/signoz/pkg/query-service/model"
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
"github.com/SigNoz/signoz/pkg/query-service/postprocess"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
"github.com/SigNoz/signoz/pkg/valuer"
"golang.org/x/exp/slices"
)
var (
metricToUseForClusters = "k8s.node.cpu.usage"
clusterAttrsToEnrich = []string{"k8s.cluster.name"}
// TODO(srikanthccv): change this to k8s_cluster_uid after showing the missing data banner
k8sClusterUIDAttrKey = "k8s.cluster.name"
queryNamesForClusters = map[string][]string{
"cpu": {"A"},
"cpu_allocatable": {"B"},
"memory": {"C"},
"memory_allocatable": {"D"},
}
clusterQueryNames = []string{"A", "B", "C", "D"}
)
type ClustersRepo struct {
reader interfaces.Reader
querierV2 interfaces.Querier
}
func NewClustersRepo(reader interfaces.Reader, querierV2 interfaces.Querier) *ClustersRepo {
return &ClustersRepo{reader: reader, querierV2: querierV2}
}
func (n *ClustersRepo) GetClusterAttributeKeys(ctx context.Context, orgID valuer.UUID, req v3.FilterAttributeKeyRequest) (*v3.FilterAttributeKeyResponse, error) {
req.DataSource = v3.DataSourceMetrics
req.AggregateAttribute = metricToUseForClusters
if req.Limit == 0 {
req.Limit = 50
}
attributeKeysResponse, err := n.reader.GetMetricAttributeKeys(ctx, orgID, &req)
if err != nil {
return nil, err
}
return attributeKeysResponse, nil
}
func (n *ClustersRepo) GetClusterAttributeValues(ctx context.Context, orgID valuer.UUID, req v3.FilterAttributeValueRequest) (*v3.FilterAttributeValueResponse, error) {
req.DataSource = v3.DataSourceMetrics
req.AggregateAttribute = metricToUseForClusters
if req.Limit == 0 {
req.Limit = 50
}
attributeValuesResponse, err := n.reader.GetMetricAttributeValues(ctx, orgID, &req)
if err != nil {
return nil, err
}
return attributeValuesResponse, nil
}
func (p *ClustersRepo) getMetadataAttributes(ctx context.Context, req model.ClusterListRequest) (map[string]map[string]string, error) {
clusterAttrs := map[string]map[string]string{}
for _, key := range clusterAttrsToEnrich {
hasKey := false
for _, groupByKey := range req.GroupBy {
if groupByKey.Key == key {
hasKey = true
break
}
}
if !hasKey {
req.GroupBy = append(req.GroupBy, v3.AttributeKey{Key: key})
}
}
mq := v3.BuilderQuery{
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricToUseForClusters,
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
GroupBy: req.GroupBy,
}
query, err := helpers.PrepareTimeseriesFilterQuery(req.Start, req.End, &mq)
if err != nil {
return nil, err
}
query = localQueryToDistributedQuery(query)
attrsListResponse, err := p.reader.GetListResultV3(ctx, query)
if err != nil {
return nil, err
}
for _, row := range attrsListResponse {
stringData := map[string]string{}
for key, value := range row.Data {
if str, ok := value.(string); ok {
stringData[key] = str
} else if strPtr, ok := value.(*string); ok {
stringData[key] = *strPtr
}
}
clusterUID := stringData[k8sClusterUIDAttrKey]
if _, ok := clusterAttrs[clusterUID]; !ok {
clusterAttrs[clusterUID] = map[string]string{}
}
for _, key := range req.GroupBy {
clusterAttrs[clusterUID][key.Key] = stringData[key.Key]
}
}
return clusterAttrs, nil
}
func (p *ClustersRepo) getTopClusterGroups(ctx context.Context, orgID valuer.UUID, req model.ClusterListRequest, q *v3.QueryRangeParamsV3) ([]map[string]string, []map[string]string, error) {
step, timeSeriesTableName, samplesTableName := getParamsForTopClusters(req)
queryNames := queryNamesForClusters[req.OrderBy.ColumnName]
topClusterGroupsQueryRangeParams := &v3.QueryRangeParamsV3{
Start: req.Start,
End: req.End,
Step: step,
CompositeQuery: &v3.CompositeQuery{
BuilderQueries: map[string]*v3.BuilderQuery{},
QueryType: v3.QueryTypeBuilder,
PanelType: v3.PanelTypeTable,
},
}
for _, queryName := range queryNames {
query := q.CompositeQuery.BuilderQueries[queryName].Clone()
query.StepInterval = step
query.MetricTableHints = &v3.MetricTableHints{
TimeSeriesTableName: timeSeriesTableName,
SamplesTableName: samplesTableName,
}
if req.Filters != nil && len(req.Filters.Items) > 0 {
if query.Filters == nil {
query.Filters = &v3.FilterSet{Operator: "AND", Items: []v3.FilterItem{}}
}
query.Filters.Items = append(query.Filters.Items, req.Filters.Items...)
}
topClusterGroupsQueryRangeParams.CompositeQuery.BuilderQueries[queryName] = query
}
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "inframetrics",
instrumentationtypes.CodeFunctionName: "getTopClusterGroups",
})
queryResponse, _, err := p.querierV2.QueryRange(ctx, orgID, topClusterGroupsQueryRangeParams)
if err != nil {
return nil, nil, err
}
formattedResponse, err := postprocess.PostProcessResult(queryResponse, topClusterGroupsQueryRangeParams)
if err != nil {
return nil, nil, err
}
if len(formattedResponse) == 0 || len(formattedResponse[0].Series) == 0 {
return nil, nil, nil
}
if req.OrderBy.Order == v3.DirectionDesc {
sort.Slice(formattedResponse[0].Series, func(i, j int) bool {
return formattedResponse[0].Series[i].Points[0].Value > formattedResponse[0].Series[j].Points[0].Value
})
} else {
sort.Slice(formattedResponse[0].Series, func(i, j int) bool {
return formattedResponse[0].Series[i].Points[0].Value < formattedResponse[0].Series[j].Points[0].Value
})
}
max := math.Min(float64(req.Offset+req.Limit), float64(len(formattedResponse[0].Series)))
paginatedTopClusterGroupsSeries := formattedResponse[0].Series[req.Offset:int(max)]
topClusterGroups := []map[string]string{}
for _, series := range paginatedTopClusterGroupsSeries {
topClusterGroups = append(topClusterGroups, series.Labels)
}
allClusterGroups := []map[string]string{}
for _, series := range formattedResponse[0].Series {
allClusterGroups = append(allClusterGroups, series.Labels)
}
return topClusterGroups, allClusterGroups, nil
}
func (p *ClustersRepo) GetClusterList(ctx context.Context, orgID valuer.UUID, req model.ClusterListRequest) (model.ClusterListResponse, error) {
resp := model.ClusterListResponse{}
if req.Limit == 0 {
req.Limit = 10
}
if req.OrderBy == nil {
req.OrderBy = &v3.OrderBy{ColumnName: "cpu", Order: v3.DirectionDesc}
}
if req.GroupBy == nil {
req.GroupBy = []v3.AttributeKey{{Key: k8sClusterUIDAttrKey}}
resp.Type = model.ResponseTypeList
} else {
resp.Type = model.ResponseTypeGroupedList
}
step := int64(math.Max(float64(common.MinAllowedStepInterval(req.Start, req.End)), 60))
query := NodesTableListQuery.Clone()
query.Start = req.Start
query.End = req.End
query.Step = step
for _, query := range query.CompositeQuery.BuilderQueries {
query.StepInterval = step
if req.Filters != nil && len(req.Filters.Items) > 0 {
if query.Filters == nil {
query.Filters = &v3.FilterSet{Operator: "AND", Items: []v3.FilterItem{}}
}
query.Filters.Items = append(query.Filters.Items, req.Filters.Items...)
}
query.GroupBy = req.GroupBy
}
clusterAttrs, err := p.getMetadataAttributes(ctx, req)
if err != nil {
return resp, err
}
topClusterGroups, allClusterGroups, err := p.getTopClusterGroups(ctx, orgID, req, query)
if err != nil {
return resp, err
}
groupFilters := map[string][]string{}
for _, topClusterGroup := range topClusterGroups {
for k, v := range topClusterGroup {
groupFilters[k] = append(groupFilters[k], v)
}
}
for groupKey, groupValues := range groupFilters {
hasGroupFilter := false
if req.Filters != nil && len(req.Filters.Items) > 0 {
for _, filter := range req.Filters.Items {
if filter.Key.Key == groupKey {
hasGroupFilter = true
break
}
}
}
if !hasGroupFilter {
for _, query := range query.CompositeQuery.BuilderQueries {
query.Filters.Items = append(query.Filters.Items, v3.FilterItem{
Key: v3.AttributeKey{Key: groupKey},
Value: groupValues,
Operator: v3.FilterOperatorIn,
})
}
}
}
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "inframetrics",
instrumentationtypes.CodeFunctionName: "GetClusterList",
})
queryResponse, _, err := p.querierV2.QueryRange(ctx, orgID, query)
if err != nil {
return resp, err
}
formattedResponse, err := postprocess.PostProcessResult(queryResponse, query)
if err != nil {
return resp, err
}
records := []model.ClusterListRecord{}
for _, result := range formattedResponse {
for _, row := range result.Table.Rows {
record := model.ClusterListRecord{
CPUUsage: -1,
CPUAllocatable: -1,
MemoryUsage: -1,
MemoryAllocatable: -1,
}
if clusterUID, ok := row.Data[k8sClusterUIDAttrKey].(string); ok {
record.ClusterUID = clusterUID
}
if cpu, ok := row.Data["A"].(float64); ok {
record.CPUUsage = cpu
}
if cpuAllocatable, ok := row.Data["B"].(float64); ok {
record.CPUAllocatable = cpuAllocatable
}
if mem, ok := row.Data["C"].(float64); ok {
record.MemoryUsage = mem
}
if memoryAllocatable, ok := row.Data["D"].(float64); ok {
record.MemoryAllocatable = memoryAllocatable
}
record.Meta = map[string]string{}
if _, ok := clusterAttrs[record.ClusterUID]; ok && record.ClusterUID != "" {
record.Meta = clusterAttrs[record.ClusterUID]
}
for k, v := range row.Data {
if slices.Contains(clusterQueryNames, k) {
continue
}
if labelValue, ok := v.(string); ok {
record.Meta[k] = labelValue
}
}
records = append(records, record)
}
}
resp.Total = len(allClusterGroups)
resp.Records = records
resp.SortBy(req.OrderBy)
return resp, nil
}

View File

@@ -0,0 +1,208 @@
package inframetrics
import (
"fmt"
"strings"
"time"
"github.com/SigNoz/signoz/pkg/query-service/constants"
"github.com/SigNoz/signoz/pkg/query-service/model"
)
const fromWhereQuery = `
FROM %s.%s
WHERE metric_name IN (%s)
AND unix_milli >= toUnixTimestamp(now() - toIntervalMinute(60)) * 1000
`
var (
// TODO(srikanthccv): import metadata yaml from receivers and use generated files to check the metrics
podMetricNamesToCheck = []string{
"k8s.pod.cpu.usage",
"k8s.pod.memory.working_set",
"k8s.pod.cpu_request_utilization",
"k8s.pod.memory_request_utilization",
"k8s.pod.cpu_limit_utilization",
"k8s.pod.memory_limit_utilization",
"k8s.container.restarts",
"k8s.pod.phase",
}
nodeMetricNamesToCheck = []string{
"k8s.node.cpu.usage",
"k8s.node.allocatable_cpu",
"k8s.node.memory.working_set",
"k8s.node.allocatable_memory",
"k8s.node.condition_ready",
}
clusterMetricNamesToCheck = []string{
"k8s.daemonset.desired_scheduled_nodes",
"k8s.daemonset.current_scheduled_nodes",
"k8s.deployment.desired",
"k8s.deployment.available",
"k8s.job.desired_successful_pods",
"k8s.job.active_pods",
"k8s.job.failed_pods",
"k8s.job.successful_pods",
"k8s.statefulset.desired_pods",
"k8s.statefulset.current_pods",
}
optionalPodMetricNamesToCheck = []string{
"k8s.pod.cpu_request_utilization",
"k8s.pod.memory_request_utilization",
"k8s.pod.cpu_limit_utilization",
"k8s.pod.memory_limit_utilization",
}
// did they ever send _any_ pod metrics?
didSendPodMetricsQuery = `
SELECT count() FROM %s.%s WHERE metric_name IN (%s)
`
// did they ever send any node metrics?
didSendNodeMetricsQuery = `
SELECT count() FROM %s.%s WHERE metric_name IN (%s)
`
// did they ever send any cluster metrics?
didSendClusterMetricsQuery = `
SELECT count() FROM %s.%s WHERE metric_name IN (%s)
`
// if they ever sent _any_ pod metrics, we assume they know how to send pod metrics
// now, are they sending optional pod metrics such request/limit metrics?
isSendingOptionalPodMetricsQuery = `
SELECT count() FROM %s.%s WHERE metric_name IN (%s)
`
// there should be [cluster, node, namespace, one of (deployment, statefulset, daemonset, cronjob, job)] for each pod
selectQuery = fmt.Sprintf(`
SELECT
any(JSONExtractString(labels, '%s')) as k8s_cluster_name,
any(JSONExtractString(labels, '%s')) as k8s_node_name,
any(JSONExtractString(labels, '%s')) as k8s_namespace_name,
any(JSONExtractString(labels, '%s')) as k8s_deployment_name,
any(JSONExtractString(labels, '%s')) as k8s_statefulset_name,
any(JSONExtractString(labels, '%s')) as k8s_daemonset_name,
any(JSONExtractString(labels, '%s')) as k8s_cronjob_name,
any(JSONExtractString(labels, '%s')) as k8s_job_name,
JSONExtractString(labels, '%s') as k8s_pod_name
`,
"k8s.cluster.name",
"k8s.node.name",
"k8s.namespace.name",
"k8s.deployment.name",
"k8s.statefulset.name",
"k8s.daemonset.name",
"k8s.cronjob.name",
"k8s.job.name",
"k8s.pod.name",
)
filterGroupQuery = fmt.Sprintf(`
AND JSONExtractString(labels, '%s')
NOT IN ('kube-system','kube-public','kube-node-lease','metallb-system')
GROUP BY k8s_pod_name
LIMIT 1 BY k8s_cluster_name, k8s_node_name, k8s_namespace_name
`,
"k8s.namespace.name",
)
isSendingRequiredMetadataQuery = selectQuery + fromWhereQuery + filterGroupQuery
)
// getParamsForTopItems returns the step, time series table name and samples table name
// for the top items query. what are we doing here?
// we want to identify the top hosts/pods/nodes quickly, so we use pre-aggregated data
// for samples and time series tables to speed up the query
// the speed of the query depends on the number of values in group by clause, the higher
// the step interval, the faster the query will be as number of rows to group by is reduced
// here we are using the averaged value of the time series data to get the top items
func getParamsForTopItems(start, end int64) (int64, string, string) {
var step int64
var timeSeriesTableName string
var samplesTableName string
if end-start < time.Hour.Milliseconds() {
// 5 minute aggregation for any query less than 1 hour
step = 5 * 60
timeSeriesTableName = constants.SIGNOZ_TIMESERIES_v4_LOCAL_TABLENAME
samplesTableName = constants.SIGNOZ_SAMPLES_V4_AGG_5M_TABLENAME
} else if end-start < time.Hour.Milliseconds()*6 {
// 15 minute aggregation for any query less than 6 hours
step = 15 * 60
timeSeriesTableName = constants.SIGNOZ_TIMESERIES_v4_6HRS_LOCAL_TABLENAME
samplesTableName = constants.SIGNOZ_SAMPLES_V4_AGG_5M_TABLENAME
} else if end-start < time.Hour.Milliseconds()*24 {
// 1 hour aggregation for any query less than 1 day
step = 60 * 60
timeSeriesTableName = constants.SIGNOZ_TIMESERIES_v4_1DAY_LOCAL_TABLENAME
samplesTableName = constants.SIGNOZ_SAMPLES_V4_AGG_30M_TABLENAME
} else if end-start < time.Hour.Milliseconds()*7 {
// 6 hours aggregation for any query less than 1 week
step = 6 * 60 * 60
timeSeriesTableName = constants.SIGNOZ_TIMESERIES_v4_1WEEK_LOCAL_TABLENAME
samplesTableName = constants.SIGNOZ_SAMPLES_V4_AGG_30M_TABLENAME
} else {
// 12 hours aggregation for any query greater than 1 week
step = 12 * 60 * 60
timeSeriesTableName = constants.SIGNOZ_TIMESERIES_v4_1WEEK_LOCAL_TABLENAME
samplesTableName = constants.SIGNOZ_SAMPLES_V4_AGG_30M_TABLENAME
}
return step, timeSeriesTableName, samplesTableName
}
func getParamsForTopHosts(req model.HostListRequest) (int64, string, string) {
return getParamsForTopItems(req.Start, req.End)
}
func getParamsForTopProcesses(req model.ProcessListRequest) (int64, string, string) {
return getParamsForTopItems(req.Start, req.End)
}
func getParamsForTopPods(req model.PodListRequest) (int64, string, string) {
return getParamsForTopItems(req.Start, req.End)
}
func getParamsForTopNodes(req model.NodeListRequest) (int64, string, string) {
return getParamsForTopItems(req.Start, req.End)
}
func getParamsForTopNamespaces(req model.NamespaceListRequest) (int64, string, string) {
return getParamsForTopItems(req.Start, req.End)
}
func getParamsForTopClusters(req model.ClusterListRequest) (int64, string, string) {
return getParamsForTopItems(req.Start, req.End)
}
func getParamsForTopDeployments(req model.DeploymentListRequest) (int64, string, string) {
return getParamsForTopItems(req.Start, req.End)
}
func getParamsForTopDaemonSets(req model.DaemonSetListRequest) (int64, string, string) {
return getParamsForTopItems(req.Start, req.End)
}
func getParamsForTopStatefulSets(req model.StatefulSetListRequest) (int64, string, string) {
return getParamsForTopItems(req.Start, req.End)
}
func getParamsForTopJobs(req model.JobListRequest) (int64, string, string) {
return getParamsForTopItems(req.Start, req.End)
}
func getParamsForTopVolumes(req model.VolumeListRequest) (int64, string, string) {
return getParamsForTopItems(req.Start, req.End)
}
// TODO(srikanthccv): remove this
// What is happening here?
// The `PrepareTimeseriesFilterQuery` uses the local time series table for sub-query because each fingerprint
// goes to same shard.
// However, in this case, we are interested in the attributes values across all the shards.
// So, we replace the local time series table with the distributed time series table.
// See `PrepareTimeseriesFilterQuery` for more details.
func localQueryToDistributedQuery(query string) string {
return strings.Replace(query, ".time_series_v4", ".distributed_time_series_v4", 1)
}

View File

@@ -0,0 +1,457 @@
package inframetrics
import (
"context"
"math"
"sort"
"github.com/SigNoz/signoz/pkg/query-service/app/metrics/v4/helpers"
"github.com/SigNoz/signoz/pkg/query-service/common"
"github.com/SigNoz/signoz/pkg/query-service/interfaces"
"github.com/SigNoz/signoz/pkg/query-service/model"
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
"github.com/SigNoz/signoz/pkg/query-service/postprocess"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
"github.com/SigNoz/signoz/pkg/valuer"
"golang.org/x/exp/slices"
)
var (
metricToUseForDaemonSets = "k8s.pod.cpu.usage"
k8sDaemonSetNameAttrKey = "k8s.daemonset.name"
metricNamesForDaemonSets = map[string]string{
"desired_nodes": "k8s.daemonset.desired_scheduled_nodes",
"available_nodes": "k8s.daemonset.current_scheduled_nodes",
}
daemonSetAttrsToEnrich = []string{
"k8s.daemonset.name",
"k8s.namespace.name",
"k8s.cluster.name",
}
queryNamesForDaemonSets = map[string][]string{
"cpu": {"A"},
"cpu_request": {"B", "A"},
"cpu_limit": {"C", "A"},
"memory": {"D"},
"memory_request": {"E", "D"},
"memory_limit": {"F", "D"},
"restarts": {"G", "A"},
"desired_nodes": {"H"},
"available_nodes": {"I"},
}
builderQueriesForDaemonSets = map[string]*v3.BuilderQuery{
// desired nodes
"H": {
QueryName: "H",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForDaemonSets["desired_nodes"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{},
Expression: "H",
ReduceTo: v3.ReduceToOperatorLast,
TimeAggregation: v3.TimeAggregationAnyLast,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
// available nodes
"I": {
QueryName: "I",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForDaemonSets["available_nodes"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{},
Expression: "I",
ReduceTo: v3.ReduceToOperatorLast,
TimeAggregation: v3.TimeAggregationAnyLast,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
}
daemonSetQueryNames = []string{"A", "B", "C", "D", "E", "F", "G", "H", "I"}
)
type DaemonSetsRepo struct {
reader interfaces.Reader
querierV2 interfaces.Querier
}
func NewDaemonSetsRepo(reader interfaces.Reader, querierV2 interfaces.Querier) *DaemonSetsRepo {
return &DaemonSetsRepo{reader: reader, querierV2: querierV2}
}
func (d *DaemonSetsRepo) GetDaemonSetAttributeKeys(ctx context.Context, orgID valuer.UUID, req v3.FilterAttributeKeyRequest) (*v3.FilterAttributeKeyResponse, error) {
// TODO(srikanthccv): remove hardcoded metric name and support keys from any pod metric
req.DataSource = v3.DataSourceMetrics
req.AggregateAttribute = metricToUseForDaemonSets
if req.Limit == 0 {
req.Limit = 50
}
attributeKeysResponse, err := d.reader.GetMetricAttributeKeys(ctx, orgID, &req)
if err != nil {
return nil, err
}
// TODO(srikanthccv): only return resource attributes when we have a way to
// distinguish between resource attributes and other attributes.
filteredKeys := []v3.AttributeKey{}
for _, key := range attributeKeysResponse.AttributeKeys {
if slices.Contains(pointAttrsToIgnore, key.Key) {
continue
}
filteredKeys = append(filteredKeys, key)
}
return &v3.FilterAttributeKeyResponse{AttributeKeys: filteredKeys}, nil
}
func (d *DaemonSetsRepo) GetDaemonSetAttributeValues(ctx context.Context, orgID valuer.UUID, req v3.FilterAttributeValueRequest) (*v3.FilterAttributeValueResponse, error) {
req.DataSource = v3.DataSourceMetrics
req.AggregateAttribute = metricToUseForDaemonSets
if req.Limit == 0 {
req.Limit = 50
}
attributeValuesResponse, err := d.reader.GetMetricAttributeValues(ctx, orgID, &req)
if err != nil {
return nil, err
}
return attributeValuesResponse, nil
}
func (d *DaemonSetsRepo) getMetadataAttributes(ctx context.Context, req model.DaemonSetListRequest) (map[string]map[string]string, error) {
daemonSetAttrs := map[string]map[string]string{}
for _, key := range daemonSetAttrsToEnrich {
hasKey := false
for _, groupByKey := range req.GroupBy {
if groupByKey.Key == key {
hasKey = true
break
}
}
if !hasKey {
req.GroupBy = append(req.GroupBy, v3.AttributeKey{Key: key})
}
}
mq := v3.BuilderQuery{
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricToUseForDaemonSets,
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
GroupBy: req.GroupBy,
}
query, err := helpers.PrepareTimeseriesFilterQuery(req.Start, req.End, &mq)
if err != nil {
return nil, err
}
query = localQueryToDistributedQuery(query)
attrsListResponse, err := d.reader.GetListResultV3(ctx, query)
if err != nil {
return nil, err
}
for _, row := range attrsListResponse {
stringData := map[string]string{}
for key, value := range row.Data {
if str, ok := value.(string); ok {
stringData[key] = str
} else if strPtr, ok := value.(*string); ok {
stringData[key] = *strPtr
}
}
daemonSetName := stringData[k8sDaemonSetNameAttrKey]
if _, ok := daemonSetAttrs[daemonSetName]; !ok {
daemonSetAttrs[daemonSetName] = map[string]string{}
}
for _, key := range req.GroupBy {
daemonSetAttrs[daemonSetName][key.Key] = stringData[key.Key]
}
}
return daemonSetAttrs, nil
}
func (d *DaemonSetsRepo) getTopDaemonSetGroups(ctx context.Context, orgID valuer.UUID, req model.DaemonSetListRequest, q *v3.QueryRangeParamsV3) ([]map[string]string, []map[string]string, error) {
step, timeSeriesTableName, samplesTableName := getParamsForTopDaemonSets(req)
queryNames := queryNamesForDaemonSets[req.OrderBy.ColumnName]
topDaemonSetGroupsQueryRangeParams := &v3.QueryRangeParamsV3{
Start: req.Start,
End: req.End,
Step: step,
CompositeQuery: &v3.CompositeQuery{
BuilderQueries: map[string]*v3.BuilderQuery{},
QueryType: v3.QueryTypeBuilder,
PanelType: v3.PanelTypeTable,
},
}
for _, queryName := range queryNames {
query := q.CompositeQuery.BuilderQueries[queryName].Clone()
query.StepInterval = step
query.MetricTableHints = &v3.MetricTableHints{
TimeSeriesTableName: timeSeriesTableName,
SamplesTableName: samplesTableName,
}
if req.Filters != nil && len(req.Filters.Items) > 0 {
if query.Filters == nil {
query.Filters = &v3.FilterSet{Operator: "AND", Items: []v3.FilterItem{}}
}
query.Filters.Items = append(query.Filters.Items, req.Filters.Items...)
}
topDaemonSetGroupsQueryRangeParams.CompositeQuery.BuilderQueries[queryName] = query
}
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "inframetrics",
instrumentationtypes.CodeFunctionName: "getTopDaemonSetGroups",
})
queryResponse, _, err := d.querierV2.QueryRange(ctx, orgID, topDaemonSetGroupsQueryRangeParams)
if err != nil {
return nil, nil, err
}
formattedResponse, err := postprocess.PostProcessResult(queryResponse, topDaemonSetGroupsQueryRangeParams)
if err != nil {
return nil, nil, err
}
if len(formattedResponse) == 0 || len(formattedResponse[0].Series) == 0 {
return nil, nil, nil
}
if req.OrderBy.Order == v3.DirectionDesc {
sort.Slice(formattedResponse[0].Series, func(i, j int) bool {
return formattedResponse[0].Series[i].Points[0].Value > formattedResponse[0].Series[j].Points[0].Value
})
} else {
sort.Slice(formattedResponse[0].Series, func(i, j int) bool {
return formattedResponse[0].Series[i].Points[0].Value < formattedResponse[0].Series[j].Points[0].Value
})
}
limit := math.Min(float64(req.Offset+req.Limit), float64(len(formattedResponse[0].Series)))
paginatedTopDaemonSetGroupsSeries := formattedResponse[0].Series[req.Offset:int(limit)]
topDaemonSetGroups := []map[string]string{}
for _, series := range paginatedTopDaemonSetGroupsSeries {
topDaemonSetGroups = append(topDaemonSetGroups, series.Labels)
}
allDaemonSetGroups := []map[string]string{}
for _, series := range formattedResponse[0].Series {
allDaemonSetGroups = append(allDaemonSetGroups, series.Labels)
}
return topDaemonSetGroups, allDaemonSetGroups, nil
}
func (d *DaemonSetsRepo) GetDaemonSetList(ctx context.Context, orgID valuer.UUID, req model.DaemonSetListRequest) (model.DaemonSetListResponse, error) {
resp := model.DaemonSetListResponse{}
if req.Limit == 0 {
req.Limit = 10
}
if req.OrderBy == nil {
req.OrderBy = &v3.OrderBy{ColumnName: "cpu", Order: v3.DirectionDesc}
}
if req.GroupBy == nil {
req.GroupBy = []v3.AttributeKey{{Key: k8sDaemonSetNameAttrKey}}
resp.Type = model.ResponseTypeList
} else {
resp.Type = model.ResponseTypeGroupedList
}
step := int64(math.Max(float64(common.MinAllowedStepInterval(req.Start, req.End)), 60))
query := WorkloadTableListQuery.Clone()
query.Start = req.Start
query.End = req.End
query.Step = step
// add additional queries for daemon sets
for _, daemonSetQuery := range builderQueriesForDaemonSets {
query.CompositeQuery.BuilderQueries[daemonSetQuery.QueryName] = daemonSetQuery.Clone()
}
for _, query := range query.CompositeQuery.BuilderQueries {
query.StepInterval = step
if req.Filters != nil && len(req.Filters.Items) > 0 {
if query.Filters == nil {
query.Filters = &v3.FilterSet{Operator: "AND", Items: []v3.FilterItem{}}
}
query.Filters.Items = append(query.Filters.Items, req.Filters.Items...)
}
query.GroupBy = req.GroupBy
// make sure we only get records for daemon sets
query.Filters.Items = append(query.Filters.Items, v3.FilterItem{
Key: v3.AttributeKey{Key: k8sDaemonSetNameAttrKey},
Operator: v3.FilterOperatorExists,
})
}
daemonSetAttrs, err := d.getMetadataAttributes(ctx, req)
if err != nil {
return resp, err
}
topDaemonSetGroups, allDaemonSetGroups, err := d.getTopDaemonSetGroups(ctx, orgID, req, query)
if err != nil {
return resp, err
}
groupFilters := map[string][]string{}
for _, topDaemonSetGroup := range topDaemonSetGroups {
for k, v := range topDaemonSetGroup {
groupFilters[k] = append(groupFilters[k], v)
}
}
for groupKey, groupValues := range groupFilters {
hasGroupFilter := false
if req.Filters != nil && len(req.Filters.Items) > 0 {
for _, filter := range req.Filters.Items {
if filter.Key.Key == groupKey {
hasGroupFilter = true
break
}
}
}
if !hasGroupFilter {
for _, query := range query.CompositeQuery.BuilderQueries {
query.Filters.Items = append(query.Filters.Items, v3.FilterItem{
Key: v3.AttributeKey{Key: groupKey},
Value: groupValues,
Operator: v3.FilterOperatorIn,
})
}
}
}
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "inframetrics",
instrumentationtypes.CodeFunctionName: "GetDaemonSetList",
})
queryResponse, _, err := d.querierV2.QueryRange(ctx, orgID, query)
if err != nil {
return resp, err
}
formattedResponse, err := postprocess.PostProcessResult(queryResponse, query)
if err != nil {
return resp, err
}
records := []model.DaemonSetListRecord{}
for _, result := range formattedResponse {
for _, row := range result.Table.Rows {
record := model.DaemonSetListRecord{
DaemonSetName: "",
CPUUsage: -1,
CPURequest: -1,
CPULimit: -1,
MemoryUsage: -1,
MemoryRequest: -1,
MemoryLimit: -1,
DesiredNodes: -1,
AvailableNodes: -1,
}
if daemonSetName, ok := row.Data[k8sDaemonSetNameAttrKey].(string); ok {
record.DaemonSetName = daemonSetName
}
if cpu, ok := row.Data["A"].(float64); ok {
record.CPUUsage = cpu
}
if cpuRequest, ok := row.Data["B"].(float64); ok {
record.CPURequest = cpuRequest
}
if cpuLimit, ok := row.Data["C"].(float64); ok {
record.CPULimit = cpuLimit
}
if memory, ok := row.Data["D"].(float64); ok {
record.MemoryUsage = memory
}
if memoryRequest, ok := row.Data["E"].(float64); ok {
record.MemoryRequest = memoryRequest
}
if memoryLimit, ok := row.Data["F"].(float64); ok {
record.MemoryLimit = memoryLimit
}
if restarts, ok := row.Data["G"].(float64); ok {
record.Restarts = int(restarts)
}
if desiredNodes, ok := row.Data["H"].(float64); ok {
record.DesiredNodes = int(desiredNodes)
}
if availableNodes, ok := row.Data["I"].(float64); ok {
record.AvailableNodes = int(availableNodes)
}
record.Meta = map[string]string{}
if _, ok := daemonSetAttrs[record.DaemonSetName]; ok && record.DaemonSetName != "" {
record.Meta = daemonSetAttrs[record.DaemonSetName]
}
for k, v := range row.Data {
if slices.Contains(daemonSetQueryNames, k) {
continue
}
if labelValue, ok := v.(string); ok {
record.Meta[k] = labelValue
}
}
records = append(records, record)
}
}
resp.Total = len(allDaemonSetGroups)
resp.Records = records
resp.SortBy(req.OrderBy)
return resp, nil
}

View File

@@ -0,0 +1,453 @@
package inframetrics
import (
"context"
"math"
"sort"
"github.com/SigNoz/signoz/pkg/query-service/app/metrics/v4/helpers"
"github.com/SigNoz/signoz/pkg/query-service/common"
"github.com/SigNoz/signoz/pkg/query-service/interfaces"
"github.com/SigNoz/signoz/pkg/query-service/model"
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
"github.com/SigNoz/signoz/pkg/query-service/postprocess"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
"github.com/SigNoz/signoz/pkg/valuer"
"golang.org/x/exp/slices"
)
var (
metricToUseForDeployments = "k8s.pod.cpu.usage"
k8sDeploymentNameAttrKey = "k8s.deployment.name"
metricNamesForDeployments = map[string]string{
"desired_pods": "k8s.deployment.desired",
"available_pods": "k8s.deployment.available",
}
deploymentAttrsToEnrich = []string{
"k8s.deployment.name",
"k8s.namespace.name",
"k8s.cluster.name",
}
queryNamesForDeployments = map[string][]string{
"cpu": {"A"},
"cpu_request": {"B", "A"},
"cpu_limit": {"C", "A"},
"memory": {"D"},
"memory_request": {"E", "D"},
"memory_limit": {"F", "D"},
"restarts": {"G", "A"},
"desired_pods": {"H"},
"available_pods": {"I"},
}
builderQueriesForDeployments = map[string]*v3.BuilderQuery{
// desired pods
"H": {
QueryName: "H",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForDeployments["desired_pods"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{},
Expression: "H",
ReduceTo: v3.ReduceToOperatorLast,
TimeAggregation: v3.TimeAggregationAnyLast,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
// available pods
"I": {
QueryName: "I",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForDeployments["available_pods"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{},
Expression: "I",
ReduceTo: v3.ReduceToOperatorLast,
TimeAggregation: v3.TimeAggregationAnyLast,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
}
deploymentQueryNames = []string{"A", "B", "C", "D", "E", "F", "G", "H", "I"}
)
type DeploymentsRepo struct {
reader interfaces.Reader
querierV2 interfaces.Querier
}
func NewDeploymentsRepo(reader interfaces.Reader, querierV2 interfaces.Querier) *DeploymentsRepo {
return &DeploymentsRepo{reader: reader, querierV2: querierV2}
}
func (d *DeploymentsRepo) GetDeploymentAttributeKeys(ctx context.Context, orgID valuer.UUID, req v3.FilterAttributeKeyRequest) (*v3.FilterAttributeKeyResponse, error) {
// TODO(srikanthccv): remove hardcoded metric name and support keys from any pod metric
req.DataSource = v3.DataSourceMetrics
req.AggregateAttribute = metricToUseForDeployments
if req.Limit == 0 {
req.Limit = 50
}
attributeKeysResponse, err := d.reader.GetMetricAttributeKeys(ctx, orgID, &req)
if err != nil {
return nil, err
}
// TODO(srikanthccv): only return resource attributes when we have a way to
// distinguish between resource attributes and other attributes.
filteredKeys := []v3.AttributeKey{}
for _, key := range attributeKeysResponse.AttributeKeys {
if slices.Contains(pointAttrsToIgnore, key.Key) {
continue
}
filteredKeys = append(filteredKeys, key)
}
return &v3.FilterAttributeKeyResponse{AttributeKeys: filteredKeys}, nil
}
func (d *DeploymentsRepo) GetDeploymentAttributeValues(ctx context.Context, orgID valuer.UUID, req v3.FilterAttributeValueRequest) (*v3.FilterAttributeValueResponse, error) {
req.DataSource = v3.DataSourceMetrics
req.AggregateAttribute = metricToUseForDeployments
if req.Limit == 0 {
req.Limit = 50
}
attributeValuesResponse, err := d.reader.GetMetricAttributeValues(ctx, orgID, &req)
if err != nil {
return nil, err
}
return attributeValuesResponse, nil
}
func (d *DeploymentsRepo) getMetadataAttributes(ctx context.Context, req model.DeploymentListRequest) (map[string]map[string]string, error) {
deploymentAttrs := map[string]map[string]string{}
for _, key := range deploymentAttrsToEnrich {
hasKey := false
for _, groupByKey := range req.GroupBy {
if groupByKey.Key == key {
hasKey = true
break
}
}
if !hasKey {
req.GroupBy = append(req.GroupBy, v3.AttributeKey{Key: key})
}
}
mq := v3.BuilderQuery{
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricToUseForDeployments,
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
GroupBy: req.GroupBy,
}
query, err := helpers.PrepareTimeseriesFilterQuery(req.Start, req.End, &mq)
if err != nil {
return nil, err
}
query = localQueryToDistributedQuery(query)
attrsListResponse, err := d.reader.GetListResultV3(ctx, query)
if err != nil {
return nil, err
}
for _, row := range attrsListResponse {
stringData := map[string]string{}
for key, value := range row.Data {
if str, ok := value.(string); ok {
stringData[key] = str
} else if strPtr, ok := value.(*string); ok {
stringData[key] = *strPtr
}
}
deploymentName := stringData[k8sDeploymentNameAttrKey]
if _, ok := deploymentAttrs[deploymentName]; !ok {
deploymentAttrs[deploymentName] = map[string]string{}
}
for _, key := range req.GroupBy {
deploymentAttrs[deploymentName][key.Key] = stringData[key.Key]
}
}
return deploymentAttrs, nil
}
func (d *DeploymentsRepo) getTopDeploymentGroups(ctx context.Context, orgID valuer.UUID, req model.DeploymentListRequest, q *v3.QueryRangeParamsV3) ([]map[string]string, []map[string]string, error) {
step, timeSeriesTableName, samplesTableName := getParamsForTopDeployments(req)
queryNames := queryNamesForDeployments[req.OrderBy.ColumnName]
topDeploymentGroupsQueryRangeParams := &v3.QueryRangeParamsV3{
Start: req.Start,
End: req.End,
Step: step,
CompositeQuery: &v3.CompositeQuery{
BuilderQueries: map[string]*v3.BuilderQuery{},
QueryType: v3.QueryTypeBuilder,
PanelType: v3.PanelTypeTable,
},
}
for _, queryName := range queryNames {
query := q.CompositeQuery.BuilderQueries[queryName].Clone()
query.StepInterval = step
query.MetricTableHints = &v3.MetricTableHints{
TimeSeriesTableName: timeSeriesTableName,
SamplesTableName: samplesTableName,
}
if req.Filters != nil && len(req.Filters.Items) > 0 {
if query.Filters == nil {
query.Filters = &v3.FilterSet{Operator: "AND", Items: []v3.FilterItem{}}
}
query.Filters.Items = append(query.Filters.Items, req.Filters.Items...)
}
topDeploymentGroupsQueryRangeParams.CompositeQuery.BuilderQueries[queryName] = query
}
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "inframetrics",
instrumentationtypes.CodeFunctionName: "getTopDeploymentGroups",
})
queryResponse, _, err := d.querierV2.QueryRange(ctx, orgID, topDeploymentGroupsQueryRangeParams)
if err != nil {
return nil, nil, err
}
formattedResponse, err := postprocess.PostProcessResult(queryResponse, topDeploymentGroupsQueryRangeParams)
if err != nil {
return nil, nil, err
}
if len(formattedResponse) == 0 || len(formattedResponse[0].Series) == 0 {
return nil, nil, nil
}
if req.OrderBy.Order == v3.DirectionDesc {
sort.Slice(formattedResponse[0].Series, func(i, j int) bool {
return formattedResponse[0].Series[i].Points[0].Value > formattedResponse[0].Series[j].Points[0].Value
})
} else {
sort.Slice(formattedResponse[0].Series, func(i, j int) bool {
return formattedResponse[0].Series[i].Points[0].Value < formattedResponse[0].Series[j].Points[0].Value
})
}
limit := math.Min(float64(req.Offset+req.Limit), float64(len(formattedResponse[0].Series)))
paginatedTopDeploymentGroupsSeries := formattedResponse[0].Series[req.Offset:int(limit)]
topDeploymentGroups := []map[string]string{}
for _, series := range paginatedTopDeploymentGroupsSeries {
topDeploymentGroups = append(topDeploymentGroups, series.Labels)
}
allDeploymentGroups := []map[string]string{}
for _, series := range formattedResponse[0].Series {
allDeploymentGroups = append(allDeploymentGroups, series.Labels)
}
return topDeploymentGroups, allDeploymentGroups, nil
}
func (d *DeploymentsRepo) GetDeploymentList(ctx context.Context, orgID valuer.UUID, req model.DeploymentListRequest) (model.DeploymentListResponse, error) {
resp := model.DeploymentListResponse{}
if req.Limit == 0 {
req.Limit = 10
}
if req.OrderBy == nil {
req.OrderBy = &v3.OrderBy{ColumnName: "cpu", Order: v3.DirectionDesc}
}
if req.GroupBy == nil {
req.GroupBy = []v3.AttributeKey{{Key: k8sDeploymentNameAttrKey}}
resp.Type = model.ResponseTypeList
} else {
resp.Type = model.ResponseTypeGroupedList
}
step := int64(math.Max(float64(common.MinAllowedStepInterval(req.Start, req.End)), 60))
query := WorkloadTableListQuery.Clone()
query.Start = req.Start
query.End = req.End
query.Step = step
// add additional queries for deployments
for _, deploymentQuery := range builderQueriesForDeployments {
query.CompositeQuery.BuilderQueries[deploymentQuery.QueryName] = deploymentQuery.Clone()
}
for _, query := range query.CompositeQuery.BuilderQueries {
query.StepInterval = step
if req.Filters != nil && len(req.Filters.Items) > 0 {
if query.Filters == nil {
query.Filters = &v3.FilterSet{Operator: "AND", Items: []v3.FilterItem{}}
}
query.Filters.Items = append(query.Filters.Items, req.Filters.Items...)
}
query.GroupBy = req.GroupBy
// make sure we only get records for deployments
query.Filters.Items = append(query.Filters.Items, v3.FilterItem{
Key: v3.AttributeKey{Key: k8sDeploymentNameAttrKey},
Operator: v3.FilterOperatorExists,
})
}
deploymentAttrs, err := d.getMetadataAttributes(ctx, req)
if err != nil {
return resp, err
}
topDeploymentGroups, allDeploymentGroups, err := d.getTopDeploymentGroups(ctx, orgID, req, query)
if err != nil {
return resp, err
}
groupFilters := map[string][]string{}
for _, topDeploymentGroup := range topDeploymentGroups {
for k, v := range topDeploymentGroup {
groupFilters[k] = append(groupFilters[k], v)
}
}
for groupKey, groupValues := range groupFilters {
hasGroupFilter := false
if req.Filters != nil && len(req.Filters.Items) > 0 {
for _, filter := range req.Filters.Items {
if filter.Key.Key == groupKey {
hasGroupFilter = true
break
}
}
}
if !hasGroupFilter {
for _, query := range query.CompositeQuery.BuilderQueries {
query.Filters.Items = append(query.Filters.Items, v3.FilterItem{
Key: v3.AttributeKey{Key: groupKey},
Value: groupValues,
Operator: v3.FilterOperatorIn,
})
}
}
}
queryResponse, _, err := d.querierV2.QueryRange(ctx, orgID, query)
if err != nil {
return resp, err
}
formattedResponse, err := postprocess.PostProcessResult(queryResponse, query)
if err != nil {
return resp, err
}
records := []model.DeploymentListRecord{}
for _, result := range formattedResponse {
for _, row := range result.Table.Rows {
record := model.DeploymentListRecord{
DeploymentName: "",
CPUUsage: -1,
CPURequest: -1,
CPULimit: -1,
MemoryUsage: -1,
MemoryRequest: -1,
MemoryLimit: -1,
DesiredPods: -1,
AvailablePods: -1,
}
if deploymentName, ok := row.Data[k8sDeploymentNameAttrKey].(string); ok {
record.DeploymentName = deploymentName
}
if cpu, ok := row.Data["A"].(float64); ok {
record.CPUUsage = cpu
}
if cpuRequest, ok := row.Data["B"].(float64); ok {
record.CPURequest = cpuRequest
}
if cpuLimit, ok := row.Data["C"].(float64); ok {
record.CPULimit = cpuLimit
}
if memory, ok := row.Data["D"].(float64); ok {
record.MemoryUsage = memory
}
if memoryRequest, ok := row.Data["E"].(float64); ok {
record.MemoryRequest = memoryRequest
}
if memoryLimit, ok := row.Data["F"].(float64); ok {
record.MemoryLimit = memoryLimit
}
if restarts, ok := row.Data["G"].(float64); ok {
record.Restarts = int(restarts)
}
if desiredPods, ok := row.Data["H"].(float64); ok {
record.DesiredPods = int(desiredPods)
}
if availablePods, ok := row.Data["I"].(float64); ok {
record.AvailablePods = int(availablePods)
}
record.Meta = map[string]string{}
if _, ok := deploymentAttrs[record.DeploymentName]; ok && record.DeploymentName != "" {
record.Meta = deploymentAttrs[record.DeploymentName]
}
for k, v := range row.Data {
if slices.Contains(deploymentQueryNames, k) {
continue
}
if labelValue, ok := v.(string); ok {
record.Meta[k] = labelValue
}
}
records = append(records, record)
}
}
resp.Total = len(allDeploymentGroups)
resp.Records = records
resp.SortBy(req.OrderBy)
return resp, nil
}

View File

@@ -0,0 +1,550 @@
package inframetrics
import (
"context"
"errors"
"fmt"
"math"
"sort"
"strings"
"time"
"log/slog"
"github.com/SigNoz/signoz/pkg/query-service/app/metrics/v4/helpers"
"github.com/SigNoz/signoz/pkg/query-service/common"
"github.com/SigNoz/signoz/pkg/query-service/constants"
"github.com/SigNoz/signoz/pkg/query-service/interfaces"
"github.com/SigNoz/signoz/pkg/query-service/model"
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
"github.com/SigNoz/signoz/pkg/query-service/postprocess"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
"github.com/SigNoz/signoz/pkg/valuer"
"golang.org/x/exp/maps"
"golang.org/x/exp/slices"
)
type HostsRepo struct {
reader interfaces.Reader
querierV2 interfaces.Querier
}
var (
// we don't have a way to get the resource attributes from the current time series table
// but we only want to suggest resource attributes for system metrics,
// this is a list of attributes that we skip from all labels as they are data point attributes
// TODO(srikanthccv): remove this once we have a way to get resource attributes
pointAttrsToIgnore = []string{
"state",
"cpu",
"device",
"direction",
"mode",
"mountpoint",
"type",
"os.type",
"process.cgroup",
"process.command",
"process.command_line",
"process.executable.name",
"process.executable.path",
"process.owner",
"process.parent_pid",
"process.pid",
}
queryNamesForTopHosts = map[string][]string{
"cpu": {"A", "B", "F1"},
"memory": {"C", "D", "F2"},
"wait": {"E", "F", "F3"},
"load15": {"G"},
}
// TODO(srikanthccv): remove hardcoded metric name and support keys from any system metric
metricToUseForHostAttributes = "system.cpu.load_average.15m"
hostNameAttrKey = "host.name"
agentNameToIgnore = "k8s-infra-otel-agent"
hostAttrsToEnrich = []string{
"os.type",
}
metricNamesForHosts = map[string]string{
"filesystem": "system.filesystem.usage",
"cpu": "system.cpu.time",
"memory": "system.memory.usage",
"load15": "system.cpu.load_average.15m",
"wait": "system.cpu.time",
}
uniqueMetricNamesForHosts = []string{
"system.uptime",
"system.cpu.time",
"system.cpu.load_average.1m",
"system.cpu.load_average.5m",
"system.cpu.load_average.15m",
"system.memory.usage",
"system.paging.usage",
"system.paging.faults",
"system.paging.operations",
"system.disk.io",
"system.disk.operations",
"system.disk.io_time",
"system.disk.operation_time",
"system.disk.merged",
"system.disk.pending_operations",
"system.disk.weighted_io_time",
"system.filesystem.usage",
"system.filesystem.inodes.usage",
"system.network.io",
"system.network.errors",
"system.network.connections",
"system.network.dropped",
"system.network.packets",
"system.processes.count",
"system.processes.created",
"process.cpu.time",
"process.disk.io",
"process.memory.usage",
"process.memory.virtual",
"nfs.client.net.count",
"nfs.client.net.tcp.connection.accepted",
"nfs.client.operation.count",
"nfs.client.procedure.count",
"nfs.client.rpc.authrefresh.count",
"nfs.client.rpc.count",
"nfs.client.rpc.retransmit.count",
"nfs.server.fh.stale.count",
"nfs.server.io",
"nfs.server.net.count",
"nfs.server.net.tcp.connection.accepted",
"nfs.server.operation.count",
"nfs.server.procedure.count",
"nfs.server.repcache.requests",
"nfs.server.rpc.count",
"nfs.server.thread.count",
}
)
func NewHostsRepo(reader interfaces.Reader, querierV2 interfaces.Querier) *HostsRepo {
return &HostsRepo{reader: reader, querierV2: querierV2}
}
func (h *HostsRepo) GetHostAttributeKeys(ctx context.Context, orgID valuer.UUID, req v3.FilterAttributeKeyRequest) (*v3.FilterAttributeKeyResponse, error) {
// TODO(srikanthccv): remove hardcoded metric name and support keys from any system metric
req.DataSource = v3.DataSourceMetrics
req.AggregateAttribute = metricToUseForHostAttributes
if req.Limit == 0 {
req.Limit = 50
}
attributeKeysResponse, err := h.reader.GetMetricAttributeKeys(ctx, orgID, &req)
if err != nil {
return nil, err
}
// TODO(srikanthccv): only return resource attributes when we have a way to
// distinguish between resource attributes and other attributes.
filteredKeys := []v3.AttributeKey{}
for _, key := range attributeKeysResponse.AttributeKeys {
if slices.Contains(pointAttrsToIgnore, key.Key) {
continue
}
filteredKeys = append(filteredKeys, key)
}
return &v3.FilterAttributeKeyResponse{AttributeKeys: filteredKeys}, nil
}
func (h *HostsRepo) GetHostAttributeValues(ctx context.Context, orgID valuer.UUID, req v3.FilterAttributeValueRequest) (*v3.FilterAttributeValueResponse, error) {
req.DataSource = v3.DataSourceMetrics
req.AggregateAttribute = metricToUseForHostAttributes
if req.Limit == 0 {
req.Limit = 50
}
attributeValuesResponse, err := h.reader.GetMetricAttributeValues(ctx, orgID, &req)
if err != nil {
return nil, err
}
if req.FilterAttributeKey != hostNameAttrKey {
return attributeValuesResponse, nil
}
hostNames := []string{}
for _, attributeValue := range attributeValuesResponse.StringAttributeValues {
if strings.Contains(attributeValue, agentNameToIgnore) {
continue
}
hostNames = append(hostNames, attributeValue)
}
return &v3.FilterAttributeValueResponse{StringAttributeValues: hostNames}, nil
}
func (h *HostsRepo) getActiveHosts(ctx context.Context) (map[string]bool, error) {
tenMinAgo := time.Now().Add(-10 * time.Minute).UTC().UnixMilli()
return h.reader.GetActiveHostsFromMetricMetadata(ctx, uniqueMetricNamesForHosts, hostNameAttrKey, tenMinAgo)
}
func (h *HostsRepo) getMetadataAttributes(ctx context.Context, req model.HostListRequest) (map[string]map[string]string, error) {
hostAttrs := map[string]map[string]string{}
for _, key := range hostAttrsToEnrich {
hasKey := false
for _, groupByKey := range req.GroupBy {
if groupByKey.Key == key {
hasKey = true
break
}
}
if !hasKey {
req.GroupBy = append(req.GroupBy, v3.AttributeKey{Key: key})
}
}
mq := v3.BuilderQuery{
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricToUseForHostAttributes,
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
GroupBy: req.GroupBy,
}
query, err := helpers.PrepareTimeseriesFilterQuery(req.Start, req.End, &mq)
if err != nil {
return nil, err
}
query = localQueryToDistributedQuery(query)
attrsListResponse, err := h.reader.GetListResultV3(ctx, query)
if err != nil {
return nil, err
}
for _, row := range attrsListResponse {
stringData := map[string]string{}
for key, value := range row.Data {
if str, ok := value.(string); ok {
stringData[key] = str
} else if strPtr, ok := value.(*string); ok {
stringData[key] = *strPtr
}
}
hostName := stringData[hostNameAttrKey]
if _, ok := hostAttrs[hostName]; !ok {
hostAttrs[hostName] = map[string]string{}
}
for _, key := range req.GroupBy {
hostAttrs[hostName][key.Key] = stringData[key.Key]
}
}
return hostAttrs, nil
}
func (h *HostsRepo) getTopHostGroups(ctx context.Context, orgID valuer.UUID, req model.HostListRequest, q *v3.QueryRangeParamsV3) ([]map[string]string, []map[string]string, error) {
step, timeSeriesTableName, samplesTableName := getParamsForTopHosts(req)
queryNames := queryNamesForTopHosts[req.OrderBy.ColumnName]
topHostGroupsQueryRangeParams := &v3.QueryRangeParamsV3{
Start: req.Start,
End: req.End,
Step: step,
CompositeQuery: &v3.CompositeQuery{
BuilderQueries: map[string]*v3.BuilderQuery{},
QueryType: v3.QueryTypeBuilder,
PanelType: v3.PanelTypeTable,
},
}
for _, queryName := range queryNames {
query := q.CompositeQuery.BuilderQueries[queryName].Clone()
query.StepInterval = step
query.MetricTableHints = &v3.MetricTableHints{
TimeSeriesTableName: timeSeriesTableName,
SamplesTableName: samplesTableName,
}
if req.Filters != nil && len(req.Filters.Items) > 0 {
query.Filters.Items = append(query.Filters.Items, req.Filters.Items...)
}
topHostGroupsQueryRangeParams.CompositeQuery.BuilderQueries[queryName] = query
}
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "inframetrics",
instrumentationtypes.CodeFunctionName: "getTopHostGroups",
})
queryResponse, _, err := h.querierV2.QueryRange(ctx, orgID, topHostGroupsQueryRangeParams)
if err != nil {
return nil, nil, err
}
formattedResponse, err := postprocess.PostProcessResult(queryResponse, topHostGroupsQueryRangeParams)
if err != nil {
return nil, nil, err
}
if len(formattedResponse) == 0 || len(formattedResponse[0].Series) == 0 {
return nil, nil, nil
}
if req.OrderBy.Order == v3.DirectionDesc {
sort.Slice(formattedResponse[0].Series, func(i, j int) bool {
return formattedResponse[0].Series[i].Points[0].Value > formattedResponse[0].Series[j].Points[0].Value
})
} else {
sort.Slice(formattedResponse[0].Series, func(i, j int) bool {
return formattedResponse[0].Series[i].Points[0].Value < formattedResponse[0].Series[j].Points[0].Value
})
}
limit := math.Min(float64(req.Offset+req.Limit), float64(len(formattedResponse[0].Series)))
paginatedTopHostGroupsSeries := formattedResponse[0].Series[req.Offset:int(limit)]
topHostGroups := []map[string]string{}
for _, series := range paginatedTopHostGroupsSeries {
topHostGroups = append(topHostGroups, series.Labels)
}
allHostGroups := []map[string]string{}
for _, series := range formattedResponse[0].Series {
allHostGroups = append(allHostGroups, series.Labels)
}
return topHostGroups, allHostGroups, nil
}
// GetHostMetricsExistenceAndEarliestTime returns (count, minFirstReportedUnixMilli, error) for host metrics
// in distributed_metadata. Uses metricNamesForHosts plus system.filesystem.usage.
func (h *HostsRepo) GetHostMetricsExistenceAndEarliestTime(ctx context.Context, req model.HostListRequest) (uint64, uint64, error) {
names := []string{}
for _, metricName := range metricNamesForHosts {
names = append(names, metricName)
}
return h.reader.GetMetricsExistenceAndEarliestTime(ctx, names)
}
func (h *HostsRepo) IsSendingK8SAgentMetrics(ctx context.Context, req model.HostListRequest) ([]string, []string, error) {
names := []string{}
for _, metricName := range metricNamesForHosts {
names = append(names, metricName)
}
namesStr := "'" + strings.Join(names, "','") + "'"
queryForRecentFingerprints := fmt.Sprintf(`
SELECT DISTINCT fingerprint
FROM %s.%s
WHERE metric_name IN (%s)
AND unix_milli >= toUnixTimestamp(now() - INTERVAL 5 MINUTE) * 1000`,
constants.SIGNOZ_METRIC_DBNAME, constants.SIGNOZ_SAMPLES_V4_TABLENAME, namesStr)
query := fmt.Sprintf(`
SELECT DISTINCT JSONExtractString(labels, '%s') as k8s_cluster_name, JSONExtractString(labels, '%s') as k8s_node_name
FROM %s.%s
WHERE metric_name IN (%s)
AND unix_milli >= toUnixTimestamp(now() - INTERVAL 60 MINUTE) * 1000
AND JSONExtractString(labels, '%s') LIKE '%%-otel-agent%%'
AND fingerprint GLOBAL IN (%s)`,
"k8s.cluster.name", "k8s.node.name",
constants.SIGNOZ_METRIC_DBNAME, constants.SIGNOZ_TIMESERIES_V4_TABLENAME, namesStr, "host.name", queryForRecentFingerprints)
result, err := h.reader.GetListResultV3(ctx, query)
if err != nil {
return nil, nil, err
}
clusterNames := make(map[string]struct{})
nodeNames := make(map[string]struct{})
for _, row := range result {
switch v := row.Data["k8s.cluster.name"].(type) {
case string:
clusterNames[v] = struct{}{}
case *string:
clusterNames[*v] = struct{}{}
}
switch v := row.Data["k8s.node.name"].(type) {
case string:
nodeNames[v] = struct{}{}
case *string:
nodeNames[*v] = struct{}{}
}
}
return maps.Keys(clusterNames), maps.Keys(nodeNames), nil
}
func (h *HostsRepo) GetHostList(ctx context.Context, orgID valuer.UUID, req model.HostListRequest) (model.HostListResponse, error) {
resp := model.HostListResponse{}
if req.Limit == 0 {
req.Limit = 10
}
// default to cpu order by
if req.OrderBy == nil {
req.OrderBy = &v3.OrderBy{ColumnName: "cpu", Order: v3.DirectionDesc}
}
// default to host name group by
if len(req.GroupBy) == 0 {
req.GroupBy = []v3.AttributeKey{{Key: hostNameAttrKey}}
resp.Type = model.ResponseTypeList
} else {
resp.Type = model.ResponseTypeGroupedList
}
// don't fail the request if we can't get these values
if clusterNames, nodeNames, err := h.IsSendingK8SAgentMetrics(ctx, req); err == nil {
resp.IsSendingK8SAgentMetrics = len(clusterNames) > 0 || len(nodeNames) > 0
resp.ClusterNames = clusterNames
resp.NodeNames = nodeNames
}
// 1. Check if any host metrics exist and get earliest retention time
// if no hosts metrics exist, that means we should show the onboarding guide on UI, and return early.
// 2. If host metrics exist, but req.End is earlier than the earliest time of host metrics as read from
// metadata table, then we should convey the same to the user and return early
if count, minFirstReportedUnixMilli, err := h.GetHostMetricsExistenceAndEarliestTime(ctx, req); err == nil {
if count == 0 {
resp.SentAnyHostMetricsData = false
resp.Records = []model.HostListRecord{}
resp.Total = 0
return resp, nil
}
resp.SentAnyHostMetricsData = true
if req.End < int64(minFirstReportedUnixMilli) {
resp.EndTimeBeforeRetention = true
resp.Records = []model.HostListRecord{}
resp.Total = 0
return resp, nil
}
}
step := int64(math.Max(float64(common.MinAllowedStepInterval(req.Start, req.End)), 60))
if step <= 0 {
slog.ErrorContext(ctx, "step is less than or equal to 0", "step", step)
return resp, errors.New("step is less than or equal to 0")
}
query := HostsTableListQuery.Clone()
query.Start = req.Start
query.End = req.End
query.Step = step
for _, query := range query.CompositeQuery.BuilderQueries {
query.StepInterval = step
if req.Filters != nil && len(req.Filters.Items) > 0 {
query.Filters.Items = append(query.Filters.Items, req.Filters.Items...)
}
query.GroupBy = req.GroupBy
}
hostAttrs, err := h.getMetadataAttributes(ctx, req)
if err != nil {
return resp, err
}
activeHosts, err := h.getActiveHosts(ctx)
if err != nil {
return resp, err
}
topHostGroups, allHostGroups, err := h.getTopHostGroups(ctx, orgID, req, query)
if err != nil {
return resp, err
}
groupFilters := map[string][]string{}
for _, topHostGroup := range topHostGroups {
for k, v := range topHostGroup {
groupFilters[k] = append(groupFilters[k], v)
}
}
for groupKey, groupValues := range groupFilters {
hasGroupFilter := false
if req.Filters != nil && len(req.Filters.Items) > 0 {
for _, filter := range req.Filters.Items {
if filter.Key.Key == groupKey {
hasGroupFilter = true
break
}
}
}
if !hasGroupFilter {
for _, query := range query.CompositeQuery.BuilderQueries {
query.Filters.Items = append(query.Filters.Items, v3.FilterItem{
Key: v3.AttributeKey{Key: groupKey},
Value: groupValues,
Operator: v3.FilterOperatorIn,
})
}
}
}
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "inframetrics",
instrumentationtypes.CodeFunctionName: "GetHostList",
})
queryResponse, _, err := h.querierV2.QueryRange(ctx, orgID, query)
if err != nil {
return resp, err
}
formattedResponse, err := postprocess.PostProcessResult(queryResponse, query)
if err != nil {
return resp, err
}
records := []model.HostListRecord{}
for _, result := range formattedResponse {
for _, row := range result.Table.Rows {
record := model.HostListRecord{
CPU: -1,
Memory: -1,
Wait: -1,
Load15: -1,
}
if hostName, ok := row.Data[hostNameAttrKey].(string); ok {
record.HostName = hostName
}
if cpu, ok := row.Data["F1"].(float64); ok {
record.CPU = cpu
}
if memory, ok := row.Data["F2"].(float64); ok {
record.Memory = memory
}
if wait, ok := row.Data["F3"].(float64); ok {
record.Wait = wait
}
if load15, ok := row.Data["G"].(float64); ok {
record.Load15 = load15
}
record.Meta = map[string]string{}
if _, ok := hostAttrs[record.HostName]; ok {
record.Meta = hostAttrs[record.HostName]
}
if osType, ok := record.Meta["os.type"]; ok {
record.OS = osType
}
record.Active = activeHosts[record.HostName]
records = append(records, record)
}
}
resp.Total = len(allHostGroups)
resp.Records = records
resp.SortBy(req.OrderBy)
return resp, nil
}

View File

@@ -0,0 +1,314 @@
package inframetrics
import v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
var HostsTableListQuery = v3.QueryRangeParamsV3{
CompositeQuery: &v3.CompositeQuery{
BuilderQueries: map[string]*v3.BuilderQuery{
"A": {
QueryName: "A",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForHosts["cpu"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Cumulative,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{
{
Key: v3.AttributeKey{
Key: "state",
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeTag,
},
Operator: v3.FilterOperatorNotEqual,
Value: "idle",
},
{
Key: v3.AttributeKey{
Key: hostNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
Operator: v3.FilterOperatorNotContains,
Value: agentNameToIgnore,
},
},
},
GroupBy: []v3.AttributeKey{
{
Key: hostNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "A",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationRate,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: true,
},
"B": {
QueryName: "B",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForHosts["cpu"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Cumulative,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{
{
Key: v3.AttributeKey{
Key: hostNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
Operator: v3.FilterOperatorNotContains,
Value: agentNameToIgnore,
},
},
},
GroupBy: []v3.AttributeKey{
{
Key: hostNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "B",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationRate,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: true,
},
"F1": {
QueryName: "F1",
Expression: "A/B",
Legend: "CPU Usage (%)",
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
},
"C": {
QueryName: "C",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForHosts["memory"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Cumulative,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{
{
Key: v3.AttributeKey{
Key: "state",
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeTag,
},
Operator: v3.FilterOperatorEqual,
Value: "used",
},
{
Key: v3.AttributeKey{
Key: hostNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
Operator: v3.FilterOperatorNotContains,
Value: agentNameToIgnore,
},
},
},
GroupBy: []v3.AttributeKey{
{
Key: hostNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "C",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: true,
},
"D": {
QueryName: "D",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForHosts["memory"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Cumulative,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{
{
Key: v3.AttributeKey{
Key: hostNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
Operator: v3.FilterOperatorNotContains,
Value: agentNameToIgnore,
},
},
},
GroupBy: []v3.AttributeKey{
{
Key: hostNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "D",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: true,
},
"F2": {
QueryName: "F2",
Expression: "C/D",
Legend: "Memory Usage (%)",
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
},
"E": {
QueryName: "E",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForHosts["wait"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Cumulative,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{
{
Key: v3.AttributeKey{
Key: "state",
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeTag,
},
Operator: v3.FilterOperatorEqual,
Value: "wait",
},
{
Key: v3.AttributeKey{
Key: hostNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
Operator: v3.FilterOperatorNotContains,
Value: agentNameToIgnore,
},
},
},
GroupBy: []v3.AttributeKey{
{
Key: hostNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "E",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationRate,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: true,
},
"F": {
QueryName: "F",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForHosts["wait"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Cumulative,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{
{
Key: v3.AttributeKey{
Key: hostNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
Operator: v3.FilterOperatorNotContains,
Value: agentNameToIgnore,
},
},
},
GroupBy: []v3.AttributeKey{
{
Key: hostNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "F",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationRate,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: true,
},
"F3": {
QueryName: "F3",
Expression: "E/F",
Legend: "CPU Wait Time (%)",
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
},
"G": {
QueryName: "G",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForHosts["load15"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{
{
Key: v3.AttributeKey{
Key: hostNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
Operator: v3.FilterOperatorNotContains,
Value: agentNameToIgnore,
},
},
},
GroupBy: []v3.AttributeKey{
{
Key: hostNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "G",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
Legend: "CPU Load Average (15m)",
},
},
PanelType: v3.PanelTypeTable,
QueryType: v3.QueryTypeBuilder,
},
Version: "v4",
FormatForWeb: true,
}

View File

@@ -0,0 +1,511 @@
package inframetrics
import (
"context"
"math"
"sort"
"github.com/SigNoz/signoz/pkg/query-service/app/metrics/v4/helpers"
"github.com/SigNoz/signoz/pkg/query-service/common"
"github.com/SigNoz/signoz/pkg/query-service/interfaces"
"github.com/SigNoz/signoz/pkg/query-service/model"
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
"github.com/SigNoz/signoz/pkg/query-service/postprocess"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
"github.com/SigNoz/signoz/pkg/valuer"
"golang.org/x/exp/slices"
)
var (
metricToUseForJobs = "k8s.job.desired_successful_pods"
k8sJobNameAttrKey = "k8s.job.name"
metricNamesForJobs = map[string]string{
"desired_successful_pods": "k8s.job.desired_successful_pods",
"active_pods": "k8s.job.active_pods",
"failed_pods": "k8s.job.failed_pods",
"successful_pods": "k8s.job.successful_pods",
}
jobAttrsToEnrich = []string{
"k8s.job.name",
"k8s.namespace.name",
"k8s.cluster.name",
}
queryNamesForJobs = map[string][]string{
"cpu": {"A"},
"cpu_request": {"B", "A"},
"cpu_limit": {"C", "A"},
"memory": {"D"},
"memory_request": {"E", "D"},
"memory_limit": {"F", "D"},
"restarts": {"G", "A"},
"desired_successful_pods": {"H"},
"active_pods": {"I"},
"failed_pods": {"J"},
"successful_pods": {"K"},
}
builderQueriesForJobs = map[string]*v3.BuilderQuery{
// desired nodes
"H": {
QueryName: "H",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForJobs["desired_successful_pods"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{},
Expression: "H",
ReduceTo: v3.ReduceToOperatorLast,
TimeAggregation: v3.TimeAggregationAnyLast,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
// available nodes
"I": {
QueryName: "I",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForJobs["active_pods"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{},
Expression: "I",
ReduceTo: v3.ReduceToOperatorLast,
TimeAggregation: v3.TimeAggregationAnyLast,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
// failed pods
"J": {
QueryName: "J",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForJobs["failed_pods"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{},
Expression: "J",
ReduceTo: v3.ReduceToOperatorLast,
TimeAggregation: v3.TimeAggregationAnyLast,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
// successful pods
"K": {
QueryName: "K",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForJobs["successful_pods"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{},
Expression: "K",
ReduceTo: v3.ReduceToOperatorLast,
TimeAggregation: v3.TimeAggregationAnyLast,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
}
jobQueryNames = []string{"A", "B", "C", "D", "E", "F", "G", "H", "I", "J", "K"}
)
type JobsRepo struct {
reader interfaces.Reader
querierV2 interfaces.Querier
}
func NewJobsRepo(reader interfaces.Reader, querierV2 interfaces.Querier) *JobsRepo {
return &JobsRepo{reader: reader, querierV2: querierV2}
}
func (d *JobsRepo) GetJobAttributeKeys(ctx context.Context, orgID valuer.UUID, req v3.FilterAttributeKeyRequest) (*v3.FilterAttributeKeyResponse, error) {
// TODO(srikanthccv): remove hardcoded metric name and support keys from any pod metric
req.DataSource = v3.DataSourceMetrics
req.AggregateAttribute = metricToUseForJobs
if req.Limit == 0 {
req.Limit = 50
}
attributeKeysResponse, err := d.reader.GetMetricAttributeKeys(ctx, orgID, &req)
if err != nil {
return nil, err
}
// TODO(srikanthccv): only return resource attributes when we have a way to
// distinguish between resource attributes and other attributes.
filteredKeys := []v3.AttributeKey{}
for _, key := range attributeKeysResponse.AttributeKeys {
if slices.Contains(pointAttrsToIgnore, key.Key) {
continue
}
filteredKeys = append(filteredKeys, key)
}
return &v3.FilterAttributeKeyResponse{AttributeKeys: filteredKeys}, nil
}
func (d *JobsRepo) GetJobAttributeValues(ctx context.Context, orgID valuer.UUID, req v3.FilterAttributeValueRequest) (*v3.FilterAttributeValueResponse, error) {
req.DataSource = v3.DataSourceMetrics
req.AggregateAttribute = metricToUseForJobs
if req.Limit == 0 {
req.Limit = 50
}
attributeValuesResponse, err := d.reader.GetMetricAttributeValues(ctx, orgID, &req)
if err != nil {
return nil, err
}
return attributeValuesResponse, nil
}
func (d *JobsRepo) getMetadataAttributes(ctx context.Context, req model.JobListRequest) (map[string]map[string]string, error) {
jobAttrs := map[string]map[string]string{}
for _, key := range jobAttrsToEnrich {
hasKey := false
for _, groupByKey := range req.GroupBy {
if groupByKey.Key == key {
hasKey = true
break
}
}
if !hasKey {
req.GroupBy = append(req.GroupBy, v3.AttributeKey{Key: key})
}
}
mq := v3.BuilderQuery{
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricToUseForJobs,
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
GroupBy: req.GroupBy,
}
query, err := helpers.PrepareTimeseriesFilterQuery(req.Start, req.End, &mq)
if err != nil {
return nil, err
}
query = localQueryToDistributedQuery(query)
attrsListResponse, err := d.reader.GetListResultV3(ctx, query)
if err != nil {
return nil, err
}
for _, row := range attrsListResponse {
stringData := map[string]string{}
for key, value := range row.Data {
if str, ok := value.(string); ok {
stringData[key] = str
} else if strPtr, ok := value.(*string); ok {
stringData[key] = *strPtr
}
}
jobName := stringData[k8sJobNameAttrKey]
if _, ok := jobAttrs[jobName]; !ok {
jobAttrs[jobName] = map[string]string{}
}
for _, key := range req.GroupBy {
jobAttrs[jobName][key.Key] = stringData[key.Key]
}
}
return jobAttrs, nil
}
func (d *JobsRepo) getTopJobGroups(ctx context.Context, orgID valuer.UUID, req model.JobListRequest, q *v3.QueryRangeParamsV3) ([]map[string]string, []map[string]string, error) {
step, timeSeriesTableName, samplesTableName := getParamsForTopJobs(req)
queryNames := queryNamesForJobs[req.OrderBy.ColumnName]
topJobGroupsQueryRangeParams := &v3.QueryRangeParamsV3{
Start: req.Start,
End: req.End,
Step: step,
CompositeQuery: &v3.CompositeQuery{
BuilderQueries: map[string]*v3.BuilderQuery{},
QueryType: v3.QueryTypeBuilder,
PanelType: v3.PanelTypeTable,
},
}
for _, queryName := range queryNames {
query := q.CompositeQuery.BuilderQueries[queryName].Clone()
query.StepInterval = step
query.MetricTableHints = &v3.MetricTableHints{
TimeSeriesTableName: timeSeriesTableName,
SamplesTableName: samplesTableName,
}
if req.Filters != nil && len(req.Filters.Items) > 0 {
if query.Filters == nil {
query.Filters = &v3.FilterSet{Operator: "AND", Items: []v3.FilterItem{}}
}
query.Filters.Items = append(query.Filters.Items, req.Filters.Items...)
}
topJobGroupsQueryRangeParams.CompositeQuery.BuilderQueries[queryName] = query
}
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "inframetrics",
instrumentationtypes.CodeFunctionName: "getTopJobGroups",
})
queryResponse, _, err := d.querierV2.QueryRange(ctx, orgID, topJobGroupsQueryRangeParams)
if err != nil {
return nil, nil, err
}
formattedResponse, err := postprocess.PostProcessResult(queryResponse, topJobGroupsQueryRangeParams)
if err != nil {
return nil, nil, err
}
if len(formattedResponse) == 0 || len(formattedResponse[0].Series) == 0 {
return nil, nil, nil
}
if req.OrderBy.Order == v3.DirectionDesc {
sort.Slice(formattedResponse[0].Series, func(i, j int) bool {
return formattedResponse[0].Series[i].Points[0].Value > formattedResponse[0].Series[j].Points[0].Value
})
} else {
sort.Slice(formattedResponse[0].Series, func(i, j int) bool {
return formattedResponse[0].Series[i].Points[0].Value < formattedResponse[0].Series[j].Points[0].Value
})
}
limit := math.Min(float64(req.Offset+req.Limit), float64(len(formattedResponse[0].Series)))
paginatedTopJobGroupsSeries := formattedResponse[0].Series[req.Offset:int(limit)]
topJobGroups := []map[string]string{}
for _, series := range paginatedTopJobGroupsSeries {
topJobGroups = append(topJobGroups, series.Labels)
}
allJobGroups := []map[string]string{}
for _, series := range formattedResponse[0].Series {
allJobGroups = append(allJobGroups, series.Labels)
}
return topJobGroups, allJobGroups, nil
}
func (d *JobsRepo) GetJobList(ctx context.Context, orgID valuer.UUID, req model.JobListRequest) (model.JobListResponse, error) {
resp := model.JobListResponse{}
if req.Limit == 0 {
req.Limit = 10
}
if req.OrderBy == nil {
req.OrderBy = &v3.OrderBy{ColumnName: "desired_pods", Order: v3.DirectionDesc}
}
if req.GroupBy == nil {
req.GroupBy = []v3.AttributeKey{{Key: k8sJobNameAttrKey}}
resp.Type = model.ResponseTypeList
} else {
resp.Type = model.ResponseTypeGroupedList
}
step := int64(math.Max(float64(common.MinAllowedStepInterval(req.Start, req.End)), 60))
query := WorkloadTableListQuery.Clone()
query.Start = req.Start
query.End = req.End
query.Step = step
// add additional queries for jobs
for _, jobQuery := range builderQueriesForJobs {
query.CompositeQuery.BuilderQueries[jobQuery.QueryName] = jobQuery.Clone()
}
for _, query := range query.CompositeQuery.BuilderQueries {
query.StepInterval = step
if req.Filters != nil && len(req.Filters.Items) > 0 {
if query.Filters == nil {
query.Filters = &v3.FilterSet{Operator: "AND", Items: []v3.FilterItem{}}
}
query.Filters.Items = append(query.Filters.Items, req.Filters.Items...)
}
query.GroupBy = req.GroupBy
// make sure we only get records for jobs
query.Filters.Items = append(query.Filters.Items, v3.FilterItem{
Key: v3.AttributeKey{Key: k8sJobNameAttrKey},
Operator: v3.FilterOperatorExists,
})
}
jobAttrs, err := d.getMetadataAttributes(ctx, req)
if err != nil {
return resp, err
}
topJobGroups, allJobGroups, err := d.getTopJobGroups(ctx, orgID, req, query)
if err != nil {
return resp, err
}
groupFilters := map[string][]string{}
for _, topJobGroup := range topJobGroups {
for k, v := range topJobGroup {
groupFilters[k] = append(groupFilters[k], v)
}
}
for groupKey, groupValues := range groupFilters {
hasGroupFilter := false
if req.Filters != nil && len(req.Filters.Items) > 0 {
for _, filter := range req.Filters.Items {
if filter.Key.Key == groupKey {
hasGroupFilter = true
break
}
}
}
if !hasGroupFilter {
for _, query := range query.CompositeQuery.BuilderQueries {
query.Filters.Items = append(query.Filters.Items, v3.FilterItem{
Key: v3.AttributeKey{Key: groupKey},
Value: groupValues,
Operator: v3.FilterOperatorIn,
})
}
}
}
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "inframetrics",
instrumentationtypes.CodeFunctionName: "GetJobList",
})
queryResponse, _, err := d.querierV2.QueryRange(ctx, orgID, query)
if err != nil {
return resp, err
}
formattedResponse, err := postprocess.PostProcessResult(queryResponse, query)
if err != nil {
return resp, err
}
records := []model.JobListRecord{}
for _, result := range formattedResponse {
for _, row := range result.Table.Rows {
record := model.JobListRecord{
JobName: "",
CPUUsage: -1,
CPURequest: -1,
CPULimit: -1,
MemoryUsage: -1,
MemoryRequest: -1,
MemoryLimit: -1,
DesiredSuccessfulPods: -1,
ActivePods: -1,
FailedPods: -1,
SuccessfulPods: -1,
}
if jobName, ok := row.Data[k8sJobNameAttrKey].(string); ok {
record.JobName = jobName
}
if cpu, ok := row.Data["A"].(float64); ok {
record.CPUUsage = cpu
}
if cpuRequest, ok := row.Data["B"].(float64); ok {
record.CPURequest = cpuRequest
}
if cpuLimit, ok := row.Data["C"].(float64); ok {
record.CPULimit = cpuLimit
}
if memory, ok := row.Data["D"].(float64); ok {
record.MemoryUsage = memory
}
if memoryRequest, ok := row.Data["E"].(float64); ok {
record.MemoryRequest = memoryRequest
}
if memoryLimit, ok := row.Data["F"].(float64); ok {
record.MemoryLimit = memoryLimit
}
if restarts, ok := row.Data["G"].(float64); ok {
record.Restarts = int(restarts)
}
if desiredSuccessfulPods, ok := row.Data["H"].(float64); ok {
record.DesiredSuccessfulPods = int(desiredSuccessfulPods)
}
if activePods, ok := row.Data["I"].(float64); ok {
record.ActivePods = int(activePods)
}
if failedPods, ok := row.Data["J"].(float64); ok {
record.FailedPods = int(failedPods)
}
if successfulPods, ok := row.Data["K"].(float64); ok {
record.SuccessfulPods = int(successfulPods)
}
record.Meta = map[string]string{}
if _, ok := jobAttrs[record.JobName]; ok && record.JobName != "" {
record.Meta = jobAttrs[record.JobName]
}
for k, v := range row.Data {
if slices.Contains(jobQueryNames, k) {
continue
}
if labelValue, ok := v.(string); ok {
record.Meta[k] = labelValue
}
}
records = append(records, record)
}
}
resp.Total = len(allJobGroups)
resp.Records = records
resp.SortBy(req.OrderBy)
return resp, nil
}

View File

@@ -0,0 +1,358 @@
package inframetrics
import (
"context"
"math"
"sort"
"github.com/SigNoz/signoz/pkg/query-service/app/metrics/v4/helpers"
"github.com/SigNoz/signoz/pkg/query-service/common"
"github.com/SigNoz/signoz/pkg/query-service/interfaces"
"github.com/SigNoz/signoz/pkg/query-service/model"
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
"github.com/SigNoz/signoz/pkg/query-service/postprocess"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
"github.com/SigNoz/signoz/pkg/valuer"
"golang.org/x/exp/slices"
)
var (
metricToUseForNamespaces = "k8s.pod.cpu.usage"
namespaceAttrsToEnrich = []string{
"k8s.namespace.name",
"k8s.cluster.name",
}
queryNamesForNamespaces = map[string][]string{
"cpu": {"A"},
"memory": {"D"},
"pod_phase": {"H", "I", "J", "K"},
}
namespaceQueryNames = []string{"A", "D", "H", "I", "J", "K"}
attributesKeysForNamespaces = []v3.AttributeKey{
{Key: "k8s.namespace.name"},
{Key: "k8s.cluster.name"},
}
k8sNamespaceNameAttrKey = "k8s.namespace.name"
)
type NamespacesRepo struct {
reader interfaces.Reader
querierV2 interfaces.Querier
}
func NewNamespacesRepo(reader interfaces.Reader, querierV2 interfaces.Querier) *NamespacesRepo {
return &NamespacesRepo{reader: reader, querierV2: querierV2}
}
func (p *NamespacesRepo) GetNamespaceAttributeKeys(ctx context.Context, req v3.FilterAttributeKeyRequest) (*v3.FilterAttributeKeyResponse, error) {
return &v3.FilterAttributeKeyResponse{AttributeKeys: attributesKeysForNamespaces}, nil
}
func (p *NamespacesRepo) GetNamespaceAttributeValues(ctx context.Context, orgID valuer.UUID, req v3.FilterAttributeValueRequest) (*v3.FilterAttributeValueResponse, error) {
req.DataSource = v3.DataSourceMetrics
req.AggregateAttribute = metricToUseForNamespaces
if req.Limit == 0 {
req.Limit = 50
}
attributeValuesResponse, err := p.reader.GetMetricAttributeValues(ctx, orgID, &req)
if err != nil {
return nil, err
}
return attributeValuesResponse, nil
}
func (p *NamespacesRepo) getMetadataAttributes(ctx context.Context, req model.NamespaceListRequest) (map[string]map[string]string, error) {
namespaceAttrs := map[string]map[string]string{}
for _, key := range namespaceAttrsToEnrich {
hasKey := false
for _, groupByKey := range req.GroupBy {
if groupByKey.Key == key {
hasKey = true
break
}
}
if !hasKey {
req.GroupBy = append(req.GroupBy, v3.AttributeKey{Key: key})
}
}
mq := v3.BuilderQuery{
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricToUseForNamespaces,
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
GroupBy: req.GroupBy,
}
query, err := helpers.PrepareTimeseriesFilterQuery(req.Start, req.End, &mq)
if err != nil {
return nil, err
}
query = localQueryToDistributedQuery(query)
attrsListResponse, err := p.reader.GetListResultV3(ctx, query)
if err != nil {
return nil, err
}
for _, row := range attrsListResponse {
stringData := map[string]string{}
for key, value := range row.Data {
if str, ok := value.(string); ok {
stringData[key] = str
} else if strPtr, ok := value.(*string); ok {
stringData[key] = *strPtr
}
}
namespaceName := stringData[k8sNamespaceNameAttrKey]
if _, ok := namespaceAttrs[namespaceName]; !ok {
namespaceAttrs[namespaceName] = map[string]string{}
}
for _, key := range req.GroupBy {
namespaceAttrs[namespaceName][key.Key] = stringData[key.Key]
}
}
return namespaceAttrs, nil
}
func (p *NamespacesRepo) getTopNamespaceGroups(ctx context.Context, orgID valuer.UUID, req model.NamespaceListRequest, q *v3.QueryRangeParamsV3) ([]map[string]string, []map[string]string, error) {
step, timeSeriesTableName, samplesTableName := getParamsForTopNamespaces(req)
queryNames := queryNamesForNamespaces[req.OrderBy.ColumnName]
topNamespaceGroupsQueryRangeParams := &v3.QueryRangeParamsV3{
Start: req.Start,
End: req.End,
Step: step,
CompositeQuery: &v3.CompositeQuery{
BuilderQueries: map[string]*v3.BuilderQuery{},
QueryType: v3.QueryTypeBuilder,
PanelType: v3.PanelTypeTable,
},
}
for _, queryName := range queryNames {
query := q.CompositeQuery.BuilderQueries[queryName].Clone()
query.StepInterval = step
query.MetricTableHints = &v3.MetricTableHints{
TimeSeriesTableName: timeSeriesTableName,
SamplesTableName: samplesTableName,
}
if req.Filters != nil && len(req.Filters.Items) > 0 {
if query.Filters == nil {
query.Filters = &v3.FilterSet{Operator: "AND", Items: []v3.FilterItem{}}
}
query.Filters.Items = append(query.Filters.Items, req.Filters.Items...)
}
topNamespaceGroupsQueryRangeParams.CompositeQuery.BuilderQueries[queryName] = query
}
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "inframetrics",
instrumentationtypes.CodeFunctionName: "getTopNamespaceGroups",
})
queryResponse, _, err := p.querierV2.QueryRange(ctx, orgID, topNamespaceGroupsQueryRangeParams)
if err != nil {
return nil, nil, err
}
formattedResponse, err := postprocess.PostProcessResult(queryResponse, topNamespaceGroupsQueryRangeParams)
if err != nil {
return nil, nil, err
}
if len(formattedResponse) == 0 || len(formattedResponse[0].Series) == 0 {
return nil, nil, nil
}
if req.OrderBy.Order == v3.DirectionDesc {
sort.Slice(formattedResponse[0].Series, func(i, j int) bool {
return formattedResponse[0].Series[i].Points[0].Value > formattedResponse[0].Series[j].Points[0].Value
})
} else {
sort.Slice(formattedResponse[0].Series, func(i, j int) bool {
return formattedResponse[0].Series[i].Points[0].Value < formattedResponse[0].Series[j].Points[0].Value
})
}
limit := math.Min(float64(req.Offset+req.Limit), float64(len(formattedResponse[0].Series)))
paginatedTopNamespaceGroupsSeries := formattedResponse[0].Series[req.Offset:int(limit)]
topNamespaceGroups := []map[string]string{}
for _, series := range paginatedTopNamespaceGroupsSeries {
topNamespaceGroups = append(topNamespaceGroups, series.Labels)
}
allNamespaceGroups := []map[string]string{}
for _, series := range formattedResponse[0].Series {
allNamespaceGroups = append(allNamespaceGroups, series.Labels)
}
return topNamespaceGroups, allNamespaceGroups, nil
}
func (p *NamespacesRepo) GetNamespaceList(ctx context.Context, orgID valuer.UUID, req model.NamespaceListRequest) (model.NamespaceListResponse, error) {
resp := model.NamespaceListResponse{}
if req.Limit == 0 {
req.Limit = 10
}
if req.OrderBy == nil {
req.OrderBy = &v3.OrderBy{ColumnName: "cpu", Order: v3.DirectionDesc}
}
if req.GroupBy == nil {
req.GroupBy = []v3.AttributeKey{{Key: k8sNamespaceNameAttrKey}}
resp.Type = model.ResponseTypeList
} else {
resp.Type = model.ResponseTypeGroupedList
}
step := int64(math.Max(float64(common.MinAllowedStepInterval(req.Start, req.End)), 60))
query := PodsTableListQuery.Clone()
query.Start = req.Start
query.End = req.End
query.Step = step
for _, q := range query.CompositeQuery.BuilderQueries {
if !slices.Contains(namespaceQueryNames, q.QueryName) {
delete(query.CompositeQuery.BuilderQueries, q.QueryName)
}
q.StepInterval = step
if req.Filters != nil && len(req.Filters.Items) > 0 {
if q.Filters == nil {
q.Filters = &v3.FilterSet{Operator: "AND", Items: []v3.FilterItem{}}
}
q.Filters.Items = append(q.Filters.Items, req.Filters.Items...)
}
q.GroupBy = req.GroupBy
}
namespaceAttrs, err := p.getMetadataAttributes(ctx, req)
if err != nil {
return resp, err
}
topNamespaceGroups, allNamespaceGroups, err := p.getTopNamespaceGroups(ctx, orgID, req, query)
if err != nil {
return resp, err
}
groupFilters := map[string][]string{}
for _, topNamespaceGroup := range topNamespaceGroups {
for k, v := range topNamespaceGroup {
groupFilters[k] = append(groupFilters[k], v)
}
}
for groupKey, groupValues := range groupFilters {
hasGroupFilter := false
if req.Filters != nil && len(req.Filters.Items) > 0 {
for _, filter := range req.Filters.Items {
if filter.Key.Key == groupKey {
hasGroupFilter = true
break
}
}
}
if !hasGroupFilter {
for _, query := range query.CompositeQuery.BuilderQueries {
query.Filters.Items = append(query.Filters.Items, v3.FilterItem{
Key: v3.AttributeKey{Key: groupKey},
Value: groupValues,
Operator: v3.FilterOperatorIn,
})
}
}
}
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "inframetrics",
instrumentationtypes.CodeFunctionName: "GetNamespaceList",
})
queryResponse, _, err := p.querierV2.QueryRange(ctx, orgID, query)
if err != nil {
return resp, err
}
formattedResponse, err := postprocess.PostProcessResult(queryResponse, query)
if err != nil {
return resp, err
}
records := []model.NamespaceListRecord{}
for _, result := range formattedResponse {
for _, row := range result.Table.Rows {
record := model.NamespaceListRecord{
CPUUsage: -1,
MemoryUsage: -1,
}
if name, ok := row.Data[k8sNamespaceNameAttrKey].(string); ok {
record.NamespaceName = name
}
if cpu, ok := row.Data["A"].(float64); ok {
record.CPUUsage = cpu
}
if memory, ok := row.Data["D"].(float64); ok {
record.MemoryUsage = memory
}
if pending, ok := row.Data["H"].(float64); ok {
record.CountByPhase.Pending = int(pending)
}
if running, ok := row.Data["I"].(float64); ok {
record.CountByPhase.Running = int(running)
}
if succeeded, ok := row.Data["J"].(float64); ok {
record.CountByPhase.Succeeded = int(succeeded)
}
if failed, ok := row.Data["K"].(float64); ok {
record.CountByPhase.Failed = int(failed)
}
record.Meta = map[string]string{}
if _, ok := namespaceAttrs[record.NamespaceName]; ok && record.NamespaceName != "" {
record.Meta = namespaceAttrs[record.NamespaceName]
}
for k, v := range row.Data {
if slices.Contains(namespaceQueryNames, k) {
continue
}
if labelValue, ok := v.(string); ok {
record.Meta[k] = labelValue
}
}
records = append(records, record)
}
}
resp.Total = len(allNamespaceGroups)
resp.Records = records
resp.SortBy(req.OrderBy)
return resp, nil
}

View File

@@ -0,0 +1,387 @@
package inframetrics
import (
"context"
"fmt"
"math"
"sort"
"strings"
"github.com/SigNoz/signoz/pkg/query-service/app/metrics/v4/helpers"
"github.com/SigNoz/signoz/pkg/query-service/common"
"github.com/SigNoz/signoz/pkg/query-service/constants"
"github.com/SigNoz/signoz/pkg/query-service/interfaces"
"github.com/SigNoz/signoz/pkg/query-service/model"
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
"github.com/SigNoz/signoz/pkg/query-service/postprocess"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
"github.com/SigNoz/signoz/pkg/valuer"
"golang.org/x/exp/slices"
)
var (
metricToUseForNodes = "k8s.node.cpu.usage"
nodeAttrsToEnrich = []string{"k8s.node.name", "k8s.node.uid", "k8s.cluster.name"}
k8sNodeGroupAttrKey = "k8s.node.name"
queryNamesForNodes = map[string][]string{
"cpu": {"A"},
"cpu_allocatable": {"B"},
"memory": {"C"},
"memory_allocatable": {"D"},
}
nodeQueryNames = []string{"A", "B", "C", "D", "E", "F"}
metricNamesForNodes = map[string]string{
"cpu": "k8s.node.cpu.usage",
"cpu_allocatable": "k8s.node.allocatable_cpu",
"memory": "k8s.node.memory.working_set",
"memory_allocatable": "k8s.node.allocatable_memory",
"node_condition": "k8s.node.condition_ready",
}
)
type NodesRepo struct {
reader interfaces.Reader
querierV2 interfaces.Querier
}
func NewNodesRepo(reader interfaces.Reader, querierV2 interfaces.Querier) *NodesRepo {
return &NodesRepo{reader: reader, querierV2: querierV2}
}
func (n *NodesRepo) GetNodeAttributeKeys(ctx context.Context, orgID valuer.UUID, req v3.FilterAttributeKeyRequest) (*v3.FilterAttributeKeyResponse, error) {
req.DataSource = v3.DataSourceMetrics
req.AggregateAttribute = metricToUseForNodes
if req.Limit == 0 {
req.Limit = 50
}
attributeKeysResponse, err := n.reader.GetMetricAttributeKeys(ctx, orgID, &req)
if err != nil {
return nil, err
}
return attributeKeysResponse, nil
}
func (n *NodesRepo) DidSendNodeMetrics(ctx context.Context) (bool, error) {
namesStr := "'" + strings.Join(nodeMetricNamesToCheck, "','") + "'"
query := fmt.Sprintf(didSendNodeMetricsQuery,
constants.SIGNOZ_METRIC_DBNAME, constants.SIGNOZ_TIMESERIES_v4_1DAY_TABLENAME, namesStr)
count, err := n.reader.GetCountOfThings(ctx, query)
if err != nil {
return false, err
}
return count > 0, nil
}
func (n *NodesRepo) GetNodeAttributeValues(ctx context.Context, orgID valuer.UUID, req v3.FilterAttributeValueRequest) (*v3.FilterAttributeValueResponse, error) {
req.DataSource = v3.DataSourceMetrics
req.AggregateAttribute = metricToUseForNodes
if req.Limit == 0 {
req.Limit = 50
}
attributeValuesResponse, err := n.reader.GetMetricAttributeValues(ctx, orgID, &req)
if err != nil {
return nil, err
}
return attributeValuesResponse, nil
}
func (p *NodesRepo) getMetadataAttributes(ctx context.Context, req model.NodeListRequest) (map[string]map[string]string, error) {
nodeAttrs := map[string]map[string]string{}
for _, key := range nodeAttrsToEnrich {
hasKey := false
for _, groupByKey := range req.GroupBy {
if groupByKey.Key == key {
hasKey = true
break
}
}
if !hasKey {
req.GroupBy = append(req.GroupBy, v3.AttributeKey{Key: key})
}
}
mq := v3.BuilderQuery{
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricToUseForNodes,
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
GroupBy: req.GroupBy,
}
query, err := helpers.PrepareTimeseriesFilterQuery(req.Start, req.End, &mq)
if err != nil {
return nil, err
}
query = localQueryToDistributedQuery(query)
attrsListResponse, err := p.reader.GetListResultV3(ctx, query)
if err != nil {
return nil, err
}
for _, row := range attrsListResponse {
stringData := map[string]string{}
for key, value := range row.Data {
if str, ok := value.(string); ok {
stringData[key] = str
} else if strPtr, ok := value.(*string); ok {
stringData[key] = *strPtr
}
}
nodeUID := stringData[k8sNodeGroupAttrKey]
if _, ok := nodeAttrs[nodeUID]; !ok {
nodeAttrs[nodeUID] = map[string]string{}
}
for _, key := range req.GroupBy {
nodeAttrs[nodeUID][key.Key] = stringData[key.Key]
}
}
return nodeAttrs, nil
}
func (p *NodesRepo) getTopNodeGroups(ctx context.Context, orgID valuer.UUID, req model.NodeListRequest, q *v3.QueryRangeParamsV3) ([]map[string]string, []map[string]string, error) {
step, timeSeriesTableName, samplesTableName := getParamsForTopNodes(req)
queryNames := queryNamesForNodes[req.OrderBy.ColumnName]
topNodeGroupsQueryRangeParams := &v3.QueryRangeParamsV3{
Start: req.Start,
End: req.End,
Step: step,
CompositeQuery: &v3.CompositeQuery{
BuilderQueries: map[string]*v3.BuilderQuery{},
QueryType: v3.QueryTypeBuilder,
PanelType: v3.PanelTypeTable,
},
}
for _, queryName := range queryNames {
query := q.CompositeQuery.BuilderQueries[queryName].Clone()
query.StepInterval = step
query.MetricTableHints = &v3.MetricTableHints{
TimeSeriesTableName: timeSeriesTableName,
SamplesTableName: samplesTableName,
}
if req.Filters != nil && len(req.Filters.Items) > 0 {
if query.Filters == nil {
query.Filters = &v3.FilterSet{Operator: "AND", Items: []v3.FilterItem{}}
}
query.Filters.Items = append(query.Filters.Items, req.Filters.Items...)
}
topNodeGroupsQueryRangeParams.CompositeQuery.BuilderQueries[queryName] = query
}
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "inframetrics",
instrumentationtypes.CodeFunctionName: "getTopNodeGroups",
})
queryResponse, _, err := p.querierV2.QueryRange(ctx, orgID, topNodeGroupsQueryRangeParams)
if err != nil {
return nil, nil, err
}
formattedResponse, err := postprocess.PostProcessResult(queryResponse, topNodeGroupsQueryRangeParams)
if err != nil {
return nil, nil, err
}
if len(formattedResponse) == 0 || len(formattedResponse[0].Series) == 0 {
return nil, nil, nil
}
if req.OrderBy.Order == v3.DirectionDesc {
sort.Slice(formattedResponse[0].Series, func(i, j int) bool {
return formattedResponse[0].Series[i].Points[0].Value > formattedResponse[0].Series[j].Points[0].Value
})
} else {
sort.Slice(formattedResponse[0].Series, func(i, j int) bool {
return formattedResponse[0].Series[i].Points[0].Value < formattedResponse[0].Series[j].Points[0].Value
})
}
max := math.Min(float64(req.Offset+req.Limit), float64(len(formattedResponse[0].Series)))
paginatedTopNodeGroupsSeries := formattedResponse[0].Series[req.Offset:int(max)]
topNodeGroups := []map[string]string{}
for _, series := range paginatedTopNodeGroupsSeries {
topNodeGroups = append(topNodeGroups, series.Labels)
}
allNodeGroups := []map[string]string{}
for _, series := range formattedResponse[0].Series {
allNodeGroups = append(allNodeGroups, series.Labels)
}
return topNodeGroups, allNodeGroups, nil
}
func (p *NodesRepo) GetNodeList(ctx context.Context, orgID valuer.UUID, req model.NodeListRequest) (model.NodeListResponse, error) {
resp := model.NodeListResponse{}
if req.Limit == 0 {
req.Limit = 10
}
if req.OrderBy == nil {
req.OrderBy = &v3.OrderBy{ColumnName: "cpu", Order: v3.DirectionDesc}
}
if req.GroupBy == nil {
req.GroupBy = []v3.AttributeKey{{Key: k8sNodeGroupAttrKey}}
resp.Type = model.ResponseTypeList
} else {
resp.Type = model.ResponseTypeGroupedList
}
step := int64(math.Max(float64(common.MinAllowedStepInterval(req.Start, req.End)), 60))
query := NodesTableListQuery.Clone()
query.Start = req.Start
query.End = req.End
query.Step = step
for _, query := range query.CompositeQuery.BuilderQueries {
query.StepInterval = step
if req.Filters != nil && len(req.Filters.Items) > 0 {
if query.Filters == nil {
query.Filters = &v3.FilterSet{Operator: "AND", Items: []v3.FilterItem{}}
}
query.Filters.Items = append(query.Filters.Items, req.Filters.Items...)
}
query.GroupBy = req.GroupBy
}
nodeAttrs, err := p.getMetadataAttributes(ctx, req)
if err != nil {
return resp, err
}
topNodeGroups, allNodeGroups, err := p.getTopNodeGroups(ctx, orgID, req, query)
if err != nil {
return resp, err
}
groupFilters := map[string][]string{}
for _, topNodeGroup := range topNodeGroups {
for k, v := range topNodeGroup {
groupFilters[k] = append(groupFilters[k], v)
}
}
for groupKey, groupValues := range groupFilters {
hasGroupFilter := false
if req.Filters != nil && len(req.Filters.Items) > 0 {
for _, filter := range req.Filters.Items {
if filter.Key.Key == groupKey {
hasGroupFilter = true
break
}
}
}
if !hasGroupFilter {
for _, query := range query.CompositeQuery.BuilderQueries {
query.Filters.Items = append(query.Filters.Items, v3.FilterItem{
Key: v3.AttributeKey{Key: groupKey},
Value: groupValues,
Operator: v3.FilterOperatorIn,
})
}
}
}
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "inframetrics",
instrumentationtypes.CodeFunctionName: "GetNodeList",
})
queryResponse, _, err := p.querierV2.QueryRange(ctx, orgID, query)
if err != nil {
return resp, err
}
formattedResponse, err := postprocess.PostProcessResult(queryResponse, query)
if err != nil {
return resp, err
}
records := []model.NodeListRecord{}
for _, result := range formattedResponse {
for _, row := range result.Table.Rows {
record := model.NodeListRecord{
NodeCPUUsage: -1,
NodeCPUAllocatable: -1,
NodeMemoryUsage: -1,
NodeMemoryAllocatable: -1,
}
if nodeUID, ok := row.Data[k8sNodeGroupAttrKey].(string); ok {
record.NodeUID = nodeUID
}
if cpu, ok := row.Data["A"].(float64); ok {
record.NodeCPUUsage = cpu
}
if cpuAllocatable, ok := row.Data["B"].(float64); ok {
record.NodeCPUAllocatable = cpuAllocatable
}
if mem, ok := row.Data["C"].(float64); ok {
record.NodeMemoryUsage = mem
}
if memory, ok := row.Data["D"].(float64); ok {
record.NodeMemoryAllocatable = memory
}
if ready, ok := row.Data["E"].(float64); ok {
record.CountByCondition.Ready = int(ready)
}
if notReady, ok := row.Data["F"].(float64); ok {
record.CountByCondition.NotReady = int(notReady)
}
record.Meta = map[string]string{}
if _, ok := nodeAttrs[record.NodeUID]; ok && record.NodeUID != "" {
record.Meta = nodeAttrs[record.NodeUID]
}
for k, v := range row.Data {
if slices.Contains(nodeQueryNames, k) {
continue
}
if labelValue, ok := v.(string); ok {
record.Meta[k] = labelValue
}
}
records = append(records, record)
}
}
resp.Total = len(allNodeGroups)
resp.Records = records
resp.SortBy(req.OrderBy)
return resp, nil
}

View File

@@ -0,0 +1,186 @@
package inframetrics
import v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
var NodesTableListQuery = v3.QueryRangeParamsV3{
CompositeQuery: &v3.CompositeQuery{
BuilderQueries: map[string]*v3.BuilderQuery{
// node cpu utilization
"A": {
QueryName: "A",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForNodes["cpu"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{
{
Key: k8sNodeGroupAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "A",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
// node cpu allocatable
"B": {
QueryName: "B",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForNodes["cpu_allocatable"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{
{
Key: k8sNodeGroupAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "B",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAnyLast,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
// node memory utilization
"C": {
QueryName: "C",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForNodes["memory"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{
{
Key: k8sNodeGroupAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "C",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
// node memory allocatable
"D": {
QueryName: "D",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForNodes["memory_allocatable"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{
{
Key: k8sNodeGroupAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "D",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAnyLast,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
// node conditions - Ready
"E": {
QueryName: "E",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForNodes["node_condition"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{
{
Key: v3.AttributeKey{
Key: "__value",
},
Operator: v3.FilterOperatorEqual,
Value: 1,
},
},
},
GroupBy: []v3.AttributeKey{
{
Key: k8sNodeGroupAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "E",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAnyLast,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
// node conditions - NotReady
"F": {
QueryName: "F",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForNodes["node_condition"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{
{
Key: v3.AttributeKey{
Key: "__value",
},
Operator: v3.FilterOperatorEqual,
Value: 0,
},
},
},
GroupBy: []v3.AttributeKey{
{
Key: k8sNodeGroupAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "F",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAnyLast,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
},
PanelType: v3.PanelTypeTable,
QueryType: v3.QueryTypeBuilder,
},
Version: "v4",
FormatForWeb: true,
}

View File

@@ -0,0 +1,555 @@
package inframetrics
import (
"context"
"fmt"
"math"
"sort"
"strings"
"github.com/SigNoz/signoz/pkg/query-service/app/metrics/v4/helpers"
"github.com/SigNoz/signoz/pkg/query-service/common"
"github.com/SigNoz/signoz/pkg/query-service/constants"
"github.com/SigNoz/signoz/pkg/query-service/interfaces"
"github.com/SigNoz/signoz/pkg/query-service/model"
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
"github.com/SigNoz/signoz/pkg/query-service/postprocess"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
"github.com/SigNoz/signoz/pkg/valuer"
"golang.org/x/exp/slices"
)
var (
metricToUseForPods = "k8s.pod.cpu.usage"
podAttrsToEnrich = []string{
"k8s.pod.uid",
"k8s.pod.name",
"k8s.namespace.name",
"k8s.node.name",
"k8s.deployment.name",
"k8s.statefulset.name",
"k8s.daemonset.name",
"k8s.job.name",
"k8s.cronjob.name",
"k8s.cluster.name",
}
k8sPodUIDAttrKey = "k8s.pod.uid"
queryNamesForPods = map[string][]string{
"cpu": {"A"},
"cpu_request": {"B", "A"},
"cpu_limit": {"C", "A"},
"memory": {"D"},
"memory_request": {"E", "D"},
"memory_limit": {"F", "D"},
"restarts": {"G", "A"},
"pod_phase": {"H", "I", "J", "K"},
}
podQueryNames = []string{"A", "B", "C", "D", "E", "F", "G", "H", "I", "J", "K"}
metricNamesForPods = map[string]string{
"cpu": "k8s.pod.cpu.usage",
"cpu_request": "k8s.pod.cpu_request_utilization",
"cpu_limit": "k8s.pod.cpu_limit_utilization",
"memory": "k8s.pod.memory.working_set",
"memory_request": "k8s.pod.memory_request_utilization",
"memory_limit": "k8s.pod.memory_limit_utilization",
"restarts": "k8s.container.restarts",
"pod_phase": "k8s.pod.phase",
}
)
type PodsRepo struct {
reader interfaces.Reader
querierV2 interfaces.Querier
}
func NewPodsRepo(reader interfaces.Reader, querierV2 interfaces.Querier) *PodsRepo {
return &PodsRepo{reader: reader, querierV2: querierV2}
}
func (p *PodsRepo) GetPodAttributeKeys(ctx context.Context, orgID valuer.UUID, req v3.FilterAttributeKeyRequest) (*v3.FilterAttributeKeyResponse, error) {
// TODO(srikanthccv): remove hardcoded metric name and support keys from any pod metric
req.DataSource = v3.DataSourceMetrics
req.AggregateAttribute = metricToUseForPods
if req.Limit == 0 {
req.Limit = 50
}
attributeKeysResponse, err := p.reader.GetMetricAttributeKeys(ctx, orgID, &req)
if err != nil {
return nil, err
}
// TODO(srikanthccv): only return resource attributes when we have a way to
// distinguish between resource attributes and other attributes.
filteredKeys := []v3.AttributeKey{}
for _, key := range attributeKeysResponse.AttributeKeys {
if slices.Contains(pointAttrsToIgnore, key.Key) {
continue
}
filteredKeys = append(filteredKeys, key)
}
return &v3.FilterAttributeKeyResponse{AttributeKeys: filteredKeys}, nil
}
func (p *PodsRepo) GetPodAttributeValues(ctx context.Context, orgID valuer.UUID, req v3.FilterAttributeValueRequest) (*v3.FilterAttributeValueResponse, error) {
req.DataSource = v3.DataSourceMetrics
req.AggregateAttribute = metricToUseForPods
if req.Limit == 0 {
req.Limit = 50
}
attributeValuesResponse, err := p.reader.GetMetricAttributeValues(ctx, orgID, &req)
if err != nil {
return nil, err
}
return attributeValuesResponse, nil
}
func (p *PodsRepo) DidSendPodMetrics(ctx context.Context) (bool, error) {
namesStr := "'" + strings.Join(podMetricNamesToCheck, "','") + "'"
query := fmt.Sprintf(didSendPodMetricsQuery,
constants.SIGNOZ_METRIC_DBNAME, constants.SIGNOZ_TIMESERIES_v4_1DAY_TABLENAME, namesStr)
count, err := p.reader.GetCountOfThings(ctx, query)
if err != nil {
return false, err
}
return count > 0, nil
}
func (p *PodsRepo) DidSendClusterMetrics(ctx context.Context) (bool, error) {
namesStr := "'" + strings.Join(clusterMetricNamesToCheck, "','") + "'"
query := fmt.Sprintf(didSendClusterMetricsQuery,
constants.SIGNOZ_METRIC_DBNAME, constants.SIGNOZ_TIMESERIES_v4_1DAY_TABLENAME, namesStr)
count, err := p.reader.GetCountOfThings(ctx, query)
if err != nil {
return false, err
}
return count > 0, nil
}
func (p *PodsRepo) IsSendingOptionalPodMetrics(ctx context.Context) (bool, error) {
namesStr := "'" + strings.Join(optionalPodMetricNamesToCheck, "','") + "'"
query := fmt.Sprintf(isSendingOptionalPodMetricsQuery,
constants.SIGNOZ_METRIC_DBNAME, constants.SIGNOZ_TIMESERIES_v4_1DAY_TABLENAME, namesStr)
count, err := p.reader.GetCountOfThings(ctx, query)
if err != nil {
return false, err
}
return count > 0, nil
}
func (p *PodsRepo) SendingRequiredMetadata(ctx context.Context) ([]model.PodOnboardingStatus, error) {
namesStr := "'" + strings.Join(podMetricNamesToCheck, "','") + "'"
query := fmt.Sprintf(isSendingRequiredMetadataQuery,
constants.SIGNOZ_METRIC_DBNAME, constants.SIGNOZ_TIMESERIES_V4_TABLENAME, namesStr)
result, err := p.reader.GetListResultV3(ctx, query)
if err != nil {
return nil, err
}
statuses := []model.PodOnboardingStatus{}
// for each pod, check if we have all the required metadata
for _, row := range result {
status := model.PodOnboardingStatus{}
switch v := row.Data["k8s.cluster.name"].(type) {
case string:
status.HasClusterName = true
status.ClusterName = v
case *string:
status.HasClusterName = *v != ""
status.ClusterName = *v
}
switch v := row.Data["k8s.node.name"].(type) {
case string:
status.HasNodeName = true
status.NodeName = v
case *string:
status.HasNodeName = *v != ""
status.NodeName = *v
}
switch v := row.Data["k8s.namespace.name"].(type) {
case string:
status.HasNamespaceName = true
status.NamespaceName = v
case *string:
status.HasNamespaceName = *v != ""
status.NamespaceName = *v
}
switch v := row.Data["k8s.deployment.name"].(type) {
case string:
status.HasDeploymentName = true
case *string:
status.HasDeploymentName = *v != ""
}
switch v := row.Data["k8s.statefulset.name"].(type) {
case string:
status.HasStatefulsetName = true
case *string:
status.HasStatefulsetName = *v != ""
}
switch v := row.Data["k8s.daemonset.name"].(type) {
case string:
status.HasDaemonsetName = true
case *string:
status.HasDaemonsetName = *v != ""
}
switch v := row.Data["k8s.cronjob.name"].(type) {
case string:
status.HasCronjobName = true
case *string:
status.HasCronjobName = *v != ""
}
switch v := row.Data["k8s.job.name"].(type) {
case string:
status.HasJobName = true
case *string:
status.HasJobName = *v != ""
}
switch v := row.Data["k8s.pod.name"].(type) {
case string:
status.PodName = v
case *string:
status.PodName = *v
}
if !status.HasClusterName ||
!status.HasNodeName ||
!status.HasNamespaceName ||
(!status.HasDeploymentName && !status.HasStatefulsetName && !status.HasDaemonsetName && !status.HasCronjobName && !status.HasJobName) {
statuses = append(statuses, status)
}
}
return statuses, nil
}
func (p *PodsRepo) getMetadataAttributes(ctx context.Context, req model.PodListRequest) (map[string]map[string]string, error) {
podAttrs := map[string]map[string]string{}
for _, key := range podAttrsToEnrich {
hasKey := false
for _, groupByKey := range req.GroupBy {
if groupByKey.Key == key {
hasKey = true
break
}
}
if !hasKey {
req.GroupBy = append(req.GroupBy, v3.AttributeKey{Key: key})
}
}
mq := v3.BuilderQuery{
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricToUseForPods,
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
GroupBy: req.GroupBy,
}
query, err := helpers.PrepareTimeseriesFilterQuery(req.Start, req.End, &mq)
if err != nil {
return nil, err
}
query = localQueryToDistributedQuery(query)
attrsListResponse, err := p.reader.GetListResultV3(ctx, query)
if err != nil {
return nil, err
}
for _, row := range attrsListResponse {
stringData := map[string]string{}
for key, value := range row.Data {
if str, ok := value.(string); ok {
stringData[key] = str
} else if strPtr, ok := value.(*string); ok {
stringData[key] = *strPtr
}
}
podName := stringData[k8sPodUIDAttrKey]
if _, ok := podAttrs[podName]; !ok {
podAttrs[podName] = map[string]string{}
}
for _, key := range req.GroupBy {
podAttrs[podName][key.Key] = stringData[key.Key]
}
}
return podAttrs, nil
}
func (p *PodsRepo) getTopPodGroups(ctx context.Context, orgID valuer.UUID, req model.PodListRequest, q *v3.QueryRangeParamsV3) ([]map[string]string, []map[string]string, error) {
step, timeSeriesTableName, samplesTableName := getParamsForTopPods(req)
queryNames := queryNamesForPods[req.OrderBy.ColumnName]
topPodGroupsQueryRangeParams := &v3.QueryRangeParamsV3{
Start: req.Start,
End: req.End,
Step: step,
CompositeQuery: &v3.CompositeQuery{
BuilderQueries: map[string]*v3.BuilderQuery{},
QueryType: v3.QueryTypeBuilder,
PanelType: v3.PanelTypeTable,
},
}
for _, queryName := range queryNames {
query := q.CompositeQuery.BuilderQueries[queryName].Clone()
query.StepInterval = step
query.MetricTableHints = &v3.MetricTableHints{
TimeSeriesTableName: timeSeriesTableName,
SamplesTableName: samplesTableName,
}
if req.Filters != nil && len(req.Filters.Items) > 0 {
if query.Filters == nil {
query.Filters = &v3.FilterSet{Operator: "AND", Items: []v3.FilterItem{}}
}
query.Filters.Items = append(query.Filters.Items, req.Filters.Items...)
}
topPodGroupsQueryRangeParams.CompositeQuery.BuilderQueries[queryName] = query
}
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "inframetrics",
instrumentationtypes.CodeFunctionName: "getTopPodGroups",
})
queryResponse, _, err := p.querierV2.QueryRange(ctx, orgID, topPodGroupsQueryRangeParams)
if err != nil {
return nil, nil, err
}
formattedResponse, err := postprocess.PostProcessResult(queryResponse, topPodGroupsQueryRangeParams)
if err != nil {
return nil, nil, err
}
if len(formattedResponse) == 0 || len(formattedResponse[0].Series) == 0 {
return nil, nil, nil
}
if req.OrderBy.Order == v3.DirectionDesc {
sort.Slice(formattedResponse[0].Series, func(i, j int) bool {
return formattedResponse[0].Series[i].Points[0].Value > formattedResponse[0].Series[j].Points[0].Value
})
} else {
sort.Slice(formattedResponse[0].Series, func(i, j int) bool {
return formattedResponse[0].Series[i].Points[0].Value < formattedResponse[0].Series[j].Points[0].Value
})
}
limit := math.Min(float64(req.Offset+req.Limit), float64(len(formattedResponse[0].Series)))
paginatedTopPodGroupsSeries := formattedResponse[0].Series[req.Offset:int(limit)]
topPodGroups := []map[string]string{}
for _, series := range paginatedTopPodGroupsSeries {
topPodGroups = append(topPodGroups, series.Labels)
}
allPodGroups := []map[string]string{}
for _, series := range formattedResponse[0].Series {
allPodGroups = append(allPodGroups, series.Labels)
}
return topPodGroups, allPodGroups, nil
}
func (p *PodsRepo) GetPodList(ctx context.Context, orgID valuer.UUID, req model.PodListRequest) (model.PodListResponse, error) {
resp := model.PodListResponse{}
if req.Limit == 0 {
req.Limit = 10
}
if req.OrderBy == nil {
req.OrderBy = &v3.OrderBy{ColumnName: "cpu", Order: v3.DirectionDesc}
}
if req.GroupBy == nil {
req.GroupBy = []v3.AttributeKey{{Key: k8sPodUIDAttrKey}}
resp.Type = model.ResponseTypeList
} else {
resp.Type = model.ResponseTypeGroupedList
}
step := int64(math.Max(float64(common.MinAllowedStepInterval(req.Start, req.End)), 60))
query := PodsTableListQuery.Clone()
query.Start = req.Start
query.End = req.End
query.Step = step
for _, query := range query.CompositeQuery.BuilderQueries {
query.StepInterval = step
if req.Filters != nil && len(req.Filters.Items) > 0 {
if query.Filters == nil {
query.Filters = &v3.FilterSet{Operator: "AND", Items: []v3.FilterItem{}}
}
query.Filters.Items = append(query.Filters.Items, req.Filters.Items...)
}
query.GroupBy = req.GroupBy
}
podAttrs, err := p.getMetadataAttributes(ctx, req)
if err != nil {
return resp, err
}
topPodGroups, allPodGroups, err := p.getTopPodGroups(ctx, orgID, req, query)
if err != nil {
return resp, err
}
groupFilters := map[string][]string{}
for _, topPodGroup := range topPodGroups {
for k, v := range topPodGroup {
groupFilters[k] = append(groupFilters[k], v)
}
}
for groupKey, groupValues := range groupFilters {
hasGroupFilter := false
if req.Filters != nil && len(req.Filters.Items) > 0 {
for _, filter := range req.Filters.Items {
if filter.Key.Key == groupKey {
hasGroupFilter = true
break
}
}
}
if !hasGroupFilter {
for _, query := range query.CompositeQuery.BuilderQueries {
query.Filters.Items = append(query.Filters.Items, v3.FilterItem{
Key: v3.AttributeKey{Key: groupKey},
Value: groupValues,
Operator: v3.FilterOperatorIn,
})
}
}
}
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "inframetrics",
instrumentationtypes.CodeFunctionName: "GetPodList",
})
queryResponse, _, err := p.querierV2.QueryRange(ctx, orgID, query)
if err != nil {
return resp, err
}
formattedResponse, err := postprocess.PostProcessResult(queryResponse, query)
if err != nil {
return resp, err
}
records := []model.PodListRecord{}
for _, result := range formattedResponse {
for _, row := range result.Table.Rows {
record := model.PodListRecord{
PodCPU: -1,
PodCPURequest: -1,
PodCPULimit: -1,
PodMemory: -1,
PodMemoryRequest: -1,
PodMemoryLimit: -1,
RestartCount: -1,
}
if podUID, ok := row.Data[k8sPodUIDAttrKey].(string); ok {
record.PodUID = podUID
}
if cpu, ok := row.Data["A"].(float64); ok {
record.PodCPU = cpu
}
if cpuRequest, ok := row.Data["B"].(float64); ok {
record.PodCPURequest = cpuRequest
}
if cpuLimit, ok := row.Data["C"].(float64); ok {
record.PodCPULimit = cpuLimit
}
if memory, ok := row.Data["D"].(float64); ok {
record.PodMemory = memory
}
if memoryRequest, ok := row.Data["E"].(float64); ok {
record.PodMemoryRequest = memoryRequest
}
if memoryLimit, ok := row.Data["F"].(float64); ok {
record.PodMemoryLimit = memoryLimit
}
if restarts, ok := row.Data["G"].(float64); ok {
record.RestartCount = int(restarts)
}
if pending, ok := row.Data["H"].(float64); ok {
record.CountByPhase.Pending = int(pending)
}
if running, ok := row.Data["I"].(float64); ok {
record.CountByPhase.Running = int(running)
}
if succeeded, ok := row.Data["J"].(float64); ok {
record.CountByPhase.Succeeded = int(succeeded)
}
if failed, ok := row.Data["K"].(float64); ok {
record.CountByPhase.Failed = int(failed)
}
record.Meta = map[string]string{}
if _, ok := podAttrs[record.PodUID]; ok && record.PodUID != "" {
record.Meta = podAttrs[record.PodUID]
}
for k, v := range row.Data {
if slices.Contains(podQueryNames, k) {
continue
}
if labelValue, ok := v.(string); ok {
record.Meta[k] = labelValue
}
}
records = append(records, record)
}
}
resp.Total = len(allPodGroups)
resp.Records = records
resp.SortBy(req.OrderBy)
return resp, nil
}

View File

@@ -0,0 +1,332 @@
package inframetrics
import v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
var PodsTableListQuery = v3.QueryRangeParamsV3{
CompositeQuery: &v3.CompositeQuery{
BuilderQueries: map[string]*v3.BuilderQuery{
// pod cpu utilization
"A": {
QueryName: "A",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForPods["cpu"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{
{
Key: k8sPodUIDAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "A",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
// pod cpu request utilization
"B": {
QueryName: "B",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForPods["cpu_request"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{
{
Key: k8sPodUIDAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "B",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationAvg,
Disabled: false,
},
// pod cpu limit utilization
"C": {
QueryName: "C",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForPods["cpu_limit"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{
{
Key: k8sPodUIDAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "C",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationAvg,
Disabled: false,
},
// pod memory utilization
"D": {
QueryName: "D",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForPods["memory"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{
{
Key: k8sPodUIDAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "D",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
// pod memory request utilization
"E": {
QueryName: "E",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForPods["memory_request"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{
{
Key: k8sPodUIDAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "E",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationAvg,
Disabled: false,
},
// pod memory limit utilization
"F": {
QueryName: "F",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForPods["memory_limit"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{
{
Key: k8sPodUIDAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "F",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationAvg,
Disabled: false,
},
"G": {
QueryName: "G",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForPods["restarts"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{
{
Key: k8sPodUIDAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "G",
ReduceTo: v3.ReduceToOperatorSum,
TimeAggregation: v3.TimeAggregationAnyLast,
SpaceAggregation: v3.SpaceAggregationMax,
Functions: []v3.Function{{Name: v3.FunctionNameRunningDiff}},
Disabled: false,
},
// pod phase pending
"H": {
QueryName: "H",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForPods["pod_phase"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{
{
Key: v3.AttributeKey{
Key: "__value",
},
Operator: v3.FilterOperatorEqual,
Value: 1,
},
},
},
GroupBy: []v3.AttributeKey{
{
Key: k8sPodUIDAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "H",
ReduceTo: v3.ReduceToOperatorLast,
TimeAggregation: v3.TimeAggregationAnyLast,
SpaceAggregation: v3.SpaceAggregationCount,
Disabled: false,
},
// pod phase running
"I": {
QueryName: "I",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForPods["pod_phase"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{
{
Key: v3.AttributeKey{
Key: "__value",
},
Operator: v3.FilterOperatorEqual,
Value: 2,
},
},
},
GroupBy: []v3.AttributeKey{
{
Key: k8sPodUIDAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "I",
ReduceTo: v3.ReduceToOperatorLast,
TimeAggregation: v3.TimeAggregationAnyLast,
SpaceAggregation: v3.SpaceAggregationCount,
Disabled: false,
},
// pod phase succeeded
"J": {
QueryName: "J",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForPods["pod_phase"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{
{
Key: v3.AttributeKey{
Key: "__value",
},
Operator: v3.FilterOperatorEqual,
Value: 3,
},
},
},
GroupBy: []v3.AttributeKey{
{
Key: k8sPodUIDAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "J",
ReduceTo: v3.ReduceToOperatorLast,
TimeAggregation: v3.TimeAggregationAnyLast,
SpaceAggregation: v3.SpaceAggregationCount,
Disabled: false,
},
// pod phase failed
"K": {
QueryName: "K",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForPods["pod_phase"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{
{
Key: v3.AttributeKey{
Key: "__value",
},
Operator: v3.FilterOperatorEqual,
Value: 4,
},
},
},
GroupBy: []v3.AttributeKey{
{
Key: k8sPodUIDAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "K",
ReduceTo: v3.ReduceToOperatorLast,
TimeAggregation: v3.TimeAggregationAnyLast,
SpaceAggregation: v3.SpaceAggregationCount,
Disabled: false,
},
},
PanelType: v3.PanelTypeTable,
QueryType: v3.QueryTypeBuilder,
},
Version: "v4",
FormatForWeb: true,
}

View File

@@ -0,0 +1,73 @@
package inframetrics
import v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
var ProcessesTableListQuery = v3.QueryRangeParamsV3{
CompositeQuery: &v3.CompositeQuery{
BuilderQueries: map[string]*v3.BuilderQuery{
"A": {
QueryName: "A",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForProcesses["cpu"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Cumulative,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{
{
Key: processPIDAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "A",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationRate,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: true,
},
"F1": {
QueryName: "F1",
Expression: "A",
Legend: "Process CPU Usage (%)",
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
},
"C": {
QueryName: "C",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForProcesses["memory"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Cumulative,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{
{
Key: processPIDAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "C",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
},
PanelType: v3.PanelTypeTable,
QueryType: v3.QueryTypeBuilder,
},
Version: "v4",
FormatForWeb: true,
}

View File

@@ -0,0 +1,342 @@
package inframetrics
import (
"context"
"math"
"sort"
"github.com/SigNoz/signoz/pkg/query-service/app/metrics/v4/helpers"
"github.com/SigNoz/signoz/pkg/query-service/common"
"github.com/SigNoz/signoz/pkg/query-service/interfaces"
"github.com/SigNoz/signoz/pkg/query-service/model"
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
"github.com/SigNoz/signoz/pkg/query-service/postprocess"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
"github.com/SigNoz/signoz/pkg/valuer"
"golang.org/x/exp/slices"
)
var (
queryNamesForTopProcesses = map[string][]string{
"cpu": {"A"},
"memory": {"C"},
}
processPIDAttrKey = "process.pid"
metricNamesForProcesses = map[string]string{
"cpu": "process.cpu.time",
"memory": "process.memory.usage",
}
metricToUseForProcessAttributes = "process.memory.usage"
processNameAttrKey = "process.executable.name"
processCMDAttrKey = "process.command"
processCMDLineAttrKey = "process.command_line"
)
type ProcessesRepo struct {
reader interfaces.Reader
querierV2 interfaces.Querier
}
func NewProcessesRepo(reader interfaces.Reader, querierV2 interfaces.Querier) *ProcessesRepo {
return &ProcessesRepo{reader: reader, querierV2: querierV2}
}
func (p *ProcessesRepo) GetProcessAttributeKeys(ctx context.Context, orgID valuer.UUID, req v3.FilterAttributeKeyRequest) (*v3.FilterAttributeKeyResponse, error) {
// TODO(srikanthccv): remove hardcoded metric name and support keys from any system metric
req.DataSource = v3.DataSourceMetrics
req.AggregateAttribute = "process.memory.usage"
if req.Limit == 0 {
req.Limit = 50
}
attributeKeysResponse, err := p.reader.GetMetricAttributeKeys(ctx, orgID, &req)
if err != nil {
return nil, err
}
// TODO(srikanthccv): only return resource attributes when we have a way to
// distinguish between resource attributes and other attributes.
filteredKeys := []v3.AttributeKey{}
for _, key := range attributeKeysResponse.AttributeKeys {
if slices.Contains(pointAttrsToIgnore, key.Key) {
continue
}
filteredKeys = append(filteredKeys, key)
}
return &v3.FilterAttributeKeyResponse{AttributeKeys: filteredKeys}, nil
}
func (p *ProcessesRepo) GetProcessAttributeValues(ctx context.Context, orgID valuer.UUID, req v3.FilterAttributeValueRequest) (*v3.FilterAttributeValueResponse, error) {
req.DataSource = v3.DataSourceMetrics
req.AggregateAttribute = "process.memory.usage"
if req.Limit == 0 {
req.Limit = 50
}
attributeValuesResponse, err := p.reader.GetMetricAttributeValues(ctx, orgID, &req)
if err != nil {
return nil, err
}
return attributeValuesResponse, nil
}
func (p *ProcessesRepo) getMetadataAttributes(ctx context.Context,
req model.ProcessListRequest) (map[string]map[string]string, error) {
processAttrs := map[string]map[string]string{}
keysToAdd := []string{"process.pid", "process.executable.name", "process.command", "process.command_line"}
for _, key := range keysToAdd {
hasKey := false
for _, groupByKey := range req.GroupBy {
if groupByKey.Key == key {
hasKey = true
break
}
}
if !hasKey {
req.GroupBy = append(req.GroupBy, v3.AttributeKey{Key: key})
}
}
mq := v3.BuilderQuery{
AggregateAttribute: v3.AttributeKey{
Key: metricToUseForProcessAttributes,
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Cumulative,
GroupBy: req.GroupBy,
}
query, err := helpers.PrepareTimeseriesFilterQuery(req.Start, req.End, &mq)
if err != nil {
return nil, err
}
query = localQueryToDistributedQuery(query)
attrsListResponse, err := p.reader.GetListResultV3(ctx, query)
if err != nil {
return nil, err
}
for _, row := range attrsListResponse {
stringData := map[string]string{}
for key, value := range row.Data {
if str, ok := value.(string); ok {
stringData[key] = str
} else if strPtr, ok := value.(*string); ok {
stringData[key] = *strPtr
}
}
processID := stringData[processPIDAttrKey]
if _, ok := processAttrs[processID]; !ok {
processAttrs[processID] = map[string]string{}
}
for _, key := range req.GroupBy {
processAttrs[processID][key.Key] = stringData[key.Key]
}
}
return processAttrs, nil
}
func (p *ProcessesRepo) getTopProcessGroups(ctx context.Context, orgID valuer.UUID, req model.ProcessListRequest, q *v3.QueryRangeParamsV3) ([]map[string]string, []map[string]string, error) {
step, timeSeriesTableName, samplesTableName := getParamsForTopProcesses(req)
queryNames := queryNamesForTopProcesses[req.OrderBy.ColumnName]
topProcessGroupsQueryRangeParams := &v3.QueryRangeParamsV3{
Start: req.Start,
End: req.End,
Step: step,
CompositeQuery: &v3.CompositeQuery{
BuilderQueries: map[string]*v3.BuilderQuery{},
QueryType: v3.QueryTypeBuilder,
PanelType: v3.PanelTypeTable,
},
}
for _, queryName := range queryNames {
query := q.CompositeQuery.BuilderQueries[queryName].Clone()
query.StepInterval = step
query.MetricTableHints = &v3.MetricTableHints{
TimeSeriesTableName: timeSeriesTableName,
SamplesTableName: samplesTableName,
}
if req.Filters != nil && len(req.Filters.Items) > 0 {
query.Filters.Items = append(query.Filters.Items, req.Filters.Items...)
}
topProcessGroupsQueryRangeParams.CompositeQuery.BuilderQueries[queryName] = query
}
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "inframetrics",
instrumentationtypes.CodeFunctionName: "getTopProcessGroups",
})
queryResponse, _, err := p.querierV2.QueryRange(ctx, orgID, topProcessGroupsQueryRangeParams)
if err != nil {
return nil, nil, err
}
formattedResponse, err := postprocess.PostProcessResult(queryResponse, topProcessGroupsQueryRangeParams)
if err != nil {
return nil, nil, err
}
if len(formattedResponse) == 0 || len(formattedResponse[0].Series) == 0 {
return nil, nil, nil
}
if req.OrderBy.Order == v3.DirectionDesc {
sort.Slice(formattedResponse[0].Series, func(i, j int) bool {
return formattedResponse[0].Series[i].Points[0].Value > formattedResponse[0].Series[j].Points[0].Value
})
} else {
sort.Slice(formattedResponse[0].Series, func(i, j int) bool {
return formattedResponse[0].Series[i].Points[0].Value < formattedResponse[0].Series[j].Points[0].Value
})
}
limit := math.Min(float64(req.Offset+req.Limit), float64(len(formattedResponse[0].Series)))
paginatedTopProcessGroupsSeries := formattedResponse[0].Series[req.Offset:int(limit)]
topProcessGroups := []map[string]string{}
for _, series := range paginatedTopProcessGroupsSeries {
topProcessGroups = append(topProcessGroups, series.Labels)
}
allProcessGroups := []map[string]string{}
for _, series := range formattedResponse[0].Series {
allProcessGroups = append(allProcessGroups, series.Labels)
}
return topProcessGroups, allProcessGroups, nil
}
func (p *ProcessesRepo) GetProcessList(ctx context.Context, orgID valuer.UUID, req model.ProcessListRequest) (model.ProcessListResponse, error) {
resp := model.ProcessListResponse{}
if req.Limit == 0 {
req.Limit = 10
}
// default to cpu order by
if req.OrderBy == nil {
req.OrderBy = &v3.OrderBy{ColumnName: "cpu", Order: v3.DirectionDesc}
}
// default to process pid group by
if len(req.GroupBy) == 0 {
req.GroupBy = []v3.AttributeKey{{Key: processPIDAttrKey}}
resp.Type = model.ResponseTypeList
} else {
resp.Type = model.ResponseTypeGroupedList
}
step := int64(math.Max(float64(common.MinAllowedStepInterval(req.Start, req.End)), 60))
query := ProcessesTableListQuery.Clone()
query.Start = req.Start
query.End = req.End
query.Step = step
for _, query := range query.CompositeQuery.BuilderQueries {
query.StepInterval = step
if req.Filters != nil && len(req.Filters.Items) > 0 {
query.Filters.Items = append(query.Filters.Items, req.Filters.Items...)
}
query.GroupBy = req.GroupBy
}
processAttrs, err := p.getMetadataAttributes(ctx, req)
if err != nil {
return resp, err
}
topProcessGroups, allProcessGroups, err := p.getTopProcessGroups(ctx, orgID, req, query)
if err != nil {
return resp, err
}
groupFilters := map[string][]string{}
for _, topProcessGroup := range topProcessGroups {
for k, v := range topProcessGroup {
groupFilters[k] = append(groupFilters[k], v)
}
}
for groupKey, groupValues := range groupFilters {
hasGroupFilter := false
if req.Filters != nil && len(req.Filters.Items) > 0 {
for _, filter := range req.Filters.Items {
if filter.Key.Key == groupKey {
hasGroupFilter = true
break
}
}
}
if !hasGroupFilter {
for _, query := range query.CompositeQuery.BuilderQueries {
query.Filters.Items = append(query.Filters.Items, v3.FilterItem{
Key: v3.AttributeKey{Key: groupKey},
Value: groupValues,
Operator: v3.FilterOperatorIn,
})
}
}
}
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "inframetrics",
instrumentationtypes.CodeFunctionName: "GetProcessList",
})
queryResponse, _, err := p.querierV2.QueryRange(ctx, orgID, query)
if err != nil {
return resp, err
}
formattedResponse, err := postprocess.PostProcessResult(queryResponse, query)
if err != nil {
return resp, err
}
records := []model.ProcessListRecord{}
for _, result := range formattedResponse {
for _, row := range result.Table.Rows {
record := model.ProcessListRecord{
ProcessCPU: -1,
ProcessMemory: -1,
}
pid, ok := row.Data[processPIDAttrKey].(string)
if ok {
record.ProcessID = pid
}
processCPU, ok := row.Data["F1"].(float64)
if ok {
record.ProcessCPU = processCPU
}
processMemory, ok := row.Data["C"].(float64)
if ok {
record.ProcessMemory = processMemory
}
record.Meta = processAttrs[record.ProcessID]
record.ProcessName = record.Meta[processNameAttrKey]
record.ProcessCMD = record.Meta[processCMDAttrKey]
record.ProcessCMDLine = record.Meta[processCMDLineAttrKey]
records = append(records, record)
}
}
resp.Total = len(allProcessGroups)
resp.Records = records
return resp, nil
}

View File

@@ -0,0 +1,391 @@
package inframetrics
import (
"context"
"math"
"sort"
"github.com/SigNoz/signoz/pkg/query-service/app/metrics/v4/helpers"
"github.com/SigNoz/signoz/pkg/query-service/common"
"github.com/SigNoz/signoz/pkg/query-service/interfaces"
"github.com/SigNoz/signoz/pkg/query-service/model"
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
"github.com/SigNoz/signoz/pkg/query-service/postprocess"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
"github.com/SigNoz/signoz/pkg/valuer"
"golang.org/x/exp/slices"
)
var (
metricToUseForVolumes = "k8s.volume.available"
volumeAttrsToEnrich = []string{
"k8s.pod.uid",
"k8s.pod.name",
"k8s.namespace.name",
"k8s.node.name",
"k8s.statefulset.name",
"k8s.cluster.name",
"k8s.persistentvolumeclaim.name",
}
k8sPersistentVolumeClaimNameAttrKey = "k8s.persistentvolumeclaim.name"
queryNamesForVolumes = map[string][]string{
"available": {"A"},
"capacity": {"B", "A"},
"usage": {"F1", "B", "A"},
"inodes": {"C", "A"},
"inodes_free": {"D", "A"},
"inodes_used": {"E", "A"},
}
volumeQueryNames = []string{"A", "B", "C", "D", "E", "F1"}
metricNamesForVolumes = map[string]string{
"available": "k8s.volume.available",
"capacity": "k8s.volume.capacity",
"inodes": "k8s.volume.inodes",
"inodes_free": "k8s.volume.inodes.free",
"inodes_used": "k8s.volume.inodes.used",
}
)
type PvcsRepo struct {
reader interfaces.Reader
querierV2 interfaces.Querier
}
func NewPvcsRepo(reader interfaces.Reader, querierV2 interfaces.Querier) *PvcsRepo {
return &PvcsRepo{reader: reader, querierV2: querierV2}
}
func (p *PvcsRepo) GetPvcAttributeKeys(ctx context.Context, orgID valuer.UUID, req v3.FilterAttributeKeyRequest) (*v3.FilterAttributeKeyResponse, error) {
req.DataSource = v3.DataSourceMetrics
req.AggregateAttribute = metricToUseForVolumes
if req.Limit == 0 {
req.Limit = 50
}
attributeKeysResponse, err := p.reader.GetMetricAttributeKeys(ctx, orgID, &req)
if err != nil {
return nil, err
}
// TODO(srikanthccv): only return resource attributes when we have a way to
// distinguish between resource attributes and other attributes.
filteredKeys := []v3.AttributeKey{}
for _, key := range attributeKeysResponse.AttributeKeys {
if slices.Contains(pointAttrsToIgnore, key.Key) {
continue
}
filteredKeys = append(filteredKeys, key)
}
return &v3.FilterAttributeKeyResponse{AttributeKeys: filteredKeys}, nil
}
func (p *PvcsRepo) GetPvcAttributeValues(ctx context.Context, orgID valuer.UUID, req v3.FilterAttributeValueRequest) (*v3.FilterAttributeValueResponse, error) {
req.DataSource = v3.DataSourceMetrics
req.AggregateAttribute = metricToUseForVolumes
if req.Limit == 0 {
req.Limit = 50
}
attributeValuesResponse, err := p.reader.GetMetricAttributeValues(ctx, orgID, &req)
if err != nil {
return nil, err
}
return attributeValuesResponse, nil
}
func (p *PvcsRepo) getMetadataAttributes(ctx context.Context, req model.VolumeListRequest) (map[string]map[string]string, error) {
volumeAttrs := map[string]map[string]string{}
for _, key := range volumeAttrsToEnrich {
hasKey := false
for _, groupByKey := range req.GroupBy {
if groupByKey.Key == key {
hasKey = true
break
}
}
if !hasKey {
req.GroupBy = append(req.GroupBy, v3.AttributeKey{Key: key})
}
}
mq := v3.BuilderQuery{
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricToUseForVolumes,
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
GroupBy: req.GroupBy,
}
query, err := helpers.PrepareTimeseriesFilterQuery(req.Start, req.End, &mq)
if err != nil {
return nil, err
}
query = localQueryToDistributedQuery(query)
attrsListResponse, err := p.reader.GetListResultV3(ctx, query)
if err != nil {
return nil, err
}
for _, row := range attrsListResponse {
stringData := map[string]string{}
for key, value := range row.Data {
if str, ok := value.(string); ok {
stringData[key] = str
} else if strPtr, ok := value.(*string); ok {
stringData[key] = *strPtr
}
}
volumeName := stringData[k8sPersistentVolumeClaimNameAttrKey]
if _, ok := volumeAttrs[volumeName]; !ok {
volumeAttrs[volumeName] = map[string]string{}
}
for _, key := range req.GroupBy {
volumeAttrs[volumeName][key.Key] = stringData[key.Key]
}
}
return volumeAttrs, nil
}
func (p *PvcsRepo) getTopVolumeGroups(ctx context.Context, orgID valuer.UUID, req model.VolumeListRequest, q *v3.QueryRangeParamsV3) ([]map[string]string, []map[string]string, error) {
step, timeSeriesTableName, samplesTableName := getParamsForTopVolumes(req)
queryNames := queryNamesForVolumes[req.OrderBy.ColumnName]
topVolumeGroupsQueryRangeParams := &v3.QueryRangeParamsV3{
Start: req.Start,
End: req.End,
Step: step,
CompositeQuery: &v3.CompositeQuery{
BuilderQueries: map[string]*v3.BuilderQuery{},
QueryType: v3.QueryTypeBuilder,
PanelType: v3.PanelTypeTable,
},
}
for _, queryName := range queryNames {
query := q.CompositeQuery.BuilderQueries[queryName].Clone()
query.StepInterval = step
query.MetricTableHints = &v3.MetricTableHints{
TimeSeriesTableName: timeSeriesTableName,
SamplesTableName: samplesTableName,
}
if req.Filters != nil && len(req.Filters.Items) > 0 {
if query.Filters == nil {
query.Filters = &v3.FilterSet{Operator: "AND", Items: []v3.FilterItem{}}
}
query.Filters.Items = append(query.Filters.Items, req.Filters.Items...)
}
topVolumeGroupsQueryRangeParams.CompositeQuery.BuilderQueries[queryName] = query
}
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "inframetrics",
instrumentationtypes.CodeFunctionName: "getTopVolumeGroups",
})
queryResponse, _, err := p.querierV2.QueryRange(ctx, orgID, topVolumeGroupsQueryRangeParams)
if err != nil {
return nil, nil, err
}
formattedResponse, err := postprocess.PostProcessResult(queryResponse, topVolumeGroupsQueryRangeParams)
if err != nil {
return nil, nil, err
}
if len(formattedResponse) == 0 || len(formattedResponse[0].Series) == 0 {
return nil, nil, nil
}
if req.OrderBy.Order == v3.DirectionDesc {
sort.Slice(formattedResponse[0].Series, func(i, j int) bool {
return formattedResponse[0].Series[i].Points[0].Value > formattedResponse[0].Series[j].Points[0].Value
})
} else {
sort.Slice(formattedResponse[0].Series, func(i, j int) bool {
return formattedResponse[0].Series[i].Points[0].Value < formattedResponse[0].Series[j].Points[0].Value
})
}
limit := math.Min(float64(req.Offset+req.Limit), float64(len(formattedResponse[0].Series)))
paginatedTopVolumeGroupsSeries := formattedResponse[0].Series[req.Offset:int(limit)]
topVolumeGroups := []map[string]string{}
for _, series := range paginatedTopVolumeGroupsSeries {
topVolumeGroups = append(topVolumeGroups, series.Labels)
}
allVolumeGroups := []map[string]string{}
for _, series := range formattedResponse[0].Series {
allVolumeGroups = append(allVolumeGroups, series.Labels)
}
return topVolumeGroups, allVolumeGroups, nil
}
func (p *PvcsRepo) GetPvcList(ctx context.Context, orgID valuer.UUID, req model.VolumeListRequest) (model.VolumeListResponse, error) {
resp := model.VolumeListResponse{}
if req.Limit == 0 {
req.Limit = 10
}
if req.OrderBy == nil {
req.OrderBy = &v3.OrderBy{ColumnName: "usage", Order: v3.DirectionDesc}
}
if req.GroupBy == nil {
req.GroupBy = []v3.AttributeKey{{Key: k8sPersistentVolumeClaimNameAttrKey}}
resp.Type = model.ResponseTypeList
} else {
resp.Type = model.ResponseTypeGroupedList
}
step := int64(math.Max(float64(common.MinAllowedStepInterval(req.Start, req.End)), 60))
query := PvcsTableListQuery.Clone()
query.Start = req.Start
query.End = req.End
query.Step = step
for _, query := range query.CompositeQuery.BuilderQueries {
query.StepInterval = step
if req.Filters != nil && len(req.Filters.Items) > 0 {
if query.Filters == nil {
query.Filters = &v3.FilterSet{Operator: "AND", Items: []v3.FilterItem{}}
}
query.Filters.Items = append(query.Filters.Items, req.Filters.Items...)
}
query.GroupBy = req.GroupBy
}
volumeAttrs, err := p.getMetadataAttributes(ctx, req)
if err != nil {
return resp, err
}
topVolumeGroups, allVolumeGroups, err := p.getTopVolumeGroups(ctx, orgID, req, query)
if err != nil {
return resp, err
}
groupFilters := map[string][]string{}
for _, topVolumeGroup := range topVolumeGroups {
for k, v := range topVolumeGroup {
groupFilters[k] = append(groupFilters[k], v)
}
}
for groupKey, groupValues := range groupFilters {
hasGroupFilter := false
if req.Filters != nil && len(req.Filters.Items) > 0 {
for _, filter := range req.Filters.Items {
if filter.Key.Key == groupKey {
hasGroupFilter = true
break
}
}
}
if !hasGroupFilter {
for _, query := range query.CompositeQuery.BuilderQueries {
query.Filters.Items = append(query.Filters.Items, v3.FilterItem{
Key: v3.AttributeKey{Key: groupKey},
Value: groupValues,
Operator: v3.FilterOperatorIn,
})
}
}
}
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "inframetrics",
instrumentationtypes.CodeFunctionName: "GetPvcList",
})
queryResponse, _, err := p.querierV2.QueryRange(ctx, orgID, query)
if err != nil {
return resp, err
}
formattedResponse, err := postprocess.PostProcessResult(queryResponse, query)
if err != nil {
return resp, err
}
records := []model.VolumeListRecord{}
for _, result := range formattedResponse {
for _, row := range result.Table.Rows {
record := model.VolumeListRecord{
VolumeUsage: -1,
VolumeAvailable: -1,
VolumeCapacity: -1,
VolumeInodes: -1,
VolumeInodesFree: -1,
VolumeInodesUsed: -1,
Meta: map[string]string{},
}
if volumeName, ok := row.Data[k8sPersistentVolumeClaimNameAttrKey].(string); ok {
record.PersistentVolumeClaimName = volumeName
}
if volumeAvailable, ok := row.Data["A"].(float64); ok {
record.VolumeAvailable = volumeAvailable
}
if volumeCapacity, ok := row.Data["B"].(float64); ok {
record.VolumeCapacity = volumeCapacity
}
if volumeInodes, ok := row.Data["C"].(float64); ok {
record.VolumeInodes = volumeInodes
}
if volumeInodesFree, ok := row.Data["D"].(float64); ok {
record.VolumeInodesFree = volumeInodesFree
}
if volumeInodesUsed, ok := row.Data["E"].(float64); ok {
record.VolumeInodesUsed = volumeInodesUsed
}
record.VolumeUsage = record.VolumeCapacity - record.VolumeAvailable
record.Meta = map[string]string{}
if _, ok := volumeAttrs[record.PersistentVolumeClaimName]; ok && record.PersistentVolumeClaimName != "" {
record.Meta = volumeAttrs[record.PersistentVolumeClaimName]
}
for k, v := range row.Data {
if slices.Contains(volumeQueryNames, k) {
continue
}
if labelValue, ok := v.(string); ok {
record.Meta[k] = labelValue
}
}
records = append(records, record)
}
}
resp.Total = len(allVolumeGroups)
resp.Records = records
resp.SortBy(req.OrderBy)
return resp, nil
}

View File

@@ -0,0 +1,204 @@
package inframetrics
import v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
var PvcsTableListQuery = v3.QueryRangeParamsV3{
CompositeQuery: &v3.CompositeQuery{
BuilderQueries: map[string]*v3.BuilderQuery{
// k8s.volume.available
"A": {
QueryName: "A",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForVolumes["available"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{
{
Key: v3.AttributeKey{
Key: k8sPersistentVolumeClaimNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
Operator: v3.FilterOperatorNotEqual,
Value: "",
},
},
},
GroupBy: []v3.AttributeKey{
{
Key: k8sPersistentVolumeClaimNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "A",
ReduceTo: v3.ReduceToOperatorLast,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
// k8s.volume.capacity
"B": {
QueryName: "B",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForVolumes["capacity"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{
{
Key: v3.AttributeKey{
Key: k8sPersistentVolumeClaimNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
Operator: v3.FilterOperatorNotEqual,
Value: "",
},
},
},
GroupBy: []v3.AttributeKey{
{
Key: k8sPersistentVolumeClaimNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "B",
ReduceTo: v3.ReduceToOperatorLast,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
"F1": {
QueryName: "F1",
DataSource: v3.DataSourceMetrics,
Expression: "B - A",
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
ReduceTo: v3.ReduceToOperatorLast,
},
// k8s.volume.inodes
"C": {
QueryName: "C",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForVolumes["inodes"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{
{
Key: v3.AttributeKey{
Key: k8sPersistentVolumeClaimNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
Operator: v3.FilterOperatorNotEqual,
Value: "",
},
},
},
GroupBy: []v3.AttributeKey{
{
Key: k8sPersistentVolumeClaimNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "C",
ReduceTo: v3.ReduceToOperatorLast,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
// k8s.volume.inodes_free
"D": {
QueryName: "D",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForVolumes["inodes_free"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{
{
Key: v3.AttributeKey{
Key: k8sPersistentVolumeClaimNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
Operator: v3.FilterOperatorNotEqual,
Value: "",
},
},
},
GroupBy: []v3.AttributeKey{
{
Key: k8sPersistentVolumeClaimNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "D",
ReduceTo: v3.ReduceToOperatorLast,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
// k8s.volume.inodes_used
"E": {
QueryName: "E",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForVolumes["inodes_used"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{
{
Key: v3.AttributeKey{
Key: k8sPersistentVolumeClaimNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
Operator: v3.FilterOperatorNotEqual,
Value: "",
},
},
},
GroupBy: []v3.AttributeKey{
{
Key: k8sPersistentVolumeClaimNameAttrKey,
DataType: v3.AttributeKeyDataTypeString,
Type: v3.AttributeKeyTypeResource,
},
},
Expression: "E",
ReduceTo: v3.ReduceToOperatorLast,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
},
PanelType: v3.PanelTypeTable,
QueryType: v3.QueryTypeBuilder,
},
Version: "v4",
FormatForWeb: true,
}

View File

@@ -0,0 +1,457 @@
package inframetrics
import (
"context"
"math"
"sort"
"github.com/SigNoz/signoz/pkg/query-service/app/metrics/v4/helpers"
"github.com/SigNoz/signoz/pkg/query-service/common"
"github.com/SigNoz/signoz/pkg/query-service/interfaces"
"github.com/SigNoz/signoz/pkg/query-service/model"
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
"github.com/SigNoz/signoz/pkg/query-service/postprocess"
"github.com/SigNoz/signoz/pkg/types/ctxtypes"
"github.com/SigNoz/signoz/pkg/types/instrumentationtypes"
"github.com/SigNoz/signoz/pkg/valuer"
"golang.org/x/exp/slices"
)
var (
metricToUseForStatefulSets = "k8s.pod.cpu.usage"
k8sStatefulSetNameAttrKey = "k8s.statefulset.name"
metricNamesForStatefulSets = map[string]string{
"desired_pods": "k8s.statefulset.desired_pods",
"available_pods": "k8s.statefulset.current_pods",
}
statefulSetAttrsToEnrich = []string{
"k8s.statefulset.name",
"k8s.namespace.name",
"k8s.cluster.name",
}
queryNamesForStatefulSets = map[string][]string{
"cpu": {"A"},
"cpu_request": {"B", "A"},
"cpu_limit": {"C", "A"},
"memory": {"D"},
"memory_request": {"E", "D"},
"memory_limit": {"F", "D"},
"restarts": {"G", "A"},
"desired_pods": {"H"},
"available_pods": {"I"},
}
builderQueriesForStatefulSets = map[string]*v3.BuilderQuery{
// desired pods
"H": {
QueryName: "H",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForStatefulSets["desired_pods"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{},
Expression: "H",
ReduceTo: v3.ReduceToOperatorLast,
TimeAggregation: v3.TimeAggregationAnyLast,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
// available pods
"I": {
QueryName: "I",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForStatefulSets["available_pods"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{},
Expression: "I",
ReduceTo: v3.ReduceToOperatorLast,
TimeAggregation: v3.TimeAggregationAnyLast,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
}
statefulSetQueryNames = []string{"A", "B", "C", "D", "E", "F", "G", "H", "I"}
)
type StatefulSetsRepo struct {
reader interfaces.Reader
querierV2 interfaces.Querier
}
func NewStatefulSetsRepo(reader interfaces.Reader, querierV2 interfaces.Querier) *StatefulSetsRepo {
return &StatefulSetsRepo{reader: reader, querierV2: querierV2}
}
func (d *StatefulSetsRepo) GetStatefulSetAttributeKeys(ctx context.Context, orgID valuer.UUID, req v3.FilterAttributeKeyRequest) (*v3.FilterAttributeKeyResponse, error) {
// TODO(srikanthccv): remove hardcoded metric name and support keys from any pod metric
req.DataSource = v3.DataSourceMetrics
req.AggregateAttribute = metricToUseForStatefulSets
if req.Limit == 0 {
req.Limit = 50
}
attributeKeysResponse, err := d.reader.GetMetricAttributeKeys(ctx, orgID, &req)
if err != nil {
return nil, err
}
// TODO(srikanthccv): only return resource attributes when we have a way to
// distinguish between resource attributes and other attributes.
filteredKeys := []v3.AttributeKey{}
for _, key := range attributeKeysResponse.AttributeKeys {
if slices.Contains(pointAttrsToIgnore, key.Key) {
continue
}
filteredKeys = append(filteredKeys, key)
}
return &v3.FilterAttributeKeyResponse{AttributeKeys: filteredKeys}, nil
}
func (d *StatefulSetsRepo) GetStatefulSetAttributeValues(ctx context.Context, orgID valuer.UUID, req v3.FilterAttributeValueRequest) (*v3.FilterAttributeValueResponse, error) {
req.DataSource = v3.DataSourceMetrics
req.AggregateAttribute = metricToUseForStatefulSets
if req.Limit == 0 {
req.Limit = 50
}
attributeValuesResponse, err := d.reader.GetMetricAttributeValues(ctx, orgID, &req)
if err != nil {
return nil, err
}
return attributeValuesResponse, nil
}
func (d *StatefulSetsRepo) getMetadataAttributes(ctx context.Context, req model.StatefulSetListRequest) (map[string]map[string]string, error) {
statefulSetAttrs := map[string]map[string]string{}
for _, key := range statefulSetAttrsToEnrich {
hasKey := false
for _, groupByKey := range req.GroupBy {
if groupByKey.Key == key {
hasKey = true
break
}
}
if !hasKey {
req.GroupBy = append(req.GroupBy, v3.AttributeKey{Key: key})
}
}
mq := v3.BuilderQuery{
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricToUseForStatefulSets,
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
GroupBy: req.GroupBy,
}
query, err := helpers.PrepareTimeseriesFilterQuery(req.Start, req.End, &mq)
if err != nil {
return nil, err
}
query = localQueryToDistributedQuery(query)
attrsListResponse, err := d.reader.GetListResultV3(ctx, query)
if err != nil {
return nil, err
}
for _, row := range attrsListResponse {
stringData := map[string]string{}
for key, value := range row.Data {
if str, ok := value.(string); ok {
stringData[key] = str
} else if strPtr, ok := value.(*string); ok {
stringData[key] = *strPtr
}
}
statefulSetName := stringData[k8sStatefulSetNameAttrKey]
if _, ok := statefulSetAttrs[statefulSetName]; !ok {
statefulSetAttrs[statefulSetName] = map[string]string{}
}
for _, key := range req.GroupBy {
statefulSetAttrs[statefulSetName][key.Key] = stringData[key.Key]
}
}
return statefulSetAttrs, nil
}
func (d *StatefulSetsRepo) getTopStatefulSetGroups(ctx context.Context, orgID valuer.UUID, req model.StatefulSetListRequest, q *v3.QueryRangeParamsV3) ([]map[string]string, []map[string]string, error) {
step, timeSeriesTableName, samplesTableName := getParamsForTopStatefulSets(req)
queryNames := queryNamesForStatefulSets[req.OrderBy.ColumnName]
topStatefulSetGroupsQueryRangeParams := &v3.QueryRangeParamsV3{
Start: req.Start,
End: req.End,
Step: step,
CompositeQuery: &v3.CompositeQuery{
BuilderQueries: map[string]*v3.BuilderQuery{},
QueryType: v3.QueryTypeBuilder,
PanelType: v3.PanelTypeTable,
},
}
for _, queryName := range queryNames {
query := q.CompositeQuery.BuilderQueries[queryName].Clone()
query.StepInterval = step
query.MetricTableHints = &v3.MetricTableHints{
TimeSeriesTableName: timeSeriesTableName,
SamplesTableName: samplesTableName,
}
if req.Filters != nil && len(req.Filters.Items) > 0 {
if query.Filters == nil {
query.Filters = &v3.FilterSet{Operator: "AND", Items: []v3.FilterItem{}}
}
query.Filters.Items = append(query.Filters.Items, req.Filters.Items...)
}
topStatefulSetGroupsQueryRangeParams.CompositeQuery.BuilderQueries[queryName] = query
}
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "inframetrics",
instrumentationtypes.CodeFunctionName: "getTopStatefulSetGroups",
})
queryResponse, _, err := d.querierV2.QueryRange(ctx, orgID, topStatefulSetGroupsQueryRangeParams)
if err != nil {
return nil, nil, err
}
formattedResponse, err := postprocess.PostProcessResult(queryResponse, topStatefulSetGroupsQueryRangeParams)
if err != nil {
return nil, nil, err
}
if len(formattedResponse) == 0 || len(formattedResponse[0].Series) == 0 {
return nil, nil, nil
}
if req.OrderBy.Order == v3.DirectionDesc {
sort.Slice(formattedResponse[0].Series, func(i, j int) bool {
return formattedResponse[0].Series[i].Points[0].Value > formattedResponse[0].Series[j].Points[0].Value
})
} else {
sort.Slice(formattedResponse[0].Series, func(i, j int) bool {
return formattedResponse[0].Series[i].Points[0].Value < formattedResponse[0].Series[j].Points[0].Value
})
}
limit := math.Min(float64(req.Offset+req.Limit), float64(len(formattedResponse[0].Series)))
paginatedTopStatefulSetGroupsSeries := formattedResponse[0].Series[req.Offset:int(limit)]
topStatefulSetGroups := []map[string]string{}
for _, series := range paginatedTopStatefulSetGroupsSeries {
topStatefulSetGroups = append(topStatefulSetGroups, series.Labels)
}
allStatefulSetGroups := []map[string]string{}
for _, series := range formattedResponse[0].Series {
allStatefulSetGroups = append(allStatefulSetGroups, series.Labels)
}
return topStatefulSetGroups, allStatefulSetGroups, nil
}
func (d *StatefulSetsRepo) GetStatefulSetList(ctx context.Context, orgID valuer.UUID, req model.StatefulSetListRequest) (model.StatefulSetListResponse, error) {
resp := model.StatefulSetListResponse{}
if req.Limit == 0 {
req.Limit = 10
}
if req.OrderBy == nil {
req.OrderBy = &v3.OrderBy{ColumnName: "cpu", Order: v3.DirectionDesc}
}
if req.GroupBy == nil {
req.GroupBy = []v3.AttributeKey{{Key: k8sStatefulSetNameAttrKey}}
resp.Type = model.ResponseTypeList
} else {
resp.Type = model.ResponseTypeGroupedList
}
step := int64(math.Max(float64(common.MinAllowedStepInterval(req.Start, req.End)), 60))
query := WorkloadTableListQuery.Clone()
query.Start = req.Start
query.End = req.End
query.Step = step
// add additional queries for stateful sets
for _, statefulSetQuery := range builderQueriesForStatefulSets {
query.CompositeQuery.BuilderQueries[statefulSetQuery.QueryName] = statefulSetQuery.Clone()
}
for _, query := range query.CompositeQuery.BuilderQueries {
query.StepInterval = step
if req.Filters != nil && len(req.Filters.Items) > 0 {
if query.Filters == nil {
query.Filters = &v3.FilterSet{Operator: "AND", Items: []v3.FilterItem{}}
}
query.Filters.Items = append(query.Filters.Items, req.Filters.Items...)
}
query.GroupBy = req.GroupBy
// make sure we only get records for daemon sets
query.Filters.Items = append(query.Filters.Items, v3.FilterItem{
Key: v3.AttributeKey{Key: k8sStatefulSetNameAttrKey},
Operator: v3.FilterOperatorExists,
})
}
statefulSetAttrs, err := d.getMetadataAttributes(ctx, req)
if err != nil {
return resp, err
}
topStatefulSetGroups, allStatefulSetGroups, err := d.getTopStatefulSetGroups(ctx, orgID, req, query)
if err != nil {
return resp, err
}
groupFilters := map[string][]string{}
for _, topStatefulSetGroup := range topStatefulSetGroups {
for k, v := range topStatefulSetGroup {
groupFilters[k] = append(groupFilters[k], v)
}
}
for groupKey, groupValues := range groupFilters {
hasGroupFilter := false
if req.Filters != nil && len(req.Filters.Items) > 0 {
for _, filter := range req.Filters.Items {
if filter.Key.Key == groupKey {
hasGroupFilter = true
break
}
}
}
if !hasGroupFilter {
for _, query := range query.CompositeQuery.BuilderQueries {
query.Filters.Items = append(query.Filters.Items, v3.FilterItem{
Key: v3.AttributeKey{Key: groupKey},
Value: groupValues,
Operator: v3.FilterOperatorIn,
})
}
}
}
ctx = ctxtypes.NewContextWithCommentVals(ctx, map[string]string{
instrumentationtypes.CodeNamespace: "inframetrics",
instrumentationtypes.CodeFunctionName: "GetStatefulSetList",
})
queryResponse, _, err := d.querierV2.QueryRange(ctx, orgID, query)
if err != nil {
return resp, err
}
formattedResponse, err := postprocess.PostProcessResult(queryResponse, query)
if err != nil {
return resp, err
}
records := []model.StatefulSetListRecord{}
for _, result := range formattedResponse {
for _, row := range result.Table.Rows {
record := model.StatefulSetListRecord{
StatefulSetName: "",
CPUUsage: -1,
CPURequest: -1,
CPULimit: -1,
MemoryUsage: -1,
MemoryRequest: -1,
MemoryLimit: -1,
DesiredPods: -1,
AvailablePods: -1,
}
if statefulSetName, ok := row.Data[k8sStatefulSetNameAttrKey].(string); ok {
record.StatefulSetName = statefulSetName
}
if cpu, ok := row.Data["A"].(float64); ok {
record.CPUUsage = cpu
}
if cpuRequest, ok := row.Data["B"].(float64); ok {
record.CPURequest = cpuRequest
}
if cpuLimit, ok := row.Data["C"].(float64); ok {
record.CPULimit = cpuLimit
}
if memory, ok := row.Data["D"].(float64); ok {
record.MemoryUsage = memory
}
if memoryRequest, ok := row.Data["E"].(float64); ok {
record.MemoryRequest = memoryRequest
}
if memoryLimit, ok := row.Data["F"].(float64); ok {
record.MemoryLimit = memoryLimit
}
if restarts, ok := row.Data["G"].(float64); ok {
record.Restarts = int(restarts)
}
if desiredPods, ok := row.Data["H"].(float64); ok {
record.DesiredPods = int(desiredPods)
}
if availablePods, ok := row.Data["I"].(float64); ok {
record.AvailablePods = int(availablePods)
}
record.Meta = map[string]string{}
if _, ok := statefulSetAttrs[record.StatefulSetName]; ok && record.StatefulSetName != "" {
record.Meta = statefulSetAttrs[record.StatefulSetName]
}
for k, v := range row.Data {
if slices.Contains(statefulSetQueryNames, k) {
continue
}
if labelValue, ok := v.(string); ok {
record.Meta[k] = labelValue
}
}
records = append(records, record)
}
}
resp.Total = len(allStatefulSetGroups)
resp.Records = records
resp.SortBy(req.OrderBy)
return resp, nil
}

View File

@@ -0,0 +1,166 @@
package inframetrics
import v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
var (
metricNamesForWorkloads = map[string]string{
"cpu": "k8s.pod.cpu.usage",
"cpu_request": "k8s.pod.cpu_request_utilization",
"cpu_limit": "k8s.pod.cpu_limit_utilization",
"memory": "k8s.pod.memory.working_set",
"memory_request": "k8s.pod.memory_request_utilization",
"memory_limit": "k8s.pod.memory_limit_utilization",
"restarts": "k8s.container.restarts",
}
)
var WorkloadTableListQuery = v3.QueryRangeParamsV3{
CompositeQuery: &v3.CompositeQuery{
BuilderQueries: map[string]*v3.BuilderQuery{
// pod cpu utilization
"A": {
QueryName: "A",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForWorkloads["cpu"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{},
Expression: "A",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
// pod cpu request utilization
"B": {
QueryName: "B",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForWorkloads["cpu_request"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{},
Expression: "B",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationAvg,
Disabled: false,
},
// pod cpu limit utilization
"C": {
QueryName: "C",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForWorkloads["cpu_limit"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{},
Expression: "C",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationAvg,
Disabled: false,
},
// pod memory utilization
"D": {
QueryName: "D",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForWorkloads["memory"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{},
Expression: "D",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
// pod memory request utilization
"E": {
QueryName: "E",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForWorkloads["memory_request"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{},
Expression: "E",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationAvg,
Disabled: false,
},
// pod memory limit utilization
"F": {
QueryName: "F",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForWorkloads["memory_limit"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{},
Expression: "F",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationAvg,
Disabled: false,
},
"G": {
QueryName: "G",
DataSource: v3.DataSourceMetrics,
AggregateAttribute: v3.AttributeKey{
Key: metricNamesForWorkloads["restarts"],
DataType: v3.AttributeKeyDataTypeFloat64,
},
Temporality: v3.Unspecified,
Filters: &v3.FilterSet{
Operator: "AND",
Items: []v3.FilterItem{},
},
GroupBy: []v3.AttributeKey{},
Expression: "G",
ReduceTo: v3.ReduceToOperatorSum,
TimeAggregation: v3.TimeAggregationAnyLast,
SpaceAggregation: v3.SpaceAggregationMax,
Functions: []v3.Function{{Name: v3.FunctionNameRunningDiff}},
Disabled: false,
},
},
PanelType: v3.PanelTypeTable,
QueryType: v3.QueryTypeBuilder,
},
Version: "v4",
FormatForWeb: true,
}

View File

@@ -166,6 +166,7 @@ func (s *Server) createPublicServer(api *APIHandler, web web.Web) (*http.Server,
api.RegisterLogsRoutes(r, am)
api.RegisterIntegrationRoutes(r, am)
api.RegisterQueryRangeV3Routes(r, am)
api.RegisterInfraMetricsRoutes(r, am)
api.RegisterWebSocketPaths(r, am)
api.RegisterQueryRangeV4Routes(r, am)
api.RegisterMessagingQueuesRoutes(r, am)

View File

@@ -95,6 +95,11 @@ type Reader interface {
ReportQueryStartForProgressTracking(queryId string) (reportQueryFinished func(), apiErr *model.ApiError)
SubscribeToQueryProgress(queryId string) (<-chan model.QueryProgress, func(), *model.ApiError)
GetCountOfThings(ctx context.Context, query string) (uint64, error)
GetActiveHostsFromMetricMetadata(ctx context.Context, metricNames []string, hostNameAttr string, sinceUnixMilli int64) (map[string]bool, error)
GetMetricsExistenceAndEarliestTime(ctx context.Context, metricNames []string) (uint64, uint64, error)
//trace
GetTraceFields(ctx context.Context) (*model.GetFieldsResponse, *model.ApiError)
UpdateTraceField(ctx context.Context, field *model.UpdateField) *model.ApiError

View File

@@ -0,0 +1,757 @@
package model
import (
"sort"
v3 "github.com/SigNoz/signoz/pkg/query-service/model/v3"
)
type (
ResponseType string
)
const (
ResponseTypeList ResponseType = "list"
ResponseTypeGroupedList ResponseType = "grouped_list"
)
type HostListRequest struct {
Start int64 `json:"start"` // epoch time in ms
End int64 `json:"end"` // epoch time in ms
Filters *v3.FilterSet `json:"filters"`
GroupBy []v3.AttributeKey `json:"groupBy"`
OrderBy *v3.OrderBy `json:"orderBy"`
Offset int `json:"offset"`
Limit int `json:"limit"`
}
type HostListRecord struct {
HostName string `json:"hostName"`
Active bool `json:"active"`
OS string `json:"os"`
CPU float64 `json:"cpu"`
Memory float64 `json:"memory"`
Wait float64 `json:"wait"`
Load15 float64 `json:"load15"`
Meta map[string]string `json:"meta"`
}
type HostListResponse struct {
Type ResponseType `json:"type"`
Records []HostListRecord `json:"records"`
Total int `json:"total"`
SentAnyHostMetricsData bool `json:"sentAnyHostMetricsData"`
IsSendingK8SAgentMetrics bool `json:"isSendingK8SAgentMetrics"`
ClusterNames []string `json:"clusterNames"`
NodeNames []string `json:"nodeNames"`
EndTimeBeforeRetention bool `json:"endTimeBeforeRetention"`
}
func (r *HostListResponse) SortBy(orderBy *v3.OrderBy) {
switch orderBy.ColumnName {
case "cpu":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].CPU > r.Records[j].CPU
})
case "memory":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].Memory > r.Records[j].Memory
})
case "load15":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].Load15 > r.Records[j].Load15
})
case "wait":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].Wait > r.Records[j].Wait
})
}
// the default is descending
if orderBy.Order == v3.DirectionAsc {
// reverse the list
for i, j := 0, len(r.Records)-1; i < j; i, j = i+1, j-1 {
r.Records[i], r.Records[j] = r.Records[j], r.Records[i]
}
}
}
type ProcessListRequest struct {
Start int64 `json:"start"` // epoch time in ms
End int64 `json:"end"` // epoch time in ms
Filters *v3.FilterSet `json:"filters"`
GroupBy []v3.AttributeKey `json:"groupBy"`
OrderBy *v3.OrderBy `json:"orderBy"`
Offset int `json:"offset"`
Limit int `json:"limit"`
}
type ProcessListResponse struct {
Type ResponseType `json:"type"`
Records []ProcessListRecord `json:"records"`
Total int `json:"total"`
}
type ProcessListRecord struct {
ProcessName string `json:"processName"`
ProcessID string `json:"processID"`
ProcessCMD string `json:"processCMD"`
ProcessCMDLine string `json:"processCMDLine"`
ProcessCPU float64 `json:"processCPU"`
ProcessMemory float64 `json:"processMemory"`
Meta map[string]string `json:"meta"`
}
type PodListRequest struct {
Start int64 `json:"start"` // epoch time in ms
End int64 `json:"end"` // epoch time in ms
Filters *v3.FilterSet `json:"filters"`
GroupBy []v3.AttributeKey `json:"groupBy"`
OrderBy *v3.OrderBy `json:"orderBy"`
Offset int `json:"offset"`
Limit int `json:"limit"`
}
type PodListResponse struct {
Type ResponseType `json:"type"`
Records []PodListRecord `json:"records"`
Total int `json:"total"`
}
func (r *PodListResponse) SortBy(orderBy *v3.OrderBy) {
switch orderBy.ColumnName {
case "cpu":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].PodCPU > r.Records[j].PodCPU
})
case "cpu_request":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].PodCPURequest > r.Records[j].PodCPURequest
})
case "cpu_limit":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].PodCPULimit > r.Records[j].PodCPULimit
})
case "memory":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].PodMemory > r.Records[j].PodMemory
})
case "memory_request":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].PodMemoryRequest > r.Records[j].PodMemoryRequest
})
case "memory_limit":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].PodMemoryLimit > r.Records[j].PodMemoryLimit
})
case "restarts":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].RestartCount > r.Records[j].RestartCount
})
}
// the default is descending
if orderBy.Order == v3.DirectionAsc {
// reverse the list
for i, j := 0, len(r.Records)-1; i < j; i, j = i+1, j-1 {
r.Records[i], r.Records[j] = r.Records[j], r.Records[i]
}
}
}
type PodListRecord struct {
PodUID string `json:"podUID,omitempty"`
PodCPU float64 `json:"podCPU"`
PodCPURequest float64 `json:"podCPURequest"`
PodCPULimit float64 `json:"podCPULimit"`
PodMemory float64 `json:"podMemory"`
PodMemoryRequest float64 `json:"podMemoryRequest"`
PodMemoryLimit float64 `json:"podMemoryLimit"`
RestartCount int `json:"restartCount"`
Meta map[string]string `json:"meta"`
CountByPhase PodCountByPhase `json:"countByPhase"`
}
type PodCountByPhase struct {
Pending int `json:"pending"`
Running int `json:"running"`
Succeeded int `json:"succeeded"`
Failed int `json:"failed"`
Unknown int `json:"unknown"`
}
type NodeListRequest struct {
Start int64 `json:"start"` // epoch time in ms
End int64 `json:"end"` // epoch time in ms
Filters *v3.FilterSet `json:"filters"`
GroupBy []v3.AttributeKey `json:"groupBy"`
OrderBy *v3.OrderBy `json:"orderBy"`
Offset int `json:"offset"`
Limit int `json:"limit"`
}
type NodeListResponse struct {
Type ResponseType `json:"type"`
Records []NodeListRecord `json:"records"`
Total int `json:"total"`
}
func (r *NodeListResponse) SortBy(orderBy *v3.OrderBy) {
switch orderBy.ColumnName {
case "cpu":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].NodeCPUUsage > r.Records[j].NodeCPUUsage
})
case "cpu_allocatable":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].NodeCPUAllocatable > r.Records[j].NodeCPUAllocatable
})
case "memory":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].NodeMemoryUsage > r.Records[j].NodeMemoryUsage
})
case "memory_allocatable":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].NodeMemoryAllocatable > r.Records[j].NodeMemoryAllocatable
})
}
// the default is descending
if orderBy.Order == v3.DirectionAsc {
// reverse the list
for i, j := 0, len(r.Records)-1; i < j; i, j = i+1, j-1 {
r.Records[i], r.Records[j] = r.Records[j], r.Records[i]
}
}
}
type NodeCountByCondition struct {
Ready int `json:"ready"`
NotReady int `json:"notReady"`
Unknown int `json:"unknown"`
}
type NodeListRecord struct {
NodeUID string `json:"nodeUID,omitempty"`
NodeCPUUsage float64 `json:"nodeCPUUsage"`
NodeCPUAllocatable float64 `json:"nodeCPUAllocatable"`
NodeMemoryUsage float64 `json:"nodeMemoryUsage"`
NodeMemoryAllocatable float64 `json:"nodeMemoryAllocatable"`
CountByCondition NodeCountByCondition `json:"countByCondition"`
Meta map[string]string `json:"meta"`
}
type NamespaceListRequest struct {
Start int64 `json:"start"` // epoch time in ms
End int64 `json:"end"` // epoch time in ms
Filters *v3.FilterSet `json:"filters"`
GroupBy []v3.AttributeKey `json:"groupBy"`
OrderBy *v3.OrderBy `json:"orderBy"`
Offset int `json:"offset"`
Limit int `json:"limit"`
}
type NamespaceListResponse struct {
Type ResponseType `json:"type"`
Records []NamespaceListRecord `json:"records"`
Total int `json:"total"`
}
func (r *NamespaceListResponse) SortBy(orderBy *v3.OrderBy) {
switch orderBy.ColumnName {
case "cpu":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].CPUUsage > r.Records[j].CPUUsage
})
case "memory":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].MemoryUsage > r.Records[j].MemoryUsage
})
case "pod_phase":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].CountByPhase.Pending > r.Records[j].CountByPhase.Pending
})
}
// the default is descending
if orderBy.Order == v3.DirectionAsc {
// reverse the list
for i, j := 0, len(r.Records)-1; i < j; i, j = i+1, j-1 {
r.Records[i], r.Records[j] = r.Records[j], r.Records[i]
}
}
}
type NamespaceListRecord struct {
NamespaceName string `json:"namespaceName"`
CPUUsage float64 `json:"cpuUsage"`
MemoryUsage float64 `json:"memoryUsage"`
CountByPhase PodCountByPhase `json:"countByPhase"`
Meta map[string]string `json:"meta"`
}
type ClusterListRequest struct {
Start int64 `json:"start"` // epoch time in ms
End int64 `json:"end"` // epoch time in ms
Filters *v3.FilterSet `json:"filters"`
GroupBy []v3.AttributeKey `json:"groupBy"`
OrderBy *v3.OrderBy `json:"orderBy"`
Offset int `json:"offset"`
Limit int `json:"limit"`
}
type ClusterListResponse struct {
Type ResponseType `json:"type"`
Records []ClusterListRecord `json:"records"`
Total int `json:"total"`
}
func (r *ClusterListResponse) SortBy(orderBy *v3.OrderBy) {
switch orderBy.ColumnName {
case "cpu":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].CPUUsage > r.Records[j].CPUUsage
})
case "cpu_allocatable":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].CPUAllocatable > r.Records[j].CPUAllocatable
})
case "memory":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].MemoryUsage > r.Records[j].MemoryUsage
})
case "memory_allocatable":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].MemoryAllocatable > r.Records[j].MemoryAllocatable
})
}
// the default is descending
if orderBy.Order == v3.DirectionAsc {
// reverse the list
for i, j := 0, len(r.Records)-1; i < j; i, j = i+1, j-1 {
r.Records[i], r.Records[j] = r.Records[j], r.Records[i]
}
}
}
type ClusterListRecord struct {
ClusterUID string `json:"clusterUID"`
CPUUsage float64 `json:"cpuUsage"`
CPUAllocatable float64 `json:"cpuAllocatable"`
MemoryUsage float64 `json:"memoryUsage"`
MemoryAllocatable float64 `json:"memoryAllocatable"`
Meta map[string]string `json:"meta"`
}
type DeploymentListRequest struct {
Start int64 `json:"start"` // epoch time in ms
End int64 `json:"end"` // epoch time in ms
Filters *v3.FilterSet `json:"filters"`
GroupBy []v3.AttributeKey `json:"groupBy"`
OrderBy *v3.OrderBy `json:"orderBy"`
Offset int `json:"offset"`
Limit int `json:"limit"`
}
type DeploymentListResponse struct {
Type ResponseType `json:"type"`
Records []DeploymentListRecord `json:"records"`
Total int `json:"total"`
}
func (r *DeploymentListResponse) SortBy(orderBy *v3.OrderBy) {
switch orderBy.ColumnName {
case "cpu":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].CPUUsage > r.Records[j].CPUUsage
})
case "cpu_request":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].CPURequest > r.Records[j].CPURequest
})
case "cpu_limit":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].CPULimit > r.Records[j].CPULimit
})
case "memory":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].MemoryUsage > r.Records[j].MemoryUsage
})
case "memory_request":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].MemoryRequest > r.Records[j].MemoryRequest
})
case "memory_limit":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].MemoryLimit > r.Records[j].MemoryLimit
})
case "desired_pods":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].DesiredPods > r.Records[j].DesiredPods
})
case "available_pods":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].AvailablePods > r.Records[j].AvailablePods
})
case "restarts":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].Restarts > r.Records[j].Restarts
})
}
// the default is descending
if orderBy.Order == v3.DirectionAsc {
// reverse the list
for i, j := 0, len(r.Records)-1; i < j; i, j = i+1, j-1 {
r.Records[i], r.Records[j] = r.Records[j], r.Records[i]
}
}
}
type DeploymentListRecord struct {
DeploymentName string `json:"deploymentName"`
CPUUsage float64 `json:"cpuUsage"`
MemoryUsage float64 `json:"memoryUsage"`
DesiredPods int `json:"desiredPods"`
AvailablePods int `json:"availablePods"`
CPURequest float64 `json:"cpuRequest"`
MemoryRequest float64 `json:"memoryRequest"`
CPULimit float64 `json:"cpuLimit"`
MemoryLimit float64 `json:"memoryLimit"`
Restarts int `json:"restarts"`
Meta map[string]string `json:"meta"`
}
type DaemonSetListRequest struct {
Start int64 `json:"start"` // epoch time in ms
End int64 `json:"end"` // epoch time in ms
Filters *v3.FilterSet `json:"filters"`
GroupBy []v3.AttributeKey `json:"groupBy"`
OrderBy *v3.OrderBy `json:"orderBy"`
Offset int `json:"offset"`
Limit int `json:"limit"`
}
type DaemonSetListResponse struct {
Type ResponseType `json:"type"`
Records []DaemonSetListRecord `json:"records"`
Total int `json:"total"`
}
func (r *DaemonSetListResponse) SortBy(orderBy *v3.OrderBy) {
switch orderBy.ColumnName {
case "cpu":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].CPUUsage > r.Records[j].CPUUsage
})
case "cpu_request":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].CPURequest > r.Records[j].CPURequest
})
case "cpu_limit":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].CPULimit > r.Records[j].CPULimit
})
case "memory":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].MemoryUsage > r.Records[j].MemoryUsage
})
case "memory_request":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].MemoryRequest > r.Records[j].MemoryRequest
})
case "memory_limit":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].MemoryLimit > r.Records[j].MemoryLimit
})
case "restarts":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].Restarts > r.Records[j].Restarts
})
case "desired_nodes":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].DesiredNodes > r.Records[j].DesiredNodes
})
case "available_nodes":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].AvailableNodes > r.Records[j].AvailableNodes
})
}
// the default is descending
if orderBy.Order == v3.DirectionAsc {
// reverse the list
for i, j := 0, len(r.Records)-1; i < j; i, j = i+1, j-1 {
r.Records[i], r.Records[j] = r.Records[j], r.Records[i]
}
}
}
type DaemonSetListRecord struct {
DaemonSetName string `json:"daemonSetName"`
CPUUsage float64 `json:"cpuUsage"`
MemoryUsage float64 `json:"memoryUsage"`
CPURequest float64 `json:"cpuRequest"`
MemoryRequest float64 `json:"memoryRequest"`
CPULimit float64 `json:"cpuLimit"`
MemoryLimit float64 `json:"memoryLimit"`
Restarts int `json:"restarts"`
DesiredNodes int `json:"desiredNodes"`
AvailableNodes int `json:"availableNodes"`
Meta map[string]string `json:"meta"`
}
type StatefulSetListRequest struct {
Start int64 `json:"start"` // epoch time in ms
End int64 `json:"end"` // epoch time in ms
Filters *v3.FilterSet `json:"filters"`
GroupBy []v3.AttributeKey `json:"groupBy"`
OrderBy *v3.OrderBy `json:"orderBy"`
Offset int `json:"offset"`
Limit int `json:"limit"`
}
type StatefulSetListResponse struct {
Type ResponseType `json:"type"`
Records []StatefulSetListRecord `json:"records"`
Total int `json:"total"`
}
func (r *StatefulSetListResponse) SortBy(orderBy *v3.OrderBy) {
switch orderBy.ColumnName {
case "cpu":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].CPUUsage > r.Records[j].CPUUsage
})
case "cpu_request":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].CPURequest > r.Records[j].CPURequest
})
case "cpu_limit":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].CPULimit > r.Records[j].CPULimit
})
case "memory":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].MemoryUsage > r.Records[j].MemoryUsage
})
case "memory_request":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].MemoryRequest > r.Records[j].MemoryRequest
})
case "memory_limit":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].MemoryLimit > r.Records[j].MemoryLimit
})
case "restarts":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].Restarts > r.Records[j].Restarts
})
case "desired_pods":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].DesiredPods > r.Records[j].DesiredPods
})
case "available_pods":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].AvailablePods > r.Records[j].AvailablePods
})
}
// the default is descending
if orderBy.Order == v3.DirectionAsc {
// reverse the list
for i, j := 0, len(r.Records)-1; i < j; i, j = i+1, j-1 {
r.Records[i], r.Records[j] = r.Records[j], r.Records[i]
}
}
}
type StatefulSetListRecord struct {
StatefulSetName string `json:"statefulSetName"`
CPUUsage float64 `json:"cpuUsage"`
MemoryUsage float64 `json:"memoryUsage"`
CPURequest float64 `json:"cpuRequest"`
MemoryRequest float64 `json:"memoryRequest"`
CPULimit float64 `json:"cpuLimit"`
MemoryLimit float64 `json:"memoryLimit"`
Restarts int `json:"restarts"`
DesiredPods int `json:"desiredPods"`
AvailablePods int `json:"availablePods"`
Meta map[string]string `json:"meta"`
}
type JobListRequest struct {
Start int64 `json:"start"` // epoch time in ms
End int64 `json:"end"` // epoch time in ms
Filters *v3.FilterSet `json:"filters"`
GroupBy []v3.AttributeKey `json:"groupBy"`
OrderBy *v3.OrderBy `json:"orderBy"`
Offset int `json:"offset"`
Limit int `json:"limit"`
}
type JobListResponse struct {
Type ResponseType `json:"type"`
Records []JobListRecord `json:"records"`
Total int `json:"total"`
}
func (r *JobListResponse) SortBy(orderBy *v3.OrderBy) {
switch orderBy.ColumnName {
case "cpu":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].CPUUsage > r.Records[j].CPUUsage
})
case "cpu_request":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].CPURequest > r.Records[j].CPURequest
})
case "cpu_limit":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].CPULimit > r.Records[j].CPULimit
})
case "memory":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].MemoryUsage > r.Records[j].MemoryUsage
})
case "memory_request":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].MemoryRequest > r.Records[j].MemoryRequest
})
case "memory_limit":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].MemoryLimit > r.Records[j].MemoryLimit
})
case "restarts":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].Restarts > r.Records[j].Restarts
})
case "desired_successful_pods":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].DesiredSuccessfulPods > r.Records[j].DesiredSuccessfulPods
})
case "active_pods":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].ActivePods > r.Records[j].ActivePods
})
case "failed_pods":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].FailedPods > r.Records[j].FailedPods
})
case "successful_pods":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].SuccessfulPods > r.Records[j].SuccessfulPods
})
}
// the default is descending
if orderBy.Order == v3.DirectionAsc {
// reverse the list
for i, j := 0, len(r.Records)-1; i < j; i, j = i+1, j-1 {
r.Records[i], r.Records[j] = r.Records[j], r.Records[i]
}
}
}
type JobListRecord struct {
JobName string `json:"jobName"`
CPUUsage float64 `json:"cpuUsage"`
MemoryUsage float64 `json:"memoryUsage"`
CPURequest float64 `json:"cpuRequest"`
MemoryRequest float64 `json:"memoryRequest"`
CPULimit float64 `json:"cpuLimit"`
MemoryLimit float64 `json:"memoryLimit"`
Restarts int `json:"restarts"`
DesiredSuccessfulPods int `json:"desiredSuccessfulPods"`
ActivePods int `json:"activePods"`
FailedPods int `json:"failedPods"`
SuccessfulPods int `json:"successfulPods"`
Meta map[string]string `json:"meta"`
}
type VolumeListRequest struct {
Start int64 `json:"start"` // epoch time in ms
End int64 `json:"end"` // epoch time in ms
Filters *v3.FilterSet `json:"filters"`
GroupBy []v3.AttributeKey `json:"groupBy"`
OrderBy *v3.OrderBy `json:"orderBy"`
Offset int `json:"offset"`
Limit int `json:"limit"`
}
type VolumeListResponse struct {
Type ResponseType `json:"type"`
Records []VolumeListRecord `json:"records"`
Total int `json:"total"`
}
func (r *VolumeListResponse) SortBy(orderBy *v3.OrderBy) {
switch orderBy.ColumnName {
case "available":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].VolumeAvailable > r.Records[j].VolumeAvailable
})
case "capacity":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].VolumeCapacity > r.Records[j].VolumeCapacity
})
case "usage":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].VolumeUsage > r.Records[j].VolumeUsage
})
case "inodes":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].VolumeInodes > r.Records[j].VolumeInodes
})
case "inodes_free":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].VolumeInodesFree > r.Records[j].VolumeInodesFree
})
case "inodes_used":
sort.Slice(r.Records, func(i, j int) bool {
return r.Records[i].VolumeInodesUsed > r.Records[j].VolumeInodesUsed
})
}
// the default is descending
if orderBy.Order == v3.DirectionAsc {
// reverse the list
for i, j := 0, len(r.Records)-1; i < j; i, j = i+1, j-1 {
r.Records[i], r.Records[j] = r.Records[j], r.Records[i]
}
}
}
type VolumeListRecord struct {
PersistentVolumeClaimName string `json:"persistentVolumeClaimName"`
VolumeAvailable float64 `json:"volumeAvailable"`
VolumeCapacity float64 `json:"volumeCapacity"`
VolumeInodes float64 `json:"volumeInodes"`
VolumeInodesFree float64 `json:"volumeInodesFree"`
VolumeInodesUsed float64 `json:"volumeInodesUsed"`
VolumeUsage float64 `json:"volumeUsage"`
Meta map[string]string `json:"meta"`
}
type PodOnboardingStatus struct {
ClusterName string `json:"clusterName"`
NodeName string `json:"nodeName"`
NamespaceName string `json:"namespaceName"`
PodName string `json:"podName"`
HasClusterName bool `json:"hasClusterName"`
HasNodeName bool `json:"hasNodeName"`
HasNamespaceName bool `json:"hasNamespaceName"`
HasDeploymentName bool `json:"hasDeploymentName"`
HasStatefulsetName bool `json:"hasStatefulsetName"`
HasDaemonsetName bool `json:"hasDaemonsetName"`
HasCronjobName bool `json:"hasCronjobName"`
HasJobName bool `json:"hasJobName"`
}
type OnboardingStatus struct {
DidSendPodMetrics bool `json:"didSendPodMetrics"`
DidSendNodeMetrics bool `json:"didSendNodeMetrics"`
DidSendClusterMetrics bool `json:"didSendClusterMetrics"`
IsSendingOptionalPodMetrics bool `json:"isSendingOptionalPodMetrics"`
IsSendingRequiredMetadata []PodOnboardingStatus `json:"isSendingRequiredMetadata"`
}

View File

@@ -48,6 +48,8 @@ import (
"github.com/SigNoz/signoz/pkg/modules/tracedetail/impltracedetail"
"github.com/SigNoz/signoz/pkg/modules/tracefunnel"
"github.com/SigNoz/signoz/pkg/modules/tracefunnel/impltracefunnel"
"github.com/SigNoz/signoz/pkg/prometheus"
"github.com/SigNoz/signoz/pkg/prometheus/promapi"
"github.com/SigNoz/signoz/pkg/querier"
"github.com/SigNoz/signoz/pkg/ruler"
"github.com/SigNoz/signoz/pkg/ruler/signozruler"
@@ -81,6 +83,7 @@ type Handlers struct {
RuleStateHistory rulestatehistory.Handler
SpanMapperHandler spanmapper.Handler
AlertmanagerHandler alertmanager.Handler
PrometheusHandler prometheus.Handler
TraceDetail tracedetail.Handler
RulerHandler ruler.Handler
LLMPricingRuleHandler llmpricingrule.Handler
@@ -101,6 +104,7 @@ func NewHandlers(
zeusService zeus.Zeus,
registryHandler factory.Handler,
alertmanagerService alertmanager.Alertmanager,
prometheusService prometheus.Prometheus,
rulerService ruler.Ruler,
statsAggregator statsreporter.Aggregator,
) Handlers {
@@ -129,6 +133,7 @@ func NewHandlers(
CloudIntegrationHandler: implcloudintegration.NewHandler(modules.CloudIntegration),
SpanMapperHandler: implspanmapper.NewHandler(modules.SpanMapper),
AlertmanagerHandler: signozalertmanager.NewHandler(alertmanagerService),
PrometheusHandler: promapi.NewHandler(providerSettings.Logger, prometheusService),
TraceDetail: impltracedetail.NewHandler(modules.TraceDetail),
RulerHandler: signozruler.NewHandler(rulerService),
LLMPricingRuleHandler: impllmpricingrule.NewHandler(modules.LLMPricingRule),

View File

@@ -63,7 +63,7 @@ func TestNewHandlers(t *testing.T) {
querierHandler := querier.NewHandler(providerSettings, nil, nil)
registryHandler := factory.NewHandler(nil)
handlers := NewHandlers(modules, providerSettings, nil, querierHandler, nil, nil, nil, nil, nil, nil, nil, registryHandler, alertmanager, nil, nil)
handlers := NewHandlers(modules, providerSettings, nil, querierHandler, nil, nil, nil, nil, nil, nil, nil, registryHandler, alertmanager, nil, nil, nil)
reflectVal := reflect.ValueOf(handlers)
for i := 0; i < reflectVal.NumField(); i++ {
f := reflectVal.Field(i)

View File

@@ -617,7 +617,7 @@ func New(
// Initialize all handlers for the modules
registryHandler := factory.NewHandler(registry)
handlers := NewHandlers(modules, providerSettings, analytics, querierHandler, licensing, global, flagger, gateway, telemetryMetadataStore, authz, zeus, registryHandler, alertmanager, rulerInstance, statsAggregator)
handlers := NewHandlers(modules, providerSettings, analytics, querierHandler, licensing, global, flagger, gateway, telemetryMetadataStore, authz, zeus, registryHandler, alertmanager, prometheus, rulerInstance, statsAggregator)
// Initialize the API server (after registry so it can access service health)
apiserverInstance, err := factory.NewProviderFromNamedMap(

66
tests/fixtures/promqltestcorpus.py vendored Normal file
View File

@@ -0,0 +1,66 @@
import json
import math
import os
from collections.abc import Callable
from datetime import UTC, datetime, timedelta
from fixtures.metrics import Metrics
TESTDATA_DIR = os.path.join(os.path.dirname(__file__), "..", "integration", "testdata", "promqltestcorpus")
CORPUS_FILE = os.path.join(TESTDATA_DIR, "corpus.json")
# Datasets sit on disjoint time windows (2h gaps, far beyond the 5m lookback)
# so one bulk ingest serves every case without cross-talk.
ISOLATION_GAP_MS = 2 * 3600 * 1000
SPECIALS = {"NaN": math.nan, "Inf": math.inf, "-Inf": -math.inf}
def ingest_promqltest_corpus(insert_metrics: Callable[[list[Metrics]], None]) -> tuple[dict, dict[int, int]]:
"""Loads the frozen corpus, lays its datasets end to end on the timeline
(newest last, ending safely in the past), ingests every sample, and
returns (corpus, dataset base timestamps).
Dataset bases are hour-aligned: registration rows are hour-bucketed, so
behavior depends on where samples fall relative to hour boundaries, and
exact known-divergences enforcement needs identical placement every run."""
with open(CORPUS_FILE, encoding="utf-8") as f:
corpus = json.load(f)
cases_by_dataset: dict[int, list[dict]] = {}
for case in corpus["cases"]:
cases_by_dataset.setdefault(case["dataset"], []).append(case)
spans = {}
for ds in corpus["datasets"]:
sample_max = max((s["samples"][-1][0] for s in ds["series"] if s["samples"]), default=0)
case_max = max((c["end_ms"] for c in cases_by_dataset.get(ds["id"], [])), default=0)
spans[ds["id"]] = max(sample_max, case_max) + corpus["meta"]["lookback_ms"]
hour_ms = 3_600_000
advances = {ds["id"]: -(-(spans[ds["id"]] + ISOLATION_GAP_MS) // hour_ms) * hour_ms for ds in corpus["datasets"]}
total = sum(advances.values())
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
cursor = (int((now - timedelta(hours=1)).timestamp() * 1000) - total) // hour_ms * hour_ms
bases: dict[int, int] = {}
metrics: list[Metrics] = []
for ds in corpus["datasets"]:
bases[ds["id"]] = cursor
for series in ds["series"]:
labels = dict(series["labels"])
metric_name = labels.pop("__name__")
for off_ms, raw in series["samples"]:
stale = raw == "stale"
metrics.append(
Metrics(
metric_name=metric_name,
labels=labels,
timestamp=datetime.fromtimestamp((cursor + off_ms) / 1000, tz=UTC),
value=0.0 if stale else (SPECIALS[raw] if isinstance(raw, str) else float(raw)),
flags=1 if stale else 0,
)
)
cursor += advances[ds["id"]]
insert_metrics(metrics)
return corpus, bases

View File

@@ -0,0 +1,138 @@
import json
import math
from collections.abc import Callable
from http import HTTPStatus
import requests
from fixtures import types
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
from fixtures.metrics import Metrics
from fixtures.promqltestcorpus import ingest_promqltest_corpus
# The same frozen corpus the promqlconformance package replays through
# /api/v5/query_range, here replayed against the /prometheus/api/v1 endpoints
# with clickhousev2 as the serving provider (see conftest.py) — the two paths
# nothing else exercises. Range cases go to query_range, where a
# RangeExecutor provider serves transpiled statements when the shape allows.
# Instant cases go to /query with a real `time` parameter, so they need no
# grid encoding.
#
# Prometheus API sample values are strings, "NaN"/"+Inf"/"-Inf" included.
SPECIALS = {"NaN": math.nan, "Inf": math.inf, "+Inf": math.inf, "-Inf": -math.inf}
QUERY_TIMEOUT = 30
def test_prometheus_api_corpus(
signoz: types.SigNoz,
create_user_admin: None, # pylint: disable=unused-argument
get_token: Callable[[str, str], str],
insert_metrics: Callable[[list[Metrics]], None],
) -> None:
corpus, bases = ingest_promqltest_corpus(insert_metrics)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
failures: list[str] = []
for case in corpus["cases"]:
# instant-coarse variants encode an instant eval as a coarse-step
# range because the v5 API cannot run true instants. This API can:
# the [base] form of the same eval goes through /query below, and the
# transpiled coarse-step serving the encoding exercises is covered
# (and its known divergences ledgered) by promqlconformance's
# clickhousev2 leg.
if case["variant"] == "instant-coarse":
continue
base = bases[case["dataset"]]
start_ms = base + case["start_ms"]
end_ms = base + case["end_ms"]
step_s = max(1, case["step_ms"] // 1000)
case_id = f"{case['source']}[{case['variant']}]"
if case["instant"]:
path, params = "/prometheus/api/v1/query", {"query": case["expr"], "time": end_ms / 1000}
else:
path, params = (
"/prometheus/api/v1/query_range",
{
"query": case["expr"],
"start": start_ms / 1000,
"end": end_ms / 1000,
"step": step_s,
},
)
response = requests.get(
signoz.self.host_configs["8080"].get(path),
params=params,
timeout=QUERY_TIMEOUT,
headers={"authorization": f"Bearer {token}"},
)
if response.status_code != HTTPStatus.OK:
failures.append(f"{case_id}: HTTP {response.status_code} for {case['expr']!r}: {response.text[:200]}")
continue
body = response.json()
if body.get("status") != "success":
failures.append(f"{case_id}: status {body.get('status')!r} for {case['expr']!r}: {json.dumps(body)[:200]}")
continue
result_type, result = body["data"]["resultType"], body["data"]["result"]
actual: dict[tuple, dict[int, float]] = {}
if result_type == "matrix":
for series in result:
points = {round(float(ts) * 1000): SPECIALS[v] if v in SPECIALS else float(v) for ts, v in series.get("values") or []}
actual[tuple(sorted((series.get("metric") or {}).items()))] = points
elif result_type == "vector":
for series in result:
ts, v = series["value"]
actual[tuple(sorted((series.get("metric") or {}).items()))] = {round(float(ts) * 1000): SPECIALS[v] if v in SPECIALS else float(v)}
elif result_type == "scalar":
ts, v = result
actual[()] = {round(float(ts) * 1000): SPECIALS[v] if v in SPECIALS else float(v)}
expected: dict[tuple, dict[int, float]] = {}
for res in case["expected"]:
points = {base + off_ms: SPECIALS[v] if isinstance(v, str) else float(v) for off_ms, v in res["points"]}
expected[tuple(sorted(res["labels"].items()))] = points
if set(actual) != set(expected):
missing = set(expected) - set(actual)
extra = set(actual) - set(expected)
failures.append(f"{case_id}: series mismatch for {case['expr']!r} (missing={sorted(missing)[:3]} extra={sorted(extra)[:3]})")
continue
mismatch = None
for lset, exp_points in expected.items():
act_points = actual[lset]
if set(act_points) != set(exp_points):
mismatch = f"{case_id}: timestamp mismatch for {case['expr']!r} series {dict(lset)} (expected {len(exp_points)} points, got {len(act_points)})"
break
for ts, exp_v in exp_points.items():
act_v = act_points[ts]
if math.isnan(act_v) or math.isnan(exp_v):
close = math.isnan(act_v) and math.isnan(exp_v)
elif math.isinf(act_v) or math.isinf(exp_v):
close = act_v == exp_v
elif act_v == exp_v:
close = True
else:
# Expected values carry the v5 API's rounding (>=1: three
# decimal places; <1: three significant digits); this API
# returns raw floats. One rounding quantum covers the
# largest possible rounding difference.
scale = max(abs(act_v), abs(exp_v))
if scale >= 1:
quantum = max(1e-3, scale * 1e-9)
else:
quantum = 10 ** (math.floor(math.log10(scale)) - 2)
close = abs(act_v - exp_v) <= quantum + 1e-12
if not close:
mismatch = f"{case_id}: value mismatch for {case['expr']!r} series {dict(lset)} at {ts}: expected {exp_v}, got {act_v}"
break
if mismatch:
break
if mismatch:
failures.append(mismatch)
for f_line in failures:
print("DIVERGED", f_line)
assert not failures, f"{len(failures)} corpus cases diverged:\n" + "\n".join(failures[:25])

View File

@@ -0,0 +1,37 @@
import pytest
from testcontainers.core.container import Network
from fixtures import types
from fixtures.signoz import create_signoz
@pytest.fixture(name="signoz", scope="package")
def signoz_promapi_v2(
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:
"""
SigNoz with clickhousev2 as the serving prometheus provider. The corpus
replays against the /prometheus/api/v1 endpoints, so this package covers
the two paths nothing else serves: v2 as the provider (range queries
transpile when the shape allows), and the Prometheus HTTP API contract.
"""
return create_signoz(
network=network,
zeus=zeus,
gateway=gateway,
sqlstore=sqlstore,
clickhouse=clickhouse,
request=request,
pytestconfig=pytestconfig,
cache_key="signoz-promapi-v2",
env_overrides={
"SIGNOZ_PROMETHEUS_PROVIDER": "clickhousev2",
},
)

View File

@@ -2,21 +2,21 @@ import json
import math
import os
from collections.abc import Callable
from datetime import UTC, datetime, timedelta
from http import HTTPStatus
from fixtures import types
from fixtures.auth import USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD
from fixtures.metrics import Metrics
from fixtures.promqltestcorpus import ingest_promqltest_corpus
from fixtures.querier import get_all_series, make_query_request
TESTDATA_DIR = os.path.join(os.path.dirname(__file__), "..", "..", "testdata")
# Frozen corpus extracted from Prometheus' own promql/promqltest testdata by
# scripts/promqltestcorpus (upstream load scripts + the vendored reference engine).
# Unlike live-vs-live parity suites, the oracle is this committed file, so the suite
# keeps working when the serving path itself is the thing being changed — the one
# situation where comparing two live paths against each other is blind.
CORPUS_FILE = os.path.join(TESTDATA_DIR, "promqltestcorpus", "corpus.json")
# The corpus (see fixtures/promqltestcorpus.py) is frozen from Prometheus' own
# promql/promqltest testdata by scripts/promqltestcorpus (upstream load scripts
# + the vendored reference engine). Unlike live-vs-live parity suites, the
# oracle is a committed file, so the suite keeps working when the serving path
# itself is the thing being changed — the one situation where comparing two
# live paths against each other is blind.
# One ledger per leg, enforced exactly in both directions. The default leg's
# ledger is empty and pinned there; the clickhousev2 ledger is the rollout
@@ -40,9 +40,6 @@ LEGS: list[tuple[str, dict | None]] = [
("clickhousev2", {"X-SigNoz-PromQL-Provider": "clickhousev2"}),
]
# Datasets sit on disjoint time windows (2h gaps, far beyond the 5m lookback) so
# one bulk ingest serves every case without cross-talk.
ISOLATION_GAP_MS = 2 * 3600 * 1000
SPECIALS = {"NaN": math.nan, "Inf": math.inf, "-Inf": -math.inf}
@@ -52,51 +49,7 @@ def test_upstream_promqltest_corpus(
get_token: Callable[[str, str], str],
insert_metrics: Callable[[list[Metrics]], None],
) -> None:
with open(CORPUS_FILE, encoding="utf-8") as f:
corpus = json.load(f)
cases_by_dataset: dict[int, list[dict]] = {}
for case in corpus["cases"]:
cases_by_dataset.setdefault(case["dataset"], []).append(case)
# Lay datasets end to end on the timeline, newest last, ending safely in
# the past; spans are per-dataset so the whole corpus stays within days.
spans = {}
for ds in corpus["datasets"]:
sample_max = max((s["samples"][-1][0] for s in ds["series"] if s["samples"]), default=0)
case_max = max((c["end_ms"] for c in cases_by_dataset.get(ds["id"], [])), default=0)
spans[ds["id"]] = max(sample_max, case_max) + corpus["meta"]["lookback_ms"]
# Hour-aligned dataset bases: registration rows are hour-bucketed, so
# behavior depends on where samples fall relative to hour boundaries —
# the exact known-divergences enforcement needs that identical every run.
hour_ms = 3_600_000
advances = {ds["id"]: -(-(spans[ds["id"]] + ISOLATION_GAP_MS) // hour_ms) * hour_ms for ds in corpus["datasets"]}
total = sum(advances.values())
now = datetime.now(tz=UTC).replace(second=0, microsecond=0)
cursor = (int((now - timedelta(hours=1)).timestamp() * 1000) - total) // hour_ms * hour_ms
bases: dict[int, int] = {}
metrics: list[Metrics] = []
for ds in corpus["datasets"]:
bases[ds["id"]] = cursor
for series in ds["series"]:
labels = dict(series["labels"])
metric_name = labels.pop("__name__")
for off_ms, raw in series["samples"]:
stale = raw == "stale"
metrics.append(
Metrics(
metric_name=metric_name,
labels=labels,
timestamp=datetime.fromtimestamp((cursor + off_ms) / 1000, tz=UTC),
value=0.0 if stale else (SPECIALS[raw] if isinstance(raw, str) else float(raw)),
flags=1 if stale else 0,
)
)
cursor += advances[ds["id"]]
insert_metrics(metrics)
corpus, bases = ingest_promqltest_corpus(insert_metrics)
token = get_token(USER_ADMIN_EMAIL, USER_ADMIN_PASSWORD)
failures: dict[str, list[str]] = {leg: [] for leg, _ in LEGS}