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