fix: JobCollector reuse FE RPC connection (#645)

Author: Ryan19929

cmd/ccr_syncer/ccr_syncer.go

@@ -228,7 +228,7 @@ func main() {
 	}()
 
 	// Step 9: start job collector
-	jobCollector := NewJobCollector(db)
+	jobCollector := NewJobCollector(db, hostInfo, factory)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()

cmd/ccr_syncer/job_collector.go

@@ -22,20 +22,27 @@ import (
 
 	"github.com/selectdb/ccr_syncer/pkg/ccr"
 	"github.com/selectdb/ccr_syncer/pkg/ccr/base"
-	"github.com/selectdb/ccr_syncer/pkg/rpc"
 	"github.com/selectdb/ccr_syncer/pkg/storage"
 	"github.com/selectdb/ccr_syncer/pkg/xerror"
 	"github.com/selectdb/ccr_syncer/pkg/xmetrics"
 	log "github.com/sirupsen/logrus"
 )
 
 type JobCollector struct {
-	db   storage.DB
-	stop chan struct{}
+	db       storage.DB
+	hostInfo string
+	factory  *ccr.Factory
+	stop     chan struct{}
 }
 
-func NewJobCollector(db storage.DB) *JobCollector {
-	return &JobCollector{db: db, stop: make(chan struct{})}
+func NewJobCollector(db storage.DB, hostInfo string, factory *ccr.Factory) *JobCollector {
+	log.Infof("JobCollector initialized with hostInfo: %s", hostInfo)
+	return &JobCollector{
+		db:       db,
+		hostInfo: hostInfo,
+		factory:  factory,
+		stop:     make(chan struct{}),
+	}
 }
 
 func (c *JobCollector) Collect() {
@@ -59,11 +66,13 @@ func (c *JobCollector) Stop() {
 }
 
 func (c *JobCollector) updateMetrics() error {
-	jobs, err := c.db.GetJobs()
+	_, jobs, err := c.db.GetStampAndJobs(c.hostInfo)
 	if err != nil {
-		return xerror.Wrapf(err, xerror.Normal, "get all jobs failed")
+		return xerror.Wrapf(err, xerror.Normal, "get jobs by belong_to %s failed", c.hostInfo)
 	}
 
+	log.Debugf("JobCollector fetched %d jobs for hostInfo: %s", len(jobs), c.hostInfo)
+
 	runningJobNum := 0
 	for _, jobName := range jobs {
 		jobInfo, err := loadJobInfo(jobName, c.db)
@@ -84,7 +93,7 @@ func (c *JobCollector) updateMetrics() error {
 
 		srcSpec := &jobInfo.Src
 		commitSeq := jobProgress.CommitSeq
-		lag, interval, err := getJobLag(srcSpec, commitSeq)
+		lag, interval, err := c.getJobLag(srcSpec, commitSeq)
 		if err != nil {
 			log.Warnf("get job %s lag failed: %+v", jobName, err)
 			continue
@@ -128,8 +137,8 @@ func loadJobInfo(jobName string, db storage.DB) (*ccr.Job, error) {
 	return &job, nil
 }
 
-func getJobLag(spec *base.Spec, commitSeq int64) (int64, float64, error) {
-	feRpc, err := rpc.NewFeRpc(spec)
+func (c *JobCollector) getJobLag(spec *base.Spec, commitSeq int64) (int64, float64, error) {
+	feRpc, err := c.factory.NewFeRpc(spec)
 	if err != nil {
 		return 0, 0, xerror.Wrapf(err, xerror.Normal, "new fe rpc failed")
 	}

pkg/rpc/rpc_factory.go

@@ -23,6 +23,7 @@ import (
 	"github.com/selectdb/ccr_syncer/pkg/ccr/base"
 	beservice "github.com/selectdb/ccr_syncer/pkg/rpc/kitex_gen/backendservice/backendservice"
 	"github.com/selectdb/ccr_syncer/pkg/xerror"
+	log "github.com/sirupsen/logrus"
 
 	"github.com/cloudwego/kitex/client"
 )
@@ -33,7 +34,7 @@ type IRpcFactory interface {
 }
 
 type RpcFactory struct {
-	feRpcs     map[*base.Spec]IFeRpc
+	feRpcs     map[string]IFeRpc // key: connection string (cluster+host+port)
 	feRpcsLock sync.Mutex
 
 	beRpcs     map[base.Backend]IBeRpc
@@ -42,32 +43,43 @@ type RpcFactory struct {
 
 func NewRpcFactory() IRpcFactory {
 	return &RpcFactory{
-		feRpcs: make(map[*base.Spec]IFeRpc),
+		feRpcs: make(map[string]IFeRpc),
 		beRpcs: make(map[base.Backend]IBeRpc),
 	}
 }
 
+// getFeKey generates a cache key for FE RPC based on connection info
+// Key format: "cluster@host:port:thrift_port"
+func getFeKey(spec *base.Spec) string {
+	return fmt.Sprintf("%s@%s:%s:%s", spec.Cluster, spec.Host, spec.Port, spec.ThriftPort)
+}
+
 func (rf *RpcFactory) NewFeRpc(spec *base.Spec) (IFeRpc, error) {
 	// valid spec
 	if err := spec.Valid(); err != nil {
 		return nil, err
 	}
 
+	// Generate cache key based on connection info (cluster + FE address)
+	key := getFeKey(spec)
+
 	rf.feRpcsLock.Lock()
-	if feRpc, ok := rf.feRpcs[spec]; ok {
+	if feRpc, ok := rf.feRpcs[key]; ok {
 		rf.feRpcsLock.Unlock()
+		log.Debugf("RpcFactory: reused cached FeRpc for %s (cache hit)", key)
 		return feRpc, nil
 	}
 	rf.feRpcsLock.Unlock()
 
+	log.Debugf("RpcFactory: creating new FeRpc for %s (cache miss)", key)
 	feRpc, err := NewFeRpc(spec)
 	if err != nil {
 		return nil, err
 	}
 
 	rf.feRpcsLock.Lock()
 	defer rf.feRpcsLock.Unlock()
-	rf.feRpcs[spec] = feRpc
+	rf.feRpcs[key] = feRpc
 	return feRpc, nil
 }
 

pkg/service/http_service.go

@@ -298,7 +298,7 @@ func (s *HttpService) getLagHandler(w http.ResponseWriter, r *http.Request) {
 	}
 
 	srcSpec := &job.Src
-	feRpc, err := rpc.NewFeRpc(srcSpec)
+	feRpc, err := s.jobManager.GetFactory().NewFeRpc(srcSpec)
 	if err != nil {
 		log.Warnf("new fe rpc failed: %+v", err)
 		lagResult = &result{