update deployment

synced-pool.go

@@ -1,14 +1,44 @@
 package main
 
 import (
+	"context"
 	"encoding/binary"
 	"fmt"
 	"io"
 	"log"
 	"net"
 	"strings"
+
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/client-go/kubernetes"
 )
 
+type Discovery interface {
+	Discover() ([]string, error)
+}
+
+type K8sDiscovery struct {
+	ctx    context.Context
+	client *kubernetes.Clientset
+}
+
+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{
+		LabelSelector: "pool=cart",
+	})
+	return nil, nil
+}
+
+func NewK8sDiscovery(client *kubernetes.Clientset) *K8sDiscovery {
+	return &K8sDiscovery{
+		ctx:    context.Background(),
+		client: client,
+	}
+}
+
 type Quorum interface {
 	Negotiate(knownHosts []string) ([]string, error)
 	ListChanged([]CartId) error
@@ -21,6 +51,7 @@ type RemoteHost struct {
 }
 
 type SyncedPool struct {
+	Discovery   Discovery
 	listener    net.Listener
 	Hostname    string
 	local       *GrainLocalPool
@@ -28,13 +59,14 @@ type SyncedPool struct {
 	remoteIndex map[CartId]*RemoteGrainPool
 }
 
-func NewSyncedPool(local *GrainLocalPool, hostname string) (*SyncedPool, error) {
+func NewSyncedPool(local *GrainLocalPool, hostname string, d Discovery) (*SyncedPool, error) {
 	listen := fmt.Sprintf("%s:1338", hostname)
 	l, err := net.Listen("tcp", listen)
 	if err != nil {
 		return nil, err
 	}
 	pool := &SyncedPool{
+		Discovery:   d,
 		Hostname:    hostname,
 		local:       local,
 		listener:    l,