Merge pull request #53 from KunfengHe/br_release_sdk_v3

support for MRS & DAYU & LakeFormation

Author: KunfengHe

README.md

@@ -2,6 +2,8 @@
 
 Prometheus cloudeye exporter for [Huaweicloud](https://www.huaweicloud.com/).
 
+Note: The plug-in is applicable only to the Huaweicloud regions.
+
 [中文](./README_cn.md)
 
 ## Download
@@ -39,7 +41,7 @@ Usage of ./cloudeye-exporter:
 ```
 
 ## Example of config file(clouds.yml)
-The "URL" value can be get from [Identity and Access Management (IAM) endpoint list](https://developer.huaweicloud.com/en-us/endpoint).
+The "URL" value can be get from [Identity and Access Management (IAM) endpoint list](https://developer.huaweicloud.com/endpoint?IAM).
 ```
 global:
   prefix: "huaweicloud"
@@ -48,7 +50,7 @@ global:
   scrape_batch_size: 300
 
 auth:
-  auth_url: "https://iam.xxx.yyy.com/v3"
+  auth_url: "https://iam.{region_id}.myhuaweicloud.com/v3"
   project_name: "{project_name}"
   access_key: "{access_key}"
   secret_key: "{secret_key}"

README_cn.md

@@ -3,6 +3,7 @@
 
 [华为云](https://www.huaweicloud.com/)云监控的 Prometheus Exporter.
 
+注意：该插件仅适用于华为云局点。
 
 ## 介绍
 Prometheus是用于展示大型测量数据的开源可视化工具，在工业监控、气象监控、家居自动化和过程管理等领域也有着较广泛的用户基础。将华为云Cloudeye服务接入 prometheus后，您可以利用 prometheus更好地监控和分析来自 Cloudeye服务的数据。
@@ -46,6 +47,9 @@ Prometheus是用于展示大型测量数据的开源可视化工具，在工业
 |VPC终端节点|SYS.VPCEP |√|RMS|
 |图引擎服务GES|SYS.GES|√|RMS|
 |数据库安全服务DBSS|SYS.DBSS |√|RMS|
+|MapReduce服务|SYS.MRS |√|RMS/云服务|
+|湖仓构建服务|SYS.LakeFormation |√|RMS/云服务|
+|智能数据湖运营平台|SYS.DAYU |√|云服务|
 
 注：自定义标签时，key只能包含大写字母、小写字母以及中划线
 
@@ -64,20 +68,20 @@ Prometheus是用于展示大型测量数据的开源可视化工具，在工业
 # 参考命令：
 mkdir cloudeye-exporter
 cd cloudeye-exporter
-wget https://github.com/huaweicloud/cloudeye-exporter/releases/download/v2.0.4/cloudeye-exporter.v2.0.4.tar.gz
-tar -xzvf cloudeye-exporter.v2.0.4.tar.gz
+wget https://github.com/huaweicloud/cloudeye-exporter/releases/download/v2.0.5/cloudeye-exporter.v2.0.5.tar.gz
+tar -xzvf cloudeye-exporter.v2.0.5.tar.gz
 ```
 2. 编辑clouds.yml文件配置公有云信息
 ```
 global:
   port: ":8087"
   scrape_batch_size: 300
 auth:
-auth_url: "https://iam.cn-north-1.myhuaweicloud.com/v3"
-project_name: "cn-north-1"
-access_key: ""
-secret_key: ""
-region: "cn-north-1"
+  auth_url: "https://iam.{region_id}.myhuaweicloud.com/v3"
+  project_name: "cn-north-1"
+  access_key: ""
+  secret_key: ""
+  region: "cn-north-1"
 ```
 注：默认的监控端口为8087.
 

collector/as.go

@@ -4,8 +4,6 @@ import (
 	"time"
 
 	cesmodel "github.com/huaweicloud/huaweicloud-sdk-go-v3/services/ces/v1/model"
-
-	"github.com/huaweicloud/cloudeye-exporter/logs"
 )
 
 var asInfo serversInfo
@@ -42,17 +40,5 @@ func (getter ASInfo) GetResourceInfo() (map[string]labelInfo, []cesmodel.MetricI
 }
 
 func getAllASFromRMS() ([]ResourceBaseInfo, error) {
-	resp, err := listResources("as", "scalingGroups")
-	if err != nil {
-		logs.Logger.Errorf("Failed to list resource of as.scalingGroups, error: %s", err.Error())
-		return nil, err
-	}
-	scalingGroups := make([]ResourceBaseInfo, len(resp))
-	for index, resource := range resp {
-		scalingGroups[index].ID = *resource.Id
-		scalingGroups[index].Name = *resource.Name
-		scalingGroups[index].EpId = *resource.EpId
-		scalingGroups[index].Tags = resource.Tags
-	}
-	return scalingGroups, nil
+	return getResourcesBaseInfoFromRMS("as", "scalingGroups")
 }

collector/cbr.go

@@ -57,21 +57,7 @@ func (getter CBRInfo) GetResourceInfo() (map[string]labelInfo, []model.MetricInf
 }
 
 func getAllCbrInstancesFromRMS() ([]ResourceBaseInfo, error) {
-	resp, err := listResources("cbr", "vault")
-	if err != nil {
-		logs.Logger.Errorf("Failed to list resource of cbr.vault, error: %s", err.Error())
-		return nil, err
-	}
-	cbrInstances := make([]ResourceBaseInfo, 0, len(resp))
-	for _, resource := range resp {
-		cbrInstances = append(cbrInstances, ResourceBaseInfo{
-			ID:   *resource.Id,
-			Name: *resource.Name,
-			EpId: *resource.EpId,
-			Tags: resource.Tags,
-		})
-	}
-	return cbrInstances, nil
+	return getResourcesBaseInfoFromRMS("cbr", "vault")
 }
 
 func getAllCbrInstancesFromCBR() ([]ResourceBaseInfo, error) {

collector/dayu.go

@@ -0,0 +1,132 @@
+package collector
+
+import (
+	"errors"
+	"net/http"
+	"time"
+
+	"github.com/huaweicloud/huaweicloud-sdk-go-v3/core/def"
+	"github.com/huaweicloud/huaweicloud-sdk-go-v3/services/ces/v1/model"
+
+	"github.com/huaweicloud/cloudeye-exporter/logs"
+)
+
+var dayuInfo serversInfo
+
+type DayuInfo struct{}
+
+func (getter DayuInfo) GetResourceInfo() (map[string]labelInfo, []model.MetricInfoList) {
+	resourceInfos := map[string]labelInfo{}
+	filterMetrics := make([]model.MetricInfoList, 0)
+	dayuInfo.Lock()
+	defer dayuInfo.Unlock()
+	if dayuInfo.LabelInfo == nil || time.Now().Unix() > dayuInfo.TTL {
+		streams, err := getAllStreams()
+		if err != nil {
+			logs.Logger.Error("Get all dis Streams error:", err.Error())
+			return dayuInfo.LabelInfo, dayuInfo.FilterMetrics
+		}
+
+		sysConfigMap := getMetricConfigMap("SYS.DAYU")
+		for _, stream := range streams {
+			if metricNames, ok := sysConfigMap["stream_id"]; ok {
+				metrics := buildSingleDimensionMetrics(metricNames, "SYS.DAYU", "stream_id", stream.ID)
+				filterMetrics = append(filterMetrics, metrics...)
+				info := labelInfo{
+					Name:  []string{"name", "epId"},
+					Value: []string{stream.Name, stream.EpId},
+				}
+				keys, values := getTags(stream.Tags)
+				info.Name = append(info.Name, keys...)
+				info.Value = append(info.Value, values...)
+				resourceInfos[GetResourceKeyFromMetricInfo(metrics[0])] = info
+			}
+		}
+
+		dayuInfo.LabelInfo = resourceInfos
+		dayuInfo.FilterMetrics = filterMetrics
+		dayuInfo.TTL = time.Now().Add(TTL).Unix()
+	}
+	return dayuInfo.LabelInfo, dayuInfo.FilterMetrics
+}
+
+type StreamInfo struct {
+	Private                    bool   `json:"private"`
+	StreamID                   string `json:"stream_id"`
+	StreamName                 string `json:"stream_name"`
+	CreateTime                 int64  `json:"create_time"`
+	RetentionPeriod            int    `json:"retention_period"`
+	Status                     string `json:"status"`
+	StreamType                 string `json:"stream_type"`
+	DataType                   string `json:"data_type"`
+	PartitionCount             int    `json:"partition_count"`
+	Tags                       []Tag  `json:"tags"`
+	SysTags                    []Tag  `json:"sys_tags"`
+	AutoScaleEnabled           bool   `json:"auto_scale_enabled"`
+	AutoScaleMinPartitionCount int    `json:"auto_scale_min_partition_count"`
+	AutoScaleMaxPartitionCount int    `json:"auto_scale_max_partition_count"`
+}
+
+type ListStreamsResp struct {
+	TotalNumber    int          `json:"total_number"`
+	StreamNames    []string     `json:"stream_names"`
+	StreamInfoList []StreamInfo `json:"stream_info_list"`
+	HasMoreStreams bool         `json:"has_more_streams"`
+	HttpStatusCode int          `json:"-"`
+}
+
+type ListStreamsRep struct {
+	Limit           string `json:"limit"`
+	StartStreamName string `json:"start_stream_name"`
+}
+
+func genReqDefForListStreams() *def.HttpRequestDef {
+	reqDefBuilder := def.NewHttpRequestDefBuilder().WithMethod(http.MethodGet).WithPath("/v2/{project_id}/streams").
+		WithResponse(new(ListStreamsResp)).WithContentType("application/json")
+
+	reqDefBuilder.WithRequestField(def.NewFieldDef().WithName("Limit").WithJsonTag("limit").WithLocationType(def.Query))
+	reqDefBuilder.WithRequestField(def.NewFieldDef().WithName("StartStreamName").WithJsonTag("start_stream_name").WithLocationType(def.Query))
+	return reqDefBuilder.Build()
+}
+
+func listStreams() ([]StreamInfo, error) {
+	disClient := getHcClient(getEndpoint("dis", "v2"))
+	request := &ListStreamsRep{Limit: "100"}
+	var streams []StreamInfo
+	for {
+		resp, err := disClient.Sync(request, genReqDefForListStreams())
+		if err != nil {
+			logs.Logger.Errorf("Failed to get list streams : %s", err.Error())
+			return nil, err
+		}
+		response, ok := resp.(*ListStreamsResp)
+		if !ok {
+			err := errors.New("resp type is not ServiceDetail")
+			logs.Logger.Errorf("Failed to get list streams : %s", err.Error())
+			return nil, err
+		}
+		streams = append(streams, response.StreamInfoList...)
+		if !response.HasMoreStreams {
+			break
+		}
+		request.StartStreamName = response.StreamNames[len(response.StreamNames)-1]
+	}
+	return streams, nil
+}
+
+func getAllStreams() ([]ResourceBaseInfo, error) {
+	streams, err := listStreams()
+	if err != nil {
+		logs.Logger.Errorf("Failed to get list streams : %s", err.Error())
+		return nil, err
+	}
+
+	resources := make([]ResourceBaseInfo, len(streams))
+	for i, stream := range streams {
+		resources[i].ID = stream.StreamID
+		resources[i].Name = stream.StreamName
+		resources[i].EpId = fmtTags(stream.SysTags)["_sys_enterprise_project_id"]
+		resources[i].Tags = fmtTags(stream.Tags)
+	}
+	return resources, nil
+}

collector/dcaas.go

@@ -182,17 +182,5 @@ func getDcaasVifFromRMS() ([]VifInfo, error) {
 }
 
 func getDcaasVgwFromRMS() ([]ResourceBaseInfo, error) {
-	resp, err := listResources("dcaas", "vgw")
-	if err != nil {
-		logs.Logger.Errorf("Failed to list resource of dcaas.vgw, error: %s", err.Error())
-		return nil, err
-	}
-	vgws := make([]ResourceBaseInfo, len(resp))
-	for index, resource := range resp {
-		vgws[index].ID = *resource.Id
-		vgws[index].Name = *resource.Name
-		vgws[index].EpId = *resource.EpId
-		vgws[index].Tags = resource.Tags
-	}
-	return vgws, nil
+	return getResourcesBaseInfoFromRMS("dcaas", "vgw")
 }

collector/ddos.go

@@ -46,19 +46,5 @@ func (getter DDOSInfo) GetResourceInfo() (map[string]labelInfo, []model.MetricIn
 }
 
 func getAllDDosInstancesFromRMS() ([]ResourceBaseInfo, error) {
-	resp, err := listResources("aad", "instances")
-	if err != nil {
-		logs.Logger.Errorf("Failed to list resource of aad.instances, error: %s", err.Error())
-		return nil, err
-	}
-	instances := make([]ResourceBaseInfo, 0, len(resp))
-	for _, resource := range resp {
-		instances = append(instances, ResourceBaseInfo{
-			ID:   *resource.Id,
-			Name: *resource.Name,
-			EpId: *resource.EpId,
-			Tags: resource.Tags,
-		})
-	}
-	return instances, nil
+	return getResourcesBaseInfoFromRMS("aad", "instances")
 }

collector/dli.go

@@ -85,18 +85,7 @@ func buildFlinkJobsInfo(sysConfigMap map[string][]string, filterMetrics *[]model
 }
 
 func getQueuesFromRMS() ([]ResourceBaseInfo, error) {
-	resp, err := listResources("dli", "queues")
-	if err != nil {
-		return nil, err
-	}
-	instances := make([]ResourceBaseInfo, len(resp))
-	for index, resource := range resp {
-		instances[index].ID = *resource.Id
-		instances[index].Name = *resource.Name
-		instances[index].EpId = *resource.EpId
-		instances[index].Tags = resource.Tags
-	}
-	return instances, nil
+	return getResourcesBaseInfoFromRMS("dli", "queues")
 }
 
 type ListFlinkJobsRequest struct {

collector/dms.go

@@ -6,7 +6,6 @@ import (
 	"github.com/huaweicloud/huaweicloud-sdk-go-v3/core/auth/basic"
 	"github.com/huaweicloud/huaweicloud-sdk-go-v3/core/config"
 	"github.com/huaweicloud/huaweicloud-sdk-go-v3/services/ces/v1/model"
-	rmsmodel "github.com/huaweicloud/huaweicloud-sdk-go-v3/services/rms/v1/model"
 	rocketmq "github.com/huaweicloud/huaweicloud-sdk-go-v3/services/rocketmq/v2"
 	rocketmqmodel "github.com/huaweicloud/huaweicloud-sdk-go-v3/services/rocketmq/v2/model"
 
@@ -62,27 +61,20 @@ func getDMSResourceAndMetrics() (map[string]labelInfo, []model.MetricInfoList) {
 }
 
 func getDMSInstanceFromRMS() []ResourceBaseInfo {
-	var resources []rmsmodel.ResourceEntity
-	kafkaResp, err := listResources("dms", "kafkas")
+	instances := make([]ResourceBaseInfo, 0)
+
+	kafkaResp, err := getResourcesBaseInfoFromRMS("dms", "kafkas")
 	if err != nil {
 		logs.Logger.Errorf("Get all dms kafkas : %s", err.Error())
 	} else {
-		resources = append(resources, kafkaResp...)
+		instances = append(instances, kafkaResp...)
 	}
 
-	rabbitResp, err := listResources("dms", "rabbitmqs")
+	rabbitResp, err := getResourcesBaseInfoFromRMS("dms", "rabbitmqs")
 	if err != nil {
 		logs.Logger.Errorf("Get all dms rabbitmqs: %s", err.Error())
 	} else {
-		resources = append(resources, rabbitResp...)
-	}
-
-	instances := make([]ResourceBaseInfo, len(resources))
-	for index, resource := range resources {
-		instances[index].ID = *resource.Id
-		instances[index].Name = *resource.Name
-		instances[index].EpId = *resource.EpId
-		instances[index].Tags = resource.Tags
+		instances = append(instances, rabbitResp...)
 	}
 
 	rocketMqs, err := getRocketMQInstances()

collector/extensions.go

@@ -47,6 +47,9 @@ var (
 		"SYS.GES":           GESInfo{},
 		"SYS.DBSS":          DBSSInfo{},
 		"SYS.CC":            CCInfo{},
+		"SYS.LakeFormation": LakeFormationInfo{},
+		"SYS.MRS":           MRSInfo{},
+		"SYS.DAYU":          DayuInfo{},
 	}
 )