mirror of
https://github.com/rcourtman/Pulse.git
synced 2026-10-26 05:29:45 +00:00
Assemble retained metric series directly by statement-local number (#3205)
**Assemble retained metric series directly by statement-local number** Avoid per-observation string resolution and nested-map flushes for interleaved filtered fleets. Bind only nonempty series after each read and bound the direct buffers, preserving the existing wide-sparse fallback, SQL snapshots, tiers and exact observations. Verify against single-resource readers on both layouts. **Cap retained chart observations before allocating wire points** Share the existing time-nearest selection by original indices, so chart conversion materialises only selected observations. Preserve exact timestamps, values, tie breaks, degenerate spans and oldest-source metadata. Check the frozen former sampler across 1,440 cases and the exported fallback contract. Contract-Neutral: Exact chart observations, wire schema, sampling policy and guest/storage lifecycle contracts are unchanged; only index selection and allocation order change. --- Fleet candidate `20261024T123325Z-core-runtime` from the `core-runtime` lane. Reviewed tip `bf9f0ed9822c8ee72ca25b5caecb35a2cd4a5881` on `main`. [How Pulse handles triage](https://github.com/rcourtman/Pulse/blob/main/docs/AI_TRANSPARENCY.md)
This commit is contained in:
@@ -818,6 +818,25 @@ executes a terminal-free Unix recorder and inspects explicit Windows parameter
|
||||
binding. Browser evidence is UI-only; server/native deregistration and Windows
|
||||
failure retention require their own owner result, not source acceptance.
|
||||
|
||||
### Chart caps select before wire conversion (24 October 2026)
|
||||
|
||||
The existing time-nearest chart cap selects indices in the original series
|
||||
before allocating wire-format points. Its equal-time targets, latest-time tie
|
||||
break, first/last observations and degenerate-span index fallback are unchanged.
|
||||
No average, synthetic timestamp or new sampling policy is introduced; zero and
|
||||
selected extrema retain their exact values. Oldest-data metadata still examines
|
||||
every eligible source observation, including one omitted from display. Empty
|
||||
series remain empty and each converted response owns its point slices. The
|
||||
retained reader, LTTB stage, source/alias choice, scopes and cache budgets stay
|
||||
unchanged.
|
||||
|
||||
`TestChartPointSelectionMatchesOriginalCap` compares the frozen former algorithm
|
||||
with both wire and source-point paths over mixed cadence, millisecond ties,
|
||||
negative epochs, unsorted/degenerate spans and caps from unlimited to one point.
|
||||
`TestContract_ChartIndexFallbackPreservesObservations` checks the exported fallback
|
||||
with equal-time, fractional, negative and explicit-zero observations. Source
|
||||
parity and consuming browser evidence remain separate acceptance obligations.
|
||||
|
||||
## Canonical Files
|
||||
|
||||
1a. `docs/API.md`
|
||||
|
||||
@@ -626,6 +626,25 @@ when the row is clicked: no document listener, timer, signal, metric derivation
|
||||
or History work is added per row. `GuestRow.test.tsx` covers the drag, the
|
||||
in-place click and chevron activation over a selection.
|
||||
|
||||
### Chart caps select before wire conversion (24 October 2026)
|
||||
|
||||
The existing time-nearest chart cap selects indices in the original series
|
||||
before allocating wire-format points. Its equal-time targets, latest-time tie
|
||||
break, first/last observations and degenerate-span index fallback are unchanged.
|
||||
No average, synthetic timestamp or new sampling policy is introduced; zero and
|
||||
selected extrema retain their exact values. Oldest-data metadata still examines
|
||||
every eligible source observation, including one omitted from display. Empty
|
||||
series remain empty and each converted response owns its point slices. The
|
||||
retained reader, LTTB stage, source/alias choice, scopes and cache budgets stay
|
||||
unchanged.
|
||||
|
||||
`TestChartPointSelectionMatchesOriginalCap` compares the frozen former algorithm
|
||||
with both wire and source-point paths over mixed cadence, millisecond ties,
|
||||
negative epochs, unsorted/degenerate spans and caps from unlimited to one point.
|
||||
`TestContract_ChartIndexFallbackPreservesObservations` checks the exported fallback
|
||||
with equal-time, fractional, negative and explicit-zero observations. Source
|
||||
parity and consuming browser evidence remain separate acceptance obligations.
|
||||
|
||||
## Canonical Files
|
||||
|
||||
- `frontend-modern/src/utils/workloadDiskUsage.ts` — aggregate disk numeric/current-reading admission.
|
||||
@@ -2762,8 +2781,17 @@ resource list; never persist or expose it as resource identity. With multiple
|
||||
requested metrics, return one series number instead of repeatedly decoding the
|
||||
metric name too. Resolve both dimensions against the current bound resource
|
||||
and metric lists, including on a SQL-template cache hit or a new chunk. Metric
|
||||
names remain bound data in the projection, not interpolated SQL. Full and
|
||||
value-only raw projections share this shape; unfiltered, single-resource and
|
||||
names remain bound data in the projection, not interpolated SQL. Accumulate
|
||||
filtered raw observations directly by series number, resolving names once per
|
||||
nonempty series after the read instead of flushing interleaved nested maps for
|
||||
every point. Empty resources and metrics stay absent. At most 4,096 slice
|
||||
headers are reserved per chunk; wider sparse filters retain the map assembler.
|
||||
Both paths return detached observations and restore stable timestamp order
|
||||
before Docker projection, with no new sampling or source-selection policy.
|
||||
`TestRetainedNumberedAssemblyMatchesSingleResourceReads` compares compact,
|
||||
reordered and wide-sparse fleets with single-resource reads on both layouts,
|
||||
including interleaved series, overlap exclusion, zero and nullable extrema.
|
||||
Full and value-only raw projections share this shape; unfiltered, single-resource and
|
||||
ordered display-bucket reads keep their existing row dimensions.
|
||||
`TestRetainedRawFleetSeriesNumbersBindCurrentMetrics` checks both layouts and
|
||||
projections, parameter-like identities, reordered metric bindings and exact
|
||||
|
||||
@@ -2,6 +2,10 @@ package chartapi
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"math"
|
||||
"math/rand"
|
||||
"reflect"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
@@ -651,3 +655,175 @@ func TestEmptyDockerData(t *testing.T) {
|
||||
t.Errorf("dockerData should be an empty object, not missing")
|
||||
}
|
||||
}
|
||||
|
||||
// A frozen copy of the former wire-point cap is the oracle: source-point
|
||||
// selection must preserve its exact timestamps/values, not just a point count.
|
||||
func referenceChartCapByIndex(points []MetricPoint, maxPoints int) []MetricPoint {
|
||||
if len(points) <= maxPoints || maxPoints <= 0 {
|
||||
return points
|
||||
}
|
||||
if maxPoints == 1 {
|
||||
return []MetricPoint{points[len(points)-1]}
|
||||
}
|
||||
|
||||
result := make([]MetricPoint, 0, maxPoints)
|
||||
step := float64(len(points)-1) / float64(maxPoints-1)
|
||||
prevIndex := -1
|
||||
|
||||
for i := 0; i < maxPoints; i++ {
|
||||
index := int(float64(i)*step + 0.5)
|
||||
if index <= prevIndex {
|
||||
index = prevIndex + 1
|
||||
}
|
||||
if index >= len(points) {
|
||||
index = len(points) - 1
|
||||
}
|
||||
result = append(result, points[index])
|
||||
prevIndex = index
|
||||
}
|
||||
|
||||
if result[len(result)-1].Timestamp != points[len(points)-1].Timestamp {
|
||||
result[len(result)-1] = points[len(points)-1]
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func referenceChartCap(points []MetricPoint, maxPoints int) []MetricPoint {
|
||||
if len(points) <= maxPoints || maxPoints <= 0 {
|
||||
return points
|
||||
}
|
||||
if maxPoints == 1 {
|
||||
return []MetricPoint{points[len(points)-1]}
|
||||
}
|
||||
|
||||
startTimestamp := points[0].Timestamp
|
||||
endTimestamp := points[len(points)-1].Timestamp
|
||||
if endTimestamp <= startTimestamp {
|
||||
return referenceChartCapByIndex(points, maxPoints)
|
||||
}
|
||||
|
||||
bucketSpan := float64(endTimestamp-startTimestamp) / float64(maxPoints-1)
|
||||
if bucketSpan < 1 {
|
||||
return referenceChartCapByIndex(points, maxPoints)
|
||||
}
|
||||
|
||||
type timeBucketRepresentative struct {
|
||||
point MetricPoint
|
||||
distance float64
|
||||
ok bool
|
||||
}
|
||||
|
||||
buckets := make([]timeBucketRepresentative, maxPoints)
|
||||
for _, point := range points {
|
||||
index := int(math.Round(float64(point.Timestamp-startTimestamp) / bucketSpan))
|
||||
if index < 0 {
|
||||
index = 0
|
||||
}
|
||||
if index >= maxPoints {
|
||||
index = maxPoints - 1
|
||||
}
|
||||
|
||||
targetTimestamp := float64(startTimestamp) + bucketSpan*float64(index)
|
||||
distance := math.Abs(float64(point.Timestamp) - targetTimestamp)
|
||||
current := buckets[index]
|
||||
if !current.ok ||
|
||||
distance < current.distance ||
|
||||
(distance == current.distance && point.Timestamp > current.point.Timestamp) {
|
||||
buckets[index] = timeBucketRepresentative{
|
||||
point: point,
|
||||
distance: distance,
|
||||
ok: true,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
result := make([]MetricPoint, 0, maxPoints)
|
||||
result = append(result, points[0])
|
||||
lastAddedTimestamp := points[0].Timestamp
|
||||
for index := 1; index < maxPoints-1; index++ {
|
||||
bucket := buckets[index]
|
||||
if !bucket.ok {
|
||||
continue
|
||||
}
|
||||
if bucket.point.Timestamp <= lastAddedTimestamp {
|
||||
continue
|
||||
}
|
||||
result = append(result, bucket.point)
|
||||
lastAddedTimestamp = bucket.point.Timestamp
|
||||
}
|
||||
|
||||
lastPoint := points[len(points)-1]
|
||||
if lastPoint.Timestamp <= lastAddedTimestamp {
|
||||
result[len(result)-1] = lastPoint
|
||||
return result
|
||||
}
|
||||
|
||||
result = append(result, lastPoint)
|
||||
return result
|
||||
}
|
||||
|
||||
func TestChartPointSelectionMatchesOriginalCap(t *testing.T) {
|
||||
t.Parallel()
|
||||
rng := rand.New(rand.NewSource(20261024))
|
||||
for scenario := 0; scenario < 160; scenario++ {
|
||||
n := []int{0, 1, 2, 3, 17, 121, 240, 500, 777}[scenario%9]
|
||||
points := make([]MetricPoint, n)
|
||||
stamp := int64(1_800_000_000_123)
|
||||
if scenario%11 == 0 {
|
||||
stamp = -100000
|
||||
}
|
||||
for i := range points {
|
||||
step := int64(rng.Intn(4000))
|
||||
if i > n/2 {
|
||||
step %= 5 // dense recent tail, duplicate/millisecond timestamps
|
||||
}
|
||||
stamp += step
|
||||
points[i] = MetricPoint{Timestamp: stamp, Value: float64(rng.Intn(201)-100) / 7}
|
||||
}
|
||||
if n > 2 {
|
||||
points[n-1].Value = 0
|
||||
switch scenario % 5 {
|
||||
case 0:
|
||||
for i := range points {
|
||||
points[i].Timestamp = stamp // no positive span: index fallback
|
||||
}
|
||||
case 1:
|
||||
points[0], points[n-1] = points[n-1], points[0]
|
||||
case 2:
|
||||
points[n/2].Timestamp = points[0].Timestamp - 7 // unselected minimum
|
||||
}
|
||||
}
|
||||
for _, cap := range []int{-1, 0, 1, 2, 3, 30, 36, 120, 500} {
|
||||
t.Run(fmt.Sprintf("%d/cap=%d", scenario, cap), func(t *testing.T) {
|
||||
want := referenceChartCap(points, cap)
|
||||
if got := capMetricPointSeries(points, cap); !reflect.DeepEqual(got, want) {
|
||||
t.Fatal("time cap changed its selected observations")
|
||||
}
|
||||
if got, want := capMetricPointSeriesByIndex(points, cap), referenceChartCapByIndex(points, cap); !reflect.DeepEqual(got, want) {
|
||||
t.Fatal("index fallback changed its selected observations")
|
||||
}
|
||||
source := make([]monitoring.MetricPoint, n)
|
||||
wantOldest := int64(math.MaxInt64)
|
||||
for i, point := range points {
|
||||
source[i] = monitoring.MetricPoint{Timestamp: time.UnixMilli(point.Timestamp), Value: point.Value}
|
||||
if point.Timestamp < wantOldest {
|
||||
wantOldest = point.Timestamp
|
||||
}
|
||||
}
|
||||
oldest := int64(math.MaxInt64)
|
||||
got := convertMetricsForChart(map[string][]monitoring.MetricPoint{
|
||||
"cpu": source, "not-a-sparkline": {{Timestamp: time.UnixMilli(math.MinInt64), Value: -999}},
|
||||
}, &oldest, cap)
|
||||
if len(got) != 1 || !reflect.DeepEqual(got["cpu"], want) || oldest != wantOldest {
|
||||
t.Fatalf("source conversion/cap/oldest changed: oldest %d, want %d", oldest, wantOldest)
|
||||
}
|
||||
if len(got["cpu"]) > 0 {
|
||||
got["cpu"][0].Value = 9999
|
||||
if source[0].Value != points[0].Value {
|
||||
t.Fatal("wire output aliases retained source")
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -623,28 +623,10 @@ func capMetricPointSeriesByIndex(points []MetricPoint, maxPoints int) []MetricPo
|
||||
if len(points) <= maxPoints || maxPoints <= 0 {
|
||||
return points
|
||||
}
|
||||
if maxPoints == 1 {
|
||||
return []MetricPoint{points[len(points)-1]}
|
||||
}
|
||||
|
||||
result := make([]MetricPoint, 0, maxPoints)
|
||||
step := float64(len(points)-1) / float64(maxPoints-1)
|
||||
prevIndex := -1
|
||||
|
||||
for i := 0; i < maxPoints; i++ {
|
||||
index := int(float64(i)*step + 0.5)
|
||||
if index <= prevIndex {
|
||||
index = prevIndex + 1
|
||||
}
|
||||
if index >= len(points) {
|
||||
index = len(points) - 1
|
||||
}
|
||||
result = append(result, points[index])
|
||||
prevIndex = index
|
||||
}
|
||||
|
||||
if result[len(result)-1].Timestamp != points[len(points)-1].Timestamp {
|
||||
result[len(result)-1] = points[len(points)-1]
|
||||
indices := chartPointIndicesByIndex(len(points), maxPoints)
|
||||
result := make([]MetricPoint, len(indices))
|
||||
for i, index := range indices {
|
||||
result[i] = points[index]
|
||||
}
|
||||
return result
|
||||
}
|
||||
@@ -660,6 +642,89 @@ const (
|
||||
|
||||
const InfrastructureSummaryMaxSeriesPoints = infrastructureSummaryMaxSeriesPoints
|
||||
|
||||
func chartPointIndicesByIndex(count, maxPoints int) []int {
|
||||
if maxPoints == 1 {
|
||||
return []int{count - 1}
|
||||
}
|
||||
indices := make([]int, 0, maxPoints)
|
||||
step := float64(count-1) / float64(maxPoints-1)
|
||||
prevIndex := -1
|
||||
for i := 0; i < maxPoints; i++ {
|
||||
index := int(float64(i)*step + 0.5)
|
||||
if index <= prevIndex {
|
||||
index = prevIndex + 1
|
||||
}
|
||||
if index >= count {
|
||||
index = count - 1
|
||||
}
|
||||
indices = append(indices, index)
|
||||
prevIndex = index
|
||||
}
|
||||
return indices
|
||||
}
|
||||
|
||||
// chartPointIndices selects original observations, not averages or synthetic
|
||||
// timestamps. Call only when count > maxPoints > 0. Keeping indices lets the
|
||||
// retained reader apply the same cap before allocating wire-format points.
|
||||
func chartPointIndices(count, maxPoints int, timestampAt func(int) int64) []int {
|
||||
if maxPoints == 1 {
|
||||
return []int{count - 1}
|
||||
}
|
||||
startTimestamp, endTimestamp := timestampAt(0), timestampAt(count-1)
|
||||
if endTimestamp <= startTimestamp {
|
||||
return chartPointIndicesByIndex(count, maxPoints)
|
||||
}
|
||||
bucketSpan := float64(endTimestamp-startTimestamp) / float64(maxPoints-1)
|
||||
if bucketSpan < 1 {
|
||||
return chartPointIndicesByIndex(count, maxPoints)
|
||||
}
|
||||
|
||||
type representative struct {
|
||||
indexPlusOne int // zero means no observation in this time bucket
|
||||
distance float64
|
||||
}
|
||||
buckets := make([]representative, maxPoints)
|
||||
for pointIndex := 0; pointIndex < count; pointIndex++ {
|
||||
timestamp := timestampAt(pointIndex)
|
||||
index := int(math.Round(float64(timestamp-startTimestamp) / bucketSpan))
|
||||
if index < 0 {
|
||||
index = 0
|
||||
}
|
||||
if index >= maxPoints {
|
||||
index = maxPoints - 1
|
||||
}
|
||||
targetTimestamp := float64(startTimestamp) + bucketSpan*float64(index)
|
||||
distance := math.Abs(float64(timestamp) - targetTimestamp)
|
||||
current := buckets[index]
|
||||
if current.indexPlusOne == 0 || distance < current.distance ||
|
||||
(distance == current.distance && timestamp > timestampAt(current.indexPlusOne-1)) {
|
||||
buckets[index] = representative{indexPlusOne: pointIndex + 1, distance: distance}
|
||||
}
|
||||
}
|
||||
|
||||
result := make([]int, 0, maxPoints)
|
||||
result = append(result, 0)
|
||||
lastAddedTimestamp := startTimestamp
|
||||
for index := 1; index < maxPoints-1; index++ {
|
||||
bucket := buckets[index]
|
||||
if bucket.indexPlusOne == 0 {
|
||||
continue
|
||||
}
|
||||
pointIndex := bucket.indexPlusOne - 1
|
||||
timestamp := timestampAt(pointIndex)
|
||||
if timestamp <= lastAddedTimestamp {
|
||||
continue
|
||||
}
|
||||
result = append(result, pointIndex)
|
||||
lastAddedTimestamp = timestamp
|
||||
}
|
||||
if endTimestamp <= lastAddedTimestamp {
|
||||
result[len(result)-1] = count - 1
|
||||
return result
|
||||
}
|
||||
return append(result, count-1)
|
||||
}
|
||||
|
||||
// capMetricPointSeries keeps mixed-cadence series visually proportional across
|
||||
// the selected time window. Index-based capping over-selects recent dense
|
||||
// samples, which bunches the right edge on long ranges.
|
||||
@@ -667,73 +732,11 @@ func capMetricPointSeries(points []MetricPoint, maxPoints int) []MetricPoint {
|
||||
if len(points) <= maxPoints || maxPoints <= 0 {
|
||||
return points
|
||||
}
|
||||
if maxPoints == 1 {
|
||||
return []MetricPoint{points[len(points)-1]}
|
||||
indices := chartPointIndices(len(points), maxPoints, func(i int) int64 { return points[i].Timestamp })
|
||||
result := make([]MetricPoint, len(indices))
|
||||
for i, index := range indices {
|
||||
result[i] = points[index]
|
||||
}
|
||||
|
||||
startTimestamp := points[0].Timestamp
|
||||
endTimestamp := points[len(points)-1].Timestamp
|
||||
if endTimestamp <= startTimestamp {
|
||||
return capMetricPointSeriesByIndex(points, maxPoints)
|
||||
}
|
||||
|
||||
bucketSpan := float64(endTimestamp-startTimestamp) / float64(maxPoints-1)
|
||||
if bucketSpan < 1 {
|
||||
return capMetricPointSeriesByIndex(points, maxPoints)
|
||||
}
|
||||
|
||||
type timeBucketRepresentative struct {
|
||||
point MetricPoint
|
||||
distance float64
|
||||
ok bool
|
||||
}
|
||||
|
||||
buckets := make([]timeBucketRepresentative, maxPoints)
|
||||
for _, point := range points {
|
||||
index := int(math.Round(float64(point.Timestamp-startTimestamp) / bucketSpan))
|
||||
if index < 0 {
|
||||
index = 0
|
||||
}
|
||||
if index >= maxPoints {
|
||||
index = maxPoints - 1
|
||||
}
|
||||
|
||||
targetTimestamp := float64(startTimestamp) + bucketSpan*float64(index)
|
||||
distance := math.Abs(float64(point.Timestamp) - targetTimestamp)
|
||||
current := buckets[index]
|
||||
if !current.ok ||
|
||||
distance < current.distance ||
|
||||
(distance == current.distance && point.Timestamp > current.point.Timestamp) {
|
||||
buckets[index] = timeBucketRepresentative{
|
||||
point: point,
|
||||
distance: distance,
|
||||
ok: true,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
result := make([]MetricPoint, 0, maxPoints)
|
||||
result = append(result, points[0])
|
||||
lastAddedTimestamp := points[0].Timestamp
|
||||
for index := 1; index < maxPoints-1; index++ {
|
||||
bucket := buckets[index]
|
||||
if !bucket.ok {
|
||||
continue
|
||||
}
|
||||
if bucket.point.Timestamp <= lastAddedTimestamp {
|
||||
continue
|
||||
}
|
||||
result = append(result, bucket.point)
|
||||
lastAddedTimestamp = bucket.point.Timestamp
|
||||
}
|
||||
|
||||
lastPoint := points[len(points)-1]
|
||||
if lastPoint.Timestamp <= lastAddedTimestamp {
|
||||
result[len(result)-1] = lastPoint
|
||||
return result
|
||||
}
|
||||
|
||||
result = append(result, lastPoint)
|
||||
return result
|
||||
}
|
||||
|
||||
@@ -1103,18 +1106,34 @@ func convertMetricsForChart(
|
||||
if !sparklineMetrics[metricType] {
|
||||
continue
|
||||
}
|
||||
points := make([]MetricPoint, len(metricPoints))
|
||||
for i, point := range metricPoints {
|
||||
ts := point.Timestamp.UnixMilli()
|
||||
if ts < *oldestTimestamp {
|
||||
if len(metricPoints) <= maxPoints || maxPoints <= 0 {
|
||||
points := make([]MetricPoint, len(metricPoints))
|
||||
for i, point := range metricPoints {
|
||||
ts := point.Timestamp.UnixMilli()
|
||||
if ts < *oldestTimestamp {
|
||||
*oldestTimestamp = ts
|
||||
}
|
||||
points[i] = MetricPoint{Timestamp: ts, Value: point.Value}
|
||||
}
|
||||
converted[metricType] = points
|
||||
continue
|
||||
}
|
||||
// Oldest-data metadata describes the source observations, not just
|
||||
// those selected for display (also for degenerate/unsorted inputs).
|
||||
for _, point := range metricPoints {
|
||||
if ts := point.Timestamp.UnixMilli(); ts < *oldestTimestamp {
|
||||
*oldestTimestamp = ts
|
||||
}
|
||||
points[i] = MetricPoint{
|
||||
Timestamp: ts,
|
||||
Value: point.Value,
|
||||
}
|
||||
}
|
||||
converted[metricType] = capMetricPointSeries(points, maxPoints)
|
||||
indices := chartPointIndices(len(metricPoints), maxPoints, func(i int) int64 {
|
||||
return metricPoints[i].Timestamp.UnixMilli()
|
||||
})
|
||||
points := make([]MetricPoint, len(indices))
|
||||
for i, index := range indices {
|
||||
point := metricPoints[index]
|
||||
points[i] = MetricPoint{Timestamp: point.Timestamp.UnixMilli(), Value: point.Value}
|
||||
}
|
||||
converted[metricType] = points
|
||||
}
|
||||
return converted
|
||||
}
|
||||
|
||||
@@ -27781,3 +27781,29 @@ func TestContractSSOAmbiguousSavePreservesDetail(t *testing.T) {
|
||||
t.Fatal("detail lost its access selection or exposed the saved secret")
|
||||
}
|
||||
}
|
||||
|
||||
// The cap's degenerate-time fallback selects actual observations. Equal
|
||||
// timestamps cannot authorise averaging values or replacing explicit zero.
|
||||
func TestContract_ChartIndexFallbackPreservesObservations(t *testing.T) {
|
||||
points := []chartapi.MetricPoint{
|
||||
{Timestamp: 1800000000123, Value: 0.125},
|
||||
{Timestamp: 1800000000123, Value: 91.75},
|
||||
{Timestamp: 1800000000123, Value: -7.5},
|
||||
{Timestamp: 1800000000123, Value: 63.875},
|
||||
{Timestamp: 1800000000123, Value: 0},
|
||||
}
|
||||
for _, tc := range []struct {
|
||||
cap int
|
||||
indices []int
|
||||
}{{0, []int{0, 1, 2, 3, 4}}, {1, []int{4}}, {3, []int{0, 2, 4}}, {4, []int{0, 1, 3, 4}}} {
|
||||
got := chartapi.CapMetricPointSeriesByIndex(points, tc.cap)
|
||||
if len(got) != len(tc.indices) {
|
||||
t.Fatalf("cap %d: %d observations, want %d", tc.cap, len(got), len(tc.indices))
|
||||
}
|
||||
for i, index := range tc.indices {
|
||||
if got[i] != points[index] {
|
||||
t.Fatalf("cap %d index %d: observation changed: %+v", tc.cap, index, got[i])
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+36
-6
@@ -1482,6 +1482,10 @@ func (s *Store) QueryAll(resourceType, resourceID string, start, end time.Time,
|
||||
// queries fast.
|
||||
const queryAllBatchChunkSize = 500
|
||||
|
||||
// Bound the empty slice headers for direct ordinal assembly. Unusually wide,
|
||||
// sparse metric filters keep the existing map assembler instead.
|
||||
const maxRetainedNumberedSeries = 4096
|
||||
|
||||
// QueryAllBatch retrieves all metric types for multiple resources of the same
|
||||
// type in a single query. Returns map[resourceID]map[metricType][]MetricPoint.
|
||||
// This avoids N+1 query patterns when loading charts for many resources.
|
||||
@@ -2000,6 +2004,14 @@ func queryRetainedChunkAttempt[T any](s *Store, resourceType string, resourceIDs
|
||||
|
||||
result := make(map[string]map[string][]T, len(resourceIDs))
|
||||
seriesCapacity := estimateQueryAllBatchSeriesCapacity(start, end, stepSecs)
|
||||
// Filtered raw fleets already return a statement-local series number.
|
||||
// Accumulate by that number directly: time-major rows interleave metrics,
|
||||
// so resolving strings and flushing nested maps on every row defeats the
|
||||
// compact transport. Bind names once per nonempty series after the read.
|
||||
var numberedSeries [][]T
|
||||
if seriesOrdinals && len(resourceIDs)*len(metricTypes) <= maxRetainedNumberedSeries {
|
||||
numberedSeries = make([][]T, len(resourceIDs)*len(metricTypes))
|
||||
}
|
||||
|
||||
// Consecutive points in one series append directly to its slice. Flush on
|
||||
// a series change so interleaved metrics can resume their existing slices
|
||||
@@ -2077,13 +2089,15 @@ func queryRetainedChunkAttempt[T any](s *Store, resourceType string, resourceIDs
|
||||
if resourceNumber < 0 || resourceNumber >= len(resourceIDs)*len(metricTypes) {
|
||||
return nil, fmt.Errorf("invalid retained series number %d", resourceNumber)
|
||||
}
|
||||
resourceIndex /= len(metricTypes)
|
||||
metricType = metricTypes[resourceNumber%len(metricTypes)]
|
||||
}
|
||||
if resourceIndex < 0 || resourceIndex >= len(resourceIDs) {
|
||||
if numberedSeries == nil {
|
||||
resourceID = resourceIDs[resourceNumber/len(metricTypes)]
|
||||
metricType = metricTypes[resourceNumber%len(metricTypes)]
|
||||
}
|
||||
} else if resourceIndex < 0 || resourceIndex >= len(resourceIDs) {
|
||||
return nil, fmt.Errorf("invalid retained resource number %d for %d resources", resourceIndex, len(resourceIDs))
|
||||
} else {
|
||||
resourceID = resourceIDs[resourceIndex]
|
||||
}
|
||||
resourceID = resourceIDs[resourceIndex]
|
||||
}
|
||||
if !projection.valuesOnly {
|
||||
p.Min, p.Max = p.Value, p.Value
|
||||
@@ -2096,7 +2110,11 @@ func queryRetainedChunkAttempt[T any](s *Store, resourceType string, resourceIDs
|
||||
}
|
||||
if !streamBuckets {
|
||||
p.Timestamp = time.Unix(ts, 0)
|
||||
appendPoint(resourceID, metricType, p)
|
||||
if numberedSeries != nil {
|
||||
numberedSeries[resourceNumber] = append(numberedSeries[resourceNumber], projection.point(p))
|
||||
} else {
|
||||
appendPoint(resourceID, metricType, p)
|
||||
}
|
||||
continue
|
||||
}
|
||||
start := (ts / stepSecs) * stepSecs
|
||||
@@ -2120,6 +2138,18 @@ func queryRetainedChunkAttempt[T any](s *Store, resourceType string, resourceIDs
|
||||
|
||||
flushBucket()
|
||||
flushSeries()
|
||||
for number, points := range numberedSeries {
|
||||
if len(points) == 0 {
|
||||
continue
|
||||
}
|
||||
id := resourceIDs[number/len(metricTypes)]
|
||||
metrics := result[id]
|
||||
if metrics == nil {
|
||||
metrics = make(map[string][]T, len(metricTypes))
|
||||
result[id] = metrics
|
||||
}
|
||||
metrics[metricTypes[number%len(metricTypes)]] = points
|
||||
}
|
||||
if resourceOrdinals {
|
||||
for _, metrics := range result {
|
||||
for _, points := range metrics {
|
||||
|
||||
@@ -736,3 +736,102 @@ func TestRetainedRawFleetSeriesNumbersBindCurrentMetrics(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Direct ordinal accumulation must be indistinguishable from the existing
|
||||
// single-resource assembler, including interleaved rows and empty series. Wide
|
||||
// sparse filters exercise the bounded fallback without losing statement scope.
|
||||
func TestRetainedNumberedAssemblyMatchesSingleResourceReads(t *testing.T) {
|
||||
for _, clustered := range []bool{false, true} {
|
||||
t.Run(fmt.Sprintf("clustered=%t", clustered), func(t *testing.T) {
|
||||
db := newPlanTestDB(t)
|
||||
if clustered {
|
||||
if _, err := db.Exec(`DROP TABLE metrics; CREATE TABLE metrics (` + metricsTableColumns + `) WITHOUT ROWID; CREATE INDEX idx_metrics_tier_time ON metrics(tier,timestamp)`); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
store := &Store{db: pdb.Wrap(db, "numbered-series-assembly")}
|
||||
store.clusteredMetrics.Store(clustered)
|
||||
base := time.Unix(1_800_003_600, 0)
|
||||
end := base.Add(6 * time.Hour)
|
||||
ids := []string{"b':p2", "a:p4", "missing"}
|
||||
names := []string{"memory", "cpu", "netin", "absent"}
|
||||
tx, err := db.Begin()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
insert, err := tx.Prepare(`INSERT INTO metrics(resource_type,resource_id,metric_type,tier,timestamp,value,min_value,max_value) VALUES ('node',?,?,?,?,?,?,?)`)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for i, id := range ids[:2] {
|
||||
for j, name := range names[:3] {
|
||||
// Minute-tier rows come first; the raw first/tail points must
|
||||
// still be restored to chronological order, never averaged.
|
||||
for p := 0; p < 300; p++ {
|
||||
value := float64(i*1000 + j*300 + p)
|
||||
if _, err := insert.Exec(id, name, "minute", base.Unix()+int64(p*60+60), value, value-7, value+9); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
for _, p := range []struct{ offset, value int64 }{{1, 12}, {65, 99999}, {18061, 0}} {
|
||||
if _, err := insert.Exec(id, name, "raw", base.Unix()+p.offset, p.value, nil, nil); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
insert.Close()
|
||||
if err := tx.Commit(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
for _, shape := range []string{"compact", "rebound", "wide-sparse"} {
|
||||
t.Run(shape, func(t *testing.T) {
|
||||
requested := append([]string(nil), ids...)
|
||||
filters := append([]string(nil), names...)
|
||||
if shape == "rebound" {
|
||||
requested[0], requested[1] = requested[1], requested[0]
|
||||
filters[0], filters[2] = filters[2], filters[0]
|
||||
}
|
||||
if shape == "wide-sparse" {
|
||||
for len(requested) < queryAllBatchChunkSize {
|
||||
requested = append(requested, fmt.Sprintf("missing-%d", len(requested)))
|
||||
}
|
||||
for len(filters) <= maxRetainedNumberedSeries/queryAllBatchChunkSize {
|
||||
filters = append(filters, fmt.Sprintf("absent-%d", len(filters)))
|
||||
}
|
||||
}
|
||||
full, err := store.QueryMetricTypesBatch("node", requested, filters, base, end, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(full) != 2 {
|
||||
t.Fatalf("empty resource materialised: %d resources", len(full))
|
||||
}
|
||||
for _, id := range ids[:2] {
|
||||
single, err := store.QueryMetricTypesBatch("node", []string{id}, filters, base, end, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if !reflect.DeepEqual(full[id], single[id]) {
|
||||
t.Fatalf("fleet differs from single-resource read for %s", id)
|
||||
}
|
||||
for _, points := range full[id] {
|
||||
if len(points) != 302 || points[0].Value != 12 || points[len(points)-1].Value != 0 || points[len(points)-1].Min != 0 || points[1].Min != points[1].Value-7 {
|
||||
t.Fatalf("chronology/zero/extrema/overlap changed: %d points", len(points))
|
||||
}
|
||||
}
|
||||
}
|
||||
values, err := store.QueryMetricValuesBatch("node", requested, filters, base, end, 0)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
assertMetricValueProjection(t, full, values)
|
||||
full[ids[0]][names[0]][0].Value = -999
|
||||
if values[ids[0]][names[0]][0].Value != 12 {
|
||||
t.Fatal("reader outputs alias each other")
|
||||
}
|
||||
})
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user