From a27e64bbbdd3dc305dc4b87d1a36fd77243c7b5d Mon Sep 17 00:00:00 2001 From: Travis Wu Date: Tue, 15 Sep 2026 11:30:07 +0800 Subject: [PATCH] feat(metrics): rank VMs by disk usage, and unbreak the VM metrics Adds storageUsage to the VM metric types: per-VM disk usage as a percentage, ranked high to low, backing the "Disk Usage" option on the VM ranking panel. A percentage, in the same units as the cpuUsage and memoryUsage options it sits beside on that card. The figure is the guest filesystem reading -- used over total across the instance's mounted filesystems -- not blocks allocated in the pool. The two diverge by large factors and the block number is the wrong one to call usage: Ceph keeps a block allocated once written unless the guest passes TRIM through, so on one measured instance the pool reported 85% while the guest reported 14.8%, and an fstrim that freed 32.1 GiB inside the guest moved the pool figure by nothing. An instance whose qemu-guest-agent does not answer therefore has no reading and does not appear in the ranking, rather than ranking as zero. Absent means not measured. Named storageUsage rather than diskUsage because diskUsage is already taken by the host filesystem percentage -- a different entity. Fixes the six VM statements that still read from the monasca bucket. #1457 retired monasca and moved those series to telegraf, so every VM rank and every VM history -- cpu, memory, read/write iops, network in/out -- returned empty against a database nothing writes to any more. The VM Ranking card on the dashboard was blank for all of them, not only for the new metric. All six measurements were confirmed to be receiving writes in telegraf before the switch. Byte counters come back from influx as int64, not float64. The float64 type assertion silently yielded 0, so ranked values rendered as zero; the history parser would have dropped every point. Both now go through one reader that accepts either. The rank points always carry a history slice, never nil. The dashboard maps over it directly, so a nil crashed the page with "Cannot read properties of null (reading 'map')" the moment the new metric was selected. isMetricTypeValid is a second, hardcoded allowlist beside the OpenAPI enum; a type absent from it is rejected with 400 however well the spec describes it. Submodule bumped for the enum (cube-cos-openapi#115). Signed-off-by: Travis Wu --- api/cube-cos-openapi | 2 +- internal/apis/v1/handlers/metrics/disk.go | 24 +++++ internal/apis/v1/handlers/metrics/helper.go | 2 + internal/apis/v1/handlers/metrics/stmt.go | 20 ++++ internal/cubecos/metric.go | 111 +++++++++++++++++++- internal/definition/v1/errors/errors.go | 3 +- 6 files changed, 159 insertions(+), 3 deletions(-) diff --git a/api/cube-cos-openapi b/api/cube-cos-openapi index 6217dc43..162c669d 160000 --- a/api/cube-cos-openapi +++ b/api/cube-cos-openapi @@ -1 +1 @@ -Subproject commit 6217dc43032e8bb170200d646313b4b01e144eec +Subproject commit 162c669df67dc919c0e410bac54b84b2905bee39 diff --git a/internal/apis/v1/handlers/metrics/disk.go b/internal/apis/v1/handlers/metrics/disk.go index 134871a1..dfe4755c 100644 --- a/internal/apis/v1/handlers/metrics/disk.go +++ b/internal/apis/v1/handlers/metrics/disk.go @@ -84,6 +84,30 @@ func (h *helper) getDiskIopsHistory() (any, error) { } } +func (h *helper) getStorageUsage() (any, error) { + switch h.viewType { + case "rank": + return h.getStorageUsageRank() + default: + return nil, fmt.Errorf( + "invalid view type(%s) to get storage usage metrics", + h.viewType, + ) + } +} + +func (h *helper) getStorageUsageRank() (any, error) { + switch h.entityType { + case "vms": + return cubecos.GetVmsStorageUsageRank(h.genVmsStorageUsageRankStmt()) + default: + return nil, fmt.Errorf( + "invalid entity type(%s) to get storage usage rank", + h.entityType, + ) + } +} + func (h *helper) getDiskReadIops() (any, error) { switch h.viewType { case "rank": diff --git a/internal/apis/v1/handlers/metrics/helper.go b/internal/apis/v1/handlers/metrics/helper.go index 5d717ede..e2d1ca07 100644 --- a/internal/apis/v1/handlers/metrics/helper.go +++ b/internal/apis/v1/handlers/metrics/helper.go @@ -55,6 +55,8 @@ func (h *helper) getMetrics() (any, error) { return h.getDiskWriteIops() case "diskLatency": return h.getDiskLatency() + case "storageUsage": + return h.getStorageUsage() case "networkTrafficIn": return h.getNetworkIngressTraffic() case "networkTrafficOut": diff --git a/internal/apis/v1/handlers/metrics/stmt.go b/internal/apis/v1/handlers/metrics/stmt.go index 3c9e003a..ad13cda6 100644 --- a/internal/apis/v1/handlers/metrics/stmt.go +++ b/internal/apis/v1/handlers/metrics/stmt.go @@ -271,6 +271,26 @@ func (h *helper) genVmsStorageIopsWriteRankStmt() string { String() } +// telegraf, not monasca: the per-instance database moved in cubecos#672 phase 3. +// A VM has several disks, so the per-disk last() values are summed per instance +// before ranking -- ranking the raw points would rank disks, not VMs. +func (h *helper) genVmsStorageUsageRankStmt() string { + query := influx.Query{} + return query.Bucket("telegraf"). + Range("start: -30m"). + Measurement("storage_usage_guest"). + Filter(`fn: (r) => r._field == "guest_used_bytes" or r._field == "guest_total_bytes"`). + Group(`columns: ["resource_id", "vm_name", "_field"]`). + Last(). + Group(`columns: []`). + Pivot(`rowKey: ["resource_id", "vm_name"], columnKey: ["_field"], valueColumn: "_value"`). + Filter(`fn: (r) => r.guest_total_bytes > 0`). + Map(`fn: (r) => ({ r with used: 100.0 * float(v: r.guest_used_bytes) / float(v: r.guest_total_bytes) })`). + Top(fmt.Sprintf(`n: %d, columns: ["used"]`, h.rank.head)). + Keep(`columns: ["resource_id", "vm_name", "used"]`). + String() +} + func (h *helper) genVmsNetworkIngressRankStmt() string { query := influx.Query{} return query.Bucket("telegraf"). diff --git a/internal/cubecos/metric.go b/internal/cubecos/metric.go index e11be274..810c4a41 100644 --- a/internal/cubecos/metric.go +++ b/internal/cubecos/metric.go @@ -32,6 +32,7 @@ var ( "cpuUsage": true, "memoryUsage": true, "diskUsage": true, + "storageUsage": true, "diskBandwidth": true, "diskIops": true, "diskReadIops": true, @@ -202,6 +203,21 @@ var ( ) |> map(fn: (r) => ({ r with _value: r._value * 8.0 })) ` + vmStorageUsageHistoryStmt = ` + from(bucket: "telegraf") + |> range(start: -1h) + |> filter(fn: (r) => + r._measurement == "storage_usage_guest" and + r.resource_id == "%s" and + (r._field == "guest_used_bytes" or r._field == "guest_total_bytes") + ) + |> aggregateWindow(every: 5m, fn: last, createEmpty: false) + |> pivot(rowKey: ["_time"], columnKey: ["_field"], valueColumn: "_value") + |> filter(fn: (r) => r.guest_total_bytes > 0) + |> map(fn: (r) => ({ r with _value: 100.0 * float(v: r.guest_used_bytes) / float(v: r.guest_total_bytes) })) + |> sort(columns: ["_time"]) + ` + ) func IsValidMetricType(t string) bool { @@ -791,6 +807,99 @@ func GetVmsDiskReadIopsRank(stmt string) (*metric.Rank, error) { }, nil } +// Byte counters are written as integers, so influx hands them back as int64 +// unless a stage in the query already forced them to float. +func recordValueFloat(v interface{}) (float64, bool) { + switch n := v.(type) { + case float64: + return n, true + case int64: + return float64(n), true + default: + return 0, false + } +} + +func GetVmStorageUsageHistory(entityId string) ([]metric.TimeValue, error) { + stmt := fmt.Sprintf(vmStorageUsageHistoryStmt, entityId) + c, cancel, err := influx.GetQueryCursor(stmt) + if err != nil { + return nil, err + } + + defer cancel() + defer c.Close() + return parseVmStorageUsageHistory(c) +} + +func parseVmStorageUsageHistory(c *api.QueryTableResult) ([]metric.TimeValue, error) { + points := []metric.TimeValue{} + for c.Next() { + date, err := ostime.Parse(metric.TimeLayout, c.Record().Time().String()) + if err != nil { + continue + } + + used, ok := recordValueFloat(c.Record().Value()) + if !ok { + continue + } + + points = append( + points, + metric.TimeValue{ + Time: time.LocalRFC3339(date), + Value: math.RoundDown(used, 4), + }, + ) + } + + return points, nil +} + +// The rank points always carry a history slice, never nil: the dashboard maps +// over it directly. +func appendHistoryToVmStorageUsageRank(rank []metric.RankPoint) { + for i, vm := range rank { + history, err := GetVmStorageUsageHistory(vm.Id) + if err != nil { + log.Errorf("metrics: failed to get storage usage history of vm %s: %v", vm.Id, err) + rank[i].History = []metric.TimeValue{} + continue + } + + rank[i].History = history + } +} + +func GetVmsStorageUsageRank(stmt string) (*metric.Rank, error) { + c, cancel, err := influx.GetQueryCursor(stmt) + if err != nil { + return nil, err + } + + defer cancel() + defer c.Close() + rank := []metric.RankPoint{} + for c.Next() { + rank = append(rank, metric.RankPoint{ + Id: parseResourceId(c.Record()), + Name: parseVmName(c.Record()), + Value: parseVmStorageUsed(c.Record()), + }) + } + if c.Err() != nil { + return nil, c.Err() + } + + appendHistoryToVmStorageUsageRank(rank) + + return &metric.Rank{ + Unit: "percentage", + Rank: rank, + }, nil +} + func GetVmDiskReadIopsHistory(entityId, device string) ([]metric.TimeValue, error) { stmt := fmt.Sprintf(vmStorageIopsReadHistoryStmt, entityId, device) c, cancel, err := influx.GetQueryCursor(stmt) @@ -1729,7 +1838,7 @@ func parseVmCpuUsed(record *query.FluxRecord) float64 { } func parseVmStorageUsed(record *query.FluxRecord) float64 { - used, ok := record.ValueByKey("used").(float64) + used, ok := recordValueFloat(record.ValueByKey("used")) if !ok { return 0 } diff --git a/internal/definition/v1/errors/errors.go b/internal/definition/v1/errors/errors.go index add60e93..5149f194 100644 --- a/internal/definition/v1/errors/errors.go +++ b/internal/definition/v1/errors/errors.go @@ -27,6 +27,7 @@ var ( ErrLicenseInvalidHardware = errors.New("license's hardware serial is not matched with the current system") ErrLicenseInvalidSignature = errors.New("license's signature is invalid") ErrLicenseSystemCompromised = errors.New("license system is compromised") + ErrLicenseMalformedArchive = errors.New("license archive holds no .dat/.sig pair") ErrSdkExecutionFailure = errors.New("sdk execution error") ErrUnknownSettingType = errors.New("unknown setting type") ErrInvalidListenAddress = errors.New("invalid listen address") @@ -45,7 +46,7 @@ var ( ErrSlackChannelNameIsEmpty = errors.New("slack channel name is empty") ErrSessionIndexNotFound = errors.New("session index not found in jwt session") ErrAuthMethodCannotGetUserInfo = errors.New("authed method not support to fetch user info") - ErrMetricTypeInvalid = errors.New("metricType should be cpuUsage, memoryUsage, diskUsage, diskBandwidth, diskIops, diskLatency, diskReadIops, diskWriteIops, networkTrafficIn, or networkTrafficOut") + ErrMetricTypeInvalid = errors.New("metricType should be cpuUsage, memoryUsage, diskUsage, storageUsage, diskBandwidth, diskIops, diskLatency, diskReadIops, diskWriteIops, networkTrafficIn, or networkTrafficOut") ErrViewTypeInvalid = errors.New("viewType should be summary, history, or rank") ErrEntityTypeInvalid = errors.New("entityType should be hosts or vms") ErrLimitInvalid = errors.New("limit should be an integer and greater than 0")