refactor discovery

cmd/cart/k8s-host-discovery.go

@@ -1,65 +1,9 @@
 package main
 
 import (
-	"log"
-
 	"git.k6n.net/mats/go-cart-actor/pkg/discovery"
-	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/rest"
 )
 
-func GetDiscovery() discovery.Discovery {
-	if podIp == "" {
-		return nil
-	}
-
-	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)
-	}
-	timeout := int64(30)
-	// Scope discovery to this pod's namespace so it only needs a namespaced Role
-	// (pods: get/list/watch), not a cluster-wide ClusterRole.
-	return discovery.NewK8sDiscoveryInNamespace(client, discovery.InClusterNamespace(), v1.ListOptions{
-		LabelSelector:  "actor-pool=cart",
-		TimeoutSeconds: &timeout,
-	})
-}
-
 func UseDiscovery(pool discovery.DiscoveryTarget) {
-
+	discovery.StartK8sDiscovery(podIp, "actor-pool=cart", 30, pool)
-	go func(hw discovery.Discovery) {
-		if hw == nil {
-			log.Print("No discovery service available")
-			return
-		}
-		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 pool.IsKnown(evt.Host) {
-					log.Printf("Host %s is not ready, removing", evt.Host)
-					pool.RemoveHost(evt.Host)
-				}
-			default:
-				if !pool.IsKnown(evt.Host) {
-					log.Printf("Discovered host %s", evt.Host)
-					pool.AddRemoteHost(evt.Host)
-				}
-			}
-		}
-	}(GetDiscovery())
 }

cmd/cart/pool-server.go

@@ -76,6 +76,9 @@ func (s *PoolServer) GetCartHandler(w http.ResponseWriter, r *http.Request, id c
 func (s *PoolServer) AddSkuToCartHandler(w http.ResponseWriter, r *http.Request, id cart.CartId) error {
 	sku := r.PathValue("sku")
 	country := getCountryFromHost(r.Host)
+	if s.idx == nil {
+		log.Printf("No index present")
+	}
 	if s.idx != nil && s.idx.IsDeleted(sku) {
 		// Bus-deleted SKU — skip the HTTP fetch and reject the add. Mirrors the
 		// product-fetcher's "product service returned %d for sku %s" shape so
@@ -100,6 +103,9 @@ func (s *PoolServer) AddSkuToCartHandler(w http.ResponseWriter, r *http.Request,
 		}
 	}
 	if msg == nil {
+		if s.idx == nil {
+			log.Printf("No item found in index %s", sku)
+		}
 		var err error
 		msg, err = GetItemAddMessage(r.Context(), sku, 1, country, nil)
 		if err != nil {

cmd/checkout/k8s-host-discovery.go

@@ -1,65 +1,9 @@
 package main
 
 import (
-	"log"
-
 	"git.k6n.net/mats/go-cart-actor/pkg/discovery"
-	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/rest"
 )
 
-func GetDiscovery() discovery.Discovery {
-	if podIp == "" {
-		return nil
-	}
-
-	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)
-	}
-	timeout := int64(30)
-	// Scope discovery to this pod's namespace so it only needs a namespaced Role
-	// (pods: get/list/watch), not a cluster-wide ClusterRole.
-	return discovery.NewK8sDiscoveryInNamespace(client, discovery.InClusterNamespace(), v1.ListOptions{
-		LabelSelector:  "actor-pool=checkout",
-		TimeoutSeconds: &timeout,
-	})
-}
-
 func UseDiscovery(pool discovery.DiscoveryTarget) {
-
+	discovery.StartK8sDiscovery(podIp, "actor-pool=checkout", 30, pool)
-	go func(hw discovery.Discovery) {
-		if hw == nil {
-			log.Print("No discovery service available")
-			return
-		}
-		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 pool.IsKnown(evt.Host) {
-					log.Printf("Host %s is not ready, removing", evt.Host)
-					pool.RemoveHost(evt.Host)
-				}
-			default:
-				if !pool.IsKnown(evt.Host) {
-					log.Printf("Discovered host %s", evt.Host)
-					pool.AddRemoteHost(evt.Host)
-				}
-			}
-		}
-	}(GetDiscovery())
 }

cmd/order/k8s-host-discovery.go

@@ -0,0 +1,9 @@
+package main
+
+import (
+	"git.k6n.net/mats/go-cart-actor/pkg/discovery"
+)
+
+func UseDiscovery(pool discovery.DiscoveryTarget) {
+	discovery.StartK8sDiscovery(podIp, "actor-pool=order", 30, pool)
+}

cmd/order/main.go

@@ -27,6 +27,7 @@ import (
 	"git.k6n.net/mats/go-cart-actor/pkg/order"
 	ordermcp "git.k6n.net/mats/go-cart-actor/pkg/order/mcp"
 	"git.k6n.net/mats/go-cart-actor/pkg/outbox"
+	"git.k6n.net/mats/go-cart-actor/pkg/proxy"
 	messages "git.k6n.net/mats/go-cart-actor/proto/order"
 	"git.k6n.net/mats/platform/config"
 	"git.k6n.net/mats/platform/rabbit"
@@ -36,6 +37,9 @@ import (
 	"google.golang.org/protobuf/proto"
 )
 
+var podIp = os.Getenv("POD_IP")
+var name = os.Getenv("POD_NAME")
+
 type server struct {
 	pool                  *actor.SimpleGrainPool[order.OrderGrain]
 	applier               *orderedApplier
@@ -66,8 +70,13 @@ func main() {
 	order.RegisterMutations(reg)
 	storage := actor.NewDiskStorage[order.OrderGrain](dataDir, reg)
 
+	hostname := podIp
+	if hostname == "" {
+		hostname = "order-1"
+	}
+
 	pool, err := actor.NewSimpleGrainPool(actor.GrainPoolConfig[order.OrderGrain]{
-		Hostname: "order-1",
+		Hostname: hostname,
 		Spawn: func(ctx context.Context, id uint64) (actor.Grain[order.OrderGrain], error) {
 			g := order.NewOrderGrain(id, time.Now())
 			// Replay any persisted history so the resident grain is current.
@@ -76,8 +85,8 @@ func main() {
 			}
 			return g, nil
 		},
-		SpawnHost: func(string) (actor.Host[order.OrderGrain], error) {
+		SpawnHost: func(host string) (actor.Host[order.OrderGrain], error) {
-			return nil, fmt.Errorf("order service is single-node")
+			return proxy.NewRemoteHost[order.OrderGrain](host)
 		},
 		Destroy:  func(actor.Grain[order.OrderGrain]) error { return nil },
 		TTL:      time.Hour,
@@ -95,6 +104,15 @@ func main() {
 		log.Fatalf("create pool: %v", err)
 	}
 
+	controlPlaneConfig := actor.DefaultServerConfig()
+	grpcSrv, err := actor.NewControlServer[order.OrderGrain](controlPlaneConfig, pool)
+	if err != nil {
+		log.Fatalf("Error starting control plane gRPC server: %v\n", err)
+	}
+	defer grpcSrv.GracefulStop()
+
+	UseDiscovery(pool)
+
 	applier := &orderedApplier{pool: pool, storage: storage}
 	provider := selectProvider(logger)
 

cmd/profile/k8s-host-discovery.go

@@ -1,68 +1,9 @@
 package main
 
 import (
-	"log"
-
 	"git.k6n.net/mats/go-cart-actor/pkg/discovery"
-	v1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/client-go/kubernetes"
-	"k8s.io/client-go/rest"
 )
 
-// GetDiscovery returns the k8s pod-watcher that finds peer profile replicas, or
-// nil when POD_IP is unset (single-node / local dev — clustering disabled). The
-// watch is scoped to this pod's own namespace so it only needs a namespaced Role
-// (pods: get/list/watch), not a cluster-wide role.
-func GetDiscovery() discovery.Discovery {
-	if podIp == "" {
-		return nil
-	}
-
-	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)
-	}
-	timeout := int64(30)
-	return discovery.NewK8sDiscoveryInNamespace(client, discovery.InClusterNamespace(), v1.ListOptions{
-		LabelSelector:  "actor-pool=profile",
-		TimeoutSeconds: &timeout,
-	})
-}
-
-// UseDiscovery starts the watch loop that adds/removes peer hosts from the pool
-// as profile replicas become ready or go away. A nil discovery (POD_IP unset)
-// logs and returns, leaving the pool in single-node mode.
 func UseDiscovery(pool discovery.DiscoveryTarget) {
-	go func(hw discovery.Discovery) {
+	discovery.StartK8sDiscovery(podIp, "actor-pool=profile", 30, pool)
-		if hw == nil {
-			log.Print("No discovery service available")
-			return
-		}
-		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 pool.IsKnown(evt.Host) {
-					log.Printf("Host %s is not ready, removing", evt.Host)
-					pool.RemoveHost(evt.Host)
-				}
-			default:
-				if !pool.IsKnown(evt.Host) {
-					log.Printf("Discovered host %s", evt.Host)
-					pool.AddRemoteHost(evt.Host)
-				}
-			}
-		}
-	}(GetDiscovery())
 }

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)
+				}
+			}
+		}
+	}()
+}