diff --git a/cmd/cart/k8s-host-discovery.go b/cmd/cart/k8s-host-discovery.go index 16c9e3c..fc93994 100644 --- a/cmd/cart/k8s-host-discovery.go +++ b/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) { - - 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()) + discovery.StartK8sDiscovery(podIp, "actor-pool=cart", 30, pool) } diff --git a/cmd/checkout/k8s-host-discovery.go b/cmd/checkout/k8s-host-discovery.go index 9844b76..70db1a8 100644 --- a/cmd/checkout/k8s-host-discovery.go +++ b/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) { - - 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()) + discovery.StartK8sDiscovery(podIp, "actor-pool=checkout", 30, pool) } diff --git a/cmd/order/k8s-host-discovery.go b/cmd/order/k8s-host-discovery.go new file mode 100644 index 0000000..03368c7 --- /dev/null +++ b/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) +} diff --git a/cmd/order/main.go b/cmd/order/main.go index 52b792e..9cd1488 100644 --- a/cmd/order/main.go +++ b/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) { - return nil, fmt.Errorf("order service is single-node") + SpawnHost: func(host string) (actor.Host[order.OrderGrain], error) { + 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) diff --git a/cmd/profile/k8s-host-discovery.go b/cmd/profile/k8s-host-discovery.go index 41e475b..bdbde9d 100644 --- a/cmd/profile/k8s-host-discovery.go +++ b/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) { - 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()) + discovery.StartK8sDiscovery(podIp, "actor-pool=profile", 30, pool) } diff --git a/pkg/discovery/discovery.go b/pkg/discovery/discovery.go index b61f5c8..9611731 100644 --- a/pkg/discovery/discovery.go +++ b/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) + } + } + } + }() +}