chore: clusterinfo pkg manages consumers cluster info

Author: rneuter

collector/indices.go

@@ -461,17 +461,6 @@ func NewIndices(logger *slog.Logger, client *http.Client, url *url.URL, shards b
 		},
 	}
 
-	// start go routine to fetch clusterinfo updates and save them to lastClusterinfo
-	go func() {
-		logger.Debug("starting cluster info receive loop")
-		for ci := range indices.clusterInfoCh {
-			if ci != nil {
-				logger.Debug("received cluster info update", "cluster", ci.ClusterName)
-				indices.lastClusterInfo = ci
-			}
-		}
-		logger.Debug("exiting cluster info receive loop")
-	}()
 	return indices
 }
 
@@ -481,6 +470,10 @@ func (i *Indices) ClusterLabelUpdates() *chan *clusterinfo.Response {
 	return &i.clusterInfoCh
 }
 
+func (i *Indices) SetClusterInfo(r *clusterinfo.Response) {
+	i.lastClusterInfo = r
+}
+
 // String implements the stringer interface. It is part of the clusterinfo.consumer interface
 func (i *Indices) String() string {
 	return namespace + "indices"

collector/shards.go

@@ -52,6 +52,10 @@ func (s *Shards) ClusterLabelUpdates() *chan *clusterinfo.Response {
 	return &s.clusterInfoCh
 }
 
+func (s *Shards) SetClusterInfo(r *clusterinfo.Response) {
+	s.lastClusterInfo = r
+}
+
 // String implements the stringer interface. It is part of the clusterinfo.consumer interface
 func (s *Shards) String() string {
 	return namespace + "shards"
@@ -111,18 +115,6 @@ func NewShards(logger *slog.Logger, client *http.Client, url *url.URL) *Shards {
 		}),
 	}
 
-	// start go routine to fetch clusterinfo updates and save them to lastClusterinfo
-	go func() {
-		logger.Debug("starting cluster info receive loop")
-		for ci := range shards.clusterInfoCh {
-			if ci != nil {
-				logger.Debug("received cluster info update", "cluster", ci.ClusterName)
-				shards.lastClusterInfo = ci
-			}
-		}
-		logger.Debug("exiting cluster info receive loop")
-	}()
-
 	return shards
 }
 

collector/slm.go

@@ -139,18 +139,6 @@ func NewSLM(logger *slog.Logger, u *url.URL, hc *http.Client, ci *clusterinfo.Re
 		return slm, err
 	}
 
-	// start go routine to fetch clusterinfo updates and save them to lastClusterinfo
-	go func() {
-		logger.Debug("starting cluster info receive loop")
-		for ci := range slm.clusterInfoCh {
-			if ci != nil {
-				logger.Debug("received cluster info update", "cluster", ci.ClusterName)
-				slm.lastClusterInfo = ci
-			}
-		}
-		logger.Debug("exiting cluster info receive loop")
-	}()
-
 	return slm, nil
 }
 
@@ -174,6 +162,10 @@ func (s *SLM) ClusterLabelUpdates() *chan *clusterinfo.Response {
 	return &s.clusterInfoCh
 }
 
+func (s *SLM) SetClusterInfo(r *clusterinfo.Response) {
+	s.lastClusterInfo = r
+}
+
 func (s *SLM) String() string {
 	return namespace + "slm"
 }

pkg/clusterinfo/clusterinfo.go

@@ -46,6 +46,7 @@ type consumer interface {
 	ClusterLabelUpdates() *chan *Response
 	// String implements the stringer interface
 	String() string
+	SetClusterInfo(*Response)
 }
 
 // Retriever periodically gets the cluster info from the / endpoint and
@@ -160,6 +161,18 @@ func (r *Retriever) RegisterConsumer(c consumer) error {
 		return ErrConsumerAlreadyRegistered
 	}
 	r.consumerChannels[c.String()] = c.ClusterLabelUpdates()
+
+	// start go routine to fetch clusterinfo updates and save them to the consumer
+	go func() {
+		r.logger.Debug("starting cluster info receive loop", "consumer", c.String())
+		for ci := range *c.ClusterLabelUpdates() {
+			if ci != nil {
+				r.logger.Debug("received cluster info update", "consumer", c.String(), "cluster", ci.ClusterName)
+				c.SetClusterInfo(ci)
+			}
+		}
+		r.logger.Debug("exiting cluster info receive loop", "consumer", c.String())
+	}()
 	return nil
 }
 

pkg/clusterinfo/clusterinfo_test.go

@@ -113,6 +113,12 @@ func (mc *mockConsumer) ClusterLabelUpdates() *chan *Response {
 	return &mc.ch
 }
 
+func (mc *mockConsumer) SetClusterInfo(r *Response) {
+	mc.mu.Lock()
+	mc.data = r
+	mc.mu.Unlock()
+}
+
 func TestNew(t *testing.T) {
 	u, err := url.Parse("http://localhost:9200")
 	if err != nil {