feat: update genesis data store

server/controller/cloud/genesis/genesis.go

@@ -29,6 +29,7 @@ import (
 	"github.com/deepflowio/deepflow/server/controller/common"
 	mysqlmodel "github.com/deepflowio/deepflow/server/controller/db/mysql/model"
 	"github.com/deepflowio/deepflow/server/controller/genesis"
+	gcommon "github.com/deepflowio/deepflow/server/controller/genesis/common"
 	"github.com/deepflowio/deepflow/server/controller/model"
 	"github.com/deepflowio/deepflow/server/controller/statsd"
 	"github.com/deepflowio/deepflow/server/libs/logger"
@@ -50,7 +51,7 @@ type Genesis struct {
 	defaultVpcName  string
 	ips             []cloudmodel.IP
 	subnets         []cloudmodel.Subnet
-	genesisData     genesis.GenesisSyncDataResponse
+	genesisData     gcommon.GenesisSyncDataResponse
 	cloudStatsd     statsd.CloudStatsd
 }
 
@@ -78,7 +79,7 @@ func NewGenesis(orgID int, domain mysqlmodel.Domain, cfg config.CloudConfig) (*G
 		UuidGenerate:    domain.DisplayName,
 		defaultVpcName:  cfg.GenesisDefaultVpcName,
 		regionUuid:      config.Get("region_uuid").MustString(),
-		genesisData:     genesis.GenesisSyncDataResponse{},
+		genesisData:     gcommon.GenesisSyncDataResponse{},
 		cloudStatsd:     statsd.NewCloudStatsd(),
 	}, nil
 }
@@ -104,7 +105,7 @@ func (g *Genesis) GetStatter() statsd.StatsdStatter {
 	}
 }
 
-func (g *Genesis) getGenesisData() (genesis.GenesisSyncDataResponse, error) {
+func (g *Genesis) getGenesisData() (gcommon.GenesisSyncDataResponse, error) {
 	return genesis.GenesisService.GetGenesisSyncResponse(g.orgID)
 }
 

server/controller/cloud/kubernetes_gather/kubernetes_gather_test.go

@@ -17,6 +17,7 @@
 package kubernetes_gather
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"os"
@@ -32,6 +33,7 @@ import (
 	"github.com/deepflowio/deepflow/server/controller/db/mysql"
 	mysqlmodel "github.com/deepflowio/deepflow/server/controller/db/mysql/model"
 	"github.com/deepflowio/deepflow/server/controller/genesis"
+	gcommon "github.com/deepflowio/deepflow/server/controller/genesis/common"
 )
 
 func TestKubernetes(t *testing.T) {
@@ -83,11 +85,11 @@ func TestKubernetes(t *testing.T) {
 		})
 		defer k8sInfoPatch.Reset()
 
-		g := genesis.NewGenesis(&config.ControllerConfig{})
+		g := genesis.NewGenesis(context.Background(), &config.ControllerConfig{})
 		vJsonData, _ := os.ReadFile("./testfiles/vinterfaces.json")
-		var vData genesis.GenesisSyncDataResponse
+		var vData gcommon.GenesisSyncDataResponse
 		json.Unmarshal(vJsonData, &vData)
-		vinterfacesInfoPatch := gomonkey.ApplyMethod(reflect.TypeOf(g), "GetGenesisSyncResponse", func(_ *genesis.Genesis, _ int) (genesis.GenesisSyncDataResponse, error) {
+		vinterfacesInfoPatch := gomonkey.ApplyMethod(reflect.TypeOf(g), "GetGenesisSyncResponse", func(_ *genesis.Genesis, _ int) (gcommon.GenesisSyncDataResponse, error) {
 			return vData, nil
 		})
 		defer vinterfacesInfoPatch.Reset()

server/controller/controller/controller.go

@@ -107,7 +107,7 @@ func Start(ctx context.Context, configPath, serverLogFile string, shared *server
 	}
 
 	// 初始化Redis
-	if cfg.RedisCfg.Enabled && cfg.TrisolarisCfg.NodeType == "master" {
+	if cfg.RedisCfg.Enabled {
 		router.SetInitStageForHealthChecker("Redis init")
 
 		err := redis.Init(ctx, cfg.RedisCfg)
@@ -124,8 +124,7 @@ func Start(ctx context.Context, configPath, serverLogFile string, shared *server
 
 	router.SetInitStageForHealthChecker("Genesis init")
 	// 启动genesis
-	g := genesis.NewGenesis(cfg)
-	g.Start()
+	g := genesis.NewGenesis(ctx, cfg)
 
 	// start tagrecorder before manager to prevent recorder from publishing message when tagrecorder is not ready
 	router.SetInitStageForHealthChecker("TagRecorder init")

server/controller/db/redis/redis.go

@@ -34,8 +34,9 @@ var (
 
 type Config struct {
 	ResourceAPIDatabase       int      `default:"1" yaml:"resource_api_database"`
-	ResourceAPIExpireInterval int      `default:"3600" yaml:"resource_api_expire_interval"`
 	DimensionResourceDatabase int      `default:"2" yaml:"dimension_resource_database"`
+	GenesisSyncDatabase       int      `default:"4" yaml:"genesis_sync_database"`
+	ResourceAPIExpireInterval int      `default:"3600" yaml:"resource_api_expire_interval"`
 	Host                      []string `default:"" yaml:"host"` // TODO add default value
 	Port                      uint32   `default:"6379" yaml:"port"`
 	Password                  string   `default:"deepflow" yaml:"password"`
@@ -55,6 +56,7 @@ func GetConfig() *Config { // TODO use this function
 type Client struct {
 	ResourceAPI       redis.UniversalClient
 	DimensionResource redis.UniversalClient
+	GenesisSync       redis.UniversalClient
 	Config            *Config
 }
 
@@ -108,13 +110,13 @@ func Init(ctx context.Context, cfg Config) (err error) {
 		client = &Client{
 			ResourceAPI:       createUniversalClient(cfg, cfg.ResourceAPIDatabase),
 			DimensionResource: createUniversalClient(cfg, cfg.DimensionResourceDatabase),
+			GenesisSync:       createUniversalClient(cfg, cfg.GenesisSyncDatabase),
 		}
 		_, err = client.ResourceAPI.Ping(ctx).Result()
 		if err != nil {
 			return
 		}
 		_, err = client.DimensionResource.Ping(ctx).Result()
-		return
 	})
 	return
 }

server/controller/genesis/common/const.go

@@ -17,9 +17,16 @@
 package common
 
 const (
-	TYPE_UPDATE                  = 1
-	TYPE_RENEW                   = 2
-	TYPE_EXIT                    = 3
+	SYNC_TYPE_FORMAT = "%v-%v-%v-%v" // region-orgID-type-vtapID
+)
+
+const (
+	TYPE_UPDATE = 1
+	TYPE_RENEW  = 2
+	TYPE_EXIT   = 3
+)
+
+const (
 	DEVICE_TYPE_KVM_HOST         = "kvm-host"
 	DEVICE_TYPE_KVM_VM           = "kvm-vm"
 	DEVICE_TYPE_DOCKER_HOST      = "docker-host"

server/controller/genesis/common/type.go

@@ -0,0 +1,82 @@
+/*
+ * Copyright (c) 2024 Yunshan Networks
+ *
+ * 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 common
+
+import (
+	"time"
+
+	"github.com/deepflowio/deepflow/message/common"
+	"github.com/deepflowio/deepflow/message/trident"
+	"github.com/deepflowio/deepflow/server/controller/model"
+)
+
+type GenesisSyncType interface {
+	model.GenesisVinterface | model.GenesisVpc | model.GenesisHost | model.GenesisVM | model.GenesisVIP | model.GenesisNetwork | model.GenesisPort | model.GenesisLldp | model.GenesisIP | model.GenesisProcess
+}
+
+type GenesisSyncData struct {
+	IPLastSeens map[int][]model.GenesisIP
+	VIPs        map[int][]model.GenesisVIP
+	VMs         map[int][]model.GenesisVM
+	VPCs        map[int][]model.GenesisVpc
+	Hosts       map[int][]model.GenesisHost
+	Lldps       map[int][]model.GenesisLldp
+	Ports       map[int][]model.GenesisPort
+	Networks    map[int][]model.GenesisNetwork
+	Vinterfaces map[int][]model.GenesisVinterface
+	Processes   map[int][]model.GenesisProcess
+}
+
+type GenesisSyncDataResponse struct {
+	IPLastSeens []model.GenesisIP
+	VIPs        []model.GenesisVIP
+	VMs         []model.GenesisVM
+	VPCs        []model.GenesisVpc
+	Hosts       []model.GenesisHost
+	Lldps       []model.GenesisLldp
+	Ports       []model.GenesisPort
+	Networks    []model.GenesisNetwork
+	Vinterfaces []model.GenesisVinterface
+	Processes   []model.GenesisProcess
+}
+
+type KubernetesInfo struct {
+	ORGID     int
+	ClusterID string
+	ErrorMSG  string
+	Version   uint64
+	Epoch     time.Time
+	Entries   []*common.KubernetesAPIInfo
+}
+
+type K8SRPCMessage struct {
+	ORGID       int
+	MessageType int
+	VtapID      uint32
+	Peer        string
+	Message     *trident.KubernetesAPISyncRequest
+}
+
+type VIFRPCMessage struct {
+	ORGID        int
+	MessageType  int
+	TeamID       uint32
+	VtapID       uint32
+	Peer         string
+	K8SClusterID string
+	Message      *trident.GenesisSyncRequest
+}