Compare commits

..

6 Commits

Author SHA1 Message Date
Prashant Shahi
bb2a60565b Merge branch 'main' into chore/sync-develop 2024-12-18 18:59:20 +05:30
Prashant Shahi
73fc5e45e5 Merge branch 'develop' into chore/sync-develop 2024-12-18 17:53:50 +05:30
Prashant Shahi
799c33ff82 Merge branch 'main' into chore/sync-develop 2024-12-18 17:46:44 +05:30
Prashant Shahi
e1b8205a16 Merge branch 'develop' into chore/sync-develop 2024-12-13 18:36:01 +05:30
Prashant Shahi
e21b9a561d chore: deprecate develop branch - use main
Signed-off-by: Prashant Shahi <prashant@signoz.io>
2024-12-13 18:08:36 +05:30
Prashant Shahi
c4858ec829 Merge branch 'main' into chore/sync-develop 2024-12-13 17:46:01 +05:30
37 changed files with 110 additions and 1581 deletions

View File

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

View File

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

View File

@@ -2571,58 +2571,6 @@ func (r *ClickHouseReader) FetchTemporality(ctx context.Context, metricNames []s
return metricNameToTemporality, nil
}
func (r *ClickHouseReader) GetTemporalitySwitchPoints(ctx context.Context, metricName string, startTime int64, endTime int64) ([]v3.TemporalityChangePoint, error) {
// Initialize slice to store temporality switch points
var temporalitySwitches []v3.TemporalityChangePoint
query := fmt.Sprintf(`
SELECT
temporality,
unix_milli,
lag_temporality
FROM (
SELECT
metric_name,
temporality,
unix_milli,
lagInFrame(temporality, 1, '') OVER (
PARTITION BY metric_name ORDER BY unix_milli
) AS lag_temporality
FROM %s.%s
WHERE unix_milli >= %d
AND unix_milli <= %d
AND metric_name = '%s'
) AS subquery
WHERE lag_temporality != temporality
AND lag_temporality != ''
ORDER BY unix_milli ASC;
`, signozMetricDBName, signozTSLocalTableNameV4, startTime, endTime, metricName)
rows, err := r.db.Query(ctx, query)
if err != nil {
return nil, err
}
defer rows.Close()
for rows.Next() {
var temporality string
var timestamp int64
var lagTemporality string
err := rows.Scan(&temporality, &timestamp, &lagTemporality)
if err != nil {
return nil, err
}
// Store each temporality switch point with both temporalities
temporalitySwitches = append(temporalitySwitches, v3.TemporalityChangePoint{
Timestamp: timestamp,
FromTemporality: v3.Temporality(lagTemporality),
ToTemporality: v3.Temporality(temporality),
})
}
return temporalitySwitches, nil
}
func (r *ClickHouseReader) GetTimeSeriesInfo(ctx context.Context) (map[string]interface{}, error) {
queryStr := fmt.Sprintf("SELECT countDistinct(fingerprint) as count from %s.%s where metric_name not like 'signoz_%%' group by metric_name order by count desc;", signozMetricDBName, signozTSTableNameV41Day)
@@ -2746,8 +2694,8 @@ func (r *ClickHouseReader) GetTagsInfoInLastHeartBeatInterval(ctx context.Contex
}
// remove this after sometime
func removeUnderscoreDuplicateFields(fields []model.Field) []model.Field {
lookup := map[string]model.Field{}
func removeUnderscoreDuplicateFields(fields []model.LogField) []model.LogField {
lookup := map[string]model.LogField{}
for _, v := range fields {
lookup[v.Name+v.DataType] = v
}
@@ -2758,7 +2706,7 @@ func removeUnderscoreDuplicateFields(fields []model.Field) []model.Field {
}
}
updatedFields := []model.Field{}
updatedFields := []model.LogField{}
for _, v := range lookup {
updatedFields = append(updatedFields, v)
}
@@ -2769,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.Field{},
Interesting: []model.LogField{},
}
// get attribute keys
attributes := []model.Field{}
attributes := []model.LogField{}
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 {
@@ -2781,7 +2729,7 @@ func (r *ClickHouseReader) GetLogFields(ctx context.Context) (*model.GetFieldsRe
}
// get resource keys
resources := []model.Field{}
resources := []model.LogField{}
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 {
@@ -2805,11 +2753,9 @@ func (r *ClickHouseReader) GetLogFields(ctx context.Context) (*model.GetFieldsRe
return &response, nil
}
func (r *ClickHouseReader) extractSelectedAndInterestingFields(tableStatement string, overrideFieldType string, fields *[]model.Field, response *model.GetFieldsResponse) {
func (r *ClickHouseReader) extractSelectedAndInterestingFields(tableStatement string, fieldType string, fields *[]model.LogField, response *model.GetFieldsResponse) {
for _, field := range *fields {
if overrideFieldType != "" {
field.Type = overrideFieldType
}
field.Type = fieldType
// 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)
@@ -2999,165 +2945,6 @@ 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)

View File

@@ -125,8 +125,6 @@ type APIHandler struct {
daemonsetsRepo *inframetrics.DaemonSetsRepo
statefulsetsRepo *inframetrics.StatefulSetsRepo
jobsRepo *inframetrics.JobsRepo
pvcsRepo *inframetrics.PvcsRepo
}
type APIHandlerOpts struct {
@@ -210,7 +208,6 @@ 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,
@@ -240,7 +237,6 @@ func NewAPIHandler(opts APIHandlerOpts) (*APIHandler, error) {
daemonsetsRepo: daemonsetsRepo,
statefulsetsRepo: statefulsetsRepo,
jobsRepo: jobsRepo,
pvcsRepo: pvcsRepo,
}
logsQueryBuilder := logsv3.PrepareLogsQuery
@@ -412,11 +408,6 @@ 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)
@@ -536,9 +527,6 @@ 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)
@@ -645,9 +633,6 @@ func (aH *APIHandler) PopulateTemporality(ctx context.Context, qp *v3.QueryRange
} else {
query.Temporality = v3.Unspecified
}
if len(aH.temporalityMap[query.AggregateAttribute.Key]) > 1 {
query.MultipleTemporalities = true
}
}
// we don't have temporality for this metric
if query.DataSource == v3.DataSourceMetrics && query.Temporality == "" {
@@ -675,9 +660,6 @@ func (aH *APIHandler) PopulateTemporality(ctx context.Context, qp *v3.QueryRange
} else {
query.Temporality = v3.Unspecified
}
if len(nameToTemporality[query.AggregateAttribute.Key]) > 1 {
query.MultipleTemporalities = true
}
aH.temporalityMap[query.AggregateAttribute.Key] = nameToTemporality[query.AggregateAttribute.Key]
}
}
@@ -4910,35 +4892,3 @@ 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)
}

View File

@@ -544,56 +544,3 @@ 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)
}

View File

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

View File

@@ -23,11 +23,10 @@ var (
}
queryNamesForNamespaces = map[string][]string{
"cpu": {"A"},
"memory": {"D"},
"pod_phase": {"H", "I", "J", "K"},
"cpu": {"A"},
"memory": {"D"},
}
namespaceQueryNames = []string{"A", "D", "H", "I", "J", "K"}
namespaceQueryNames = []string{"A", "D"}
attributesKeysForNamespaces = []v3.AttributeKey{
{Key: "k8s_namespace_name"},
@@ -308,19 +307,6 @@ 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]

View File

@@ -17,7 +17,7 @@ import (
var (
metricToUseForNodes = "k8s_node_cpu_utilization"
nodeAttrsToEnrich = []string{"k8s_node_name", "k8s_node_uid", "k8s_cluster_name"}
nodeAttrsToEnrich = []string{"k8s_node_name", "k8s_node_uid"}
k8sNodeUIDAttrKey = "k8s_node_uid"
@@ -27,14 +27,13 @@ var (
"memory": {"C"},
"memory_allocatable": {"D"},
}
nodeQueryNames = []string{"A", "B", "C", "D", "E", "F"}
nodeQueryNames = []string{"A", "B", "C", "D"}
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",
}
)
@@ -326,14 +325,6 @@ 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]

View File

@@ -109,74 +109,6 @@ 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,

View File

@@ -27,7 +27,6 @@ var (
"k8s_daemonset_name",
"k8s_job_name",
"k8s_cronjob_name",
"k8s_cluster_name",
}
k8sPodUIDAttrKey = "k8s_pod_uid"
@@ -40,9 +39,8 @@ 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", "H", "I", "J", "K"}
podQueryNames = []string{"A", "B", "C", "D", "E", "F", "G"}
metricNamesForPods = map[string]string{
"cpu": "k8s_pod_cpu_utilization",
@@ -52,7 +50,6 @@ var (
"memory_request": "k8s_pod_memory_request_utilization",
"memory_limit": "k8s_pod_memory_limit_utilization",
"restarts": "k8s_container_restarts",
"pod_phase": "k8s_pod_phase",
}
)
@@ -368,22 +365,6 @@ 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]

View File

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

View File

@@ -1,378 +0,0 @@
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
}

View File

@@ -1,204 +0,0 @@
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,
}

View File

@@ -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_request": "k8s_pod_cpu_request_utilization",
"cpu_limit": "k8s_pod_cpu_limit_utilization",
"memory": "k8s_pod_memory_usage",
"memory_request": "k8s_pod_memory_request_utilization",
"memory_limit": "k8s_pod_memory_limit_utilization",
"restarts": "k8s_container_restarts",
"cpu": "k8s_pod_cpu_utilization",
"cpu_req": "k8s_pod_cpu_request_utilization",
"cpu_limit": "k8s_pod_cpu_limit_utilization",
"memory": "k8s_pod_memory_usage",
"memory_req": "k8s_pod_memory_request_utilization",
"memory_limit": "k8s_pod_memory_limit_utilization",
"restarts": "k8s_container_restarts",
}
)
@@ -54,7 +54,7 @@ var WorkloadTableListQuery = v3.QueryRangeParamsV3{
Expression: "B",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationAvg,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
// pod cpu limit utilization
@@ -74,7 +74,7 @@ var WorkloadTableListQuery = v3.QueryRangeParamsV3{
Expression: "C",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationAvg,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
// pod memory utilization
@@ -114,7 +114,7 @@ var WorkloadTableListQuery = v3.QueryRangeParamsV3{
Expression: "E",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationAvg,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
// pod memory limit utilization
@@ -134,7 +134,7 @@ var WorkloadTableListQuery = v3.QueryRangeParamsV3{
Expression: "F",
ReduceTo: v3.ReduceToOperatorAvg,
TimeAggregation: v3.TimeAggregationAvg,
SpaceAggregation: v3.SpaceAggregationAvg,
SpaceAggregation: v3.SpaceAggregationSum,
Disabled: false,
},
"G": {

View File

@@ -6,8 +6,6 @@ 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
@@ -20,8 +18,6 @@ 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'
@@ -40,15 +36,13 @@ FROM
consumer_query
ORDER BY
resource_string_service$$name;
`, start, end, tsBucketStart, tsBucketEnd, queueType, topic, partition, consumerGroup, timeRange)
`, start, end, 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
@@ -60,8 +54,6 @@ 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
@@ -76,15 +68,13 @@ FROM
partition_query
ORDER BY
topic;
`, start, end, tsBucketStart, tsBucketEnd, queueType, timeRange)
`, start, end, 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
@@ -97,8 +87,6 @@ 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'
@@ -116,15 +104,14 @@ FROM
consumer_pl
ORDER BY
consumer_group;
`, start, end, tsBucketStart, tsBucketEnd, queueType, topic, partition, timeRange)
`, start, end, queueType, topic, partition, timeRange)
return query
}
// S3, producer overview
func generateProducerPartitionThroughputSQL(start, end int64, queueType string) string {
timeRange := (end - start) / 1000000000
tsBucketStart := (start / 1000000000) - 1800
tsBucketEnd := end / 1000000000 // t, svc, rps, byte*, p99, err
// t, svc, rps, byte*, p99, err
query := fmt.Sprintf(`
WITH producer_latency AS (
SELECT
@@ -137,8 +124,6 @@ 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
@@ -152,15 +137,13 @@ SELECT
COALESCE(total_requests / %d, 0) AS throughput
FROM
producer_latency
`, start, end, tsBucketStart, tsBucketEnd, queueType, timeRange)
`, start, end, 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
@@ -172,8 +155,6 @@ 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'
@@ -188,15 +169,13 @@ SELECT
COALESCE(total_requests / %d, 0) AS throughput
FROM
consumer_latency
`, start, end, tsBucketStart, tsBucketEnd, service, queueType, topic, timeRange)
`, start, end, 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
@@ -210,8 +189,6 @@ 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
@@ -228,15 +205,13 @@ FROM
consumer_latency
ORDER BY
topic;
`, start, end, tsBucketStart, tsBucketEnd, queueType, timeRange, timeRange)
`, start, end, 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
@@ -248,8 +223,6 @@ 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'
@@ -264,7 +237,7 @@ SELECT
COALESCE(total_requests / %d, 0) AS throughput
FROM
consumer_latency
`, start, end, tsBucketStart, tsBucketEnd, service, queueType, topic, timeRange)
`, start, end, service, queueType, topic, timeRange)
return query
}
@@ -320,8 +293,6 @@ 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
@@ -333,8 +304,6 @@ 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'
@@ -351,39 +320,33 @@ FROM
producer_query
ORDER BY
resource_string_service$$name;
`, start, end, tsBucketStart, tsBucketEnd, queueType, topic, partition, timeRange)
`, start, end, 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,
resources_string['service.instance.id'] AS service_instance_id,
attributes_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, tsBucketStart, tsBucketEnd, queueType, consumerGroup, partitionID)
`, timeRange, start, end, 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,
@@ -395,15 +358,11 @@ FROM
signoz_traces.distributed_signoz_index_v3
WHERE
timestamp >= '%d'
AND timestamp <= '%d'
AND ts_bucket_start >= '%d'
AND ts_bucket_start <= '%d';`, queueType, start, end, tsBucketStart, tsBucketEnd)
AND timestamp <= '%d';`, queueType, start, end)
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,
@@ -415,12 +374,10 @@ 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(resources_string, 'service.instance.id'), 1, NULL)) = 0 AS instanceid
COUNT(IF(has(attributes_string, 'service.instance.id'), 1, NULL)) = 0 AS instanceid
FROM signoz_traces.distributed_signoz_index_v3
WHERE
timestamp >= '%d'
AND timestamp <= '%d'
AND ts_bucket_start >= '%d'
AND ts_bucket_start <= '%d' ;`, queueType, start, end, tsBucketStart, tsBucketEnd)
AND timestamp <= '%d';`, queueType, start, end)
return query
}

View File

@@ -228,8 +228,8 @@ func parseColumn(s string) (*string, error) {
return &colName, nil
}
func arrayToMap(fields []model.Field) map[string]model.Field {
res := map[string]model.Field{}
func arrayToMap(fields []model.LogField) map[string]model.LogField {
res := map[string]model.LogField{}
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.Field, interestingFieldLookup map[string]model.Field) (string, error) {
func replaceFieldInToken(queryToken string, selectedFieldsLookup map[string]model.LogField, interestingFieldLookup map[string]model.LogField) (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.Field{}
field := model.LogField{}
if sfield, ok := selectedFieldsLookup[sqlColName]; ok {
field = sfield

View File

@@ -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.Field{
Selected: []model.LogField{
{
Name: "id_key",
DataType: "int64",
Type: "attributes",
},
},
Interesting: []model.Field{
Interesting: []model.LogField{
{
Name: "id.userid",
DataType: "int64",
@@ -326,7 +326,7 @@ func TestCheckIfPrevousPaginateAndModifyOrder(t *testing.T) {
}
var generateSQLQueryFields = model.GetFieldsResponse{
Selected: []model.Field{
Selected: []model.LogField{
{
Name: "field1",
DataType: "int64",
@@ -348,7 +348,7 @@ var generateSQLQueryFields = model.GetFieldsResponse{
Type: "static",
},
},
Interesting: []model.Field{
Interesting: []model.LogField{
{
Name: "FielD1",
DataType: "int64",

View File

@@ -6,7 +6,6 @@ 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 {
@@ -39,36 +38,3 @@ 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
}

View File

@@ -5,73 +5,9 @@ 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) {

View File

@@ -5,7 +5,6 @@ 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"
@@ -336,10 +335,6 @@ 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 ||

View File

@@ -20,16 +20,11 @@ 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)" +
valueFilter +
" WHERE isNaN(per_series_value) = 0" +
" GROUP BY %s" +
" ORDER BY %s"

View File

@@ -190,16 +190,11 @@ 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)" +
valueFilter +
" WHERE isNaN(per_series_value) = 0" +
" GROUP BY %s" +
" ORDER BY %s"

View File

@@ -25,16 +25,11 @@ 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)" +
valueFilter +
" WHERE isNaN(per_series_value) = 0" +
" GROUP BY %s" +
" ORDER BY %s"

View File

@@ -142,16 +142,11 @@ 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)" +
valueFilter +
" WHERE isNaN(per_series_value) = 0" +
" GROUP BY %s" +
" ORDER BY %s"

View File

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

View File

@@ -4,7 +4,6 @@ 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"
@@ -20,9 +19,6 @@ 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

View File

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

View File

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

View File

@@ -177,23 +177,7 @@ func (q *querier) runBuilderQueries(ctx context.Context, params *v3.QueryRangePa
for queryName, builderQuery := range params.CompositeQuery.BuilderQueries {
if queryName == builderQuery.Expression {
wg.Add(1)
if builderQuery.MultipleTemporalities == true {
go func() {
temporalitySwitches, err := q.reader.GetTemporalitySwitchPoints(ctx, builderQuery.AggregateAttribute.Key, params.Start, params.End)
if err != nil {
ch <- channelResult{Err: err, Name: queryName}
return
}
if len(temporalitySwitches) == 0 {
q.runBuilderQuery(ctx, builderQuery, params, cacheKeys, ch, &wg)
} else {
q.handleTemporalitySwitches(ctx, temporalitySwitches, &wg, builderQuery, params, cacheKeys, ch, queryName)
}
}()
} else {
go q.runBuilderQuery(ctx, builderQuery, params, cacheKeys, ch, &wg)
}
go q.runBuilderQuery(ctx, builderQuery, params, cacheKeys, ch, &wg)
}
}
@@ -225,58 +209,6 @@ func (q *querier) runBuilderQueries(ctx context.Context, params *v3.QueryRangePa
return results, errQueriesByName, err
}
func (q *querier) handleTemporalitySwitches(ctx context.Context, temporalitySwitches []v3.TemporalityChangePoint, wg *sync.WaitGroup, builderQuery *v3.BuilderQuery, params *v3.QueryRangeParamsV3, cacheKeys map[string]string, ch chan channelResult, queryName string) {
defer wg.Done()
tempCh := make(chan channelResult, len(temporalitySwitches)+1)
var tempWg sync.WaitGroup
// Handle each segment between switch points
for i := 0; i <= len(temporalitySwitches); i++ {
tempWg.Add(1)
go func(idx int) {
queryWithTemporality := *builderQuery
queryParams := *params
if i == 0 {
queryParams.End = temporalitySwitches[idx].Timestamp
queryWithTemporality.Temporality = temporalitySwitches[idx].FromTemporality
} else if idx < len(temporalitySwitches) {
queryParams.Start = temporalitySwitches[idx-1].Timestamp
queryParams.End = temporalitySwitches[idx].Timestamp
queryWithTemporality.Temporality = temporalitySwitches[idx].FromTemporality
queryWithTemporality.ShiftBy = 0
} else if idx == len(temporalitySwitches) {
queryParams.Start = temporalitySwitches[idx-1].Timestamp
queryParams.End = params.End
queryWithTemporality.Temporality = temporalitySwitches[idx-1].ToTemporality
}
q.runBuilderQuery(ctx, &queryWithTemporality, &queryParams, cacheKeys, tempCh, &tempWg)
}(i)
}
// Wait for all temporal queries to complete
tempWg.Wait()
close(tempCh)
// Combine results from all temporal queries
var combinedSeries []*v3.Series
var lastErr error
for result := range tempCh {
if result.Err != nil {
lastErr = result.Err
continue
}
combinedSeries = append(combinedSeries, result.Series...)
}
ch <- channelResult{
Series: combinedSeries,
Err: lastErr,
Name: queryName,
}
}
func (q *querier) runPromQueries(ctx context.Context, params *v3.QueryRangeParamsV3) ([]*v3.Result, map[string]error, error) {
channelResults := make(chan channelResult, len(params.CompositeQuery.PromQueries))
var wg sync.WaitGroup

View File

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

View File

@@ -74,19 +74,6 @@ 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
@@ -123,7 +110,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 := existsSubQueryForFixedColumn(item.Key, item.Operator)
subQuery, err := tracesV3.ExistsSubQueryForFixedColumn(item.Key, item.Operator)
if err != nil {
return "", err
}
@@ -325,7 +312,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)) + "#LIMIT_PLACEHOLDER)"
filterSubQuery = filterSubQuery + " AND " + fmt.Sprintf("(%s) GLOBAL IN (", tracesV3.GetSelectKeys(mq.AggregateOperator, mq.GroupBy)) + "%s)"
}
switch mq.AggregateOperator {
@@ -363,7 +350,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 := existsSubQueryForFixedColumn(mq.AggregateAttribute, v3.FilterOperatorExists)
subQuery, err := tracesV3.ExistsSubQueryForFixedColumn(mq.AggregateAttribute, v3.FilterOperatorExists)
if err == nil {
filterSubQuery = fmt.Sprintf("%s AND %s", filterSubQuery, subQuery)
}

View File

@@ -265,11 +265,9 @@ 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 `attribute_string_path` = '' AND http_url = '' AND `attribute_string_http$$route` = ''",
want: "mapContains(attributes_string, 'host') AND mapContains(attributes_number, 'duration') AND NOT mapContains(attributes_bool, 'isDone') AND NOT mapContains(attributes_string, 'host1') AND path = ''",
},
}
for _, tt := range tests {
@@ -685,7 +683,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 (#LIMIT_PLACEHOLDER) 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 (%s) group by `function` order by value DESC",
},
{
name: "test with limit with resources- first",
@@ -768,7 +766,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 (#LIMIT_PLACEHOLDER) group by `function`,`serviceName` order by value DESC",
"AND simpleJSONExtractString(labels, 'hostname') = 'server1' AND labels like '%hostname%server1%')) AND (`function`,`serviceName`) GLOBAL IN (%s) group by `function`,`serviceName` order by value DESC",
},
}

View File

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

View File

@@ -109,12 +109,6 @@ 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
GetTemporalitySwitchPoints(ctx context.Context, metricName string, startTime int64, endTime int64) ([]v3.TemporalityChangePoint, error)
}
type Querier interface {

View File

@@ -151,20 +151,13 @@ 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"`
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"`
NodeUID string `json:"nodeUID,omitempty"`
NodeCPUUsage float64 `json:"nodeCPUUsage"`
NodeCPUAllocatable float64 `json:"nodeCPUAllocatable"`
NodeMemoryUsage float64 `json:"nodeMemoryUsage"`
NodeMemoryAllocatable float64 `json:"nodeMemoryAllocatable"`
Meta map[string]string `json:"meta"`
}
type NamespaceListRequest struct {
@@ -187,7 +180,6 @@ 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"`
}
@@ -337,30 +329,3 @@ 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"`
}

View File

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

View File

@@ -770,47 +770,32 @@ 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"`
DataSource DataSource `json:"dataSource"`
AggregateOperator AggregateOperator `json:"aggregateOperator"`
AggregateAttribute AttributeKey `json:"aggregateAttribute,omitempty"`
Temporality Temporality `json:"temporality,omitempty"`
Filters *FilterSet `json:"filters,omitempty"`
GroupBy []AttributeKey `json:"groupBy,omitempty"`
Expression string `json:"expression"`
Disabled bool `json:"disabled"`
Having []Having `json:"having,omitempty"`
Legend string `json:"legend,omitempty"`
Limit uint64 `json:"limit"`
Offset uint64 `json:"offset"`
PageSize uint64 `json:"pageSize"`
OrderBy []OrderBy `json:"orderBy,omitempty"`
ReduceTo ReduceToOperator `json:"reduceTo,omitempty"`
SelectColumns []AttributeKey `json:"selectColumns,omitempty"`
TimeAggregation TimeAggregation `json:"timeAggregation,omitempty"`
SpaceAggregation SpaceAggregation `json:"spaceAggregation,omitempty"`
Functions []Function `json:"functions,omitempty"`
ShiftBy int64
IsAnomaly bool
QueriesUsedInFormula []string
MetricTableHints *MetricTableHints `json:"-"`
MetricValueFilter *MetricValueFilter `json:"-"`
MultipleTemporalities bool
QueryName string `json:"queryName"`
StepInterval int64 `json:"stepInterval"`
DataSource DataSource `json:"dataSource"`
AggregateOperator AggregateOperator `json:"aggregateOperator"`
AggregateAttribute AttributeKey `json:"aggregateAttribute,omitempty"`
Temporality Temporality `json:"temporality,omitempty"`
Filters *FilterSet `json:"filters,omitempty"`
GroupBy []AttributeKey `json:"groupBy,omitempty"`
Expression string `json:"expression"`
Disabled bool `json:"disabled"`
Having []Having `json:"having,omitempty"`
Legend string `json:"legend,omitempty"`
Limit uint64 `json:"limit"`
Offset uint64 `json:"offset"`
PageSize uint64 `json:"pageSize"`
OrderBy []OrderBy `json:"orderBy,omitempty"`
ReduceTo ReduceToOperator `json:"reduceTo,omitempty"`
SelectColumns []AttributeKey `json:"selectColumns,omitempty"`
TimeAggregation TimeAggregation `json:"timeAggregation,omitempty"`
SpaceAggregation SpaceAggregation `json:"spaceAggregation,omitempty"`
Functions []Function `json:"functions,omitempty"`
ShiftBy int64
IsAnomaly bool
QueriesUsedInFormula []string
MetricTableHints *MetricTableHints `json:"-"`
}
func (b *BuilderQuery) SetShiftByFromFunc() {
@@ -874,7 +859,6 @@ func (b *BuilderQuery) Clone() *BuilderQuery {
ShiftBy: b.ShiftBy,
IsAnomaly: b.IsAnomaly,
QueriesUsedInFormula: b.QueriesUsedInFormula,
MetricValueFilter: b.MetricValueFilter.Clone(),
}
}
@@ -1377,9 +1361,3 @@ type QBOptions struct {
IsLivetailQuery bool
PreferRPM bool
}
type TemporalityChangePoint struct {
Timestamp int64 `json:"timestamp"`
FromTemporality Temporality `json:"from_temporality"`
ToTemporality Temporality `json:"to_temporality"`
}