Make TidbMonitor intergrated in AutoScaler (#1747)

manifests/crd.yaml

@@ -7043,6 +7043,19 @@ spec:
                 the pd could provide it. MetricsUrl represents the url to fetch the
                 metrics info
               type: string
+            monitor:
+              description: TidbMonitorRef reference to a TidbMonitor
+              properties:
+                name:
+                  description: Name is the name of TidbMonitor object
+                  type: string
+                namespace:
+                  description: Namespace is the namespace that TidbMonitor object
+                    locates, default to the same namespace with TidbClusterAutoScaler
+                  type: string
+              required:
+              - name
+              type: object
             tidb:
               description: TidbAutoScalerSpec describes the spec for tidb auto-scaling
               properties:

pkg/apis/pingcap/v1alpha1/tidbclusterautoscaler_types.go

@@ -58,6 +58,11 @@ type TidbClusterAutoScalerSpec struct {
 	// +optional
 	MetricsUrl *string `json:"metricsUrl,omitempty"`
 
+	// TidbMonitorRef describe the target TidbMonitor, when MetricsUrl and Monitor are both set,
+	// Operator will use MetricsUrl
+	// +optional
+	Monitor *TidbMonitorRef `json:"monitor,omitempty"`
+
 	// TiKV represents the auto-scaling spec for tikv
 	// +optional
 	TiKV *TikvAutoScalerSpec `json:"tikv,omitempty"`
@@ -132,6 +137,18 @@ type BasicAutoScalerSpec struct {
 	ScaleInThreshold *int32 `json:"scaleInThreshold,omitempty"`
 }
 
+// +k8s:openapi-gen=true
+// TidbMonitorRef reference to a TidbMonitor
+type TidbMonitorRef struct {
+	// Namespace is the namespace that TidbMonitor object locates,
+	// default to the same namespace with TidbClusterAutoScaler
+	// +optional
+	Namespace string `json:"namespace,omitempty"`
+
+	// Name is the name of TidbMonitor object
+	Name string `json:"name"`
+}
+
 // TODO: sync status
 type TidbClusterAutoSclaerStatus struct {
 }

pkg/autoscaler/autoscaler/autoscaler_manager.go

@@ -21,7 +21,6 @@ import (
 	informers "github.com/pingcap/tidb-operator/pkg/client/informers/externalversions"
 	v1alpha1listers "github.com/pingcap/tidb-operator/pkg/client/listers/pingcap/v1alpha1"
 	"github.com/pingcap/tidb-operator/pkg/controller"
-	promClient "github.com/prometheus/client_golang/api"
 	"k8s.io/apimachinery/pkg/api/errors"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	utilruntime "k8s.io/apimachinery/pkg/util/runtime"
@@ -88,21 +87,14 @@ func (am *autoScalerManager) Sync(tac *v1alpha1.TidbClusterAutoScaler) error {
 }
 
 func (am *autoScalerManager) syncAutoScaling(tc *v1alpha1.TidbCluster, tac *v1alpha1.TidbClusterAutoScaler) error {
-	if tac.Spec.MetricsUrl == nil {
-		return fmt.Errorf("tidbclusterAutoScaler[%s/%s]' metrics url should be defined explicitly", tac.Namespace, tac.Name)
-	}
-	c, err := promClient.NewClient(promClient.Config{Address: *tac.Spec.MetricsUrl})
-	if err != nil {
-		return err
-	}
 	defaultTAC(tac)
 	oldTikvReplicas := tc.Spec.TiKV.Replicas
-	if err := am.syncTiKV(tc, tac, c); err != nil {
+	if err := am.syncTiKV(tc, tac); err != nil {
 		tc.Spec.TiKV.Replicas = oldTikvReplicas
 		klog.Errorf("tac[%s/%s] tikv sync failed, continue to sync next, err:%v", tac.Namespace, tac.Name, err)
 	}
 	oldTidbReplicas := tc.Spec.TiDB.Replicas
-	if err := am.syncTiDB(tc, tac, c); err != nil {
+	if err := am.syncTiDB(tc, tac); err != nil {
 		tc.Spec.TiDB.Replicas = oldTidbReplicas
 		klog.Errorf("tac[%s/%s] tidb sync failed, continue to sync next, err:%v", tac.Namespace, tac.Name, err)
 	}

pkg/autoscaler/autoscaler/calculate/calculate.go

@@ -36,6 +36,7 @@ const (
 )
 
 type SingleQuery struct {
+	Endpoint  string
 	Timestamp int64
 	Quary     string
 	Instances []string
@@ -45,7 +46,7 @@ type SingleQuery struct {
 func queryMetricsFromPrometheus(tac *v1alpha1.TidbClusterAutoScaler, client promClient.Client, sq *SingleQuery, resp *Response) error {
 	query := sq.Quary
 	timestamp := sq.Timestamp
-	req, err := http.NewRequest("GET", fmt.Sprintf("%s%s", *tac.Spec.MetricsUrl, queryPath), nil)
+	req, err := http.NewRequest("GET", fmt.Sprintf("%s%s", sq.Endpoint, queryPath), nil)
 	if err != nil {
 		return err
 	}

pkg/autoscaler/autoscaler/tidb_autoscaler.go

@@ -25,7 +25,7 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 )
 
-func (am *autoScalerManager) syncTiDB(tc *v1alpha1.TidbCluster, tac *v1alpha1.TidbClusterAutoScaler, client promClient.Client) error {
+func (am *autoScalerManager) syncTiDB(tc *v1alpha1.TidbCluster, tac *v1alpha1.TidbClusterAutoScaler) error {
 	if tac.Spec.TiDB == nil {
 		return nil
 	}
@@ -38,7 +38,7 @@ func (am *autoScalerManager) syncTiDB(tc *v1alpha1.TidbCluster, tac *v1alpha1.Ti
 	}
 	currentReplicas := tc.Spec.TiDB.Replicas
 	instances := filterTidbInstances(tc)
-	targetReplicas, err := calculateTidbMetrics(tac, sts, client, instances)
+	targetReplicas, err := calculateTidbMetrics(tac, sts, instances)
 	if err != nil {
 		return err
 	}
@@ -73,7 +73,15 @@ func updateTcTiDBAnnIfScale(tac *v1alpha1.TidbClusterAutoScaler) {
 	tac.Annotations[label.AnnTiDBLastAutoScalingTimestamp] = fmt.Sprintf("%d", time.Now().Unix())
 }
 
-func calculateTidbMetrics(tac *v1alpha1.TidbClusterAutoScaler, sts *appsv1.StatefulSet, client promClient.Client, instances []string) (int32, error) {
+func calculateTidbMetrics(tac *v1alpha1.TidbClusterAutoScaler, sts *appsv1.StatefulSet, instances []string) (int32, error) {
+	ep, err := genMetricsEndpoint(tac)
+	if err != nil {
+		return -1, err
+	}
+	client, err := promClient.NewClient(promClient.Config{Address: ep})
+	if err != nil {
+		return -1, err
+	}
 	metric := calculate.FilterMetrics(tac.Spec.TiDB.Metrics)
 	mType, err := calculate.GenMetricType(tac, metric)
 	if err != nil {
@@ -84,6 +92,7 @@ func calculateTidbMetrics(tac *v1alpha1.TidbClusterAutoScaler, sts *appsv1.State
 		return -1, err
 	}
 	sq := &calculate.SingleQuery{
+		Endpoint:  ep,
 		Timestamp: time.Now().Unix(),
 		Instances: instances,
 		Metric:    metric,

pkg/autoscaler/autoscaler/tikv_autoscaler.go

@@ -25,7 +25,7 @@ import (
 	appsv1 "k8s.io/api/apps/v1"
 )
 
-func (am *autoScalerManager) syncTiKV(tc *v1alpha1.TidbCluster, tac *v1alpha1.TidbClusterAutoScaler, client promClient.Client) error {
+func (am *autoScalerManager) syncTiKV(tc *v1alpha1.TidbCluster, tac *v1alpha1.TidbClusterAutoScaler) error {
 	if tac.Spec.TiKV == nil {
 		return nil
 	}
@@ -38,7 +38,7 @@ func (am *autoScalerManager) syncTiKV(tc *v1alpha1.TidbCluster, tac *v1alpha1.Ti
 	}
 	instances := filterTiKVInstances(tc)
 	currentReplicas := int32(len(instances))
-	targetReplicas, err := calculateTikvMetrics(tac, sts, client, instances)
+	targetReplicas, err := calculateTikvMetrics(tac, sts, instances)
 	if err != nil {
 		return err
 	}
@@ -87,7 +87,16 @@ func updateTcTiKVAnnIfScale(tac *v1alpha1.TidbClusterAutoScaler) {
 	tac.Annotations[label.AnnTiKVLastAutoScalingTimestamp] = fmt.Sprintf("%d", time.Now().Unix())
 }
 
-func calculateTikvMetrics(tac *v1alpha1.TidbClusterAutoScaler, sts *appsv1.StatefulSet, client promClient.Client, instances []string) (int32, error) {
+func calculateTikvMetrics(tac *v1alpha1.TidbClusterAutoScaler, sts *appsv1.StatefulSet, instances []string) (int32, error) {
+	ep, err := genMetricsEndpoint(tac)
+	if err != nil {
+		return -1, err
+	}
+	client, err := promClient.NewClient(promClient.Config{Address: ep})
+	if err != nil {
+		return -1, err
+	}
+
 	metric := calculate.FilterMetrics(tac.Spec.TiKV.Metrics)
 	mType, err := calculate.GenMetricType(tac, metric)
 	if err != nil {
@@ -99,6 +108,7 @@ func calculateTikvMetrics(tac *v1alpha1.TidbClusterAutoScaler, sts *appsv1.State
 		return -1, err
 	}
 	sq := &calculate.SingleQuery{
+		Endpoint:  ep,
 		Timestamp: time.Now().Unix(),
 		Instances: instances,
 		Metric:    metric,

pkg/autoscaler/autoscaler/util.go

@@ -14,6 +14,7 @@
 package autoscaler
 
 import (
+	"fmt"
 	"strconv"
 	"time"
 
@@ -174,6 +175,12 @@ func defaultTAC(tac *v1alpha1.TidbClusterAutoScaler) {
 			tac.Spec.TiDB.ScaleInIntervalSeconds = pointer.Int32Ptr(500)
 		}
 	}
+
+	if tac.Spec.Monitor != nil {
+		if len(tac.Spec.Monitor.Namespace) < 1 {
+			tac.Spec.Monitor.Namespace = tac.Namespace
+		}
+	}
 }
 
 func resetAutoScalingAnn(tac *v1alpha1.TidbClusterAutoScaler) {
@@ -198,3 +205,13 @@ func checkAndUpdateTacAnn(tac *v1alpha1.TidbClusterAutoScaler) {
 	// If not satisfied, reset tac Ann
 	resetAutoScalingAnn(tac)
 }
+
+func genMetricsEndpoint(tac *v1alpha1.TidbClusterAutoScaler) (string, error) {
+	if tac.Spec.MetricsUrl == nil && tac.Spec.Monitor == nil {
+		return "", fmt.Errorf("tac[%s/%s] metrics url or monitor should be defined explicitly", tac.Namespace, tac.Name)
+	}
+	if tac.Spec.MetricsUrl != nil {
+		return *tac.Spec.MetricsUrl, nil
+	}
+	return fmt.Sprintf("http://%s-prometheus.%s.svc:9090", tac.Spec.Monitor.Name, tac.Spec.Monitor.Namespace), nil
+}