refactor discovery

pkg/discovery/discovery.go

@@ -2,6 +2,7 @@ package discovery
 
 import (
 	"context"
+	"log"
 	"os"
 	"slices"
 	"strings"
@@ -11,6 +12,7 @@ import (
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/apimachinery/pkg/watch"
 	"k8s.io/client-go/kubernetes"
+	"k8s.io/client-go/rest"
 	"k8s.io/client-go/tools/cache"
 	toolsWatch "k8s.io/client-go/tools/watch"
 )
@@ -118,3 +120,51 @@ func NewK8sDiscoveryInNamespace(client *kubernetes.Clientset, namespace string,
 		namespace:   namespace,
 	}
 }
+
+// StartK8sDiscovery initializes kubernetes discovery for target.
+// If podIp is empty, it returns early.
+func StartK8sDiscovery(podIp string, labelSelector string, timeoutSeconds int64, target DiscoveryTarget) {
+	if podIp == "" {
+		log.Print("No discovery service available")
+		return
+	}
+
+	config, kerr := rest.InClusterConfig()
+	if kerr != nil {
+		log.Fatalf("Error creating kubernetes client: %v\n", kerr)
+	}
+	client, err := kubernetes.NewForConfig(config)
+	if err != nil {
+		log.Fatalf("Error creating client: %v\n", err)
+	}
+
+	hw := NewK8sDiscoveryInNamespace(client, InClusterNamespace(), metav1.ListOptions{
+		LabelSelector:  labelSelector,
+		TimeoutSeconds: &timeoutSeconds,
+	})
+
+	go func() {
+		ch, err := hw.Watch()
+		if err != nil {
+			log.Printf("Discovery error: %v", err)
+			return
+		}
+		for evt := range ch {
+			if evt.Host == "" {
+				continue
+			}
+			switch evt.IsReady {
+			case false:
+				if target.IsKnown(evt.Host) {
+					log.Printf("Host %s is not ready, removing", evt.Host)
+					target.RemoveHost(evt.Host)
+				}
+			default:
+				if !target.IsKnown(evt.Host) {
+					log.Printf("Discovered host %s", evt.Host)
+					target.AddRemoteHost(evt.Host)
+				}
+			}
+		}
+	}()
+}