more stuff

synced-pool.go

@@ -8,6 +8,7 @@ import (
 	"log"
 	"net"
 	"strings"
+	"time"
 
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
 	"k8s.io/client-go/kubernetes"
@@ -26,10 +27,18 @@ func (k *K8sDiscovery) Discover() ([]string, error) {
 	return k.DiscoverInNamespace("")
 }
 func (k *K8sDiscovery) DiscoverInNamespace(namespace string) ([]string, error) {
-	k.client.CoreV1().Pods(namespace).List(k.ctx, metav1.ListOptions{
+	pods, err := k.client.CoreV1().Pods(namespace).List(k.ctx, metav1.ListOptions{
-		LabelSelector: "pool=cart",
+		//LabelSelector: "pool=cart",
 	})
-	return nil, nil
+	if err != nil {
+		return nil, err
+	}
+	hosts := make([]string, 0, len(pods.Items))
+	for _, pod := range pods.Items {
+		hosts = append(hosts, pod.Name)
+		log.Printf("Found pod %s\n", pod.Name)
+	}
+	return hosts, nil
 }
 
 func NewK8sDiscovery(client *kubernetes.Clientset) *K8sDiscovery {
@@ -73,6 +82,23 @@ func NewSyncedPool(local *GrainLocalPool, hostname string, d Discovery) (*Synced
 		remotes:     make([]RemoteHost, 0),
 		remoteIndex: make(map[CartId]*RemoteGrainPool),
 	}
+	if d != nil {
+		cartPurge := time.NewTicker(time.Minute)
+		go func() {
+			<-cartPurge.C
+			hosts, err := d.Discover()
+			if err != nil {
+				log.Printf("Error discovering hosts: %v\n", err)
+				return
+			}
+			for _, h := range hosts {
+				err := pool.AddRemote(h)
+				if err != nil {
+					log.Printf("Error adding remote %s: %v\n", h, err)
+				}
+			}
+		}()
+	}
 	go func() {
 		for {
 			conn, err := l.Accept()