conprof: support ticdc backend (#167)

ref #166

Author: mornyx

component/conprof/http/api.go

@@ -115,6 +115,11 @@ func getProfileEstimateSize(component topology.Component) int {
 			100*1024 + // goroutine size
 			400*1024 + // heap size
 			30*1024 // mutex size
+	case topology.ComponentTiCDC:
+		return 100*1024 + // profile size
+			100*1024 + // goroutine size
+			400*1024 + // heap size
+			30*1024 // mutex size
 	case topology.ComponentTiKV:
 		return 200 * 1024 // profile size
 	case topology.ComponentTiFlash:
@@ -128,6 +133,7 @@ type ComponentNum struct {
 	PD      int `json:"pd"`
 	TiKV    int `json:"tikv"`
 	TiFlash int `json:"tiflash"`
+	TiCDC   int `json:"ticdc"`
 }
 
 type GroupProfiles struct {
@@ -227,6 +233,8 @@ func (sc *StatusCounterAndTargets) getComponentNum() ComponentNum {
 			compNum.TiKV = num
 		case topology.ComponentTiFlash:
 			compNum.TiFlash = num
+		case topology.ComponentTiCDC:
+			compNum.TiCDC = num
 		}
 	}
 	return compNum

component/conprof/scrape/manager.go

@@ -229,7 +229,7 @@ func (m *Manager) stopScrape(component topology.Component) {
 
 func (m *Manager) getProfilingConfig(component topology.Component) *config.ProfilingConfig {
 	switch component.Name {
-	case topology.ComponentTiDB, topology.ComponentPD:
+	case topology.ComponentTiDB, topology.ComponentPD, topology.ComponentTiCDC:
 		return goAppProfilingConfig(m.config.ContinueProfiling)
 	default:
 		return nonGoAppProfilingConfig(m.config.ContinueProfiling)

component/topology/discovery.go

@@ -2,12 +2,16 @@ package topology
 
 import (
 	"context"
+	"encoding/json"
+	"strconv"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
 
 	"github.com/pingcap/ng-monitoring/component/domain"
 	"github.com/pingcap/ng-monitoring/utils"
+	clientv3 "go.etcd.io/etcd/client/v3"
 
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb-dashboard/util/topo"
@@ -20,6 +24,7 @@ const (
 	ComponentTiKV    = "tikv"
 	ComponentTiFlash = "tiflash"
 	ComponentPD      = "pd"
+	ComponentTiCDC   = "ticdc"
 )
 
 var (
@@ -126,6 +131,7 @@ func (d *TopologyDiscoverer) fetchAllScrapeTargets(ctx context.Context) ([]Compo
 		d.getTiDBComponents,
 		d.getPDComponents,
 		d.getStoreComponents,
+		d.getTiCDCComponents,
 	}
 	components := make([]Component, 0, 8)
 	for _, fn := range fns {
@@ -220,3 +226,58 @@ func (d *TopologyDiscoverer) getStoreComponents(ctx context.Context) ([]Componen
 	}
 	return components, nil
 }
+
+func (d *TopologyDiscoverer) getTiCDCComponents(ctx context.Context) ([]Component, error) {
+	etcdCli, err := d.do.GetEtcdClient()
+	if err != nil {
+		return nil, err
+	}
+	return getTiCDCComponents(ctx, etcdCli)
+}
+
+const ticdcTopologyKeyPrefix = "/tidb/cdc/default/__cdc_meta__/capture/"
+
+type ticdcNodeItem struct {
+	ID      string `json:"id"`
+	Address string `json:"address"`
+	Version string `json:"version"`
+}
+
+func getTiCDCComponents(ctx context.Context, etcdCli *clientv3.Client) ([]Component, error) {
+	resp, err := etcdCli.Get(ctx, ticdcTopologyKeyPrefix, clientv3.WithPrefix())
+	if err != nil {
+		return nil, err
+	}
+	components := make([]Component, 0, 3)
+	for _, kv := range resp.Kvs {
+		key := string(kv.Key)
+		if !strings.HasPrefix(key, ticdcTopologyKeyPrefix) {
+			continue
+		}
+		var item ticdcNodeItem
+		if err := json.Unmarshal(kv.Value, &item); err != nil {
+			log.Warn("invalid ticdc node item in etcd", zap.Error(err))
+			continue
+		}
+		arr := strings.Split(item.Address, ":")
+		if len(arr) != 2 {
+			log.Warn("invalid ticdc node address in etcd", zap.String("address", item.Address))
+			continue
+		}
+		ip := arr[0]
+		port, err := strconv.Atoi(arr[1])
+		if err != nil {
+			log.Warn("invalid ticdc node address in etcd",
+				zap.Error(err),
+				zap.String("address", item.Address))
+			continue
+		}
+		components = append(components, Component{
+			Name:       ComponentTiCDC,
+			IP:         ip,
+			Port:       uint(port),
+			StatusPort: uint(port),
+		})
+	}
+	return components, nil
+}