volcano metrics source support elasticsearch

docs/design/usage-based-scheduling.md

@@ -17,7 +17,7 @@ Currently the pod is scheduled based on the resource request and node allocatabl
 ## Design 
 
 ### Scheduler Cache
-A separated goroutine is created in scheduler cache to talk with Prometheus which is used to collect and aggregate node usage metrics. The node usage data in cache is consumed by usage based scheduling plugin and other plugins like rescheduling plugin. The struct is as below. 
+A separated goroutine is created in scheduler cache to talk with Metrics source(like prometheus, elasticsearch) which is used to collect and aggregate node usage metrics. The node usage data in cache is consumed by usage based scheduling plugin and other plugins like rescheduling plugin. The struct is as below. 
 ```
 type NodeUsage struct {
     cpuUsageAvg map[string]float64
@@ -57,7 +57,8 @@ tiers:
       - name: nodeorder
       - name: binpack
 metrics:                             # metrics server related configuration
-  address: http://192.168.0.10:9090  # mandatory, The Prometheus server address
+  type: prometheus                   # Optional, The metrics source type, prometheus by default, support prometheus and elasticsearch
+  address: http://192.168.0.10:9090  # mandatory, The metrics source address
   interval: 30s                      # Optional, The scheduler pull metrics from Prometheus with this interval, 5s by default
   ```
 

go.mod

@@ -4,6 +4,7 @@ go 1.19
 
 require (
 	github.com/agiledragon/gomonkey/v2 v2.1.0
+	github.com/elastic/go-elasticsearch/v7 v7.17.7
 	github.com/fsnotify/fsnotify v1.5.1
 	github.com/golang/mock v1.6.0
 	github.com/google/go-cmp v0.5.8

go.sum

@@ -95,6 +95,8 @@ github.com/docker/docker v20.10.17+incompatible/go.mod h1:eEKB0N0r5NX/I1kEveEz05
 github.com/docker/go-connections v0.4.0/go.mod h1:Gbd7IOopHjR8Iph03tsViu4nIes5XhDvyHbTtUxmeec=
 github.com/docker/go-units v0.4.0/go.mod h1:fgPhTUdO+D/Jk86RDLlptpiXQzgHJF7gydDDbaIK4Dk=
 github.com/docopt/docopt-go v0.0.0-20180111231733-ee0de3bc6815/go.mod h1:WwZ+bS3ebgob9U8Nd0kOddGdZWjyMGR8Wziv+TBNwSE=
+github.com/elastic/go-elasticsearch/v7 v7.17.7 h1:pcYNfITNPusl+cLwLN6OLmVT+F73Els0nbaWOmYachs=
+github.com/elastic/go-elasticsearch/v7 v7.17.7/go.mod h1:OJ4wdbtDNk5g503kvlHLyErCgQwwzmDtaFC4XyOxXA4=
 github.com/emicklei/go-restful/v3 v3.8.0 h1:eCZ8ulSerjdAiaNpF7GxXIE7ZCMo1moN1qX+S609eVw=
 github.com/emicklei/go-restful/v3 v3.8.0/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc=
 github.com/envoyproxy/go-control-plane v0.9.4/go.mod h1:6rpuAdCZL397s3pYoYcLgu1mIlRU8Am5FuJP05cCM98=

pkg/scheduler/metrics/source/metrics_client.go

@@ -35,5 +35,9 @@ func NewMetricsClient(metricsConf map[string]string) (MetricsClient, error) {
 	if len(address) == 0 {
 		return nil, errors.New("metrics address is empty")
 	}
+	metricsType := metricsConf["type"]
+	if metricsType == "elasticsearch" {
+		return NewElasticsearchMetricsClient(address)
+	}
 	return &PrometheusMetricsClient{address: address}, nil
 }

pkg/scheduler/metrics/source/metrics_client_elasticsearch.go

@@ -0,0 +1,109 @@
+/*
+ Copyright 2023 The Volcano Authors.
+
+ 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 source
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"github.com/elastic/go-elasticsearch/v7"
+)
+
+const (
+	// esHostNameField is the field name of host name in the document
+	esHostNameField = "host.hostname"
+	// esCpuUsageField is the field name of cpu usage in the document
+	esCpuUsageField = "host.cpu.usage"
+	// esMemUsageField is the field name of mem usage in the document
+	esMemUsageField = "system.memory.actual.used.pct"
+)
+
+type ElasticsearchMetricsClient struct {
+	address string
+	es      *elasticsearch.Client
+}
+
+func NewElasticsearchMetricsClient(address string) (*ElasticsearchMetricsClient, error) {
+	e := &ElasticsearchMetricsClient{address: address}
+	var err error
+	e.es, err = elasticsearch.NewDefaultClient()
+	if err != nil {
+		return nil, err
+	}
+	return e, nil
+}
+
+func (p *ElasticsearchMetricsClient) NodeMetricsAvg(ctx context.Context, nodeName string, period string) (*NodeMetrics, error) {
+	nodeMetrics := &NodeMetrics{}
+	var buf bytes.Buffer
+	query := map[string]interface{}{
+		"size": 0,
+		"query": map[string]interface{}{
+			"bool": map[string]interface{}{
+				"must": []map[string]interface{}{
+					{
+						"range": map[string]interface{}{
+							"@timestamp": map[string]interface{}{
+								"gte": "now-" + period,
+								"lt":  "now",
+							},
+						},
+					},
+					{
+						"term": map[string]interface{}{
+							esHostNameField: nodeName,
+						},
+					},
+				},
+			},
+		},
+		"aggs": map[string]interface{}{
+			"cpu": map[string]interface{}{
+				"avg": map[string]interface{}{
+					"field": esCpuUsageField,
+				},
+			},
+			"mem": map[string]interface{}{
+				"avg": map[string]interface{}{
+					"field": esMemUsageField,
+				},
+			},
+		},
+	}
+	if err := json.NewEncoder(&buf).Encode(query); err != nil {
+		return nil, err
+	}
+	res, err := p.es.Search(
+		p.es.Search.WithContext(ctx),
+		p.es.Search.WithIndex("metricbeat-*"),
+		p.es.Search.WithBody(&buf),
+	)
+	if err != nil {
+		return nil, err
+	}
+	defer res.Body.Close()
+	var r map[string]interface{}
+	if err := json.NewDecoder(res.Body).Decode(&r); err != nil {
+		return nil, err
+	}
+	aggs := r["aggregations"].(map[string]interface{})
+	cpuUsage := aggs["cpu"].(map[string]interface{})["value"].(float64)
+	memUsage := aggs["mem"].(map[string]interface{})["value"].(float64)
+	nodeMetrics.Cpu = cpuUsage
+	nodeMetrics.Memory = memUsage
+	return nodeMetrics, nil
+}