Skip to content

Commit 0211bdd

Browse files
askyrieSongZhen0704
askyrie
authored and
SongZhen0704
committed
fix: genesis sync
1 parent 1f1ef74 commit 0211bdd

17 files changed

Lines changed: 431 additions & 161 deletions

File tree

‎server/controller/cloud/kubernetes_gather/kubernetes_gather.go‎

Lines changed: 43 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -17,6 +17,7 @@
1717
package kubernetes_gather
1818

1919
import (
20+
"errors"
2021
"regexp"
2122
"strings"
2223
"time"
@@ -33,6 +34,7 @@ import (
3334
mysql "github.com/deepflowio/deepflow/server/controller/db/metadb"
3435
mysqlmodel "github.com/deepflowio/deepflow/server/controller/db/metadb/model"
3536
"github.com/deepflowio/deepflow/server/controller/genesis"
37+
cmodel "github.com/deepflowio/deepflow/server/controller/model"
3638
"github.com/deepflowio/deepflow/server/controller/statsd"
3739
"github.com/deepflowio/deepflow/server/libs/logger"
3840
)
@@ -63,6 +65,7 @@ type KubernetesGather struct {
6365
podGroupLcuuids mapset.Set
6466
podNetworkLcuuidCIDRs networkLcuuidCIDRs
6567
nodeNetworkLcuuidCIDRs networkLcuuidCIDRs
68+
vinterfaceData []cmodel.GenesisVinterface
6669
podIPToLcuuid map[string]string
6770
nodeIPToLcuuid map[string]string
6871
namespaceToLcuuid map[string]string
@@ -96,6 +99,10 @@ func NewKubernetesGather(db *mysql.DB, domain *mysqlmodel.Domain, subDomain *mys
9699
var err error
97100

98101
domainConfigJson, err = simplejson.NewJson([]byte(domain.Config))
102+
if err != nil {
103+
log.Error(err, logger.NewORGPrefix(db.ORGID))
104+
return nil
105+
}
99106
portNameRegex := domainConfigJson.Get("node_port_name_regex").MustString()
100107
if portNameRegex == "" {
101108
portNameRegex = common.DEFAULT_PORT_NAME_REGEX
@@ -202,6 +209,7 @@ func NewKubernetesGather(db *mysql.DB, domain *mysqlmodel.Domain, subDomain *mys
202209
customTagLenMax: cfg.CustomTagLenMax,
203210
isSubDomain: isSubDomain,
204211
podGroupLcuuids: mapset.NewSet(),
212+
vinterfaceData: []cmodel.GenesisVinterface{},
205213
nodeNetworkLcuuidCIDRs: networkLcuuidCIDRs{},
206214
podNetworkLcuuidCIDRs: networkLcuuidCIDRs{},
207215
podIPToLcuuid: map[string]string{},
@@ -277,7 +285,7 @@ func (k *KubernetesGather) pgSpecGenerateConnections(nsName, pgName, pgLcuuid st
277285
if !ok {
278286
continue
279287
}
280-
cmName := ref.Get("Name").MustString()
288+
cmName := ref.Get("name").MustString()
281289
cmLcuuid, ok := k.configMapToLcuuid[[2]string{nsName, cmName}]
282290
if !ok {
283291
log.Infof("pod group (%s) imported env config map (%s) not found", pgName, cmName, logger.NewORGPrefix(k.orgID))
@@ -328,6 +336,7 @@ func (k *KubernetesGather) GetKubernetesGatherData() (model.KubernetesGatherReso
328336
// 任务循环的是同一个实例，所以这里要对关联关系进行初始化
329337
k.azLcuuid = ""
330338
k.k8sInfo = nil
339+
k.vinterfaceData = []cmodel.GenesisVinterface{}
331340
k.podNetworkLcuuidCIDRs = networkLcuuidCIDRs{}
332341
k.nodeNetworkLcuuidCIDRs = networkLcuuidCIDRs{}
333342
k.podGroupLcuuids = mapset.NewSet()
@@ -345,6 +354,39 @@ func (k *KubernetesGather) GetKubernetesGatherData() (model.KubernetesGatherReso
345354
k.nsServiceNameToService = map[string]map[string]map[string]int{}
346355
k.cloudStatsd = statsd.NewCloudStatsd()
347356

357+
if genesis.GenesisService == nil {
358+
errMsg := "genesis service is nil"
359+
log.Warning(errMsg, logger.NewORGPrefix(k.orgID))
360+
return model.KubernetesGatherResource{
361+
ErrorState: common.RESOURCE_STATE_CODE_EXIT,
362+
ErrorMessage: errMsg,
363+
}, errors.New(errMsg)
364+
}
365+
366+
k8sInfo, err := k.getKubernetesInfo()
367+
if err != nil {
368+
log.Warning(err.Error(), logger.NewORGPrefix(k.orgID))
369+
return model.KubernetesGatherResource{
370+
ErrorState: common.RESOURCE_STATE_CODE_WARNING,
371+
ErrorMessage: err.Error(),
372+
}, err
373+
}
374+
k.k8sInfo = k8sInfo
375+
376+
gsData, err := genesis.GenesisService.GetGenesisSyncResponse(k.orgID)
377+
if err != nil {
378+
return model.KubernetesGatherResource{
379+
ErrorState: common.RESOURCE_STATE_CODE_EXIT,
380+
ErrorMessage: err.Error(),
381+
}, err
382+
}
383+
for _, v := range gsData.Vinterfaces {
384+
if v.KubernetesClusterID != k.ClusterID {
385+
continue
386+
}
387+
k.vinterfaceData = append(k.vinterfaceData, v)
388+
}
389+
348390
region, err := k.getRegion()
349391
if err != nil {
350392
return model.KubernetesGatherResource{}, err
@@ -368,16 +410,6 @@ func (k *KubernetesGather) GetKubernetesGatherData() (model.KubernetesGatherReso
368410
}, err
369411
}
370412

371-
k8sInfo, err := k.getKubernetesInfo()
372-
if err != nil {
373-
log.Warning(err.Error(), logger.NewORGPrefix(k.orgID))
374-
return model.KubernetesGatherResource{
375-
ErrorState: common.RESOURCE_STATE_CODE_WARNING,
376-
ErrorMessage: err.Error(),
377-
}, err
378-
}
379-
k.k8sInfo = k8sInfo
380-
381413
podCluster, err := k.getPodCluster()
382414
if err != nil {
383415
return model.KubernetesGatherResource{}, err

‎server/controller/cloud/kubernetes_gather/vinterface_and_ip.go‎

Lines changed: 4 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -17,7 +17,6 @@
1717
package kubernetes_gather
1818

1919
import (
20-
"errors"
2120
"regexp"
2221
"sort"
2322
"strings"
@@ -28,7 +27,6 @@ import (
2827
cloudcommon "github.com/deepflowio/deepflow/server/controller/cloud/common"
2928
"github.com/deepflowio/deepflow/server/controller/cloud/model"
3029
"github.com/deepflowio/deepflow/server/controller/common"
31-
"github.com/deepflowio/deepflow/server/controller/genesis"
3230
"github.com/deepflowio/deepflow/server/libs/logger"
3331
"github.com/mikioh/ipaddr"
3432
)
@@ -59,20 +57,7 @@ func (k *KubernetesGather) getVInterfacesAndIPs() (nodeSubnets, podSubnets []mod
5957
}
6058

6159
// 获取vinterface API返回中host ip与其上所有node ip的对应关系
62-
if genesis.GenesisService == nil {
63-
err = errors.New("genesis service is nil")
64-
return
65-
}
66-
genesisData, err := genesis.GenesisService.GetGenesisSyncResponse(k.orgID)
67-
if err != nil {
68-
log.Error(err.Error(), logger.NewORGPrefix(k.orgID))
69-
return
70-
}
71-
vData := genesisData.Vinterfaces
72-
for _, vItem := range vData {
73-
if vItem.KubernetesClusterID != k.ClusterID {
74-
continue
75-
}
60+
for _, vItem := range k.vinterfaceData {
7661
deviceType := vItem.DeviceType
7762
if deviceType == "docker-host" || deviceType == "kvm-host" {
7863
hostIP := vItem.HostIP
@@ -96,10 +81,7 @@ func (k *KubernetesGather) getVInterfacesAndIPs() (nodeSubnets, podSubnets []mod
9681
}
9782
}
9883
// 生成device_uuid或uuid和pod lcuuid的对应关系
99-
for _, vItem := range vData {
100-
if vItem.KubernetesClusterID != k.ClusterID {
101-
continue
102-
}
84+
for _, vItem := range k.vinterfaceData {
10385
if vItem.DeviceType != "docker-container" {
10486
continue
10587
}
@@ -135,10 +117,7 @@ func (k *KubernetesGather) getVInterfacesAndIPs() (nodeSubnets, podSubnets []mod
135117
}
136118

137119
// 处理POD IP，生成port，ip，cidrs信息
138-
for _, vItem := range vData {
139-
if vItem.KubernetesClusterID != k.ClusterID {
140-
continue
141-
}
120+
for _, vItem := range k.vinterfaceData {
142121
if vItem.DeviceType != "docker-container" {
143122
continue
144123
}
@@ -402,10 +381,7 @@ func (k *KubernetesGather) getVInterfacesAndIPs() (nodeSubnets, podSubnets []mod
402381

403382
// 处理nodeIP，生成port，ip，cidrs信息
404383
nodeVinterfaceLcuuids := mapset.NewSet()
405-
for _, vItem := range vData {
406-
if vItem.KubernetesClusterID != k.ClusterID {
407-
continue
408-
}
384+
for _, vItem := range k.vinterfaceData {
409385
deviceType := vItem.DeviceType
410386
if deviceType != "docker-host" && deviceType != "kvm-host" {
411387
continue

‎server/controller/cloud/kubernetes_gather_task.go‎

Lines changed: 5 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -122,10 +122,14 @@ func (k *KubernetesGatherTask) run(rSignal *queue.OverwriteQueue) {
122122
kResource, err := k.kubernetesGather.GetKubernetesGatherData()
123123
// 这里因为任务内部没有对成功的状态赋值状态码，在这里统一处理了
124124
if err != nil {
125-
kResource.ErrorMessage = err.Error()
125+
if kResource.ErrorState == common.RESOURCE_STATE_CODE_EXIT {
126+
log.Infof("kubernetes gather (%s) assemble failed: %s", k.kubernetesGather.Name, err.Error(), logger.NewORGPrefix(k.orgID))
127+
return
128+
}
126129
if kResource.ErrorState == 0 {
127130
kResource.ErrorState = common.RESOURCE_STATE_CODE_EXCEPTION
128131
}
132+
kResource.ErrorMessage = err.Error()
129133
} else {
130134
kResource.ErrorState = common.RESOURCE_STATE_CODE_SUCCESS
131135
}

‎server/controller/common/const.go‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -586,6 +586,7 @@ const (
586586
RESOURCE_STATE_CODE_DELETING = 2
587587
RESOURCE_STATE_CODE_EXCEPTION = 3
588588
RESOURCE_STATE_CODE_WARNING = 4
589+
RESOURCE_STATE_CODE_EXIT = 5
589590
)
590591

591592
const (

‎server/controller/genesis/common/type.go‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@ import (
2626
)
2727

2828
type GenesisSyncType interface {
29+
GetInfo() string
2930
GetLcuuid() string
3031
GetVtapID() uint32
3132
}
@@ -79,6 +80,7 @@ type VIFRPCMessage struct {
7980
MessageType int
8081
TeamID uint32
8182
VtapID uint32
83+
Version uint64
8284
Peer string
8385
K8SClusterID string
8486
Key string

‎server/controller/genesis/config/config.go‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -19,12 +19,13 @@ package config
1919
type GenesisConfig struct {
2020
AgingTime float64 `default:"86400" yaml:"aging_time"`
2121
VinterfaceAgingTime float64 `default:"300" yaml:"vinterface_aging_time"`
22-
AgentHeartBeat float64 `default:"60" yaml:"agent_heart_beat"`
22+
AgentHeartBeat float64 `default:"10" yaml:"agent_heart_beat"`
2323
HostIPs []string `yaml:"host_ips"`
2424
LocalIPRanges []string `yaml:"local_ip_ranges"`
2525
ExcludeIPRanges []string `yaml:"exclude_ip_ranges"`
2626
QueueLengths int `default:"60" yaml:"queue_length"`
2727
DataPersistenceInterval int `default:"60" yaml:"data_persistence_interval"`
28+
LogDetailEnabled bool `default:"false" yaml:"log_detail_enabled"`
2829
MultiNSMode bool `default:"false" yaml:"multi_ns_mode"`
2930
SingleVPCMode bool `default:"false" yaml:"single_vpc_mode"`
3031
DefaultVPCName string `default:"default-public-vpc" yaml:"default_vpc_name"`

‎server/controller/genesis/grpc/agent_server.go‎

Lines changed: 25 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -195,7 +195,15 @@ func (g *SynchronizerServer) AgentGenesisSync(ctx context.Context, request *agen
195195
_, enabled := g.workloadResourceEnabledCache.Get(fmt.Sprintf("%d-%s", orgID, groupShortLcuuid))
196196

197197
platformData := request.GetPlatformData()
198-
if version == localVersion || platformData == nil {
198+
if version == localVersion {
199+
// It is necessary to verify whether there are available data for use,
200+
// If there is no data, needs to be re-reported.
201+
if v, ok := g.gsync.GetVtapUpdatedVersion(vtap); !ok || v != version {
202+
g.vtapToVersion.Store(vtap, uint64(0))
203+
log.Infof("genesis sync re-reporting, not data for keep alive, reset local version, from ip %s vtap_id %v", remote, vtapID, logger.NewORGPrefix(orgID))
204+
return &agent.GenesisSyncResponse{}, nil
205+
}
206+
199207
// If the worload-v is modified to be enabled during the period of continuous heartbeat,
200208
// it will trigger the re-reporting of the full data.
201209
if _, ok := g.workloadResourceChangeEnabledCache.Get(vtap); ok {
@@ -211,6 +219,7 @@ func (g *SynchronizerServer) AgentGenesisSync(ctx context.Context, request *agen
211219
Key: vtapKey,
212220
Peer: remote,
213221
VtapID: vtapID,
222+
Version: version,
214223
ORGID: orgID,
215224
TeamID: uint32(teamID),
216225
MessageType: common.TYPE_RENEW,
@@ -222,12 +231,27 @@ func (g *SynchronizerServer) AgentGenesisSync(ctx context.Context, request *agen
222231
return &agent.GenesisSyncResponse{Version: &localVersion}, nil
223232
}
224233

234+
if platformData == nil {
235+
log.Debugf("genesis sync received version %v platform is nil from ip %s vtap_id %v, local version %v", version, remote, vtapID, localVersion, logger.NewORGPrefix(orgID))
236+
return &agent.GenesisSyncResponse{}, nil
237+
}
238+
239+
// 当采集器为容器类型时（cluster id 非空）
240+
// - 采集器未注册（vtapID==0），即使没有 Interfaces 也需要处理 vinterface 来让采集器能够注册
241+
// - 采集器已经注册（vtapID!=0），采集器重启会出现 Interfaces 为空的情况
242+
// 为了避免 vinterface 异常增删，丢弃当前消息
243+
if k8sClusterID != "" && len(platformData.Interfaces) == 0 && vtapID != 0 {
244+
log.Infof("genesis sync received version %v message with empty interfaces from ip %s vtap_id %v, local version %v", version, remote, vtapID, localVersion, logger.NewORGPrefix(orgID))
245+
return &agent.GenesisSyncResponse{Version: &version}, nil
246+
}
247+
225248
log.Infof("genesis sync received version %v -> %v from ip %s vtap_id %v", localVersion, version, remote, vtapID, logger.NewORGPrefix(orgID))
226249
g.genesisSyncQueue.Put(
227250
common.VIFRPCMessage{
228251
Key: vtapKey,
229252
Peer: remote,
230253
VtapID: vtapID,
254+
Version: version,
231255
ORGID: orgID,
232256
TeamID: uint32(teamID),
233257
K8SClusterID: k8sClusterID,

‎server/controller/genesis/grpc/server.go‎

Lines changed: 25 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -198,12 +198,21 @@ func (g *SynchronizerServer) GenesisSync(ctx context.Context, request *trident.G
198198
}
199199

200200
platformData := request.GetPlatformData()
201-
if version == localVersion || platformData == nil {
201+
if version == localVersion {
202+
// It is necessary to verify whether there are available data for use,
203+
// If there is no data, needs to be re-reported.
204+
if v, ok := g.gsync.GetVtapUpdatedVersion(vtap); !ok || v != version {
205+
g.vtapToVersion.Store(vtap, uint64(0))
206+
log.Infof("genesis sync re-reporting, not data for keep alive, reset local version, from ip %s vtap_id %v", remote, vtapID, logger.NewORGPrefix(orgID))
207+
return &trident.GenesisSyncResponse{}, nil
208+
}
209+
202210
log.Debugf("genesis sync renew version %v from ip %s vtap_id %v", version, remote, vtapID, logger.NewORGPrefix(orgID))
203211
g.genesisSyncQueue.Put(
204212
common.VIFRPCMessage{
205213
Peer: remote,
206214
VtapID: vtapID,
215+
Version: version,
207216
ORGID: orgID,
208217
TeamID: uint32(teamID),
209218
MessageType: common.TYPE_RENEW,
@@ -214,11 +223,26 @@ func (g *SynchronizerServer) GenesisSync(ctx context.Context, request *trident.G
214223
return &trident.GenesisSyncResponse{Version: &localVersion}, nil
215224
}
216225

226+
if platformData == nil {
227+
log.Debugf("genesis sync received version %v platform is nil from ip %s vtap_id %v, local version %v", version, remote, vtapID, localVersion, logger.NewORGPrefix(orgID))
228+
return &trident.GenesisSyncResponse{}, nil
229+
}
230+
231+
// 当采集器为容器类型时（cluster id 非空）
232+
// - 采集器未注册（vtapID==0），即使没有 Interfaces 也需要处理 vinterface 来让采集器能够注册
233+
// - 采集器已经注册（vtapID!=0），采集器重启会出现 Interfaces 为空的情况
234+
// 为了避免 vinterface 异常增删，丢弃当前消息
235+
if k8sClusterID != "" && len(platformData.Interfaces) == 0 && vtapID != 0 {
236+
log.Infof("genesis sync received version %v message with empty interfaces from ip %s vtap_id %v, local version %v", version, remote, vtapID, localVersion, logger.NewORGPrefix(orgID))
237+
return &trident.GenesisSyncResponse{Version: &version}, nil
238+
}
239+
217240
log.Infof("genesis sync received version %v -> %v from ip %s vtap_id %v", localVersion, version, remote, vtapID, logger.NewORGPrefix(orgID))
218241
g.genesisSyncQueue.Put(
219242
common.VIFRPCMessage{
220243
Peer: remote,
221244
VtapID: vtapID,
245+
Version: version,
222246
ORGID: orgID,
223247
TeamID: uint32(teamID),
224248
K8SClusterID: k8sClusterID,

0 commit comments

Comments
 (0)