lywsvip/cloudpods
1
0
Fork 0
mirror of https://github.com/yunionio/cloudpods.git synced 2026-07-01 00:34:18 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix(region): apsara metric data

Browse Source
This commit is contained in:
Qu Xuan
2021-09-13 17:02:48 +08:00
parent 55cb7ef439
commit 1467c0bc84
6 changed files with 156 additions and 78 deletions
Download Patch File Download Diff File Expand all files Collapse all files

155
pkg/cloudmon/collectors/apsaramon/apsaraservice.go
View File

@@ -22,12 +22,23 @@ import (
"yunion.io/x/jsonutils"
"yunion.io/x/log"
api "yunion.io/x/onecloud/pkg/apis/compute"
"yunion.io/x/onecloud/pkg/cloudmon/collectors/common"
"yunion.io/x/onecloud/pkg/cloudprovider"
"yunion.io/x/onecloud/pkg/multicloud/apsara"
"yunion.io/x/onecloud/pkg/util/influxdb"
)
type SApsaraMetric struct {
Timestamp int64
UserId string
InstanceId string
Maximum float64
Minimum float64
Average float64
NodeId string
}
func (self *SApsaraCloudReport) collectRegionMetricOfHost(region cloudprovider.ICloudRegion, servers []jsonutils.JSONObject) error {
dataList := make([]influxdb.SMetricData, 0)
aliReg := region.(*apsara.SRegion)
@@ -35,29 +46,39 @@ func (self *SApsaraCloudReport) collectRegionMetricOfHost(region cloudprovider.I
if err != nil {
return err
}
for metricName, influxDbSpecs := range aliMetricSpecs {
rtnArray, _, err := aliReg.DescribeMetricList(metricName, "acs_ecs_dashboard", since, until, "")
for metricName, influxDbSpecs := range apsaraMetricSpecs {
metricArray, err := aliReg.FetchMetricData(metricName, "acs_ecs_dashboard", since, until)
if err != nil {
log.Errorln(err)
continue
}
if len(rtnArray) > 0 {
for _, rtnMetric := range rtnArray {
for _, server := range servers {
external_id, _ := server.GetString("external_id")
if instanceId, _ := rtnMetric.GetString("instanceId"); instanceId == external_id {
metric, err := common.FillVMCapacity(server.(*jsonutils.JSONDict))
if err != nil {
return err
}
dataList = append(dataList, metric)
serverMetric, err := self.collectMetricFromThisServer(server, rtnMetric, influxDbSpecs)
if err != nil {
return err
}
dataList = append(dataList, serverMetric)
}
instances := []api.ServerDetails{}
instanceMaps := map[string]api.ServerDetails{}
jsonutils.Update(&instances, servers)
for i := range instances {
if len(instances[i].ExternalId) == 0 {
continue
}
instanceMaps[instances[i].ExternalId] = instances[i]
}
metrics := []SApsaraMetric{}
jsonutils.Update(&metrics, metricArray)
for _, rtnMetric := range metrics {
server, ok := instanceMaps[rtnMetric.InstanceId]
if ok {
metric, err := common.FillVMCapacity(jsonutils.Marshal(server).(*jsonutils.JSONDict))
if err != nil {
return err
}
dataList = append(dataList, metric)
serverMetric, err := self.collectMetricFromThisServer(jsonutils.Marshal(server), rtnMetric, influxDbSpecs)
if err != nil {
return err
}
dataList = append(dataList, serverMetric)
}
}
}
@@ -92,24 +113,25 @@ func (self *SApsaraCloudReport) collectRegionMetricOfRedis(region cloudprovider.
}
for metricName, influxDbSpecs := range aliRedisMetricSpecs {
for _, pre := range redisDeployType {
rtnArray, _, err := aliReg.DescribeMetricList(pre+metricName, "acs_kvstore", since, until, "")
rtnArray, err := aliReg.FetchMetricData(pre+metricName, "acs_kvstore", since, until)
if err != nil {
log.Errorln(err)
continue
}
if len(rtnArray) > 0 {
for _, rtnMetric := range rtnArray {
for _, server := range servers {
external_id, _ := server.GetString("external_id")
if instanceId, _ := rtnMetric.GetString("instanceId"); instanceId == external_id {
serverMetric, err := self.collectMetricFromThisServer(server, rtnMetric, influxDbSpecs)
node_id, _ := rtnMetric.GetString("nodeId")
serverMetric.Tags = append(serverMetric.Tags, influxdb.SKeyValue{Key: "node_id", Value: node_id})
if err != nil {
return err
}
dataList = append(dataList, serverMetric)
metrics := []SApsaraMetric{}
jsonutils.Update(&metrics, rtnArray)
for _, rtnMetric := range metrics {
for _, server := range servers {
external_id, _ := server.GetString("external_id")
if rtnMetric.InstanceId == external_id {
serverMetric, err := self.collectMetricFromThisServer(server, rtnMetric, influxDbSpecs)
serverMetric.Tags = append(serverMetric.Tags, influxdb.SKeyValue{Key: "node_id", Value: rtnMetric.NodeId})
if err != nil {
return err
}
dataList = append(dataList, serverMetric)
}
}
}
@@ -126,30 +148,29 @@ func (self *SApsaraCloudReport) collectRegionMetricOfRds(region cloudprovider.IC
return err
}
for metricName, influxDbSpecs := range aliRdsMetricSpecs {
rtnArray, _, err := aliReg.DescribeMetricList(metricName, "acs_rds_dashboard", since, until, "")
rtnArray, err := aliReg.FetchMetricData(metricName, "acs_rds_dashboard", since, until)
if err != nil {
log.Errorln(err)
continue
}
if len(rtnArray) > 0 {
for _, rtnMetric := range rtnArray {
for _, server := range servers {
external_id, _ := server.GetString("external_id")
if instanceId, _ := rtnMetric.GetString("instanceId"); instanceId == external_id {
metric, err := common.FillVMCapacity(server.(*jsonutils.JSONDict))
if err != nil {
return err
}
dataList = append(dataList, metric)
serverMetric, err := self.collectMetricFromThisServer(server, rtnMetric, influxDbSpecs)
if err != nil {
return err
}
dataList = append(dataList, serverMetric)
metrics := []SApsaraMetric{}
jsonutils.Update(&metrics, rtnArray)
for _, rtnMetric := range metrics {
for _, server := range servers {
external_id, _ := server.GetString("external_id")
if rtnMetric.InstanceId == external_id {
metric, err := common.FillVMCapacity(server.(*jsonutils.JSONDict))
if err != nil {
return err
}
dataList = append(dataList, metric)
serverMetric, err := self.collectMetricFromThisServer(server, rtnMetric, influxDbSpecs)
if err != nil {
return err
}
dataList = append(dataList, serverMetric)
}
}
}
}
@@ -164,7 +185,7 @@ func (self *SApsaraCloudReport) collectRegionMetricOfOss(region cloudprovider.IC
return err
}
for metricName, influxDbSpecs := range aliOSSMetricSpecs {
rtnArray, _, err := aliReg.DescribeMetricList(metricName, "acs_oss", since, until, "")
rtnArray, err := aliReg.FetchMetricData(metricName, "acs_oss", since, until)
if err != nil {
log.Errorln(err)
continue
@@ -196,22 +217,24 @@ func (self *SApsaraCloudReport) collectRegionMetricOfElb(region cloudprovider.IC
return err
}
for metricName, influxDbSpecs := range aliElbMetricSpecs {
rtnArray, _, err := aliReg.DescribeMetricList(metricName, "acs_slb_dashboard", since, until, "")
rtnArray, err := aliReg.FetchMetricData(metricName, "acs_slb_dashboard", since, until)
if err != nil {
log.Errorln(err)
continue
}
if len(rtnArray) > 0 {
for _, rtnMetric := range rtnArray {
for _, server := range servers {
external_id, _ := server.GetString("external_id")
if instanceId, _ := rtnMetric.GetString("instanceId"); instanceId == external_id {
serverMetric, err := self.collectMetricFromThisServer(server, rtnMetric, influxDbSpecs)
if err != nil {
return err
}
dataList = append(dataList, serverMetric)
metrics := []SApsaraMetric{}
jsonutils.Update(&metrics, rtnArray)
for _, rtnMetric := range metrics {
for _, server := range servers {
external_id, _ := server.GetString("external_id")
if rtnMetric.InstanceId == external_id {
serverMetric, err := self.collectMetricFromThisServer(server, rtnMetric, influxDbSpecs)
if err != nil {
return err
}
dataList = append(dataList, serverMetric)
}
}
}
@@ -219,19 +242,15 @@ func (self *SApsaraCloudReport) collectRegionMetricOfElb(region cloudprovider.IC
return common.SendMetrics(self.Session, dataList, self.Args.Debug, "")
}
func (self *SApsaraCloudReport) collectMetricFromThisServer(server jsonutils.JSONObject, rtnMetric jsonutils.JSONObject,
func (self *SApsaraCloudReport) collectMetricFromThisServer(server jsonutils.JSONObject, rtnMetric SApsaraMetric,
influxDbSpecs []string) (influxdb.SMetricData, error) {
metric, err := self.NewMetricFromJson(server)
//metric, err := common.JsonToMetric(server.(*jsonutils.JSONDict), "", common.ServerTags, make([]string, 0))
if err != nil {
return influxdb.SMetricData{}, err
}
timestamp, _ := rtnMetric.Get("timestamp")
metric.Timestamp = time.Unix(timestamp.(*jsonutils.JSONInt).Value()/1000, 0)
fieldValue, err := rtnMetric.Float(UNIT_AVERAGE)
if err != nil {
return influxdb.SMetricData{}, err
}
metric.Timestamp = time.Unix(rtnMetric.Timestamp/1000, 0)
fieldValue := rtnMetric.Average
//根据条件拼装metric的tag和metirc信息
influxDbSpec := influxDbSpecs[2]
measurement := common.SubstringBefore(influxDbSpec, ".")

12
pkg/cloudmon/collectors/apsaramon/apsarautils.go
View File

@@ -131,6 +131,18 @@ var aliMetricSpecs = map[string][]string{
"DiskReadIOPS": {DEFAULT_STATISTICS, UNIT_CPS, INFLUXDB_FIELD_DISK_READ_IOPS},
"DiskWriteIOPS": {DEFAULT_STATISTICS, UNIT_CPS, INFLUXDB_FIELD_DISK_WRITE_IOPS},
}
var apsaraMetricSpecs = map[string][]string{
"vm.MemoryUtilization": {DEFAULT_STATISTICS, UNIT_PERCENT, INFLUXDB_FIELD_MEM_USAGE},
"vm.CPUUtilization": {DEFAULT_STATISTICS, UNIT_PERCENT, INFLUXDB_FIELD_CPU_USAGE},
"InternetInRate": {DEFAULT_STATISTICS, UNIT_BPS, INFLUXDB_FIELD_NET_BPS_RX_INTERNET},
"IntranetInRate": {DEFAULT_STATISTICS, UNIT_BPS, INFLUXDB_FIELD_NET_BPS_RX_INTRANET},
"InternetOutRate": {DEFAULT_STATISTICS, UNIT_BPS, INFLUXDB_FIELD_NET_BPS_TX_INTERNET},
"IntranetOutRate": {DEFAULT_STATISTICS, UNIT_BPS, INFLUXDB_FIELD_NET_BPS_TX_INTRANET},
"DiskReadBPS": {DEFAULT_STATISTICS, UNIT_BYTEPS, INFLUXDB_FIELD_DISK_READ_BPS},
"DiskWriteBPS": {DEFAULT_STATISTICS, UNIT_BYTEPS, INFLUXDB_FIELD_DISK_WRITE_BPS},
"DiskReadIOPS": {DEFAULT_STATISTICS, UNIT_CPS, INFLUXDB_FIELD_DISK_READ_IOPS},
"DiskWriteIOPS": {DEFAULT_STATISTICS, UNIT_CPS, INFLUXDB_FIELD_DISK_WRITE_IOPS},
}
var aliRdsMetricSpecs = map[string][]string{
"CpuUsage": {DEFAULT_STATISTICS, UNIT_PERCENT, INFLUXDB_FIELD_RDS_CPU_USAGE},
"MemoryUsage": {DEFAULT_STATISTICS, UNIT_PERCENT, INFLUXDB_FIELD_RDS_MEM_USAGE},

2
pkg/cloudmon/collectors/common/multiCloudMonUtils.go
View File

@@ -326,7 +326,7 @@ func TimeRangeFromArgs(args *ReportOptions) (since, until time.Time, err error)
return since, until, err
}
} else {
period64, err := strconv.ParseInt(args.Interval, 10, 8)
period64, err := strconv.ParseInt(args.Interval, 10, 32)
if err != nil {
return since, until, err
}

40
pkg/multicloud/apsara/monitor.go
View File

@@ -18,6 +18,8 @@ import (
"strconv"
"time"
alierr "github.com/aliyun/alibaba-cloud-sdk-go/sdk/errors"
"yunion.io/x/jsonutils"
"yunion.io/x/pkg/errors"
)
@@ -138,7 +140,7 @@ func (r *SRegion) FetchMetrics(ns string) ([]SMetricMeta, error) {
return metrics, nil
}
func (r *SRegion) DescribeMetricList(name string, ns string, since time.Time, until time.Time, nextToken string) ([]jsonutils.JSONObject, string, error) {
func (r *SRegion) DescribeMetricList(department, name string, ns string, since time.Time, until time.Time, nextToken string) ([]jsonutils.JSONObject, string, error) {
params := make(map[string]string)
params["MetricName"] = name
params["Namespace"] = ns
@@ -152,6 +154,9 @@ func (r *SRegion) DescribeMetricList(name string, ns string, since time.Time, un
if !until.IsZero() {
params["EndTime"] = strconv.FormatInt(until.Unix()*1000, 10)
}
if len(department) > 0 {
params["Department"] = department
}
body, err := r.metricsRequest("DescribeMetricList", params)
if err != nil {
return nil, "", errors.Wrap(err, "region.MetricRequest")
@@ -172,11 +177,11 @@ func (r *SRegion) DescribeMetricList(name string, ns string, since time.Time, un
return dataArray, nToken, nil
}
func (r *SRegion) FetchMetricData(name string, ns string, since time.Time, until time.Time) ([]jsonutils.JSONObject, error) {
func (r *SRegion) fetchMetricData(department string, name string, ns string, since time.Time, until time.Time) ([]jsonutils.JSONObject, error) {
data := make([]jsonutils.JSONObject, 0)
nextToken := ""
for {
datArray, next, err := r.DescribeMetricList(name, ns, since, until, nextToken)
datArray, next, err := r.DescribeMetricList(department, name, ns, since, until, nextToken)
if err != nil {
return nil, errors.Wrap(err, "r.DescribeMetricList")
}
@@ -188,3 +193,32 @@ func (r *SRegion) FetchMetricData(name string, ns string, since time.Time, until
}
return data, nil
}
func (r *SRegion) FetchMetricData(name string, ns string, since time.Time, until time.Time) ([]jsonutils.JSONObject, error) {
data := make([]jsonutils.JSONObject, 0)
part, err := r.fetchMetricData("1", name, ns, since, until)
if err == nil {
return part, nil
}
if err != nil {
if e, ok := errors.Cause(err).(*alierr.ServerError); ok && e.ErrorCode() != "NoPermission" {
return nil, errors.Wrapf(err, "fetchMetricData")
}
}
orgs, err := r.GetClient().GetOrganizationList()
if err != nil {
return nil, errors.Wrapf(err, "GetOrganizationList")
}
for i := range orgs {
if orgs[i].ParentId == "1" {
part, err := r.fetchMetricData(orgs[i].Id, name, ns, since, until)
if err != nil {
return nil, errors.Wrapf(err, "fetchMetricData")
}
data = append(data, part...)
}
}
return data, nil
}

17
pkg/multicloud/apsara/organization.go
View File

@@ -63,8 +63,7 @@ func (self *SApsaraClient) GetOrganizationTree(id int) (*SOrganizationTree, erro
id = 1
}
params := map[string]string{
"Id": fmt.Sprintf("%d", id),
"RegionId": "ln-1",
"Id": fmt.Sprintf("%d", id),
}
resp, err := self.ascmRequest("GetOrganizationTree", params)
if err != nil {
@@ -78,6 +77,20 @@ func (self *SApsaraClient) GetOrganizationTree(id int) (*SOrganizationTree, erro
return &tree, nil
}
func (self *SApsaraClient) GetOrganizationList() ([]SOrganization, error) {
params := map[string]string{"Id": "1"}
resp, err := self.ascmRequest("GetOrganizationList", params)
if err != nil {
return nil, err
}
result := []SOrganization{}
err = resp.Unmarshal(&result, "data")
if err != nil {
return nil, errors.Wrapf(err, "resp.Unmarshal")
}
return result, nil
}
func (self *SOrganizationTree) ListProjects() []SResourceGroupList {
ret := []SResourceGroupList{}
for i := range self.ResourceGroupList {

8
pkg/multicloud/apsara/shell/resourcegroup.go
View File

@@ -63,7 +63,7 @@ func init() {
Id int
}
shellutils.R(&OrganizationListOptions{}, "organization-list", "List organization tree", func(cli *apsara.SRegion, args *OrganizationListOptions) error {
shellutils.R(&OrganizationListOptions{}, "organization-tree", "List organization tree", func(cli *apsara.SRegion, args *OrganizationListOptions) error {
org, err := cli.GetClient().GetOrganizationTree(args.Id)
if err != nil {
return err
@@ -72,12 +72,12 @@ func init() {
return nil
})
shellutils.R(&OrganizationListOptions{}, "organization-project-list", "List organization tree", func(cli *apsara.SRegion, args *OrganizationListOptions) error {
org, err := cli.GetClient().GetOrganizationTree(args.Id)
shellutils.R(&OrganizationListOptions{}, "organization-list", "List organization", func(cli *apsara.SRegion, args *OrganizationListOptions) error {
orgs, err := cli.GetClient().GetOrganizationList()
if err != nil {
return err
}
printList(org.ListProjects(), 0, 0, 0, nil)
printList(orgs, 0, 0, 0, nil)
return nil
})