()
| 288 | } |
| 289 | |
| 290 | func (c *ClusterChecker) probeLoop() { |
| 291 | defer c.wg.Done() |
| 292 | log := logger.Get().With( |
| 293 | zap.String("namespace", c.namespace), |
| 294 | zap.String("clusterName", c.clusterName), |
| 295 | ) |
| 296 | |
| 297 | probeTicker := time.NewTicker(c.options.pingInterval) |
| 298 | defer probeTicker.Stop() |
| 299 | for { |
| 300 | select { |
| 301 | case <-probeTicker.C: |
| 302 | clusterInfo, err := c.clusterStore.GetCluster(c.ctx, c.namespace, c.clusterName) |
| 303 | if err != nil { |
| 304 | log.Error("Failed to get the clusterName info from the clusterStore", zap.Error(err)) |
| 305 | break |
| 306 | } |
| 307 | c.clusterMu.Lock() |
| 308 | c.cluster = clusterInfo |
| 309 | c.clusterMu.Unlock() |
| 310 | c.parallelProbeNodes(c.ctx, clusterInfo) |
| 311 | case <-c.syncCh: |
| 312 | if err := c.syncClusterToNodes(c.ctx); err != nil { |
| 313 | log.Error("Failed to sync the clusterName to the nodes", zap.Error(err)) |
| 314 | } |
| 315 | case <-c.ctx.Done(): |
| 316 | return |
| 317 | } |
| 318 | } |
| 319 | } |
| 320 | |
| 321 | func (c *ClusterChecker) updateCluster(cluster *store.Cluster) { |
| 322 | c.clusterMu.Lock() |
no test coverage detected