start work on dev branch [CI SKIP]
This commit is contained in:
@@ -2,8 +2,13 @@ package tasks
|
||||
|
||||
import (
|
||||
"context"
|
||||
"database/sql"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"os"
|
||||
"runtime"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
"vctp/db"
|
||||
"vctp/internal/metrics"
|
||||
@@ -56,6 +61,17 @@ func (c *CronTask) aggregateDailySummary(ctx context.Context, targetTime time.Ti
|
||||
}
|
||||
}
|
||||
|
||||
// If enabled, use the Go fan-out/reduce path to parallelize aggregation.
|
||||
if os.Getenv("DAILY_AGG_GO") == "1" {
|
||||
if err := c.aggregateDailySummaryGo(ctx, dayStart, dayEnd, summaryTable, force); err != nil {
|
||||
c.Logger.Warn("go-based daily aggregation failed, falling back to SQL path", "error", err)
|
||||
} else {
|
||||
metrics.RecordDailyAggregation(time.Since(jobStart), nil)
|
||||
c.Logger.Debug("Finished daily inventory aggregation (Go path)", "summary_table", summaryTable)
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
hourlySnapshots, err := report.SnapshotRecordsWithFallback(ctx, c.Database, "hourly", "inventory_hourly_", "epoch", dayStart, dayEnd)
|
||||
if err != nil {
|
||||
return err
|
||||
@@ -164,3 +180,379 @@ func (c *CronTask) aggregateDailySummary(ctx context.Context, targetTime time.Ti
|
||||
func dailySummaryTableName(t time.Time) (string, error) {
|
||||
return db.SafeTableName(fmt.Sprintf("inventory_daily_summary_%s", t.Format("20060102")))
|
||||
}
|
||||
|
||||
// aggregateDailySummaryGo performs daily aggregation by reading hourly tables in parallel,
|
||||
// reducing in Go, and writing the summary table. It mirrors the outputs of the SQL path
|
||||
// as closely as possible while improving CPU utilization on multi-core hosts.
|
||||
func (c *CronTask) aggregateDailySummaryGo(ctx context.Context, dayStart, dayEnd time.Time, summaryTable string, force bool) error {
|
||||
jobStart := time.Now()
|
||||
dbConn := c.Database.DB()
|
||||
|
||||
hourlySnapshots, err := report.SnapshotRecordsWithFallback(ctx, c.Database, "hourly", "inventory_hourly_", "epoch", dayStart, dayEnd)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
hourlySnapshots = filterRecordsInRange(hourlySnapshots, dayStart, dayEnd)
|
||||
hourlySnapshots = filterSnapshotsWithRows(ctx, dbConn, hourlySnapshots)
|
||||
if len(hourlySnapshots) == 0 {
|
||||
return fmt.Errorf("no hourly snapshot tables found for %s", dayStart.Format("2006-01-02"))
|
||||
}
|
||||
|
||||
hourlyTables := make([]string, 0, len(hourlySnapshots))
|
||||
for _, snapshot := range hourlySnapshots {
|
||||
hourlyTables = append(hourlyTables, snapshot.TableName)
|
||||
if err := db.EnsureSnapshotIndexes(ctx, dbConn, snapshot.TableName); err != nil {
|
||||
c.Logger.Warn("failed to ensure indexes on hourly table", "table", snapshot.TableName, "error", err)
|
||||
}
|
||||
}
|
||||
unionQuery, err := buildUnionQuery(hourlyTables, summaryUnionColumns, templateExclusionFilter())
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Clear existing summary if forcing.
|
||||
if rowsExist, err := db.TableHasRows(ctx, dbConn, summaryTable); err != nil {
|
||||
return err
|
||||
} else if rowsExist && !force {
|
||||
c.Logger.Debug("Daily summary already exists, skipping aggregation (Go path)", "summary_table", summaryTable)
|
||||
return nil
|
||||
} else if rowsExist && force {
|
||||
if err := clearTable(ctx, dbConn, summaryTable); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
totalSamples := len(hourlyTables)
|
||||
aggMap, err := c.scanHourlyTablesParallel(ctx, hourlySnapshots, totalSamples)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if len(aggMap) == 0 {
|
||||
return fmt.Errorf("no VM records aggregated for %s", dayStart.Format("2006-01-02"))
|
||||
}
|
||||
|
||||
// Insert aggregated rows.
|
||||
if err := c.insertDailyAggregates(ctx, summaryTable, aggMap, totalSamples); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Refine lifecycle with existing SQL helper to pick up first-after deletions.
|
||||
if err := db.RefineCreationDeletionFromUnion(ctx, dbConn, summaryTable, unionQuery); err != nil {
|
||||
c.Logger.Warn("failed to refine creation/deletion times", "error", err, "table", summaryTable)
|
||||
}
|
||||
|
||||
db.AnalyzeTableIfPostgres(ctx, dbConn, summaryTable)
|
||||
rowCount, err := db.TableRowCount(ctx, dbConn, summaryTable)
|
||||
if err != nil {
|
||||
c.Logger.Warn("unable to count daily summary rows", "error", err, "table", summaryTable)
|
||||
}
|
||||
if err := report.RegisterSnapshot(ctx, c.Database, "daily", summaryTable, dayStart, rowCount); err != nil {
|
||||
c.Logger.Warn("failed to register daily snapshot", "error", err, "table", summaryTable)
|
||||
}
|
||||
if err := c.generateReport(ctx, summaryTable); err != nil {
|
||||
c.Logger.Warn("failed to generate daily report", "error", err, "table", summaryTable)
|
||||
return err
|
||||
}
|
||||
|
||||
c.Logger.Debug("Finished daily inventory aggregation (Go path)", "summary_table", summaryTable, "duration", time.Since(jobStart))
|
||||
return nil
|
||||
}
|
||||
|
||||
type dailyAggKey struct {
|
||||
Vcenter string
|
||||
VmId string
|
||||
VmUuid string
|
||||
Name string
|
||||
}
|
||||
|
||||
type dailyAggVal struct {
|
||||
key dailyAggKey
|
||||
resourcePool string
|
||||
datacenter string
|
||||
cluster string
|
||||
folder string
|
||||
isTemplate string
|
||||
poweredOn string
|
||||
srmPlaceholder string
|
||||
creation int64
|
||||
firstSeen int64
|
||||
lastSeen int64
|
||||
sumVcpu int64
|
||||
sumRam int64
|
||||
sumDisk float64
|
||||
samples int64
|
||||
tinHits int64
|
||||
bronzeHits int64
|
||||
silverHits int64
|
||||
goldHits int64
|
||||
}
|
||||
|
||||
func (c *CronTask) scanHourlyTablesParallel(ctx context.Context, snapshots []report.SnapshotRecord, totalSamples int) (map[dailyAggKey]*dailyAggVal, error) {
|
||||
agg := make(map[dailyAggKey]*dailyAggVal, 1024)
|
||||
mu := sync.Mutex{}
|
||||
workers := runtime.NumCPU()
|
||||
if workers < 2 {
|
||||
workers = 2
|
||||
}
|
||||
if workers > len(snapshots) {
|
||||
workers = len(snapshots)
|
||||
}
|
||||
|
||||
jobs := make(chan report.SnapshotRecord, len(snapshots))
|
||||
wg := sync.WaitGroup{}
|
||||
for i := 0; i < workers; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for snap := range jobs {
|
||||
rows, err := c.scanHourlyTable(ctx, snap)
|
||||
if err != nil {
|
||||
c.Logger.Warn("failed to scan hourly table", "table", snap.TableName, "error", err)
|
||||
continue
|
||||
}
|
||||
mu.Lock()
|
||||
for k, v := range rows {
|
||||
if existing, ok := agg[k]; ok {
|
||||
mergeDailyAgg(existing, v)
|
||||
} else {
|
||||
agg[k] = v
|
||||
}
|
||||
}
|
||||
mu.Unlock()
|
||||
}
|
||||
}()
|
||||
}
|
||||
for _, snap := range snapshots {
|
||||
jobs <- snap
|
||||
}
|
||||
close(jobs)
|
||||
wg.Wait()
|
||||
return agg, nil
|
||||
}
|
||||
|
||||
func mergeDailyAgg(dst, src *dailyAggVal) {
|
||||
if src.creation > 0 && (dst.creation == 0 || src.creation < dst.creation) {
|
||||
dst.creation = src.creation
|
||||
}
|
||||
if dst.firstSeen == 0 || (src.firstSeen > 0 && src.firstSeen < dst.firstSeen) {
|
||||
dst.firstSeen = src.firstSeen
|
||||
}
|
||||
if src.lastSeen > dst.lastSeen {
|
||||
dst.lastSeen = src.lastSeen
|
||||
dst.resourcePool = src.resourcePool
|
||||
dst.datacenter = src.datacenter
|
||||
dst.cluster = src.cluster
|
||||
dst.folder = src.folder
|
||||
dst.isTemplate = src.isTemplate
|
||||
dst.poweredOn = src.poweredOn
|
||||
dst.srmPlaceholder = src.srmPlaceholder
|
||||
}
|
||||
dst.sumVcpu += src.sumVcpu
|
||||
dst.sumRam += src.sumRam
|
||||
dst.sumDisk += src.sumDisk
|
||||
dst.samples += src.samples
|
||||
dst.tinHits += src.tinHits
|
||||
dst.bronzeHits += src.bronzeHits
|
||||
dst.silverHits += src.silverHits
|
||||
dst.goldHits += src.goldHits
|
||||
}
|
||||
|
||||
func (c *CronTask) scanHourlyTable(ctx context.Context, snap report.SnapshotRecord) (map[dailyAggKey]*dailyAggVal, error) {
|
||||
dbConn := c.Database.DB()
|
||||
query := fmt.Sprintf(`
|
||||
SELECT
|
||||
"Name","Vcenter","VmId","VmUuid","ResourcePool","Datacenter","Cluster","Folder",
|
||||
COALESCE("ProvisionedDisk",0) AS disk,
|
||||
COALESCE("VcpuCount",0) AS vcpu,
|
||||
COALESCE("RamGB",0) AS ram,
|
||||
COALESCE("CreationTime",0) AS creation,
|
||||
COALESCE("DeletionTime",0) AS deletion,
|
||||
COALESCE("IsTemplate",'FALSE') AS is_template,
|
||||
COALESCE("PoweredOn",'FALSE') AS powered_on,
|
||||
COALESCE("SrmPlaceholder",'FALSE') AS srm_placeholder,
|
||||
COALESCE("SnapshotTime",0) AS snapshot_time
|
||||
FROM %s
|
||||
`, snap.TableName)
|
||||
rows, err := dbConn.QueryxContext(ctx, query)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
defer rows.Close()
|
||||
|
||||
out := make(map[dailyAggKey]*dailyAggVal, 256)
|
||||
for rows.Next() {
|
||||
var (
|
||||
name, vcenter, vmId, vmUuid, resourcePool string
|
||||
dc, cluster, folder sql.NullString
|
||||
disk sql.NullFloat64
|
||||
vcpu, ram sql.NullInt64
|
||||
creation, deletion, snapshotTime sql.NullInt64
|
||||
isTemplate, poweredOn, srmPlaceholder sql.NullString
|
||||
)
|
||||
if err := rows.Scan(&name, &vcenter, &vmId, &vmUuid, &resourcePool, &dc, &cluster, &folder, &disk, &vcpu, &ram, &creation, &deletion, &isTemplate, &poweredOn, &srmPlaceholder, &snapshotTime); err != nil {
|
||||
continue
|
||||
}
|
||||
if vcenter == "" {
|
||||
continue
|
||||
}
|
||||
// Skip templates.
|
||||
if strings.EqualFold(strings.TrimSpace(isTemplate.String), "true") || isTemplate.String == "1" {
|
||||
continue
|
||||
}
|
||||
key := dailyAggKey{
|
||||
Vcenter: vcenter,
|
||||
VmId: strings.TrimSpace(vmId),
|
||||
VmUuid: strings.TrimSpace(vmUuid),
|
||||
Name: strings.TrimSpace(name),
|
||||
}
|
||||
if key.VmId == "" && key.VmUuid == "" && key.Name == "" {
|
||||
continue
|
||||
}
|
||||
if key.VmId == "" {
|
||||
key.VmId = key.VmUuid
|
||||
}
|
||||
pool := strings.ToLower(strings.TrimSpace(resourcePool))
|
||||
hitTin := btoi(pool == "tin")
|
||||
hitBronze := btoi(pool == "bronze")
|
||||
hitSilver := btoi(pool == "silver")
|
||||
hitGold := btoi(pool == "gold")
|
||||
row := &dailyAggVal{
|
||||
key: key,
|
||||
resourcePool: resourcePool,
|
||||
datacenter: dc.String,
|
||||
cluster: cluster.String,
|
||||
folder: folder.String,
|
||||
isTemplate: isTemplate.String,
|
||||
poweredOn: poweredOn.String,
|
||||
srmPlaceholder: srmPlaceholder.String,
|
||||
creation: int64OrZero(creation),
|
||||
firstSeen: int64OrZero(snapshotTime),
|
||||
lastSeen: int64OrZero(snapshotTime),
|
||||
sumVcpu: vcpu.Int64,
|
||||
sumRam: ram.Int64,
|
||||
sumDisk: disk.Float64,
|
||||
samples: 1,
|
||||
tinHits: hitTin,
|
||||
bronzeHits: hitBronze,
|
||||
silverHits: hitSilver,
|
||||
goldHits: hitGold,
|
||||
}
|
||||
out[key] = row
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
func (c *CronTask) insertDailyAggregates(ctx context.Context, table string, agg map[dailyAggKey]*dailyAggVal, totalSamples int) error {
|
||||
dbConn := c.Database.DB()
|
||||
tx, err := dbConn.Beginx()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
|
||||
driver := strings.ToLower(dbConn.DriverName())
|
||||
placeholders := makePlaceholders(driver, 29)
|
||||
insert := fmt.Sprintf(`
|
||||
INSERT INTO %s (
|
||||
"Name","Vcenter","VmId","VmUuid","ResourcePool","Datacenter","Cluster","Folder",
|
||||
"ProvisionedDisk","VcpuCount","RamGB","IsTemplate","PoweredOn","SrmPlaceholder",
|
||||
"CreationTime","DeletionTime","SamplesPresent","AvgVcpuCount","AvgRamGB","AvgProvisionedDisk",
|
||||
"AvgIsPresent","PoolTinPct","PoolBronzePct","PoolSilverPct","PoolGoldPct","Tin","Bronze","Silver","Gold"
|
||||
) VALUES (%s)
|
||||
`, table, placeholders)
|
||||
|
||||
for _, v := range agg {
|
||||
if v.samples == 0 {
|
||||
continue
|
||||
}
|
||||
avgVcpu := float64(v.sumVcpu) / float64(v.samples)
|
||||
avgRam := float64(v.sumRam) / float64(v.samples)
|
||||
avgDisk := v.sumDisk / float64(v.samples)
|
||||
total := float64(totalSamples)
|
||||
avgPresent := 0.0
|
||||
tinPct := 0.0
|
||||
bronzePct := 0.0
|
||||
silverPct := 0.0
|
||||
goldPct := 0.0
|
||||
if total > 0 {
|
||||
avgPresent = float64(v.samples) / total
|
||||
tinPct = float64(v.tinHits) * 100 / total
|
||||
bronzePct = float64(v.bronzeHits) * 100 / total
|
||||
silverPct = float64(v.silverHits) * 100 / total
|
||||
goldPct = float64(v.goldHits) * 100 / total
|
||||
}
|
||||
args := []interface{}{
|
||||
v.key.Name,
|
||||
v.key.Vcenter,
|
||||
nullIfEmpty(v.key.VmId),
|
||||
nullIfEmpty(v.key.VmUuid),
|
||||
v.resourcePool,
|
||||
nullIfEmpty(v.datacenter),
|
||||
nullIfEmpty(v.cluster),
|
||||
nullIfEmpty(v.folder),
|
||||
v.sumDisk,
|
||||
v.sumVcpu,
|
||||
v.sumRam,
|
||||
v.isTemplate,
|
||||
v.poweredOn,
|
||||
v.srmPlaceholder,
|
||||
v.creation,
|
||||
int64(0), // deletion time refined later
|
||||
v.samples,
|
||||
avgVcpu,
|
||||
avgRam,
|
||||
avgDisk,
|
||||
avgPresent,
|
||||
tinPct, bronzePct, silverPct, goldPct,
|
||||
float64(v.tinHits), float64(v.bronzeHits), float64(v.silverHits), float64(v.goldHits),
|
||||
}
|
||||
if driver != "sqlite" {
|
||||
// Postgres expects primitive types, nulls are handled by pq via nil.
|
||||
for i, a := range args {
|
||||
if s, ok := a.(string); ok && s == "" {
|
||||
args[i] = nil
|
||||
}
|
||||
}
|
||||
}
|
||||
if _, err := tx.ExecContext(ctx, insert, args...); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
return tx.Commit()
|
||||
}
|
||||
|
||||
func int64OrZero(v sql.NullInt64) int64 {
|
||||
if v.Valid {
|
||||
return v.Int64
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func nullIfEmpty(s string) interface{} {
|
||||
if strings.TrimSpace(s) == "" {
|
||||
return nil
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
func makePlaceholders(driver string, n int) string {
|
||||
if driver == "sqlite" {
|
||||
parts := make([]string, n)
|
||||
for i := 0; i < n; i++ {
|
||||
parts[i] = "?"
|
||||
}
|
||||
return strings.Join(parts, ",")
|
||||
}
|
||||
parts := make([]string, n)
|
||||
for i := 0; i < n; i++ {
|
||||
parts[i] = fmt.Sprintf("$%d", i+1)
|
||||
}
|
||||
return strings.Join(parts, ",")
|
||||
}
|
||||
|
||||
func btoi(b bool) int64 {
|
||||
if b {
|
||||
return 1
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user