chore: add pvcs list (#6654)

chore: develop deprecation and related changes

frontend/src/pages/MessagingQueues/MQDetails/MQDetails.tsx

@@ -24,13 +24,13 @@ const MQServiceDetailTypePerView = (
 	producerLatencyOption: ProducerLatencyOptions,
 ): Record<string, MessagingQueueServiceDetailType[]> => ({
 	[MessagingQueuesViewType.consumerLag.value]: [
-		MessagingQueueServiceDetailType.ConsumerDetails,
 		MessagingQueueServiceDetailType.ProducerDetails,
+		MessagingQueueServiceDetailType.ConsumerDetails,
 		MessagingQueueServiceDetailType.NetworkLatency,
 	],
 	[MessagingQueuesViewType.partitionLatency.value]: [
-		MessagingQueueServiceDetailType.ConsumerDetails,
 		MessagingQueueServiceDetailType.ProducerDetails,
+		MessagingQueueServiceDetailType.ConsumerDetails,
 	],
 	[MessagingQueuesViewType.producerLatency.value]: [
 		producerLatencyOption === ProducerLatencyOptions.Consumers
@@ -122,7 +122,7 @@ function MessagingQueuesDetails({
 	producerLatencyOption: ProducerLatencyOptions;
 }): JSX.Element {
 	const [currentTab, setCurrentTab] = useState<MessagingQueueServiceDetailType>(
-		MessagingQueueServiceDetailType.ConsumerDetails,
+		MessagingQueueServiceDetailType.ProducerDetails,
 	);
 
 	useEffect(() => {

frontend/src/pages/MessagingQueues/MessagingQueuesUtils.ts

@@ -179,10 +179,13 @@ export const convertToNanoseconds = (timestamp: number): bigint =>
 export const getStartAndEndTimesInMilliseconds = (
 	timestamp: number,
 ): { start: number; end: number } => {
-	const FIVE_MINUTES_IN_MILLISECONDS = 5 * 60 * 1000; // 5 minutes in milliseconds - check with Shivanshu once
+	const FIVE_MINUTES_IN_MILLISECONDS = 5 * 60 * 1000; // 300,000 milliseconds
 
-	const start = Math.floor(timestamp);
+	const pointInTime = Math.floor(timestamp * 1000);
-	const end = Math.floor(start + FIVE_MINUTES_IN_MILLISECONDS);
+
+	// Convert timestamp to milliseconds and floor it
+	const start = Math.floor(pointInTime - FIVE_MINUTES_IN_MILLISECONDS);
+	const end = Math.floor(pointInTime + FIVE_MINUTES_IN_MILLISECONDS);
 
 	return { start, end };
 };
@@ -311,8 +314,8 @@ export const getMetaDataAndAPIPerView = (
 	return {
 		[MessagingQueuesViewType.consumerLag.value]: {
 			tableApiPayload: {
-				start: (selectedTimelineQuery?.start || 0) * 1e9,
+				start: (selectedTimelineQuery?.start || 0) * 1e6,
-				end: (selectedTimelineQuery?.end || 0) * 1e9,
+				end: (selectedTimelineQuery?.end || 0) * 1e6,
 				variables: {
 					partition: selectedTimelineQuery?.partition,
 					topic: selectedTimelineQuery?.topic,

pkg/query-service/app/clickhouseReader/reader.go

@@ -2694,8 +2694,8 @@ func (r *ClickHouseReader) GetTagsInfoInLastHeartBeatInterval(ctx context.Contex
 }
 
 // remove this after sometime
-func removeUnderscoreDuplicateFields(fields []model.LogField) []model.LogField {
+func removeUnderscoreDuplicateFields(fields []model.Field) []model.Field {
-	lookup := map[string]model.LogField{}
+	lookup := map[string]model.Field{}
 	for _, v := range fields {
 		lookup[v.Name+v.DataType] = v
 	}
@@ -2706,7 +2706,7 @@ func removeUnderscoreDuplicateFields(fields []model.LogField) []model.LogField {
 		}
 	}
 
-	updatedFields := []model.LogField{}
+	updatedFields := []model.Field{}
 	for _, v := range lookup {
 		updatedFields = append(updatedFields, v)
 	}
@@ -2717,11 +2717,11 @@ func (r *ClickHouseReader) GetLogFields(ctx context.Context) (*model.GetFieldsRe
 	// response will contain top level fields from the otel log model
 	response := model.GetFieldsResponse{
 		Selected:    constants.StaticSelectedLogFields,
-		Interesting: []model.LogField{},
+		Interesting: []model.Field{},
 	}
 
 	// get attribute keys
-	attributes := []model.LogField{}
+	attributes := []model.Field{}
 	query := fmt.Sprintf("SELECT DISTINCT name, datatype from %s.%s group by name, datatype", r.logsDB, r.logsAttributeKeys)
 	err := r.db.Select(ctx, &attributes, query)
 	if err != nil {
@@ -2729,7 +2729,7 @@ func (r *ClickHouseReader) GetLogFields(ctx context.Context) (*model.GetFieldsRe
 	}
 
 	// get resource keys
-	resources := []model.LogField{}
+	resources := []model.Field{}
 	query = fmt.Sprintf("SELECT DISTINCT name, datatype from %s.%s group by name, datatype", r.logsDB, r.logsResourceKeys)
 	err = r.db.Select(ctx, &resources, query)
 	if err != nil {
@@ -2753,9 +2753,11 @@ func (r *ClickHouseReader) GetLogFields(ctx context.Context) (*model.GetFieldsRe
 	return &response, nil
 }
 
-func (r *ClickHouseReader) extractSelectedAndInterestingFields(tableStatement string, fieldType string, fields *[]model.LogField, response *model.GetFieldsResponse) {
+func (r *ClickHouseReader) extractSelectedAndInterestingFields(tableStatement string, overrideFieldType string, fields *[]model.Field, response *model.GetFieldsResponse) {
 	for _, field := range *fields {
-		field.Type = fieldType
+		if overrideFieldType != "" {
+			field.Type = overrideFieldType
+		}
 		// all static fields are assumed to be selected as we don't allow changing them
 		if isColumn(r.useLogsNewSchema, tableStatement, field.Type, field.Name, field.DataType) {
 			response.Selected = append(response.Selected, field)
@@ -2945,6 +2947,165 @@ func (r *ClickHouseReader) UpdateLogField(ctx context.Context, field *model.Upda
 	return nil
 }
 
+func (r *ClickHouseReader) GetTraceFields(ctx context.Context) (*model.GetFieldsResponse, *model.ApiError) {
+	// response will contain top level fields from the otel trace model
+	response := model.GetFieldsResponse{
+		Selected:    []model.Field{},
+		Interesting: []model.Field{},
+	}
+
+	// get the top level selected fields
+	for _, field := range constants.NewStaticFieldsTraces {
+		if (v3.AttributeKey{} == field) {
+			continue
+		}
+		response.Selected = append(response.Selected, model.Field{
+			Name:     field.Key,
+			DataType: field.DataType.String(),
+			Type:     constants.Static,
+		})
+	}
+
+	// get attribute keys
+	attributes := []model.Field{}
+	query := fmt.Sprintf("SELECT tagKey, tagType, dataType from %s.%s group by tagKey, tagType, dataType", r.TraceDB, r.spanAttributesKeysTable)
+	rows, err := r.db.Query(ctx, query)
+	if err != nil {
+		return nil, &model.ApiError{Err: err, Typ: model.ErrorInternal}
+	}
+	defer rows.Close()
+
+	var tagKey string
+	var dataType string
+	var tagType string
+	for rows.Next() {
+		if err := rows.Scan(&tagKey, &tagType, &dataType); err != nil {
+			return nil, &model.ApiError{Err: err, Typ: model.ErrorInternal}
+		}
+		attributes = append(attributes, model.Field{
+			Name:     tagKey,
+			DataType: dataType,
+			Type:     tagType,
+		})
+	}
+
+	statements := []model.ShowCreateTableStatement{}
+	query = fmt.Sprintf("SHOW CREATE TABLE %s.%s", r.TraceDB, r.traceLocalTableName)
+	err = r.db.Select(ctx, &statements, query)
+	if err != nil {
+		return nil, &model.ApiError{Err: err, Typ: model.ErrorInternal}
+	}
+
+	r.extractSelectedAndInterestingFields(statements[0].Statement, "", &attributes, &response)
+
+	return &response, nil
+
+}
+
+func (r *ClickHouseReader) UpdateTraceField(ctx context.Context, field *model.UpdateField) *model.ApiError {
+	if !field.Selected {
+		return model.ForbiddenError(errors.New("removing a selected field is not allowed, please reach out to support."))
+	}
+
+	// name of the materialized column
+	colname := utils.GetClickhouseColumnNameV2(field.Type, field.DataType, field.Name)
+
+	field.DataType = strings.ToLower(field.DataType)
+
+	// dataType and chDataType of the materialized column
+	var dataTypeMap = map[string]string{
+		"string":  "string",
+		"bool":    "bool",
+		"int64":   "number",
+		"float64": "number",
+	}
+	var chDataTypeMap = map[string]string{
+		"string":  "String",
+		"bool":    "Bool",
+		"int64":   "Float64",
+		"float64": "Float64",
+	}
+	chDataType := chDataTypeMap[field.DataType]
+	dataType := dataTypeMap[field.DataType]
+
+	// typeName: tag => attributes, resource => resources
+	typeName := field.Type
+	if field.Type == string(v3.AttributeKeyTypeTag) {
+		typeName = constants.Attributes
+	} else if field.Type == string(v3.AttributeKeyTypeResource) {
+		typeName = constants.Resources
+	}
+
+	attrColName := fmt.Sprintf("%s_%s", typeName, dataType)
+	for _, table := range []string{r.traceLocalTableName, r.traceTableName} {
+		q := "ALTER TABLE %s.%s ON CLUSTER %s ADD COLUMN IF NOT EXISTS `%s` %s DEFAULT %s['%s'] CODEC(ZSTD(1))"
+		query := fmt.Sprintf(q,
+			r.TraceDB, table,
+			r.cluster,
+			colname, chDataType,
+			attrColName,
+			field.Name,
+		)
+		err := r.db.Exec(ctx, query)
+		if err != nil {
+			return &model.ApiError{Err: err, Typ: model.ErrorInternal}
+		}
+
+		query = fmt.Sprintf("ALTER TABLE %s.%s ON CLUSTER %s ADD COLUMN IF NOT EXISTS `%s_exists` bool DEFAULT if(mapContains(%s, '%s') != 0, true, false) CODEC(ZSTD(1))",
+			r.TraceDB, table,
+			r.cluster,
+			colname,
+			attrColName,
+			field.Name,
+		)
+		err = r.db.Exec(ctx, query)
+		if err != nil {
+			return &model.ApiError{Err: err, Typ: model.ErrorInternal}
+		}
+	}
+
+	// create the index
+	if strings.ToLower(field.DataType) == "bool" {
+		// there is no point in creating index for bool attributes as the cardinality is just 2
+		return nil
+	}
+
+	if field.IndexType == "" {
+		field.IndexType = constants.DefaultLogSkipIndexType
+	}
+	if field.IndexGranularity == 0 {
+		field.IndexGranularity = constants.DefaultLogSkipIndexGranularity
+	}
+	query := fmt.Sprintf("ALTER TABLE %s.%s ON CLUSTER %s ADD INDEX IF NOT EXISTS `%s_idx` (`%s`) TYPE %s  GRANULARITY %d",
+		r.TraceDB, r.traceLocalTableName,
+		r.cluster,
+		colname,
+		colname,
+		field.IndexType,
+		field.IndexGranularity,
+	)
+	err := r.db.Exec(ctx, query)
+	if err != nil {
+		return &model.ApiError{Err: err, Typ: model.ErrorInternal}
+	}
+
+	// add a default minmax index for numbers
+	if dataType == "number" {
+		query = fmt.Sprintf("ALTER TABLE %s.%s ON CLUSTER %s ADD INDEX IF NOT EXISTS `%s_minmax_idx` (`%s`) TYPE minmax  GRANULARITY 1",
+			r.TraceDB, r.traceLocalTableName,
+			r.cluster,
+			colname,
+			colname,
+		)
+		err = r.db.Exec(ctx, query)
+		if err != nil {
+			return &model.ApiError{Err: err, Typ: model.ErrorInternal}
+		}
+	}
+
+	return nil
+}
+
 func (r *ClickHouseReader) GetLogs(ctx context.Context, params *model.LogsFilterParams) (*[]model.SignozLog, *model.ApiError) {
 	response := []model.SignozLog{}
 	fields, apiErr := r.GetLogFields(ctx)

pkg/query-service/app/http_handler.go

@@ -125,6 +125,8 @@ type APIHandler struct {
 	daemonsetsRepo   *inframetrics.DaemonSetsRepo
 	statefulsetsRepo *inframetrics.StatefulSetsRepo
 	jobsRepo         *inframetrics.JobsRepo
+
+	pvcsRepo *inframetrics.PvcsRepo
 }
 
 type APIHandlerOpts struct {
@@ -208,6 +210,7 @@ func NewAPIHandler(opts APIHandlerOpts) (*APIHandler, error) {
 	daemonsetsRepo := inframetrics.NewDaemonSetsRepo(opts.Reader, querierv2)
 	statefulsetsRepo := inframetrics.NewStatefulSetsRepo(opts.Reader, querierv2)
 	jobsRepo := inframetrics.NewJobsRepo(opts.Reader, querierv2)
+	pvcsRepo := inframetrics.NewPvcsRepo(opts.Reader, querierv2)
 
 	aH := &APIHandler{
 		reader:                        opts.Reader,
@@ -237,6 +240,7 @@ func NewAPIHandler(opts APIHandlerOpts) (*APIHandler, error) {
 		daemonsetsRepo:                daemonsetsRepo,
 		statefulsetsRepo:              statefulsetsRepo,
 		jobsRepo:                      jobsRepo,
+		pvcsRepo:                      pvcsRepo,
 	}
 
 	logsQueryBuilder := logsv3.PrepareLogsQuery
@@ -408,6 +412,11 @@ func (aH *APIHandler) RegisterInfraMetricsRoutes(router *mux.Router, am *AuthMid
 	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)
@@ -527,6 +536,9 @@ func (aH *APIHandler) RegisterRoutes(router *mux.Router, am *AuthMiddleware) {
 	router.HandleFunc("/api/v1/settings/ingestion_key", am.AdminAccess(aH.insertIngestionKey)).Methods(http.MethodPost)
 	router.HandleFunc("/api/v1/settings/ingestion_key", am.ViewAccess(aH.getIngestionKeys)).Methods(http.MethodGet)
 
+	router.HandleFunc("/api/v2/traces/fields", am.ViewAccess(aH.traceFields)).Methods(http.MethodGet)
+	router.HandleFunc("/api/v2/traces/fields", am.EditAccess(aH.updateTraceField)).Methods(http.MethodPost)
+
 	router.HandleFunc("/api/v1/version", am.OpenAccess(aH.getVersion)).Methods(http.MethodGet)
 	router.HandleFunc("/api/v1/featureFlags", am.OpenAccess(aH.getFeatureFlags)).Methods(http.MethodGet)
 	router.HandleFunc("/api/v1/configs", am.OpenAccess(aH.getConfigs)).Methods(http.MethodGet)
@@ -4892,3 +4904,35 @@ func (aH *APIHandler) QueryRangeV4(w http.ResponseWriter, r *http.Request) {
 
 	aH.queryRangeV4(r.Context(), queryRangeParams, w, r)
 }
+
+func (aH *APIHandler) traceFields(w http.ResponseWriter, r *http.Request) {
+	fields, apiErr := aH.reader.GetTraceFields(r.Context())
+	if apiErr != nil {
+		RespondError(w, apiErr, "failed to fetch fields from the db")
+		return
+	}
+	aH.WriteJSON(w, r, fields)
+}
+
+func (aH *APIHandler) updateTraceField(w http.ResponseWriter, r *http.Request) {
+	field := model.UpdateField{}
+	if err := json.NewDecoder(r.Body).Decode(&field); err != nil {
+		apiErr := &model.ApiError{Typ: model.ErrorBadData, Err: err}
+		RespondError(w, apiErr, "failed to decode payload")
+		return
+	}
+
+	err := logs.ValidateUpdateFieldPayloadV2(&field)
+	if err != nil {
+		apiErr := &model.ApiError{Typ: model.ErrorBadData, Err: err}
+		RespondError(w, apiErr, "incorrect payload")
+		return
+	}
+
+	apiErr := aH.reader.UpdateTraceField(r.Context(), &field)
+	if apiErr != nil {
+		RespondError(w, apiErr, "failed to update field in the db")
+		return
+	}
+	aH.WriteJSON(w, r, field)
+}

pkg/query-service/app/infra.go

@@ -544,3 +544,56 @@ func (aH *APIHandler) getJobList(w http.ResponseWriter, r *http.Request) {
 
 	aH.Respond(w, jobList)
 }
+
+func (aH *APIHandler) getPvcList(w http.ResponseWriter, r *http.Request) {
+	ctx := r.Context()
+	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, 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
+	}
+
+	keys, err := aH.pvcsRepo.GetPvcAttributeKeys(ctx, *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
+	}
+
+	values, err := aH.pvcsRepo.GetPvcAttributeValues(ctx, *req)
+	if err != nil {
+		RespondError(w, &model.ApiError{Typ: model.ErrorInternal, Err: err}, nil)
+		return
+	}
+
+	aH.Respond(w, values)
+}

pkg/query-service/app/inframetrics/common.go

@@ -89,6 +89,10 @@ 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

pkg/query-service/app/inframetrics/namespaces.go

@@ -23,10 +23,11 @@ var (
 	}
 
 	queryNamesForNamespaces = map[string][]string{
-		"cpu":    {"A"},
+		"cpu":       {"A"},
-		"memory": {"D"},
+		"memory":    {"D"},
+		"pod_phase": {"H", "I", "J", "K"},
 	}
-	namespaceQueryNames = []string{"A", "D"}
+	namespaceQueryNames = []string{"A", "D", "H", "I", "J", "K"}
 
 	attributesKeysForNamespaces = []v3.AttributeKey{
 		{Key: "k8s_namespace_name"},
@@ -307,6 +308,19 @@ func (p *NamespacesRepo) GetNamespaceList(ctx context.Context, req model.Namespa
 				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.Meta = namespaceAttrs[record.NamespaceName]

pkg/query-service/app/inframetrics/nodes.go

@@ -17,7 +17,7 @@ import (
 var (
 	metricToUseForNodes = "k8s_node_cpu_utilization"
 
-	nodeAttrsToEnrich = []string{"k8s_node_name", "k8s_node_uid"}
+	nodeAttrsToEnrich = []string{"k8s_node_name", "k8s_node_uid", "k8s_cluster_name"}
 
 	k8sNodeUIDAttrKey = "k8s_node_uid"
 
@@ -27,13 +27,14 @@ var (
 		"memory":             {"C"},
 		"memory_allocatable": {"D"},
 	}
-	nodeQueryNames = []string{"A", "B", "C", "D"}
+	nodeQueryNames = []string{"A", "B", "C", "D", "E", "F"}
 
 	metricNamesForNodes = map[string]string{
 		"cpu":                "k8s_node_cpu_utilization",
 		"cpu_allocatable":    "k8s_node_allocatable_cpu",
 		"memory":             "k8s_node_memory_usage",
 		"memory_allocatable": "k8s_node_allocatable_memory",
+		"node_condition":     "k8s_node_condition_ready",
 	}
 )
 
@@ -325,6 +326,14 @@ func (p *NodesRepo) GetNodeList(ctx context.Context, req model.NodeListRequest)
 				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.Meta = nodeAttrs[record.NodeUID]

pkg/query-service/app/inframetrics/nodes_query.go

@@ -109,6 +109,74 @@ var NodesTableListQuery = v3.QueryRangeParamsV3{
 				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:      k8sNodeUIDAttrKey,
+						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:      k8sNodeUIDAttrKey,
+						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,

pkg/query-service/app/inframetrics/pods.go

@@ -27,6 +27,7 @@ var (
 		"k8s_daemonset_name",
 		"k8s_job_name",
 		"k8s_cronjob_name",
+		"k8s_cluster_name",
 	}
 
 	k8sPodUIDAttrKey = "k8s_pod_uid"
@@ -39,8 +40,9 @@ var (
 		"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"}
+	podQueryNames = []string{"A", "B", "C", "D", "E", "F", "G", "H", "I", "J", "K"}
 
 	metricNamesForPods = map[string]string{
 		"cpu":            "k8s_pod_cpu_utilization",
@@ -50,6 +52,7 @@ var (
 		"memory_request": "k8s_pod_memory_request_utilization",
 		"memory_limit":   "k8s_pod_memory_limit_utilization",
 		"restarts":       "k8s_container_restarts",
+		"pod_phase":      "k8s_pod_phase",
 	}
 )
 
@@ -365,6 +368,22 @@ func (p *PodsRepo) GetPodList(ctx context.Context, req model.PodListRequest) (mo
 				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.Meta = podAttrs[record.PodUID]

pkg/query-service/app/inframetrics/pods_query.go

@@ -54,7 +54,7 @@ var PodsTableListQuery = v3.QueryRangeParamsV3{
 				Expression:       "B",
 				ReduceTo:         v3.ReduceToOperatorAvg,
 				TimeAggregation:  v3.TimeAggregationAvg,
-				SpaceAggregation: v3.SpaceAggregationSum,
+				SpaceAggregation: v3.SpaceAggregationAvg,
 				Disabled:         false,
 			},
 			// pod cpu limit utilization
@@ -80,7 +80,7 @@ var PodsTableListQuery = v3.QueryRangeParamsV3{
 				Expression:       "C",
 				ReduceTo:         v3.ReduceToOperatorAvg,
 				TimeAggregation:  v3.TimeAggregationAvg,
-				SpaceAggregation: v3.SpaceAggregationSum,
+				SpaceAggregation: v3.SpaceAggregationAvg,
 				Disabled:         false,
 			},
 			// pod memory utilization
@@ -132,7 +132,7 @@ var PodsTableListQuery = v3.QueryRangeParamsV3{
 				Expression:       "E",
 				ReduceTo:         v3.ReduceToOperatorAvg,
 				TimeAggregation:  v3.TimeAggregationAvg,
-				SpaceAggregation: v3.SpaceAggregationSum,
+				SpaceAggregation: v3.SpaceAggregationAvg,
 				Disabled:         false,
 			},
 			// pod memory limit utilization
@@ -158,7 +158,7 @@ var PodsTableListQuery = v3.QueryRangeParamsV3{
 				Expression:       "F",
 				ReduceTo:         v3.ReduceToOperatorAvg,
 				TimeAggregation:  v3.TimeAggregationAvg,
-				SpaceAggregation: v3.SpaceAggregationSum,
+				SpaceAggregation: v3.SpaceAggregationAvg,
 				Disabled:         false,
 			},
 			"G": {
@@ -187,6 +187,142 @@ var PodsTableListQuery = v3.QueryRangeParamsV3{
 				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,

pkg/query-service/app/inframetrics/pvcs.go

@@ -0,0 +1,378 @@
+package inframetrics
+
+import (
+	"context"
+	"math"
+	"sort"
+
+	"go.signoz.io/signoz/pkg/query-service/app/metrics/v4/helpers"
+	"go.signoz.io/signoz/pkg/query-service/common"
+	"go.signoz.io/signoz/pkg/query-service/interfaces"
+	"go.signoz.io/signoz/pkg/query-service/model"
+	v3 "go.signoz.io/signoz/pkg/query-service/model/v3"
+	"go.signoz.io/signoz/pkg/query-service/postprocess"
+	"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, 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, &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, 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, &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, 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
+	}
+
+	queryResponse, _, err := p.querierV2.QueryRange(ctx, 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, 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, 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,
+				})
+			}
+		}
+	}
+
+	queryResponse, _, err := p.querierV2.QueryRange(ctx, 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.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
+
+	return resp, nil
+}

pkg/query-service/app/inframetrics/pvcs_query.go

@@ -0,0 +1,204 @@
+package inframetrics
+
+import v3 "go.signoz.io/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,
+}

pkg/query-service/app/inframetrics/workload_query.go

@@ -4,13 +4,13 @@ import v3 "go.signoz.io/signoz/pkg/query-service/model/v3"
 
 var (
 	metricNamesForWorkloads = map[string]string{
-		"cpu":          "k8s_pod_cpu_utilization",
+		"cpu":            "k8s_pod_cpu_utilization",
-		"cpu_req":      "k8s_pod_cpu_request_utilization",
+		"cpu_request":    "k8s_pod_cpu_request_utilization",
-		"cpu_limit":    "k8s_pod_cpu_limit_utilization",
+		"cpu_limit":      "k8s_pod_cpu_limit_utilization",
-		"memory":       "k8s_pod_memory_usage",
+		"memory":         "k8s_pod_memory_usage",
-		"memory_req":   "k8s_pod_memory_request_utilization",
+		"memory_request": "k8s_pod_memory_request_utilization",
-		"memory_limit": "k8s_pod_memory_limit_utilization",
+		"memory_limit":   "k8s_pod_memory_limit_utilization",
-		"restarts":     "k8s_container_restarts",
+		"restarts":       "k8s_container_restarts",
 	}
 )
 
@@ -54,7 +54,7 @@ var WorkloadTableListQuery = v3.QueryRangeParamsV3{
 				Expression:       "B",
 				ReduceTo:         v3.ReduceToOperatorAvg,
 				TimeAggregation:  v3.TimeAggregationAvg,
-				SpaceAggregation: v3.SpaceAggregationSum,
+				SpaceAggregation: v3.SpaceAggregationAvg,
 				Disabled:         false,
 			},
 			// pod cpu limit utilization
@@ -74,7 +74,7 @@ var WorkloadTableListQuery = v3.QueryRangeParamsV3{
 				Expression:       "C",
 				ReduceTo:         v3.ReduceToOperatorAvg,
 				TimeAggregation:  v3.TimeAggregationAvg,
-				SpaceAggregation: v3.SpaceAggregationSum,
+				SpaceAggregation: v3.SpaceAggregationAvg,
 				Disabled:         false,
 			},
 			// pod memory utilization
@@ -114,7 +114,7 @@ var WorkloadTableListQuery = v3.QueryRangeParamsV3{
 				Expression:       "E",
 				ReduceTo:         v3.ReduceToOperatorAvg,
 				TimeAggregation:  v3.TimeAggregationAvg,
-				SpaceAggregation: v3.SpaceAggregationSum,
+				SpaceAggregation: v3.SpaceAggregationAvg,
 				Disabled:         false,
 			},
 			// pod memory limit utilization
@@ -134,7 +134,7 @@ var WorkloadTableListQuery = v3.QueryRangeParamsV3{
 				Expression:       "F",
 				ReduceTo:         v3.ReduceToOperatorAvg,
 				TimeAggregation:  v3.TimeAggregationAvg,
-				SpaceAggregation: v3.SpaceAggregationSum,
+				SpaceAggregation: v3.SpaceAggregationAvg,
 				Disabled:         false,
 			},
 			"G": {

pkg/query-service/app/integrations/messagingQueues/kafka/sql.go

@@ -6,6 +6,8 @@ import (
 
 func generateConsumerSQL(start, end int64, topic, partition, consumerGroup, queueType string) string {
 	timeRange := (end - start) / 1000000000
+	tsBucketStart := (start / 1000000000) - 1800
+	tsBucketEnd := end / 1000000000
 	query := fmt.Sprintf(`
 WITH consumer_query AS (
     SELECT
@@ -18,6 +20,8 @@ WITH consumer_query AS (
     WHERE
         timestamp >= '%d'
         AND timestamp <= '%d'
+        AND ts_bucket_start >=  '%d'
+        AND ts_bucket_start <= '%d'
         AND kind = 5
         AND attribute_string_messaging$$system = '%s'
         AND attributes_string['messaging.destination.name'] = '%s'
@@ -36,13 +40,15 @@ FROM
     consumer_query
 ORDER BY
     resource_string_service$$name;
-`, start, end, queueType, topic, partition, consumerGroup, timeRange)
+`, start, end, tsBucketStart, tsBucketEnd, queueType, topic, partition, consumerGroup, timeRange)
 	return query
 }
 
 // S1 landing
 func generatePartitionLatencySQL(start, end int64, queueType string) string {
 	timeRange := (end - start) / 1000000000
+	tsBucketStart := (start / 1000000000) - 1800
+	tsBucketEnd := end / 1000000000
 	query := fmt.Sprintf(`
 WITH partition_query AS (
     SELECT
@@ -54,6 +60,8 @@ WITH partition_query AS (
     WHERE
         timestamp >= '%d'
         AND timestamp <= '%d'
+        AND ts_bucket_start >=  '%d'
+        AND ts_bucket_start <= '%d'
         AND kind = 4
         AND attribute_string_messaging$$system = '%s'
     GROUP BY topic, partition
@@ -68,13 +76,15 @@ FROM
     partition_query
 ORDER BY
     topic;
-`, start, end, queueType, timeRange)
+`, start, end, tsBucketStart, tsBucketEnd, queueType, timeRange)
 	return query
 }
 
 // S1 consumer
 func generateConsumerPartitionLatencySQL(start, end int64, topic, partition, queueType string) string {
 	timeRange := (end - start) / 1000000000
+	tsBucketStart := (start / 1000000000) - 1800
+	tsBucketEnd := end / 1000000000
 	query := fmt.Sprintf(`
 WITH consumer_pl AS (
     SELECT
@@ -87,6 +97,8 @@ WITH consumer_pl AS (
     WHERE
         timestamp >= '%d'
         AND timestamp <= '%d'
+        AND ts_bucket_start >=  '%d'
+        AND ts_bucket_start <= '%d'
         AND kind = 5
         AND attribute_string_messaging$$system = '%s'
         AND attributes_string['messaging.destination.name'] = '%s'
@@ -104,14 +116,15 @@ FROM
     consumer_pl
 ORDER BY
     consumer_group;
-`, start, end, queueType, topic, partition, timeRange)
+`, start, end, tsBucketStart, tsBucketEnd, queueType, topic, partition, timeRange)
 	return query
 }
 
 // S3, producer overview
 func generateProducerPartitionThroughputSQL(start, end int64, queueType string) string {
 	timeRange := (end - start) / 1000000000
-	// t, svc, rps, byte*, p99, err
+	tsBucketStart := (start / 1000000000) - 1800
+	tsBucketEnd := end / 1000000000 // t, svc, rps, byte*, p99, err
 	query := fmt.Sprintf(`
 WITH producer_latency AS (
     SELECT
@@ -124,6 +137,8 @@ WITH producer_latency AS (
     WHERE
         timestamp >= '%d'
         AND timestamp <= '%d'
+        AND ts_bucket_start >=  '%d'
+        AND ts_bucket_start <= '%d'
         AND kind = 4
         AND attribute_string_messaging$$system = '%s'
     GROUP BY topic, resource_string_service$$name
@@ -137,13 +152,15 @@ SELECT
     COALESCE(total_requests / %d, 0) AS throughput
 FROM
     producer_latency
-`, start, end, queueType, timeRange)
+`, start, end, tsBucketStart, tsBucketEnd, queueType, timeRange)
 	return query
 }
 
 // S3, producer topic/service overview
 func generateProducerTopicLatencySQL(start, end int64, topic, service, queueType string) string {
 	timeRange := (end - start) / 1000000000
+	tsBucketStart := (start / 1000000000) - 1800
+	tsBucketEnd := end / 1000000000
 	query := fmt.Sprintf(`
 WITH consumer_latency AS (
     SELECT
@@ -155,6 +172,8 @@ WITH consumer_latency AS (
     WHERE
         timestamp >= '%d'
         AND timestamp <= '%d'
+        AND ts_bucket_start >=  '%d'
+        AND ts_bucket_start <= '%d'
         AND kind = 4
 		AND resource_string_service$$name = '%s'
         AND attribute_string_messaging$$system = '%s'
@@ -169,13 +188,15 @@ SELECT
     COALESCE(total_requests / %d, 0) AS throughput
 FROM
     consumer_latency
-`, start, end, service, queueType, topic, timeRange)
+`, start, end, tsBucketStart, tsBucketEnd, service, queueType, topic, timeRange)
 	return query
 }
 
 // S3 consumer overview
 func generateConsumerLatencySQL(start, end int64, queueType string) string {
 	timeRange := (end - start) / 1000000000
+	tsBucketStart := (start / 1000000000) - 1800
+	tsBucketEnd := end / 1000000000
 	query := fmt.Sprintf(`
 WITH consumer_latency AS (
     SELECT
@@ -189,6 +210,8 @@ WITH consumer_latency AS (
     WHERE
         timestamp >= '%d'
         AND timestamp <= '%d'
+        AND ts_bucket_start >=  '%d'
+        AND ts_bucket_start <= '%d'
         AND kind = 5
         AND attribute_string_messaging$$system = '%s'
     GROUP BY topic, resource_string_service$$name
@@ -205,13 +228,15 @@ FROM
     consumer_latency
 ORDER BY
     topic;
-`, start, end, queueType, timeRange, timeRange)
+`, start, end, tsBucketStart, tsBucketEnd, queueType, timeRange, timeRange)
 	return query
 }
 
 // S3 consumer topic/service
 func generateConsumerServiceLatencySQL(start, end int64, topic, service, queueType string) string {
 	timeRange := (end - start) / 1000000000
+	tsBucketStart := (start / 1000000000) - 1800
+	tsBucketEnd := end / 1000000000
 	query := fmt.Sprintf(`
 WITH consumer_latency AS (
     SELECT
@@ -223,6 +248,8 @@ WITH consumer_latency AS (
     WHERE
         timestamp >= '%d'
         AND timestamp <= '%d'
+        AND ts_bucket_start >=  '%d'
+        AND ts_bucket_start <= '%d'
         AND kind = 5
 		AND resource_string_service$$name = '%s'
         AND attribute_string_messaging$$system = '%s'
@@ -237,7 +264,7 @@ SELECT
     COALESCE(total_requests / %d, 0) AS throughput
 FROM
     consumer_latency
-`, start, end, service, queueType, topic, timeRange)
+`, start, end, tsBucketStart, tsBucketEnd, service, queueType, topic, timeRange)
 	return query
 }
 
@@ -293,6 +320,8 @@ GROUP BY
 
 func generateProducerSQL(start, end int64, topic, partition, queueType string) string {
 	timeRange := (end - start) / 1000000000
+	tsBucketStart := (start / 1000000000) - 1800
+	tsBucketEnd := end / 1000000000
 	query := fmt.Sprintf(`
 WITH producer_query AS (
     SELECT
@@ -304,6 +333,8 @@ WITH producer_query AS (
     WHERE
         timestamp >= '%d'
         AND timestamp <= '%d'
+        AND ts_bucket_start >=  '%d'
+        AND ts_bucket_start <= '%d'
         AND kind = 4
         AND attribute_string_messaging$$system = '%s'
         AND attributes_string['messaging.destination.name'] = '%s'
@@ -320,33 +351,39 @@ FROM
     producer_query
 ORDER BY
     resource_string_service$$name;
-`, start, end, queueType, topic, partition, timeRange)
+`, start, end, tsBucketStart, tsBucketEnd, queueType, topic, partition, timeRange)
 	return query
 }
 
 func generateNetworkLatencyThroughputSQL(start, end int64, consumerGroup, partitionID, queueType string) string {
 	timeRange := (end - start) / 1000000000
+	tsBucketStart := (start / 1000000000) - 1800
+	tsBucketEnd := end / 1000000000
 	query := fmt.Sprintf(`
 SELECT
     attributes_string['messaging.client_id'] AS client_id,
-	attributes_string['service.instance.id'] AS service_instance_id,
+	resources_string['service.instance.id'] AS service_instance_id,
     resource_string_service$$name AS service_name,
     count(*) / %d AS throughput
 FROM signoz_traces.distributed_signoz_index_v3
 WHERE
     timestamp >= '%d'
     AND timestamp <= '%d'
+    AND ts_bucket_start >=  '%d'
+    AND ts_bucket_start <= '%d'
     AND kind = 5
     AND attribute_string_messaging$$system = '%s' 
     AND attributes_string['messaging.kafka.consumer.group'] = '%s'
     AND attributes_string['messaging.destination.partition.id'] = '%s'
 GROUP BY service_name, client_id, service_instance_id
 ORDER BY throughput DESC
-`, timeRange, start, end, queueType, consumerGroup, partitionID)
+`, timeRange, start, end, tsBucketStart, tsBucketEnd, queueType, consumerGroup, partitionID)
 	return query
 }
 
 func onboardProducersSQL(start, end int64, queueType string) string {
+	tsBucketStart := (start / 1000000000) - 1800
+	tsBucketEnd := end / 1000000000
 	query := fmt.Sprintf(`
 SELECT 
     COUNT(*) = 0 AS entries,
@@ -358,11 +395,15 @@ FROM
     signoz_traces.distributed_signoz_index_v3
 WHERE 
     timestamp >= '%d'
-    AND timestamp <= '%d';`, queueType, start, end)
+    AND timestamp <= '%d'
+    AND ts_bucket_start >=  '%d'
+    AND ts_bucket_start <= '%d';`, queueType, start, end, tsBucketStart, tsBucketEnd)
 	return query
 }
 
 func onboardConsumerSQL(start, end int64, queueType string) string {
+	tsBucketStart := (start / 1000000000) - 1800
+	tsBucketEnd := end / 1000000000
 	query := fmt.Sprintf(`
 SELECT  
     COUNT(*) = 0 AS entries,
@@ -374,10 +415,12 @@ SELECT
     COUNT(IF(has(attributes_string, 'messaging.kafka.consumer.group'), 1, NULL)) = 0 AS cgroup,
     COUNT(IF(has(attributes_number, 'messaging.message.body.size'), 1, NULL)) = 0 AS bodysize,
     COUNT(IF(has(attributes_string, 'messaging.client_id'), 1, NULL)) = 0 AS clientid,
-    COUNT(IF(has(attributes_string, 'service.instance.id'), 1, NULL)) = 0 AS instanceid
+    COUNT(IF(has(resources_string, 'service.instance.id'), 1, NULL)) = 0 AS instanceid
 FROM signoz_traces.distributed_signoz_index_v3
 WHERE 
     timestamp >= '%d'
-    AND timestamp <= '%d';`, queueType, start, end)
+    AND timestamp <= '%d'
+    AND ts_bucket_start >=  '%d'
+    AND ts_bucket_start <= '%d'	;`, queueType, start, end, tsBucketStart, tsBucketEnd)
 	return query
 }

pkg/query-service/app/logs/parser.go

@@ -228,8 +228,8 @@ func parseColumn(s string) (*string, error) {
 	return &colName, nil
 }
 
-func arrayToMap(fields []model.LogField) map[string]model.LogField {
+func arrayToMap(fields []model.Field) map[string]model.Field {
-	res := map[string]model.LogField{}
+	res := map[string]model.Field{}
 	for _, field := range fields {
 		res[field.Name] = field
 	}
@@ -251,7 +251,7 @@ func replaceInterestingFields(allFields *model.GetFieldsResponse, queryTokens []
 	return queryTokens, nil
 }
 
-func replaceFieldInToken(queryToken string, selectedFieldsLookup map[string]model.LogField, interestingFieldLookup map[string]model.LogField) (string, error) {
+func replaceFieldInToken(queryToken string, selectedFieldsLookup map[string]model.Field, interestingFieldLookup map[string]model.Field) (string, error) {
 	op := strings.TrimSpace(operatorRegex.FindString(queryToken))
 	opLower := strings.ToLower(op)
 
@@ -283,7 +283,7 @@ func replaceFieldInToken(queryToken string, selectedFieldsLookup map[string]mode
 			}
 		} else {
 			// creating the query token here as we have the metadata
-			field := model.LogField{}
+			field := model.Field{}
 
 			if sfield, ok := selectedFieldsLookup[sqlColName]; ok {
 				field = sfield

pkg/query-service/app/logs/parser_test.go

@@ -238,14 +238,14 @@ func TestParseColumn(t *testing.T) {
 func TestReplaceInterestingFields(t *testing.T) {
 	queryTokens := []string{"id.userid IN (100) ", "and id_key >= 50 ", `AND body ILIKE '%searchstring%'`}
 	allFields := model.GetFieldsResponse{
-		Selected: []model.LogField{
+		Selected: []model.Field{
 			{
 				Name:     "id_key",
 				DataType: "int64",
 				Type:     "attributes",
 			},
 		},
-		Interesting: []model.LogField{
+		Interesting: []model.Field{
 			{
 				Name:     "id.userid",
 				DataType: "int64",
@@ -326,7 +326,7 @@ func TestCheckIfPrevousPaginateAndModifyOrder(t *testing.T) {
 }
 
 var generateSQLQueryFields = model.GetFieldsResponse{
-	Selected: []model.LogField{
+	Selected: []model.Field{
 		{
 			Name:     "field1",
 			DataType: "int64",
@@ -348,7 +348,7 @@ var generateSQLQueryFields = model.GetFieldsResponse{
 			Type:     "static",
 		},
 	},
-	Interesting: []model.LogField{
+	Interesting: []model.Field{
 		{
 			Name:     "FielD1",
 			DataType: "int64",

pkg/query-service/app/logs/validator.go

@@ -6,6 +6,7 @@ import (
 
 	"go.signoz.io/signoz/pkg/query-service/constants"
 	"go.signoz.io/signoz/pkg/query-service/model"
+	v3 "go.signoz.io/signoz/pkg/query-service/model/v3"
 )
 
 func ValidateUpdateFieldPayload(field *model.UpdateField) error {
@@ -38,3 +39,36 @@ func ValidateUpdateFieldPayload(field *model.UpdateField) error {
 	}
 	return nil
 }
+
+func ValidateUpdateFieldPayloadV2(field *model.UpdateField) error {
+	if field.Name == "" {
+		return fmt.Errorf("name cannot be empty")
+	}
+	if field.Type == "" {
+		return fmt.Errorf("type cannot be empty")
+	}
+	if field.DataType == "" {
+		return fmt.Errorf("dataType cannot be empty")
+	}
+
+	// the logs api uses the old names i.e attributes and resources while traces use tag and attribute.
+	// update log api to use tag and attribute.
+	matched, err := regexp.MatchString(fmt.Sprintf("^(%s|%s)$", v3.AttributeKeyTypeTag, v3.AttributeKeyTypeResource), field.Type)
+	if err != nil {
+		return err
+	}
+	if !matched {
+		return fmt.Errorf("type %s not supported", field.Type)
+	}
+
+	if field.IndexType != "" {
+		matched, err := regexp.MatchString(`^(minmax|set\([0-9]\)|bloom_filter\((0?.?[0-9]+|1)\)|tokenbf_v1\([0-9]+,[0-9]+,[0-9]+\)|ngrambf_v1\([0-9]+,[0-9]+,[0-9]+,[0-9]+\))$`, field.IndexType)
+		if err != nil {
+			return err
+		}
+		if !matched {
+			return fmt.Errorf("index type %s not supported", field.IndexType)
+		}
+	}
+	return nil
+}

pkg/query-service/app/metrics/query_builder.go

@@ -5,9 +5,73 @@ import (
 	"reflect"
 	"strings"
 
+	v3 "go.signoz.io/signoz/pkg/query-service/model/v3"
 	"go.uber.org/zap"
 )
 
+func AddMetricValueFilter(mq *v3.BuilderQuery) *v3.MetricValueFilter {
+
+	var metricValueFilter *v3.MetricValueFilter = nil
+
+	if mq != nil && mq.Filters != nil && mq.Filters.Items != nil {
+		for _, item := range mq.Filters.Items {
+			if item.Key.Key == "__value" {
+				switch v := item.Value.(type) {
+				case float64:
+					metricValueFilter = &v3.MetricValueFilter{
+						Value: v,
+					}
+				case float32:
+					metricValueFilter = &v3.MetricValueFilter{
+						Value: float64(v),
+					}
+				case int:
+					metricValueFilter = &v3.MetricValueFilter{
+						Value: float64(v),
+					}
+				case int8:
+					metricValueFilter = &v3.MetricValueFilter{
+						Value: float64(v),
+					}
+				case int16:
+					metricValueFilter = &v3.MetricValueFilter{
+						Value: float64(v),
+					}
+				case int32:
+					metricValueFilter = &v3.MetricValueFilter{
+						Value: float64(v),
+					}
+				case int64:
+					metricValueFilter = &v3.MetricValueFilter{
+						Value: float64(v),
+					}
+				case uint:
+					metricValueFilter = &v3.MetricValueFilter{
+						Value: float64(v),
+					}
+				case uint8:
+					metricValueFilter = &v3.MetricValueFilter{
+						Value: float64(v),
+					}
+				case uint16:
+					metricValueFilter = &v3.MetricValueFilter{
+						Value: float64(v),
+					}
+				case uint32:
+					metricValueFilter = &v3.MetricValueFilter{
+						Value: float64(v),
+					}
+				case uint64:
+					metricValueFilter = &v3.MetricValueFilter{
+						Value: float64(v),
+					}
+				}
+			}
+		}
+	}
+	return metricValueFilter
+}
+
 // FormattedValue formats the value to be used in clickhouse query
 func FormattedValue(v interface{}) string {
 	switch x := v.(type) {

pkg/query-service/app/metrics/v3/query_builder.go

@@ -5,6 +5,7 @@ import (
 	"strings"
 	"time"
 
+	"go.signoz.io/signoz/pkg/query-service/app/metrics"
 	"go.signoz.io/signoz/pkg/query-service/app/metrics/v4/helpers"
 	"go.signoz.io/signoz/pkg/query-service/common"
 	"go.signoz.io/signoz/pkg/query-service/constants"
@@ -335,6 +336,10 @@ func PrepareMetricQuery(start, end int64, queryType v3.QueryType, panelType v3.P
 
 	start, end = common.AdjustedMetricTimeRange(start, end, mq.StepInterval, *mq)
 
+	if valFilter := metrics.AddMetricValueFilter(mq); valFilter != nil {
+		mq.MetricValueFilter = valFilter
+	}
+
 	// if the aggregate operator is a histogram quantile, and user has not forgotten
 	// the le tag in the group by then add the le tag to the group by
 	if mq.AggregateOperator == v3.AggregateOperatorHistQuant50 ||

pkg/query-service/app/metrics/v4/cumulative/table.go

@@ -20,11 +20,16 @@ func PrepareMetricQueryCumulativeTable(start, end, step int64, mq *v3.BuilderQue
 	orderBy := helpers.OrderByAttributeKeyTags(mq.OrderBy, mq.GroupBy)
 	selectLabels := helpers.GroupByAttributeKeyTags(mq.GroupBy...)
 
+	valueFilter := " WHERE isNaN(per_series_value) = 0"
+	if mq.MetricValueFilter != nil {
+		valueFilter += fmt.Sprintf(" AND per_series_value = %f", mq.MetricValueFilter.Value)
+	}
+
 	queryTmpl :=
 		"SELECT %s," +
 			" %s as value" +
 			" FROM (%s)" +
-			" WHERE isNaN(per_series_value) = 0" +
+			valueFilter +
 			" GROUP BY %s" +
 			" ORDER BY %s"
 

pkg/query-service/app/metrics/v4/cumulative/timeseries.go

@@ -190,11 +190,16 @@ func PrepareMetricQueryCumulativeTimeSeries(start, end, step int64, mq *v3.Build
 	orderBy := helpers.OrderByAttributeKeyTags(mq.OrderBy, mq.GroupBy)
 	selectLabels := helpers.GroupByAttributeKeyTags(mq.GroupBy...)
 
+	valueFilter := " WHERE isNaN(per_series_value) = 0"
+	if mq.MetricValueFilter != nil {
+		valueFilter += fmt.Sprintf(" AND per_series_value = %f", mq.MetricValueFilter.Value)
+	}
+
 	queryTmpl :=
 		"SELECT %s," +
 			" %s as value" +
 			" FROM (%s)" +
-			" WHERE isNaN(per_series_value) = 0" +
+			valueFilter +
 			" GROUP BY %s" +
 			" ORDER BY %s"
 

pkg/query-service/app/metrics/v4/delta/table.go

@@ -25,11 +25,16 @@ func PrepareMetricQueryDeltaTable(start, end, step int64, mq *v3.BuilderQuery) (
 	orderBy := helpers.OrderByAttributeKeyTags(mq.OrderBy, mq.GroupBy)
 	selectLabels := helpers.GroupByAttributeKeyTags(mq.GroupBy...)
 
+	valueFilter := " WHERE isNaN(per_series_value) = 0"
+	if mq.MetricValueFilter != nil {
+		valueFilter += fmt.Sprintf(" AND per_series_value = %f", mq.MetricValueFilter.Value)
+	}
+
 	queryTmpl :=
 		"SELECT %s," +
 			" %s as value" +
 			" FROM (%s)" +
-			" WHERE isNaN(per_series_value) = 0" +
+			valueFilter +
 			" GROUP BY %s" +
 			" ORDER BY %s"
 

pkg/query-service/app/metrics/v4/delta/timeseries.go

@@ -142,11 +142,16 @@ func PrepareMetricQueryDeltaTimeSeries(start, end, step int64, mq *v3.BuilderQue
 	orderBy := helpers.OrderByAttributeKeyTags(mq.OrderBy, mq.GroupBy)
 	selectLabels := helpers.GroupByAttributeKeyTags(mq.GroupBy...)
 
+	valueFilter := " WHERE isNaN(per_series_value) = 0"
+	if mq.MetricValueFilter != nil {
+		valueFilter += fmt.Sprintf(" AND per_series_value = %f", mq.MetricValueFilter.Value)
+	}
+
 	queryTmpl :=
 		"SELECT %s," +
 			" %s as value" +
 			" FROM (%s)" +
-			" WHERE isNaN(per_series_value) = 0" +
+			valueFilter +
 			" GROUP BY %s" +
 			" ORDER BY %s"
 

pkg/query-service/app/metrics/v4/helpers/sub_query.go

@@ -270,6 +270,10 @@ func PrepareTimeseriesFilterQuery(start, end int64, mq *v3.BuilderQuery) (string
 
 	if fs != nil && len(fs.Items) != 0 {
 		for _, item := range fs.Items {
+			if item.Key.Key == "__value" {
+				continue
+			}
+
 			toFormat := item.Value
 			op := v3.FilterOperator(strings.ToLower(strings.TrimSpace(string(item.Operator))))
 			if op == v3.FilterOperatorContains || op == v3.FilterOperatorNotContains {

pkg/query-service/app/metrics/v4/query_builder.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"time"
 
+	"go.signoz.io/signoz/pkg/query-service/app/metrics"
 	metricsV3 "go.signoz.io/signoz/pkg/query-service/app/metrics/v3"
 	"go.signoz.io/signoz/pkg/query-service/app/metrics/v4/cumulative"
 	"go.signoz.io/signoz/pkg/query-service/app/metrics/v4/delta"
@@ -19,6 +20,9 @@ import (
 // step is in seconds
 func PrepareMetricQuery(start, end int64, queryType v3.QueryType, panelType v3.PanelType, mq *v3.BuilderQuery, options metricsV3.Options) (string, error) {
 
+	if valFilter := metrics.AddMetricValueFilter(mq); valFilter != nil {
+		mq.MetricValueFilter = valFilter
+	}
 	start, end = common.AdjustedMetricTimeRange(start, end, mq.StepInterval, *mq)
 
 	var quantile float64

pkg/query-service/app/querier/helper.go

@@ -190,7 +190,7 @@ func (q *querier) runBuilderQuery(
 				ch <- channelResult{Err: err, Name: queryName, Query: limitQuery, Series: nil}
 				return
 			}
-			query = fmt.Sprintf(placeholderQuery, limitQuery)
+			query = strings.Replace(placeholderQuery, "#LIMIT_PLACEHOLDER", limitQuery, 1)
 		} else {
 			query, err = tracesQueryBuilder(
 				start,

pkg/query-service/app/querier/v2/helper.go

@@ -190,7 +190,7 @@ func (q *querier) runBuilderQuery(
 				ch <- channelResult{Err: err, Name: queryName, Query: limitQuery, Series: nil}
 				return
 			}
-			query = fmt.Sprintf(placeholderQuery, limitQuery)
+			query = strings.Replace(placeholderQuery, "#LIMIT_PLACEHOLDER", limitQuery, 1)
 		} else {
 			query, err = tracesQueryBuilder(
 				start,

pkg/query-service/app/traces/v4/enrich.go

@@ -34,8 +34,8 @@ func enrichKeyWithMetadata(key v3.AttributeKey, keys map[string]v3.AttributeKey)
 		return v
 	}
 
-	for _, key := range utils.GenerateEnrichmentKeys(key) {
+	for _, tkey := range utils.GenerateEnrichmentKeys(key) {
-		if val, ok := keys[key]; ok {
+		if val, ok := keys[tkey]; ok {
 			return val
 		}
 	}

pkg/query-service/app/traces/v4/query_builder.go

@@ -74,6 +74,19 @@ func getSelectLabels(groupBy []v3.AttributeKey) string {
 	return strings.Join(labels, ",")
 }
 
+// TODO(nitya): use the _exists columns as well in the future similar to logs
+func existsSubQueryForFixedColumn(key v3.AttributeKey, op v3.FilterOperator) (string, error) {
+	if key.DataType == v3.AttributeKeyDataTypeString {
+		if op == v3.FilterOperatorExists {
+			return fmt.Sprintf("%s %s ''", getColumnName(key), tracesOperatorMappingV3[v3.FilterOperatorNotEqual]), nil
+		} else {
+			return fmt.Sprintf("%s %s ''", getColumnName(key), tracesOperatorMappingV3[v3.FilterOperatorEqual]), nil
+		}
+	} else {
+		return "", fmt.Errorf("unsupported operation, exists and not exists can only be applied on custom attributes or string type columns")
+	}
+}
+
 func buildTracesFilterQuery(fs *v3.FilterSet) (string, error) {
 	var conditions []string
 
@@ -110,7 +123,7 @@ func buildTracesFilterQuery(fs *v3.FilterSet) (string, error) {
 					conditions = append(conditions, fmt.Sprintf(operator, columnName, fmtVal))
 				case v3.FilterOperatorExists, v3.FilterOperatorNotExists:
 					if item.Key.IsColumn {
-						subQuery, err := tracesV3.ExistsSubQueryForFixedColumn(item.Key, item.Operator)
+						subQuery, err := existsSubQueryForFixedColumn(item.Key, item.Operator)
 						if err != nil {
 							return "", err
 						}
@@ -312,7 +325,7 @@ func buildTracesQuery(start, end, step int64, mq *v3.BuilderQuery, panelType v3.
 	}
 
 	if options.GraphLimitQtype == constants.SecondQueryGraphLimit {
-		filterSubQuery = filterSubQuery + " AND " + fmt.Sprintf("(%s) GLOBAL IN (", tracesV3.GetSelectKeys(mq.AggregateOperator, mq.GroupBy)) + "%s)"
+		filterSubQuery = filterSubQuery + " AND " + fmt.Sprintf("(%s) GLOBAL IN (", tracesV3.GetSelectKeys(mq.AggregateOperator, mq.GroupBy)) + "#LIMIT_PLACEHOLDER)"
 	}
 
 	switch mq.AggregateOperator {
@@ -350,7 +363,7 @@ func buildTracesQuery(start, end, step int64, mq *v3.BuilderQuery, panelType v3.
 	case v3.AggregateOperatorCount:
 		if mq.AggregateAttribute.Key != "" {
 			if mq.AggregateAttribute.IsColumn {
-				subQuery, err := tracesV3.ExistsSubQueryForFixedColumn(mq.AggregateAttribute, v3.FilterOperatorExists)
+				subQuery, err := existsSubQueryForFixedColumn(mq.AggregateAttribute, v3.FilterOperatorExists)
 				if err == nil {
 					filterSubQuery = fmt.Sprintf("%s AND %s", filterSubQuery, subQuery)
 				}

pkg/query-service/app/traces/v4/query_builder_test.go

@@ -265,9 +265,11 @@ func Test_buildTracesFilterQuery(t *testing.T) {
 					{Key: v3.AttributeKey{Key: "isDone", DataType: v3.AttributeKeyDataTypeBool, Type: v3.AttributeKeyTypeTag}, Operator: v3.FilterOperatorNotExists},
 					{Key: v3.AttributeKey{Key: "host1", DataType: v3.AttributeKeyDataTypeString, Type: v3.AttributeKeyTypeTag}, Operator: v3.FilterOperatorNotExists},
 					{Key: v3.AttributeKey{Key: "path", DataType: v3.AttributeKeyDataTypeString, Type: v3.AttributeKeyTypeTag, IsColumn: true}, Operator: v3.FilterOperatorNotExists},
+					{Key: v3.AttributeKey{Key: "http_url", DataType: v3.AttributeKeyDataTypeString, Type: v3.AttributeKeyTypeTag, IsColumn: true}, Operator: v3.FilterOperatorNotExists},
+					{Key: v3.AttributeKey{Key: "http.route", DataType: v3.AttributeKeyDataTypeString, Type: v3.AttributeKeyTypeTag, IsColumn: true}, Operator: v3.FilterOperatorNotExists},
 				}},
 			},
-			want: "mapContains(attributes_string, 'host') AND mapContains(attributes_number, 'duration') AND NOT mapContains(attributes_bool, 'isDone') AND NOT mapContains(attributes_string, 'host1') AND path = ''",
+			want: "mapContains(attributes_string, 'host') AND mapContains(attributes_number, 'duration') AND NOT mapContains(attributes_bool, 'isDone') AND NOT mapContains(attributes_string, 'host1') AND `attribute_string_path` = '' AND http_url = '' AND `attribute_string_http$$route` = ''",
 		},
 	}
 	for _, tt := range tests {
@@ -683,7 +685,7 @@ func TestPrepareTracesQuery(t *testing.T) {
 				},
 			},
 			want: "SELECT  attributes_string['function'] as `function`, toFloat64(count(distinct(name))) as value from signoz_traces.distributed_signoz_index_v3 where " +
-				"(timestamp >= '1680066360726210000' AND timestamp <= '1680066458000000000') AND (ts_bucket_start >= 1680064560 AND ts_bucket_start <= 1680066458) AND mapContains(attributes_string, 'function') AND (`function`) GLOBAL IN (%s) group by `function` order by value DESC",
+				"(timestamp >= '1680066360726210000' AND timestamp <= '1680066458000000000') AND (ts_bucket_start >= 1680064560 AND ts_bucket_start <= 1680066458) AND mapContains(attributes_string, 'function') AND (`function`) GLOBAL IN (#LIMIT_PLACEHOLDER) group by `function` order by value DESC",
 		},
 		{
 			name: "test with limit with resources- first",
@@ -766,7 +768,7 @@ func TestPrepareTracesQuery(t *testing.T) {
 			want: "SELECT  `attribute_string_function` as `function`, serviceName as `serviceName`, toFloat64(count(distinct(name))) as value from signoz_traces.distributed_signoz_index_v3 " +
 				"where (timestamp >= '1680066360726210000' AND timestamp <= '1680066458000000000') AND (ts_bucket_start >= 1680064560 AND ts_bucket_start <= 1680066458) AND attributes_number['line'] = 100 " +
 				"AND (resource_fingerprint GLOBAL IN (SELECT fingerprint FROM signoz_traces.distributed_traces_v3_resource WHERE (seen_at_ts_bucket_start >= 1680064560) AND (seen_at_ts_bucket_start <= 1680066458) " +
-				"AND simpleJSONExtractString(labels, 'hostname') = 'server1' AND labels like '%hostname%server1%')) AND (`function`,`serviceName`) GLOBAL IN (%s) group by `function`,`serviceName` order by value DESC",
+				"AND simpleJSONExtractString(labels, 'hostname') = 'server1' AND labels like '%hostname%server1%')) AND (`function`,`serviceName`) GLOBAL IN (#LIMIT_PLACEHOLDER) group by `function`,`serviceName` order by value DESC",
 		},
 	}
 

pkg/query-service/constants/constants.go

@@ -90,7 +90,7 @@ func EnableHostsInfraMonitoring() bool {
 	return GetOrDefaultEnv("ENABLE_INFRA_METRICS", "true") == "true"
 }
 
-var KafkaSpanEval = GetOrDefaultEnv("KAFKA_SPAN_EVAL", "true")
+var KafkaSpanEval = GetOrDefaultEnv("KAFKA_SPAN_EVAL", "false")
 
 func IsDurationSortFeatureEnabled() bool {
 	isDurationSortFeatureEnabledStr := DurationSortFeature
@@ -290,7 +290,7 @@ const (
 	UINT8                 = "Uint8"
 )
 
-var StaticSelectedLogFields = []model.LogField{
+var StaticSelectedLogFields = []model.Field{
 	{
 		Name:     "timestamp",
 		DataType: UINT32,

pkg/query-service/interfaces/interface.go

@@ -109,6 +109,10 @@ type Reader interface {
 	SubscribeToQueryProgress(queryId string) (<-chan model.QueryProgress, func(), *model.ApiError)
 
 	GetCountOfThings(ctx context.Context, query string) (uint64, error)
+
+	//trace
+	GetTraceFields(ctx context.Context) (*model.GetFieldsResponse, *model.ApiError)
+	UpdateTraceField(ctx context.Context, field *model.UpdateField) *model.ApiError
 }
 
 type Querier interface {

pkg/query-service/model/infra.go

@@ -151,13 +151,20 @@ type NodeListResponse struct {
 	Total   int              `json:"total"`
 }
 
+type NodeCountByCondition struct {
+	Ready    int `json:"ready"`
+	NotReady int `json:"notReady"`
+	Unknown  int `json:"unknown"`
+}
+
 type NodeListRecord struct {
-	NodeUID               string            `json:"nodeUID,omitempty"`
+	NodeUID               string               `json:"nodeUID,omitempty"`
-	NodeCPUUsage          float64           `json:"nodeCPUUsage"`
+	NodeCPUUsage          float64              `json:"nodeCPUUsage"`
-	NodeCPUAllocatable    float64           `json:"nodeCPUAllocatable"`
+	NodeCPUAllocatable    float64              `json:"nodeCPUAllocatable"`
-	NodeMemoryUsage       float64           `json:"nodeMemoryUsage"`
+	NodeMemoryUsage       float64              `json:"nodeMemoryUsage"`
-	NodeMemoryAllocatable float64           `json:"nodeMemoryAllocatable"`
+	NodeMemoryAllocatable float64              `json:"nodeMemoryAllocatable"`
-	Meta                  map[string]string `json:"meta"`
+	CountByCondition      NodeCountByCondition `json:"countByCondition"`
+	Meta                  map[string]string    `json:"meta"`
 }
 
 type NamespaceListRequest struct {
@@ -180,6 +187,7 @@ 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"`
 }
 
@@ -329,3 +337,30 @@ type JobListRecord struct {
 	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"`
+}
+
+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"`
+}

pkg/query-service/model/response.go

@@ -509,15 +509,15 @@ type ShowCreateTableStatement struct {
 	Statement string `json:"statement" ch:"statement"`
 }
 
-type LogField struct {
+type Field struct {
 	Name     string `json:"name" ch:"name"`
 	DataType string `json:"dataType" ch:"datatype"`
 	Type     string `json:"type"`
 }
 
 type GetFieldsResponse struct {
-	Selected    []LogField `json:"selected"`
+	Selected    []Field `json:"selected"`
-	Interesting []LogField `json:"interesting"`
+	Interesting []Field `json:"interesting"`
 }
 
 // Represents a log record in query service requests and responses.

pkg/query-service/model/v3/v3.go

@@ -770,6 +770,19 @@ type MetricTableHints struct {
 	SamplesTableName    string
 }
 
+type MetricValueFilter struct {
+	Value float64
+}
+
+func (m *MetricValueFilter) Clone() *MetricValueFilter {
+	if m == nil {
+		return nil
+	}
+	return &MetricValueFilter{
+		Value: m.Value,
+	}
+}
+
 type BuilderQuery struct {
 	QueryName            string            `json:"queryName"`
 	StepInterval         int64             `json:"stepInterval"`
@@ -795,7 +808,8 @@ type BuilderQuery struct {
 	ShiftBy              int64
 	IsAnomaly            bool
 	QueriesUsedInFormula []string
-	MetricTableHints     *MetricTableHints `json:"-"`
+	MetricTableHints     *MetricTableHints  `json:"-"`
+	MetricValueFilter    *MetricValueFilter `json:"-"`
 }
 
 func (b *BuilderQuery) SetShiftByFromFunc() {
@@ -859,6 +873,7 @@ func (b *BuilderQuery) Clone() *BuilderQuery {
 		ShiftBy:              b.ShiftBy,
 		IsAnomaly:            b.IsAnomaly,
 		QueriesUsedInFormula: b.QueriesUsedInFormula,
+		MetricValueFilter:    b.MetricValueFilter.Clone(),
 	}
 }