fix: genesis sync update storage

Author: askyrie

server/controller/genesis/common/type.go

@@ -72,11 +72,12 @@ type K8SRPCMessage struct {
 }
 
 type VIFRPCMessage struct {
-	ORGID        int
-	MessageType  int
-	TeamID       uint32
-	VtapID       uint32
-	Peer         string
-	K8SClusterID string
-	Message      *agent.GenesisSyncRequest
+	ORGID          int
+	MessageType    int
+	TeamID         uint32
+	VtapID         uint32
+	Peer           string
+	K8SClusterID   string
+	StorageRefresh bool
+	Message        *agent.GenesisSyncRequest
 }

server/controller/genesis/config/config.go

@@ -19,6 +19,7 @@ package config
 type GenesisConfig struct {
 	AgingTime               float64  `default:"86400" yaml:"aging_time"`
 	VinterfaceAgingTime     float64  `default:"300" yaml:"vinterface_aging_time"`
+	AgentHeartBeat          float64  `default:"60" yaml:"agent_heart_beat"`
 	HostIPs                 []string `yaml:"host_ips"`
 	LocalIPRanges           []string `yaml:"local_ip_ranges"`
 	ExcludeIPRanges         []string `yaml:"exclude_ip_ranges"`

server/controller/genesis/grpc/server.go

@@ -147,6 +147,7 @@ func (g *SynchronizerServer) GenesisSync(ctx context.Context, request *agent.Gen
 	}
 	vtap := fmt.Sprintf("%d-%d", orgID, vtapID)
 
+	var refresh bool
 	var localVersion uint64 = 0
 	if vtapID == 0 {
 		log.Infof("genesis sync received message with vtap_id 0 from %s", remote, logger.NewORGPrefix(orgID))
@@ -160,9 +161,11 @@ func (g *SynchronizerServer) GenesisSync(ctx context.Context, request *agent.Gen
 			} else {
 				agingTime = g.cfg.VinterfaceAgingTime
 			}
-			if now.Sub(lastTime).Seconds() >= agingTime {
+			timeSub := now.Sub(lastTime).Seconds()
+			if timeSub >= agingTime {
 				g.vtapToVersion.Store(vtap, uint64(0))
 			}
+			refresh = timeSub >= g.cfg.AgentHeartBeat*2
 		}
 		g.vtapToLastSeen.Store(vtap, now)
 		lVersion, ok := g.vtapToVersion.Load(vtap)
@@ -176,12 +179,13 @@ func (g *SynchronizerServer) GenesisSync(ctx context.Context, request *agent.Gen
 		log.Debugf("genesis sync renew version %v from ip %s vtap_id %v", version, remote, vtapID, logger.NewORGPrefix(orgID))
 		g.genesisSyncQueue.Put(
 			common.VIFRPCMessage{
-				Peer:        remote,
-				VtapID:      vtapID,
-				ORGID:       orgID,
-				TeamID:      uint32(teamID),
-				MessageType: common.TYPE_RENEW,
-				Message:     request,
+				Peer:           remote,
+				VtapID:         vtapID,
+				ORGID:          orgID,
+				TeamID:         uint32(teamID),
+				MessageType:    common.TYPE_RENEW,
+				Message:        request,
+				StorageRefresh: refresh,
 			},
 		)
 		return &agent.GenesisSyncResponse{Version: &localVersion}, nil

server/controller/genesis/store/sync/run.go

@@ -422,7 +422,7 @@ func (g *GenesisSync) Start() {
 				if info.VtapID != 0 {
 					peerInfo, ok := genesisSyncDataByVtap[vtap]
 					if ok {
-						vStorage.Renew(info.ORGID, peerInfo)
+						vStorage.Renew(info.ORGID, info.VtapID, info.StorageRefresh, peerInfo)
 					}
 				}
 			} else if info.MessageType == common.TYPE_UPDATE {

server/controller/genesis/store/sync/store.go

@@ -57,7 +57,7 @@ func NewSyncStorage(ctx context.Context, cfg config.GenesisConfig, sChan chan co
 	}
 }
 
-func (s *SyncStorage) Renew(orgID int, data common.GenesisSyncDataResponse) {
+func (s *SyncStorage) Renew(orgID int, vtapID uint32, refresh bool, data common.GenesisSyncDataResponse) {
 	now := time.Now()
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
@@ -72,6 +72,20 @@ func (s *SyncStorage) Renew(orgID int, data common.GenesisSyncDataResponse) {
 	s.genesisSyncInfo.IPlastseens.Renew(orgID, now, data.IPLastSeens)
 	s.genesisSyncInfo.Vinterfaces.Renew(orgID, now, data.Vinterfaces)
 	s.genesisSyncInfo.Processes.Renew(orgID, now, data.Processes)
+
+	if !refresh {
+		return
+	}
+	db, err := metadb.GetDB(orgID)
+	if err != nil {
+		log.Error("get metadb session failed", logger.NewORGPrefix(orgID))
+		return
+	}
+	nodeIP := os.Getenv(ccommon.NODE_IP_KEY)
+	err = db.Model(&model.GenesisStorage{}).Where("vtap_id = ? AND node_ip <> ?", vtapID, nodeIP).Update("node_ip", nodeIP).Error
+	if err != nil {
+		log.Warningf("vtap id (%d) refresh storage to node (%s) failed: %s", vtapID, nodeIP, err, logger.NewORGPrefix(orgID))
+	}
 }
 
 func (s *SyncStorage) Update(orgID int, vtapID uint32, data common.GenesisSyncDataResponse) {

server/server.yaml

@@ -342,6 +342,8 @@ controller:
 
     # 数据持久化检测间隔，单位：秒
     data_persistence_interval: 60
+    # 采集器消息心跳时长，单位：秒
+    agent_heart_beat: 60
 
     # 采集器同步KVM时，配置的采集器IP所上报的内容会被解析
     host_ips: