Skip to content

Commit

Permalink
Revert "chore: add related values (#6619)" (#7067)
Browse files Browse the repository at this point in the history 
This reverts commit c5219ac.
  • Loading branch information
@vikrantgupta25
vikrantgupta25 authored Feb 7, 2025
1 parent 3987600 commit 12f65f4
Show file tree
Hide file tree
Showing 5 changed files with 7 additions and 218 deletions.
9 changes: 0 additions & 9 deletions pkg/query-service/app/clickhouseReader/options.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -48,9 +48,6 @@ const (
defaultTraceLocalTableName string = "signoz_index_v3"
defaultTraceResourceTableV3 string = "distributed_traces_v3_resource"
defaultTraceSummaryTable string = "distributed_trace_summary"

defaultMetadataDB string = "signoz_metadata"
defaultMetadataTable string = "distributed_attributes_metadata"
)

// NamespaceConfig is Clickhouse's internal configuration data
Expand Down Expand Up @@ -91,9 +88,6 @@ type namespaceConfig struct {
TraceLocalTableNameV3 string
TraceResourceTableV3 string
TraceSummaryTable string

MetadataDB string
MetadataTable string
}

// Connecto defines how to connect to the database
Expand Down Expand Up @@ -147,9 +141,6 @@ func NewOptions(
TraceLocalTableNameV3: defaultTraceLocalTableName,
TraceResourceTableV3: defaultTraceResourceTableV3,
TraceSummaryTable: defaultTraceSummaryTable,

MetadataDB: defaultMetadataDB,
MetadataTable: defaultMetadataTable,
},
others: make(map[string]*namespaceConfig, len(otherNamespaces)),
}
Expand Down
119 changes: 7 additions & 112 deletions pkg/query-service/app/clickhouseReader/reader.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -162,8 +162,6 @@ type ClickHouseReader struct {

fluxIntervalForTraceDetail time.Duration
cache cache.Cache
metadataDB string
metadataTable string
}

// NewTraceReader returns a TraceReader for the database
Expand Down Expand Up @@ -259,8 +257,6 @@ func NewReaderFromClickhouseConnection(

fluxIntervalForTraceDetail: fluxIntervalForTraceDetail,
cache: cache,
metadataDB: options.primary.MetadataDB,
metadataTable: options.primary.MetadataTable,
}
}

Expand Down Expand Up @@ -1459,7 +1455,7 @@ func (r *ClickHouseReader) GetWaterfallSpansForTraceWithMetadata(ctx context.Con
var serviceNameIntervalMap = map[string][]tracedetail.Interval{}
var hasMissingSpans bool

userEmail, emailErr := auth.GetEmailFromJwt(ctx)
userEmail , emailErr := auth.GetEmailFromJwt(ctx)
cachedTraceData, err := r.GetWaterfallSpansForTraceWithMetadataCache(ctx, traceID)
if err == nil {
startTime = cachedTraceData.StartTime
Expand Down Expand Up @@ -1535,8 +1531,8 @@ func (r *ClickHouseReader) GetWaterfallSpansForTraceWithMetadata(ctx context.Con
if startTime == 0 || startTimeUnixNano < startTime {
startTime = startTimeUnixNano
}
if endTime == 0 || (startTimeUnixNano+jsonItem.DurationNano) > endTime {
endTime = (startTimeUnixNano + jsonItem.DurationNano)
if endTime == 0 || (startTimeUnixNano + jsonItem.DurationNano ) > endTime {
endTime = (startTimeUnixNano + jsonItem.DurationNano )
}
if durationNano == 0 || jsonItem.DurationNano > durationNano {
durationNano = jsonItem.DurationNano
Expand Down Expand Up @@ -1713,12 +1709,12 @@ func (r *ClickHouseReader) GetFlamegraphSpansForTrace(ctx context.Context, trace
}

// metadata calculation
startTimeUnixNano := uint64(item.TimeUnixNano.UnixNano())
startTimeUnixNano := uint64(item.TimeUnixNano.UnixNano())
if startTime == 0 || startTimeUnixNano < startTime {
startTime = startTimeUnixNano
}
if endTime == 0 || (startTimeUnixNano+jsonItem.DurationNano) > endTime {
endTime = (startTimeUnixNano + jsonItem.DurationNano)
if endTime == 0 || ( startTimeUnixNano + jsonItem.DurationNano ) > endTime {
endTime = (startTimeUnixNano + jsonItem.DurationNano )
}
if durationNano == 0 || jsonItem.DurationNano > durationNano {
durationNano = jsonItem.DurationNano
Expand Down Expand Up @@ -1782,7 +1778,7 @@ func (r *ClickHouseReader) GetFlamegraphSpansForTrace(ctx context.Context, trace

trace.Spans = selectedSpansForRequest
trace.StartTimestampMillis = startTime / 1000000
trace.EndTimestampMillis = endTime / 1000000
trace.EndTimestampMillis = endTime / 1000000
return trace, nil
}

Expand Down Expand Up @@ -4140,97 +4136,6 @@ func (r *ClickHouseReader) GetLogAttributeKeys(ctx context.Context, req *v3.Filt
return &response, nil
}

func (r *ClickHouseReader) FetchRelatedValues(ctx context.Context, req *v3.FilterAttributeValueRequest) ([]string, error) {
var andConditions []string

andConditions = append(andConditions, fmt.Sprintf("unix_milli >= %d", req.StartTimeMillis))
andConditions = append(andConditions, fmt.Sprintf("unix_milli <= %d", req.EndTimeMillis))

if len(req.ExistingFilterItems) != 0 {
for _, item := range req.ExistingFilterItems {
// we only support string for related values
if item.Key.DataType != v3.AttributeKeyDataTypeString {
continue
}

var colName string
switch item.Key.Type {
case v3.AttributeKeyTypeResource:
colName = "resource_attributes"
case v3.AttributeKeyTypeTag:
colName = "attributes"
default:
// we only support resource and tag for related values as of now
continue
}
// IN doesn't make use of map value index, we convert it to = or !=
operator := item.Operator
if v3.FilterOperator(strings.ToLower(string(item.Operator))) == v3.FilterOperatorIn {
operator = "="
} else if v3.FilterOperator(strings.ToLower(string(item.Operator))) == v3.FilterOperatorNotIn {
operator = "!="
}
addCondition := func(val string) {
andConditions = append(andConditions, fmt.Sprintf("mapContains(%s, '%s') AND %s['%s'] %s %s", colName, item.Key.Key, colName, item.Key.Key, operator, val))
}
switch v := item.Value.(type) {
case string:
fmtVal := utils.ClickHouseFormattedValue(v)
addCondition(fmtVal)
case []string:
for _, val := range v {
fmtVal := utils.ClickHouseFormattedValue(val)
addCondition(fmtVal)
}
case []interface{}:
for _, val := range v {
fmtVal := utils.ClickHouseFormattedValue(val)
addCondition(fmtVal)
}
}
}
}
whereClause := strings.Join(andConditions, " AND ")

var selectColumn string
switch req.TagType {
case v3.TagTypeResource:
selectColumn = "resource_attributes" + "['" + req.FilterAttributeKey + "']"
case v3.TagTypeTag:
selectColumn = "attributes" + "['" + req.FilterAttributeKey + "']"
default:
selectColumn = "attributes" + "['" + req.FilterAttributeKey + "']"
}

filterSubQuery := fmt.Sprintf(
"SELECT DISTINCT %s FROM %s.%s WHERE %s LIMIT 100",
selectColumn,
r.metadataDB,
r.metadataTable,
whereClause,
)
zap.L().Debug("filterSubQuery for related values", zap.String("query", filterSubQuery))

rows, err := r.db.Query(ctx, filterSubQuery)
if err != nil {
return nil, fmt.Errorf("error while executing query: %s", err.Error())
}
defer rows.Close()

var attributeValues []string
for rows.Next() {
var value string
if err := rows.Scan(&value); err != nil {
return nil, fmt.Errorf("error while scanning rows: %s", err.Error())
}
if value != "" {
attributeValues = append(attributeValues, value)
}
}

return attributeValues, nil
}

func (r *ClickHouseReader) GetLogAttributeValues(ctx context.Context, req *v3.FilterAttributeValueRequest) (*v3.FilterAttributeValueResponse, error) {
var err error
var filterValueColumn string
Expand Down Expand Up @@ -4332,11 +4237,6 @@ func (r *ClickHouseReader) GetLogAttributeValues(ctx context.Context, req *v3.Fi
}
}

relatedValues, _ := r.FetchRelatedValues(ctx, req)
attributeValues.RelatedValues = &v3.FilterAttributeValueResponse{
StringAttributeValues: relatedValues,
}

return &attributeValues, nil

}
Expand Down Expand Up @@ -5017,11 +4917,6 @@ func (r *ClickHouseReader) GetTraceAttributeValues(ctx context.Context, req *v3.
}
}

relatedValues, _ := r.FetchRelatedValues(ctx, req)
attributeValues.RelatedValues = &v3.FilterAttributeValueResponse{
StringAttributeValues: relatedValues,
}

return &attributeValues, nil
}

Expand Down
30 changes: 0 additions & 30 deletions pkg/query-service/app/http_handler.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -380,7 +380,6 @@ func (aH *APIHandler) RegisterQueryRangeV3Routes(router *mux.Router, am *AuthMid
withCacheControl(AutoCompleteCacheControlAge, aH.autoCompleteAttributeKeys))).Methods(http.MethodGet)
subRouter.HandleFunc("/autocomplete/attribute_values", am.ViewAccess(
withCacheControl(AutoCompleteCacheControlAge, aH.autoCompleteAttributeValues))).Methods(http.MethodGet)
subRouter.HandleFunc("/autocomplete/attribute_values", am.ViewAccess(aH.autoCompleteAttributeValuesPost)).Methods(http.MethodPost)
subRouter.HandleFunc("/query_range", am.ViewAccess(aH.QueryRangeV3)).Methods(http.MethodPost)
subRouter.HandleFunc("/query_range/format", am.ViewAccess(aH.QueryRangeV3Format)).Methods(http.MethodPost)

Expand Down Expand Up @@ -4816,35 +4815,6 @@ func (aH *APIHandler) autoCompleteAttributeValues(w http.ResponseWriter, r *http
aH.Respond(w, response)
}

func (aH *APIHandler) autoCompleteAttributeValuesPost(w http.ResponseWriter, r *http.Request) {
var response *v3.FilterAttributeValueResponse
req, err := parseFilterAttributeValueRequestBody(r)

if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, nil)
return
}

switch req.DataSource {
case v3.DataSourceMetrics:
response, err = aH.reader.GetMetricAttributeValues(r.Context(), req)
case v3.DataSourceLogs:
response, err = aH.reader.GetLogAttributeValues(r.Context(), req)
case v3.DataSourceTraces:
response, err = aH.reader.GetTraceAttributeValues(r.Context(), req)
default:
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: fmt.Errorf("invalid data source")}, nil)
return
}

if err != nil {
RespondError(w, &model.ApiError{Typ: model.ErrorBadData, Err: err}, nil)
return
}

aH.Respond(w, response)
}

func (aH *APIHandler) getSpanKeysV3(ctx context.Context, queryRangeParams *v3.QueryRangeParamsV3) (map[string]v3.AttributeKey, error) {
data := map[string]v3.AttributeKey{}
for _, query := range queryRangeParams.CompositeQuery.BuilderQueries {
Expand Down
19 changes: 0 additions & 19 deletions pkg/query-service/app/parser.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -769,25 +769,6 @@ func parseFilterAttributeValueRequest(r *http.Request) (*v3.FilterAttributeValue
return &req, nil
}

func parseFilterAttributeValueRequestBody(r *http.Request) (*v3.FilterAttributeValueRequest, error) {

var req v3.FilterAttributeValueRequest

if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
return nil, err
}

if err := req.Validate(); err != nil {
return nil, err
}

// offset by two windows periods for start for better results
req.StartTimeMillis = req.StartTimeMillis - time.Hour.Milliseconds()*6*2
req.EndTimeMillis = req.EndTimeMillis + time.Hour.Milliseconds()*6

return &req, nil
}

func validateQueryRangeParamsV3(qp *v3.QueryRangeParamsV3) error {
err := qp.CompositeQuery.Validate()
if err != nil {
Expand Down
48 changes: 0 additions & 48 deletions pkg/query-service/model/v3/v3.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -297,8 +297,6 @@ func (q AttributeKeyDataType) String() string {
// for a selected aggregate operator, aggregate attribute, filter attribute key
// and search text.
type FilterAttributeValueRequest struct {
StartTimeMillis int64 `json:"startTimeMillis"`
EndTimeMillis int64 `json:"endTimeMillis"`
DataSource DataSource `json:"dataSource"`
AggregateOperator AggregateOperator `json:"aggregateOperator"`
AggregateAttribute string `json:"aggregateAttribute"`
Expand All @@ -307,50 +305,6 @@ type FilterAttributeValueRequest struct {
TagType TagType `json:"tagType"`
SearchText string `json:"searchText"`
Limit int `json:"limit"`
ExistingFilterItems []FilterItem `json:"existingFilterItems"`
MetricNames []string `json:"metricNames"`
}

func (f *FilterAttributeValueRequest) Validate() error {
if f.FilterAttributeKey == "" {
return fmt.Errorf("filterAttributeKey is required")
}

if f.StartTimeMillis == 0 {
return fmt.Errorf("startTimeMillis is required")
}

if f.EndTimeMillis == 0 {
return fmt.Errorf("endTimeMillis is required")
}

if f.Limit == 0 {
f.Limit = 100
}

if f.Limit > 1000 {
return fmt.Errorf("limit must be less than 1000")
}

if f.ExistingFilterItems != nil {
for _, value := range f.ExistingFilterItems {
if value.Key.Key == "" {
return fmt.Errorf("existingFilterItems must contain a valid key")
}
}
}

if err := f.DataSource.Validate(); err != nil {
return fmt.Errorf("invalid data source: %w", err)
}

if f.DataSource != DataSourceMetrics {
if err := f.AggregateOperator.Validate(); err != nil {
return fmt.Errorf("invalid aggregate operator: %w", err)
}
}

return nil
}

type AggregateAttributeResponse struct {
Expand Down Expand Up @@ -415,8 +369,6 @@ type FilterAttributeValueResponse struct {
StringAttributeValues []string `json:"stringAttributeValues"`
NumberAttributeValues []interface{} `json:"numberAttributeValues"`
BoolAttributeValues []bool `json:"boolAttributeValues"`

RelatedValues *FilterAttributeValueResponse `json:"relatedValues"`
}

type QueryRangeParamsV3 struct {
Expand Down

0 comments on commit 12f65f4

Please sign in to comment.