Merge branch 'develop' into fix/ingestion-dashboard

deploy/docker/clickhouse-setup/docker-compose-core.yaml

@@ -72,7 +72,9 @@ services:
     image: signoz/signoz-schema-migrator:${OTELCOL_TAG:-0.111.14}
     container_name: otel-migrator
     command:
+      - "sync" 
       - "--dsn=tcp://clickhouse:9000"
+      - "--up=" 
     depends_on:
       clickhouse:
         condition: service_healthy

frontend/src/container/MetricsApplication/Tabs/Overview.tsx

@@ -145,41 +145,49 @@ function Application(): JSX.Element {
 		[servicename, topLevelOperations],
 	);
 
-	const operationPerSecWidget = getWidgetQueryBuilder({
-		query: {
-			queryType: EQueryType.QUERY_BUILDER,
-			promql: [],
-			builder: operationPerSec({
-				servicename,
-				tagFilterItems,
-				topLevelOperations: topLevelOperationsRoute,
+	const operationPerSecWidget = useMemo(
+		() =>
+			getWidgetQueryBuilder({
+				query: {
+					queryType: EQueryType.QUERY_BUILDER,
+					promql: [],
+					builder: operationPerSec({
+						servicename,
+						tagFilterItems,
+						topLevelOperations: topLevelOperationsRoute,
+					}),
+					clickhouse_sql: [],
+					id: uuid(),
+				},
+				title: GraphTitle.RATE_PER_OPS,
+				panelTypes: PANEL_TYPES.TIME_SERIES,
+				yAxisUnit: 'ops',
+				id: SERVICE_CHART_ID.rps,
 			}),
-			clickhouse_sql: [],
-			id: uuid(),
-		},
-		title: GraphTitle.RATE_PER_OPS,
-		panelTypes: PANEL_TYPES.TIME_SERIES,
-		yAxisUnit: 'ops',
-		id: SERVICE_CHART_ID.rps,
-	});
+		[servicename, tagFilterItems, topLevelOperationsRoute],
+	);
 
-	const errorPercentageWidget = getWidgetQueryBuilder({
-		query: {
-			queryType: EQueryType.QUERY_BUILDER,
-			promql: [],
-			builder: errorPercentage({
-				servicename,
-				tagFilterItems,
-				topLevelOperations: topLevelOperationsRoute,
+	const errorPercentageWidget = useMemo(
+		() =>
+			getWidgetQueryBuilder({
+				query: {
+					queryType: EQueryType.QUERY_BUILDER,
+					promql: [],
+					builder: errorPercentage({
+						servicename,
+						tagFilterItems,
+						topLevelOperations: topLevelOperationsRoute,
+					}),
+					clickhouse_sql: [],
+					id: uuid(),
+				},
+				title: GraphTitle.ERROR_PERCENTAGE,
+				panelTypes: PANEL_TYPES.TIME_SERIES,
+				yAxisUnit: '%',
+				id: SERVICE_CHART_ID.errorPercentage,
 			}),
-			clickhouse_sql: [],
-			id: uuid(),
-		},
-		title: GraphTitle.ERROR_PERCENTAGE,
-		panelTypes: PANEL_TYPES.TIME_SERIES,
-		yAxisUnit: '%',
-		id: SERVICE_CHART_ID.errorPercentage,
-	});
+		[servicename, tagFilterItems, topLevelOperationsRoute],
+	);
 
 	const stepInterval = useMemo(
 		() =>

frontend/src/container/MetricsApplication/Tabs/Overview/ServiceOverview.tsx

@@ -53,24 +53,28 @@ function ServiceOverview({
 		[isSpanMetricEnable, queries],
 	);
 
-	const latencyWidget = getWidgetQueryBuilder({
-		query: {
-			queryType: EQueryType.QUERY_BUILDER,
-			promql: [],
-			builder: latency({
-				servicename,
-				tagFilterItems,
-				isSpanMetricEnable,
-				topLevelOperationsRoute,
+	const latencyWidget = useMemo(
+		() =>
+			getWidgetQueryBuilder({
+				query: {
+					queryType: EQueryType.QUERY_BUILDER,
+					promql: [],
+					builder: latency({
+						servicename,
+						tagFilterItems,
+						isSpanMetricEnable,
+						topLevelOperationsRoute,
+					}),
+					clickhouse_sql: [],
+					id: uuid(),
+				},
+				title: GraphTitle.LATENCY,
+				panelTypes: PANEL_TYPES.TIME_SERIES,
+				yAxisUnit: 'ns',
+				id: SERVICE_CHART_ID.latency,
 			}),
-			clickhouse_sql: [],
-			id: uuid(),
-		},
-		title: GraphTitle.LATENCY,
-		panelTypes: PANEL_TYPES.TIME_SERIES,
-		yAxisUnit: 'ns',
-		id: SERVICE_CHART_ID.latency,
-	});
+		[isSpanMetricEnable, servicename, tagFilterItems, topLevelOperationsRoute],
+	);
 
 	const isQueryEnabled =
 		!topLevelOperationsIsLoading && topLevelOperationsRoute.length > 0;

frontend/src/container/SideNav/SideNav.tsx

@@ -16,7 +16,7 @@ import history from 'lib/history';
 import {
 	AlertTriangle,
 	CheckSquare,
-	RocketIcon,
+	PackagePlus,
 	UserCircle,
 } from 'lucide-react';
 import { MouseEvent, useCallback, useEffect, useMemo, useState } from 'react';
@@ -395,9 +395,9 @@ function SideNav({
 								onClickGetStarted(event);
 							}}
 						>
-							<RocketIcon size={16} />
+							<PackagePlus size={16} />
 
-							<div className="license tag nav-item-label"> Get Started </div>
+							<div className="license tag nav-item-label"> New source </div>
 						</Button>
 					</div>
 				)}

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

@@ -1203,7 +1203,9 @@ func (r *ClickHouseReader) GetUsage(ctx context.Context, queryParams *model.GetU
 	return &usageItems, nil
 }
 
-func (r *ClickHouseReader) SearchTracesV2(ctx context.Context, params *model.SearchTracesParams) (*[]model.SearchSpansResult, error) {
+func (r *ClickHouseReader) SearchTracesV2(ctx context.Context, params *model.SearchTracesParams,
+	smartTraceAlgorithm func(payload []model.SearchSpanResponseItem, targetSpanId string,
+		levelUp int, levelDown int, spanLimit int) ([]model.SearchSpansResult, error)) (*[]model.SearchSpansResult, error) {
 	searchSpansResult := []model.SearchSpansResult{
 		{
 			Columns:   []string{"__time", "SpanId", "TraceId", "ServiceName", "Name", "Kind", "DurationNano", "TagsKeys", "TagsValues", "References", "Events", "HasError", "StatusMessage", "StatusCodeString", "SpanKind"},
@@ -1318,9 +1320,29 @@ func (r *ClickHouseReader) SearchTracesV2(ctx context.Context, params *model.Sea
 	end = time.Now()
 	zap.L().Debug("getTraceSQLQuery unmarshal took: ", zap.Duration("duration", end.Sub(start)))
 
-	for i, item := range searchSpanResponses {
-		spanEvents := item.GetValues()
-		searchSpansResult[0].Events[i] = spanEvents
+	err = r.featureFlags.CheckFeature(model.SmartTraceDetail)
+	smartAlgoEnabled := err == nil
+	if len(searchScanResponses) > params.SpansRenderLimit && smartAlgoEnabled {
+		start = time.Now()
+		searchSpansResult, err = smartTraceAlgorithm(searchSpanResponses, params.SpanID, params.LevelUp, params.LevelDown, params.SpansRenderLimit)
+		if err != nil {
+			return nil, err
+		}
+		end = time.Now()
+		zap.L().Debug("smartTraceAlgo took: ", zap.Duration("duration", end.Sub(start)))
+		userEmail, err := auth.GetEmailFromJwt(ctx)
+		if err == nil {
+			data := map[string]interface{}{
+				"traceSize":        len(searchScanResponses),
+				"spansRenderLimit": params.SpansRenderLimit,
+			}
+			telemetry.GetInstance().SendEvent(telemetry.TELEMETRY_EVENT_LARGE_TRACE_OPENED, data, userEmail, true, false)
+		}
+	} else {
+		for i, item := range searchSpanResponses {
+			spanEvents := item.GetValues()
+			searchSpansResult[0].Events[i] = spanEvents
+		}
 	}
 
 	searchSpansResult[0].StartTimestampMillis = startTime - (durationNano / 1000000)
@@ -1334,7 +1356,7 @@ func (r *ClickHouseReader) SearchTraces(ctx context.Context, params *model.Searc
 		levelUp int, levelDown int, spanLimit int) ([]model.SearchSpansResult, error)) (*[]model.SearchSpansResult, error) {
 
 	if r.useTraceNewSchema {
-		return r.SearchTracesV2(ctx, params)
+		return r.SearchTracesV2(ctx, params, smartTraceAlgorithm)
 	}
 
 	var countSpans uint64

pkg/query-service/app/http_handler.go

@@ -3141,14 +3141,14 @@ func (aH *APIHandler) getProducerThroughputOverview(
 		Hash: make(map[string]struct{}),
 	}
 
-	queryRangeParams, err := mq.BuildQRParamsWithCache(messagingQueue, "producer-throughput-overview", attributeCache)
+	producerQueryRangeParams, err := mq.BuildQRParamsWithCache(messagingQueue, "producer-throughput-overview", attributeCache)
 	if err != nil {
 		zap.L().Error(err.Error())
 		RespondError(w, apiErr, nil)
 		return
 	}
 
-	if err := validateQueryRangeParamsV3(queryRangeParams); err != nil {
+	if err := validateQueryRangeParamsV3(producerQueryRangeParams); err != nil {
 		zap.L().Error(err.Error())
 		RespondError(w, apiErr, nil)
 		return
@@ -3157,29 +3157,29 @@ func (aH *APIHandler) getProducerThroughputOverview(
 	var result []*v3.Result
 	var errQuriesByName map[string]error
 
-	result, errQuriesByName, err = aH.querierV2.QueryRange(r.Context(), queryRangeParams)
+	result, errQuriesByName, err = aH.querierV2.QueryRange(r.Context(), producerQueryRangeParams)
 	if err != nil {
 		apiErrObj := &model.ApiError{Typ: model.ErrorBadData, Err: err}
 		RespondError(w, apiErrObj, errQuriesByName)
 		return
 	}
 
 	for _, res := range result {
-		for _, list := range res.List {
-			serviceName, serviceNameOk := list.Data["service_name"].(*string)
-			topicName, topicNameOk := list.Data["topic"].(*string)
-			params := []string{*serviceName, *topicName}
+		for _, series := range res.Series {
+			serviceName, serviceNameOk := series.Labels["service_name"]
+			topicName, topicNameOk := series.Labels["topic"]
+			params := []string{serviceName, topicName}
 			hashKey := uniqueIdentifier(params, "#")
 			_, ok := attributeCache.Hash[hashKey]
 			if topicNameOk && serviceNameOk && !ok {
 				attributeCache.Hash[hashKey] = struct{}{}
-				attributeCache.TopicName = append(attributeCache.TopicName, *topicName)
-				attributeCache.ServiceName = append(attributeCache.ServiceName, *serviceName)
+				attributeCache.TopicName = append(attributeCache.TopicName, topicName)
+				attributeCache.ServiceName = append(attributeCache.ServiceName, serviceName)
 			}
 		}
 	}
 
-	queryRangeParams, err = mq.BuildQRParamsWithCache(messagingQueue, "producer-throughput-overview-latency", attributeCache)
+	queryRangeParams, err := mq.BuildQRParamsWithCache(messagingQueue, "producer-throughput-overview-byte-rate", attributeCache)
 	if err != nil {
 		zap.L().Error(err.Error())
 		RespondError(w, apiErr, nil)
@@ -3198,26 +3198,32 @@ func (aH *APIHandler) getProducerThroughputOverview(
 		return
 	}
 
-	latencyColumn := &v3.Result{QueryName: "latency"}
-	var latencySeries []*v3.Row
+	byteRateColumn := &v3.Result{QueryName: "byte_rate"}
+	var byteRateSeries []*v3.Series
 	for _, res := range resultFetchLatency {
-		for _, list := range res.List {
-			topic, topicOk := list.Data["topic"].(*string)
-			serviceName, serviceNameOk := list.Data["service_name"].(*string)
-			params := []string{*serviceName, *topic}
+		for _, series := range res.Series {
+			topic, topicOk := series.Labels["topic"]
+			serviceName, serviceNameOk := series.Labels["service_name"]
+			params := []string{serviceName, topic}
 			hashKey := uniqueIdentifier(params, "#")
 			_, ok := attributeCache.Hash[hashKey]
 			if topicOk && serviceNameOk && ok {
-				latencySeries = append(latencySeries, list)
+				byteRateSeries = append(byteRateSeries, series)
 			}
 		}
 	}
 
-	latencyColumn.List = latencySeries
-	result = append(result, latencyColumn)
+	byteRateColumn.Series = byteRateSeries
+	var latencyColumnResult []*v3.Result
+	latencyColumnResult = append(latencyColumnResult, byteRateColumn)
 
+	resultFetchLatency = postprocess.TransformToTableForBuilderQueries(latencyColumnResult, queryRangeParams)
+
+	result = postprocess.TransformToTableForClickHouseQueries(result)
+
+	result = append(result, resultFetchLatency[0])
 	resp := v3.QueryRangeResponse{
-		Result: resultFetchLatency,
+		Result: result,
 	}
 	aH.Respond(w, resp)
 }

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

@@ -61,14 +61,17 @@ func buildClickHouseQueryNetwork(messagingQueue *MessagingQueue, queueType strin
 
 func buildBuilderQueriesProducerBytes(unixMilliStart, unixMilliEnd int64, attributeCache *Clients) (map[string]*v3.BuilderQuery, error) {
 	bq := make(map[string]*v3.BuilderQuery)
-	queryName := fmt.Sprintf("latency")
+	queryName := fmt.Sprintf("byte_rate")
 
 	chq := &v3.BuilderQuery{
 		QueryName:    queryName,
 		StepInterval: common.MinAllowedStepInterval(unixMilliStart, unixMilliEnd),
 		DataSource:   v3.DataSourceMetrics,
 		AggregateAttribute: v3.AttributeKey{
-			Key: "kafka_producer_byte_rate",
+			Key:      "kafka_producer_byte_rate",
+			DataType: v3.AttributeKeyDataTypeFloat64,
+			Type:     v3.AttributeKeyType("Gauge"),
+			IsColumn: true,
 		},
 		AggregateOperator: v3.AggregateOperatorAvg,
 		Temporality:       v3.Unspecified,
@@ -276,15 +279,16 @@ func BuildQRParamsWithCache(messagingQueue *MessagingQueue, queryContext string,
 		cq, err = buildCompositeQuery(&v3.ClickHouseQuery{
 			Query: query,
 		}, queryContext)
-	} else if queryContext == "producer-throughput-overview-latency" {
+	} else if queryContext == "producer-throughput-overview-byte-rate" {
 		bhq, err := buildBuilderQueriesProducerBytes(unixMilliStart, unixMilliEnd, attributeCache)
 		if err != nil {
 			return nil, err
 		}
 		cq = &v3.CompositeQuery{
 			QueryType:      v3.QueryTypeBuilder,
 			BuilderQueries: bhq,
-			PanelType:      v3.PanelTypeList,
+			PanelType:      v3.PanelTypeTable,
+			FillGaps:       false,
 		}
 	}
 
@@ -315,7 +319,7 @@ func BuildClickHouseQuery(messagingQueue *MessagingQueue, queueType string, quer
 		if !ok {
 			return nil, fmt.Errorf("invalid type for Topic")
 		}
-		if queryContext != "consumer-throughput-details" {
+		if !(queryContext == "consumer-throughput-details" || queryContext == "producer-throughput-details") {
 			partition, ok = messagingQueue.Variables["partition"]
 			if !ok {
 				return nil, fmt.Errorf("invalid type for Partition")
@@ -364,7 +368,7 @@ func BuildClickHouseQuery(messagingQueue *MessagingQueue, queueType string, quer
 
 func buildCompositeQuery(chq *v3.ClickHouseQuery, queryContext string) (*v3.CompositeQuery, error) {
 
-	if queryContext == "producer-consumer-eval" || queryContext == "producer-throughput-overview" {
+	if queryContext == "producer-consumer-eval" {
 		return &v3.CompositeQuery{
 			QueryType:         v3.QueryTypeClickHouseSQL,
 			ClickHouseQueries: map[string]*v3.ClickHouseQuery{queryContext: chq},