Rank VMs by allocated storage, and unbreak the VM metrics

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":

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":

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").

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
 	}

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")