feat(monitor): add batch resource metrics query API (#24598)

Add unifiedmonitor class action `resource-metrics` with client options/CLI command.
Implement host/guest metric drivers to query cpu/mem/disk/net rates and attach alert state.

cmd/climc/shell/monitor/unifiedmonitor.go

@@ -41,4 +41,17 @@ func init() {
 		printObject(resp)
 		return nil
 	})
+
+	R(new(options.ResourceMetricsOptions), "monitor-resource-metrics", "Query batch resource metrics (cpu/mem/disk/net) and alert state", func(s *mcclient.ClientSession, opts *options.ResourceMetricsOptions) error {
+		input, err := opts.GetInput()
+		if err != nil {
+			return err
+		}
+		resp, err := monitor.UnifiedMonitorManager.PerformResourceMetrics(s, input)
+		if err != nil {
+			return errors.Wrap(err, "PerformResourceMetrics")
+		}
+		printObject(resp)
+		return nil
+	})
 }

pkg/apis/monitor/unifiedmonitor_query.go

@@ -313,3 +313,42 @@ func (c CdfQueryDataSet) Less(i, j int) bool {
 type CdfQueryOutput struct {
 	Data CdfQueryDataSet `json:"data"`
 }
+
+// ResourceMetricsQueryInput 批量查询资源监控数据的输入参数
+type ResourceMetricsQueryInput struct {
+	// 资源类型: "host" 或 "guest"
+	ResType string `json:"res_type"`
+	// 资源 ID 列表
+	ResIds []string `json:"res_ids"`
+	// 查询开始时间（默认: 1小时前）
+	StartTime time.Time `json:"start_time"`
+	// 查询结束时间（默认: now）
+	EndTime time.Time `json:"end_time"`
+	// 查询间隔（默认: 5m）
+	Interval string `json:"interval"`
+}
+
+// ResourceMetricValues 单个资源的监控数据
+type ResourceMetricValues struct {
+	// CPU 使用率 (%)
+	CpuUsage *float64 `json:"cpu_usage"`
+	// 内存使用率 (%)
+	MemUsage *float64 `json:"mem_usage"`
+	// 磁盘使用率 (%)
+	DiskUsage *float64 `json:"disk_usage"`
+	// 磁盘读速率 (Bps)
+	DiskReadRate *float64 `json:"disk_read_rate"`
+	// 磁盘写速率 (Bps)
+	DiskWriteRate *float64 `json:"disk_write_rate"`
+	// 网络入流量 (bps)
+	NetInRate *float64 `json:"net_in_rate"`
+	// 网络出流量 (bps)
+	NetOutRate *float64 `json:"net_out_rate"`
+	// 告警状态: init/attach/alerting
+	AlertState string `json:"alert_state"`
+}
+
+// ResourceMetricsQueryOutput 批量查询资源监控数据的输出
+type ResourceMetricsQueryOutput struct {
+	ResourceMetrics map[string]ResourceMetricValues `json:"resource_metrics"`
+}

pkg/mcclient/modules/monitor/mod_unifiedmonitor.go

@@ -52,3 +52,7 @@ func NewUnifiedMonitorManager() *SUnifiedMonitorManager {
 func (m *SUnifiedMonitorManager) PerformQuery(s *mcclient.ClientSession, input *monitor.MetricQueryInput) (jsonutils.JSONObject, error) {
 	return m.PerformClassAction(s, "query", jsonutils.Marshal(input))
 }
+
+func (m *SUnifiedMonitorManager) PerformResourceMetrics(s *mcclient.ClientSession, input *monitor.ResourceMetricsQueryInput) (jsonutils.JSONObject, error) {
+	return m.PerformClassAction(s, "resource-metrics", jsonutils.Marshal(input))
+}

pkg/mcclient/options/monitor/unifiedmonitor.go

@@ -26,6 +26,37 @@ import (
 	"yunion.io/x/onecloud/pkg/mcclient/modules/monitor"
 )
 
+type ResourceMetricsOptions struct {
+	RES_TYPE  string   `help:"resource type: host or guest" choices:"host|guest"`
+	ResIds    []string `help:"resource IDs" json:"res_ids" nargs:"+"`
+	StartTime string   `help:"start time (RFC3339). e.g.: 2023-12-06T21:54:42Z" json:"-"`
+	EndTime   string   `help:"end time (RFC3339). e.g.: 2023-12-18T21:54:42Z" json:"-"`
+	Interval  string   `help:"query interval. e.g.: 5m, 1h" json:"interval"`
+}
+
+func (o *ResourceMetricsOptions) GetInput() (*api.ResourceMetricsQueryInput, error) {
+	input := &api.ResourceMetricsQueryInput{
+		ResType:  o.RES_TYPE,
+		ResIds:   o.ResIds,
+		Interval: o.Interval,
+	}
+	if o.StartTime != "" {
+		t, err := time.Parse(time.RFC3339, o.StartTime)
+		if err != nil {
+			return nil, errors.Wrapf(err, "invalid start_time %q", o.StartTime)
+		}
+		input.StartTime = t
+	}
+	if o.EndTime != "" {
+		t, err := time.Parse(time.RFC3339, o.EndTime)
+		if err != nil {
+			return nil, errors.Wrapf(err, "invalid end_time %q", o.EndTime)
+		}
+		input.EndTime = t
+	}
+	return input, nil
+}
+
 type MeasurementsQueryOptions struct {
 	Scope           string `json:"scope"`
 	ProjectDomainId string `json:"project_domin_id"`

pkg/monitor/models/resource_metric_driver.go

@@ -0,0 +1,115 @@
+// Copyright 2019 Yunion
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package models
+
+import (
+	"yunion.io/x/onecloud/pkg/apis/monitor"
+)
+
+// IResourceMetricDriver 定义资源监控指标的驱动接口
+type IResourceMetricDriver interface {
+	// GetResType 返回资源类型标识
+	GetResType() string
+	// GetTagKey 返回 InfluxDB 中用于标识资源 ID 的 tag key
+	GetTagKey() string
+	// GetMetricSpecs 返回该资源类型的所有指标查询规格
+	GetMetricSpecs() []ResourceMetricSpec
+}
+
+// ResourceMetricSpec 描述一个 measurement 下需要查询的字段及其输出映射
+type ResourceMetricSpec struct {
+	Measurement string
+	Fields      []string
+	OutputKeys  []string
+}
+
+var resourceMetricDrivers = make(map[string]IResourceMetricDriver)
+
+// RegisterResourceMetricDriver 注册资源监控指标驱动
+func RegisterResourceMetricDriver(drv IResourceMetricDriver) {
+	resourceMetricDrivers[drv.GetResType()] = drv
+}
+
+// GetResourceMetricDriver 根据资源类型获取对应驱动
+func GetResourceMetricDriver(resType string) IResourceMetricDriver {
+	return resourceMetricDrivers[resType]
+}
+
+// SetResourceMetricValue 将查询结果赋值到 ResourceMetricValues 对应字段
+func SetResourceMetricValue(rv *monitor.ResourceMetricValues, key string, val float64) {
+	switch key {
+	case "cpu_usage":
+		rv.CpuUsage = &val
+	case "mem_usage":
+		rv.MemUsage = &val
+	case "disk_usage":
+		rv.DiskUsage = &val
+	case "disk_read_rate":
+		rv.DiskReadRate = &val
+	case "disk_write_rate":
+		rv.DiskWriteRate = &val
+	case "net_in_rate":
+		rv.NetInRate = &val
+	case "net_out_rate":
+		rv.NetOutRate = &val
+	}
+}
+
+// sHostMetricDriver 宿主机监控指标驱动
+type sHostMetricDriver struct{}
+
+func (d *sHostMetricDriver) GetResType() string {
+	return monitor.METRIC_RES_TYPE_HOST
+}
+
+func (d *sHostMetricDriver) GetTagKey() string {
+	return monitor.MEASUREMENT_TAG_ID[monitor.METRIC_RES_TYPE_HOST]
+}
+
+func (d *sHostMetricDriver) GetMetricSpecs() []ResourceMetricSpec {
+	return []ResourceMetricSpec{
+		{"cpu", []string{"usage_active"}, []string{"cpu_usage"}},
+		{"mem", []string{"used_percent"}, []string{"mem_usage"}},
+		{"disk", []string{"used_percent"}, []string{"disk_usage"}},
+		{"diskio", []string{"read_bps", "write_bps"}, []string{"disk_read_rate", "disk_write_rate"}},
+		{"net", []string{"bps_recv", "bps_sent"}, []string{"net_in_rate", "net_out_rate"}},
+	}
+}
+
+// sGuestMetricDriver 虚拟机监控指标驱动
+type sGuestMetricDriver struct{}
+
+func (d *sGuestMetricDriver) GetResType() string {
+	return monitor.METRIC_RES_TYPE_GUEST
+}
+
+func (d *sGuestMetricDriver) GetTagKey() string {
+	return monitor.MEASUREMENT_TAG_ID[monitor.METRIC_RES_TYPE_GUEST]
+}
+
+func (d *sGuestMetricDriver) GetMetricSpecs() []ResourceMetricSpec {
+	return []ResourceMetricSpec{
+		{"vm_cpu", []string{"usage_active"}, []string{"cpu_usage"}},
+		{"vm_mem", []string{"used_percent"}, []string{"mem_usage"}},
+		{"vm_disk", []string{"used_percent"}, []string{"disk_usage"}},
+		{"vm_diskio", []string{"read_bps", "write_bps"}, []string{"disk_read_rate", "disk_write_rate"}},
+		{"vm_netio", []string{"bps_recv", "bps_sent"}, []string{"net_in_rate", "net_out_rate"}},
+	}
+}
+
+func init() {
+	RegisterResourceMetricDriver(&sHostMetricDriver{})
+	RegisterResourceMetricDriver(&sGuestMetricDriver{})
+}

pkg/monitor/models/unifiedmonitor.go

@@ -703,3 +703,126 @@ func (self *SUnifiedMonitorManager) GetPropertyCdfQuery(ctx context.Context, use
 	}
 	return result, nil
 }
+
+func (self *SUnifiedMonitorManager) PerformResourceMetrics(
+	ctx context.Context,
+	userCred mcclient.TokenCredential,
+	query jsonutils.JSONObject,
+	data jsonutils.JSONObject,
+) (jsonutils.JSONObject, error) {
+	input := new(monitor.ResourceMetricsQueryInput)
+	if err := data.Unmarshal(input); err != nil {
+		return nil, httperrors.NewInputParameterError("unmarshal input: %v", err)
+	}
+
+	drv := GetResourceMetricDriver(input.ResType)
+	if drv == nil {
+		return nil, httperrors.NewInputParameterError("unsupported res_type %q", input.ResType)
+	}
+	if len(input.ResIds) == 0 {
+		return nil, httperrors.NewMissingParameterError("res_ids")
+	}
+
+	if input.EndTime.IsZero() {
+		input.EndTime = time.Now()
+	}
+	if input.StartTime.IsZero() {
+		input.StartTime = input.EndTime.Add(-1 * time.Hour)
+	}
+	if len(input.Interval) == 0 {
+		input.Interval = "5m"
+	}
+
+	tagKey := drv.GetTagKey()
+	specs := drv.GetMetricSpecs()
+
+	result := make(map[string]*monitor.ResourceMetricValues, len(input.ResIds))
+	for _, id := range input.ResIds {
+		result[id] = &monitor.ResourceMetricValues{}
+	}
+
+	for _, spec := range specs {
+		qInput := mod.NewMetricQueryInputWithDB(TELEGRAF_DATABASE, spec.Measurement).SkipCheckSeries(true)
+		sel := qInput.Selects()
+		for _, f := range spec.Fields {
+			sel.Select(f).MEAN()
+		}
+		where := qInput.Where()
+		where.IN(tagKey, input.ResIds)
+		qInput.GroupBy().TAG(tagKey)
+		qInput.From(input.StartTime).To(input.EndTime).Interval(input.Interval)
+
+		queryData := qInput.ToQueryData()
+		dbRtn, err := self.performQuery(ctx, userCred, queryData)
+		if err != nil {
+			log.Warningf("query %s metrics error: %v", spec.Measurement, err)
+			continue
+		}
+
+		for _, series := range dbRtn.Series {
+			resId := series.Tags[tagKey]
+			if resId == "" {
+				continue
+			}
+			rv, ok := result[resId]
+			if !ok {
+				continue
+			}
+			// 取最后一个有效数据点
+			for pi := len(series.Points) - 1; pi >= 0; pi-- {
+				point := series.Points[pi]
+				if len(point) < 2 {
+					continue
+				}
+				allValid := true
+				for fi := 0; fi < len(spec.Fields) && fi < len(point)-1; fi++ {
+					if point[fi] == nil {
+						allValid = false
+						break
+					}
+					if fval, ok := point[fi].(*float64); !ok || fval == nil {
+						allValid = false
+						break
+					}
+				}
+				if !allValid {
+					continue
+				}
+				for fi, outputKey := range spec.OutputKeys {
+					if fi < len(point)-1 {
+						val := 0.0
+						if fval, ok := point[fi].(*float64); ok && fval != nil {
+							val = *fval
+						} else if fval, ok := point[fi].(float64); ok {
+							val = fval
+						}
+						SetResourceMetricValue(rv, outputKey, val)
+					}
+				}
+				break
+			}
+		}
+	}
+
+	// 查询告警状态
+	monResources, err := MonitorResourceManager.GetMonitorResources(monitor.MonitorResourceListInput{
+		ResId: input.ResIds,
+	})
+	if err != nil {
+		log.Warningf("GetMonitorResources error: %v", err)
+	} else {
+		for _, mr := range monResources {
+			if rv, ok := result[mr.ResId]; ok {
+				rv.AlertState = mr.AlertState
+			}
+		}
+	}
+
+	output := monitor.ResourceMetricsQueryOutput{
+		ResourceMetrics: make(map[string]monitor.ResourceMetricValues, len(result)),
+	}
+	for id, rv := range result {
+		output.ResourceMetrics[id] = *rv
+	}
+	return jsonutils.Marshal(output), nil
+}