ui: Samples strips in Flamecharts visualization (#6192)

* SamplesStrips now part of the Flamechart visualization

* Linter fixes

* go test fix

* Minor UX improvements

* Dragging UX improvement

* Added a button to quickly switch ot 1m range and consistent strip interaction fixes

Author: manojVivek

gen/proto/go/parca/query/v1alpha1/query.pb.go

@@ -1503,6 +1503,11 @@
           "type": "string",
           "format": "int64",
           "description": "duration is the normalized aggregated duration the metric samples has been observed over."
+        },
+        "count": {
+          "type": "integer",
+          "format": "int32",
+          "description": "count is the number of samples aggregated into this data point."
         }
       },
       "title": "MetricsSample is a cumulative value and timestamp of a profile"

gen/proto/go/parca/query/v1alpha1/query_vtproto.pb.go

@@ -348,11 +348,6 @@ func (q *Querier) QueryRange(
 		return nil, err
 	}
 
-	// The step cannot be lower than 1s
-	if step < time.Second {
-		step = time.Second
-	}
-
 	start := startTime.UnixNano()
 	end := endTime.UnixNano()
 
@@ -401,6 +396,8 @@ func (q *Querier) queryRangeDelta(
 
 	totalSum := logicalplan.Sum(logicalplan.Col(profile.ColumnValue))
 	totalSumColumn := totalSum.Name()
+	totalCount := logicalplan.Count(logicalplan.Col(profile.ColumnValue))
+	totalCountColumn := totalCount.Name()
 	durationMin := logicalplan.Min(logicalplan.Col(profile.ColumnDuration))
 	timestampUnique := logicalplan.Unique(logicalplan.Col(profile.ColumnTimeNanos))
 
@@ -476,6 +473,7 @@ func (q *Querier) queryRangeDelta(
 				durationMin,
 				timestampUnique,
 				totalSum,
+				totalCount,
 			},
 			append([]logicalplan.Expr{
 				logicalplan.Col(TimestampBucket),
@@ -489,6 +487,7 @@ func (q *Querier) queryRangeDelta(
 				durationMin,
 			).Alias(profile.ColumnDuration),
 			totalSum,
+			totalCount,
 			logicalplan.DynCol(profile.ColumnLabels),
 			logicalplan.Col(TimestampBucket),
 		).
@@ -514,11 +513,13 @@ func (q *Querier) queryRangeDelta(
 		PerSecondValue int
 		ValueSum       int
 		Duration       int
+		Count          int
 	}{
 		Timestamp:      -1,
 		PerSecondValue: -1,
 		ValueSum:       -1,
 		Duration:       -1,
+		Count:          -1,
 	}
 
 	labelColumnIndices := []int{}
@@ -541,6 +542,9 @@ func (q *Querier) queryRangeDelta(
 				continue
 			case profile.ColumnDuration:
 				columnIndices.Duration = i
+			case totalCountColumn:
+				columnIndices.Count = i
+				continue
 			}
 
 			if strings.HasPrefix(field.Name, "labels.") {
@@ -606,13 +610,15 @@ func (q *Querier) queryRangeDelta(
 			valueSum := ar.Column(columnIndices.ValueSum).(*array.Int64).Value(i)
 			valuePerSecond := ar.Column(columnIndices.PerSecondValue).(*array.Float64).Value(i)
 			duration := ar.Column(columnIndices.Duration).(*array.Int64).Value(i)
+			count := ar.Column(columnIndices.Count).(*array.Int64).Value(i)
 
 			series := resSeries[index]
 			series.Samples = append(series.Samples, &pb.MetricsSample{
 				Timestamp:      timestamppb.New(time.Unix(0, ts)),
 				Value:          valueSum,
 				ValuePerSecond: valuePerSecond,
 				Duration:       duration,
+				Count:          int32(count),
 			})
 		}
 	}
@@ -686,7 +692,7 @@ func (q *Querier) queryRangeNonDelta(ctx context.Context, filterExpr logicalplan
 	labelColumnIndices := []int{}
 	labelSet := labels.NewScratchBuilder(64)
 	resSeries := []*pb.MetricsSeries{}
-	resSeriesBuckets := map[int]map[int64]struct{}{}
+	resSeriesBuckets := map[int]map[int64]int{}
 	labelsetToIndex := map[string]int{}
 
 	for _, ar := range records {
@@ -740,7 +746,7 @@ func (q *Querier) queryRangeNonDelta(ctx context.Context, filterExpr logicalplan
 				resSeries = append(resSeries, &pb.MetricsSeries{Labelset: &profilestorepb.LabelSet{Labels: pbLabelSet}})
 				index = len(resSeries) - 1
 				labelsetToIndex[s] = index
-				resSeriesBuckets[index] = map[int64]struct{}{}
+				resSeriesBuckets[index] = map[int64]int{}
 			}
 
 			ts := ar.Column(columnIndices[profile.ColumnTimeNanos].index).(*array.Int64).Value(i)
@@ -755,9 +761,13 @@ func (q *Querier) queryRangeNonDelta(ctx context.Context, filterExpr logicalplan
 			// This needs to be moved to FrostDB to not even query all of this data in the first place.
 			// With a scrape interval of 10s and a query range of 1d we'd query 8640 samples and at most return 960.
 			// Even worse for a week, we'd query 60480 samples and only return 1000.
-			tsBucket := ts / 1000 / int64(step.Seconds())
-			if _, found := resSeriesBuckets[index][tsBucket]; found {
-				// We already have a MetricsSample for this timestamp bucket, ignore it.
+			tsBucket := ts
+			if stepNanos := step.Nanoseconds(); stepNanos > 0 {
+				tsBucket = ts / stepNanos
+			}
+			if sampleIdx, found := resSeriesBuckets[index][tsBucket]; found {
+				// We already have a MetricsSample for this timestamp bucket, increment its count.
+				resSeries[index].Samples[sampleIdx].Count++
 				continue
 			}
 
@@ -766,9 +776,10 @@ func (q *Querier) queryRangeNonDelta(ctx context.Context, filterExpr logicalplan
 				Timestamp:      timestamppb.New(time.Unix(0, ts)),
 				Value:          value,
 				ValuePerSecond: float64(value),
+				Count:          1,
 			})
-			// Mark the timestamp bucket as filled by the above MetricsSample.
-			resSeriesBuckets[index][tsBucket] = struct{}{}
+			// Mark the timestamp bucket as filled by the above MetricsSample, storing its index.
+			resSeriesBuckets[index][tsBucket] = len(series.Samples) - 1
 		}
 	}
 

gen/proto/swagger/parca/query/v1alpha1/query.swagger.json

@@ -142,6 +142,9 @@ message MetricsSample {
 
   // duration is the normalized aggregated duration the metric samples has been observed over.
   int64 duration = 4;
+
+  // count is the number of samples aggregated into this data point.
+  int32 count = 5;
 }
 
 // MergeProfile contains parameters for a merge request

pkg/parcacol/querier.go

@@ -210,6 +210,12 @@ export interface MetricsSample {
      * @generated from protobuf field: int64 duration = 4
      */
     duration: bigint;
+    /**
+     * count is the number of samples aggregated into this data point.
+     *
+     * @generated from protobuf field: int32 count = 5
+     */
+    count: number;
 }
 /**
  * MergeProfile contains parameters for a merge request
@@ -2036,14 +2042,16 @@ class MetricsSample$Type extends MessageType<MetricsSample> {
             { no: 1, name: "timestamp", kind: "message", T: () => Timestamp },
             { no: 2, name: "value", kind: "scalar", T: 3 /*ScalarType.INT64*/, L: 0 /*LongType.BIGINT*/ },
             { no: 3, name: "value_per_second", kind: "scalar", T: 1 /*ScalarType.DOUBLE*/ },
-            { no: 4, name: "duration", kind: "scalar", T: 3 /*ScalarType.INT64*/, L: 0 /*LongType.BIGINT*/ }
+            { no: 4, name: "duration", kind: "scalar", T: 3 /*ScalarType.INT64*/, L: 0 /*LongType.BIGINT*/ },
+            { no: 5, name: "count", kind: "scalar", T: 5 /*ScalarType.INT32*/ }
         ]);
     }
     create(value?: PartialMessage<MetricsSample>): MetricsSample {
         const message = globalThis.Object.create((this.messagePrototype!));
         message.value = 0n;
         message.valuePerSecond = 0;
         message.duration = 0n;
+        message.count = 0;
         if (value !== undefined)
             reflectionMergePartial<MetricsSample>(this, message, value);
         return message;
@@ -2065,6 +2073,9 @@ class MetricsSample$Type extends MessageType<MetricsSample> {
                 case /* int64 duration */ 4:
                     message.duration = reader.int64().toBigInt();
                     break;
+                case /* int32 count */ 5:
+                    message.count = reader.int32();
+                    break;
                 default:
                     let u = options.readUnknownField;
                     if (u === "throw")
@@ -2089,6 +2100,9 @@ class MetricsSample$Type extends MessageType<MetricsSample> {
         /* int64 duration = 4; */
         if (message.duration !== 0n)
             writer.tag(4, WireType.Varint).int64(message.duration);
+        /* int32 count = 5; */
+        if (message.count !== 0)
+            writer.tag(5, WireType.Varint).int32(message.count);
         let u = options.writeUnknownFields;
         if (u !== false)
             (u == true ? UnknownFieldHandler.onWrite : u)(this.typeName, message, writer);

proto/parca/query/v1alpha1/query.proto

@@ -334,10 +334,13 @@ export const JSONParser = <T,>(val: ParamValue): T => {
   return JSON.parse(val as string);
 };
 
-export const NumberParser = (val: string): number => {
+export const NumberParser = (val: ParamValue): number => {
   if (val == null || val === '' || val === 'undefined') {
     return 0;
   }
+  if (Array.isArray(val)) {
+    return val.length > 0 ? Number(val[0]) : 0;
+  }
   return Number(val);
 };
 

ui/packages/shared/client/src/parca/query/v1alpha1/query.ts

@@ -57,6 +57,7 @@
     "react-use": "^17.5.0",
     "tailwindcss": "3.2.4",
     "tsc-watch": "6.3.1",
+    "usehooks-ts": "^3.1.1",
     "with-alpha-hex": "^1.0.6"
   },
   "devDependencies": {