WIP

cmd/root.go

@@ -373,17 +373,17 @@ func join_cluster(expectedInstance int, serviceName string) {
 	}
 
 	namespace := os.Getenv("MY_POD_NAMESPACE")
-	var endpoints *v1.EndpointsList
+	var endpoints *v1.Endpoints
 	var nodeCount int
 	for {
 		// wait until endpoints returns the expected number of instances
-		endpoints, err = clientset.CoreV1().Endpoints(namespace).List(context.TODO(), metav1.ListOptions{})
+		endpoints, err = clientset.CoreV1().Endpoints(namespace).Get(context.TODO(), serviceName, metav1.GetOptions{})
 		if err != nil {
 			log.Error("failed to retrieve endpoints; retrying...", "name", serviceName, "error", err)
 			time.Sleep(time.Second)
 			continue
 		}
-		nodeCount = len(endpoints.Items)
+		nodeCount = len(endpoints.Subsets[0].Addresses)
 		if nodeCount >= expectedInstance {
 			log.Info("reached the expected number of instances", "expected instances", expectedInstance, "current instances", nodeCount)
 			break
@@ -397,14 +397,14 @@ func join_cluster(expectedInstance int, serviceName string) {
 		panic("Failed to create memberlist: " + err.Error())
 	}
 
-	nodeStrings := make([]string, nodeCount)
-	for i, node := range endpoints.Items {
-		nodeStrings[i] = node.GetName()
+	ips := make([]string, nodeCount)
+	for i, node := range endpoints.Subsets[0].Addresses {
+		ips[i] = node.IP
 	}
 
 	// join the cluster
 	for {
-		_, err = list.Join(nodeStrings)
+		_, err = list.Join(ips)
 		if err == nil {
 			break
 		}
@@ -423,8 +423,8 @@ func join_cluster(expectedInstance int, serviceName string) {
 		time.Sleep(time.Second)
 	}
 
-	list.Leave(time.Second)
-	list.Shutdown()
+	_ = list.Leave(time.Second)
+	_ = list.Shutdown()
 }
 
 func startConsumers(ctx context.Context, consumerProto config.Protocol, wg *sync.WaitGroup) {