Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 13 additions & 3 deletions gen/proto/go/parca/query/v1alpha1/query.pb.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

27 changes: 27 additions & 0 deletions gen/proto/go/parca/query/v1alpha1/query_vtproto.pb.go

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

5 changes: 5 additions & 0 deletions gen/proto/swagger/parca/query/v1alpha1/query.swagger.json
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down
35 changes: 23 additions & 12 deletions pkg/parcacol/querier.go
Original file line number Diff line number Diff line change
Expand Up @@ -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()

Expand Down Expand Up @@ -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))

Expand Down Expand Up @@ -476,6 +473,7 @@ func (q *Querier) queryRangeDelta(
durationMin,
timestampUnique,
totalSum,
totalCount,
},
append([]logicalplan.Expr{
logicalplan.Col(TimestampBucket),
Expand All @@ -489,6 +487,7 @@ func (q *Querier) queryRangeDelta(
durationMin,
).Alias(profile.ColumnDuration),
totalSum,
totalCount,
logicalplan.DynCol(profile.ColumnLabels),
logicalplan.Col(TimestampBucket),
).
Expand All @@ -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{}
Expand All @@ -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.") {
Expand Down Expand Up @@ -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),
})
}
}
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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)
Expand All @@ -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
}

Expand All @@ -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
}
}

Expand Down
3 changes: 3 additions & 0 deletions proto/parca/query/v1alpha1/query.proto
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
16 changes: 15 additions & 1 deletion ui/packages/shared/client/src/parca/query/v1alpha1/query.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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;
Expand All @@ -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")
Expand All @@ -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);
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -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);
};

Expand Down
1 change: 1 addition & 0 deletions ui/packages/shared/profile/package.json
Original file line number Diff line number Diff line change
Expand Up @@ -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": {
Expand Down

This file was deleted.

Loading