nathan/vctp2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix daily aggregation sample count
All checks were successful
continuous-integration/drone/push Build is passing
Details

Browse Source
This commit is contained in:
Nathan Coad
2026-01-22 14:27:27 +11:00
parent baea0cc85c
commit b0592a2539
2 changed files with 43 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

44
internal/tasks/dailyAggregate.go
View File

@@ -361,14 +361,16 @@ LIMIT 1
}
c.Logger.Info("Daily aggregation deletion times", "source_inferred", inferredDeletions)
totalSamplesByVcenter := sampleCountsByVcenter(aggMap)
// Insert aggregated rows.
if err := c.insertDailyAggregates(ctx, summaryTable, aggMap, totalSamples); err != nil {
if err := c.insertDailyAggregates(ctx, summaryTable, aggMap, totalSamples, totalSamplesByVcenter); err != nil {
return err
}
c.Logger.Debug("inserted daily aggregates", "table", summaryTable, "rows", len(aggMap), "total_samples", totalSamples)
// Persist rollup cache for monthly aggregation.
if err := c.persistDailyRollup(ctx, dayStart.Unix(), aggMap, totalSamples); err != nil {
if err := c.persistDailyRollup(ctx, dayStart.Unix(), aggMap, totalSamples, totalSamplesByVcenter); err != nil {
c.Logger.Warn("failed to persist daily rollup cache", "error", err, "date", dayStart.Format("2006-01-02"))
}
@@ -736,7 +738,7 @@ WHERE "SnapshotTime" >= ? AND "SnapshotTime" < ?`
return agg, snapTimes, rows.Err()
}
func (c *CronTask) insertDailyAggregates(ctx context.Context, table string, agg map[dailyAggKey]*dailyAggVal, totalSamples int) error {
func (c *CronTask) insertDailyAggregates(ctx context.Context, table string, agg map[dailyAggKey]*dailyAggVal, totalSamples int, totalSamplesByVcenter map[string]int) error {
dbConn := c.Database.DB()
tx, err := dbConn.Beginx()
if err != nil {
@@ -759,7 +761,11 @@ INSERT INTO %s (
if v.samples == 0 {
continue
}
total := float64(totalSamples)
vcTotal := totalSamplesByVcenter[v.key.Vcenter]
if vcTotal <= 0 {
vcTotal = totalSamples
}
total := float64(vcTotal)
avgVcpu := 0.0
avgRam := 0.0
avgDisk := 0.0
@@ -858,12 +864,16 @@ func btoi(b bool) int64 {
}
// persistDailyRollup stores per-day aggregates into vm_daily_rollup to speed monthly aggregation.
func (c *CronTask) persistDailyRollup(ctx context.Context, dayUnix int64, agg map[dailyAggKey]*dailyAggVal, totalSamples int) error {
func (c *CronTask) persistDailyRollup(ctx context.Context, dayUnix int64, agg map[dailyAggKey]*dailyAggVal, totalSamples int, totalSamplesByVcenter map[string]int) error {
dbConn := c.Database.DB()
for _, v := range agg {
if strings.EqualFold(strings.TrimSpace(v.isTemplate), "true") || v.isTemplate == "1" {
continue
}
vcTotal := totalSamplesByVcenter[v.key.Vcenter]
if vcTotal <= 0 {
vcTotal = totalSamples
}
row := db.VmDailyRollupRow{
Vcenter: v.key.Vcenter,
VmId: v.key.VmId,
@@ -872,7 +882,7 @@ func (c *CronTask) persistDailyRollup(ctx context.Context, dayUnix int64, agg ma
CreationTime: v.creation,
DeletionTime: v.deletion,
SamplesPresent: v.samples,
TotalSamples: int64(totalSamples),
TotalSamples: int64(vcTotal),
SumVcpu: float64(v.sumVcpu),
SumRam: float64(v.sumRam),
SumDisk: v.sumDisk,
@@ -897,3 +907,25 @@ func (c *CronTask) persistDailyRollup(ctx context.Context, dayUnix int64, agg ma
}
return nil
}
func sampleCountsByVcenter(agg map[dailyAggKey]*dailyAggVal) map[string]int {
vcenterTimes := make(map[string]map[int64]struct{}, 8)
for _, v := range agg {
if v.key.Vcenter == "" {
continue
}
set := vcenterTimes[v.key.Vcenter]
if set == nil {
set = make(map[int64]struct{}, len(v.seen))
vcenterTimes[v.key.Vcenter] = set
}
for t := range v.seen {
set[t] = struct{}{}
}
}
counts := make(map[string]int, len(vcenterTimes))
for vc, set := range vcenterTimes {
counts[vc] = len(set)
}
return counts
}