diff --git a/cmd/main.go b/cmd/main.go index 035ca0c..bddce0c 100644 --- a/cmd/main.go +++ b/cmd/main.go @@ -37,6 +37,7 @@ import ( memgraphcomv1alpha1 "github.com/memgraph/kubernetes-operator/api/v1alpha1" "github.com/memgraph/kubernetes-operator/internal/controller" + "github.com/memgraph/kubernetes-operator/internal/memgraph" // +kubebuilder:scaffold:imports ) @@ -179,8 +180,9 @@ func main() { } if err := (&controller.MemgraphClusterReconciler{ - Client: mgr.GetClient(), - Scheme: mgr.GetScheme(), + Client: mgr.GetClient(), + Scheme: mgr.GetScheme(), + Memgraph: memgraph.NewBoltConnector(), }).SetupWithManager(mgr); err != nil { setupLog.Error(err, "Failed to create controller", "controller", "memgraphcluster") os.Exit(1) diff --git a/config/manager/kustomization.yaml b/config/manager/kustomization.yaml index 5c5f0b8..a7b129f 100644 --- a/config/manager/kustomization.yaml +++ b/config/manager/kustomization.yaml @@ -1,2 +1,8 @@ resources: - manager.yaml +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization +images: +- name: controller + newName: example.com/kubernetes-operator + newTag: v0.0.1 diff --git a/go.mod b/go.mod index 2dcf65d..7ff709f 100644 --- a/go.mod +++ b/go.mod @@ -4,6 +4,7 @@ go 1.26.0 require ( github.com/google/go-cmp v0.7.0 + github.com/neo4j/neo4j-go-driver/v5 v5.28.4 github.com/onsi/ginkgo/v2 v2.27.4 github.com/onsi/gomega v1.39.0 k8s.io/api v0.36.0 diff --git a/go.sum b/go.sum index 690c70d..ff062ed 100644 --- a/go.sum +++ b/go.sum @@ -105,6 +105,8 @@ github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee h1:W5t00kpgFd github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= +github.com/neo4j/neo4j-go-driver/v5 v5.28.4 h1:7toxehVcYkZbyxV4W3Ib9VcnyRBQPucF+VwNNmtSXi4= +github.com/neo4j/neo4j-go-driver/v5 v5.28.4/go.mod h1:Vff8OwT7QpLm7L2yYr85XNWe9Rbqlbeb9asNXJTHO4k= github.com/onsi/ginkgo/v2 v2.27.4 h1:fcEcQW/A++6aZAZQNUmNjvA9PSOzefMJBerHJ4t8v8Y= github.com/onsi/ginkgo/v2 v2.27.4/go.mod h1:ArE1D/XhNXBXCBkKOLkbsb2c81dQHCRcF5zwn/ykDRo= github.com/onsi/gomega v1.39.0 h1:y2ROC3hKFmQZJNFeGAMeHZKkjBL65mIZcvrLQBF9k6Q= diff --git a/internal/controller/fake_memgraph_test.go b/internal/controller/fake_memgraph_test.go new file mode 100644 index 0000000..e66eafa --- /dev/null +++ b/internal/controller/fake_memgraph_test.go @@ -0,0 +1,163 @@ +/* +Copyright 2026. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package controller + +import ( + "context" + "fmt" + "slices" + "sync" + + "github.com/memgraph/kubernetes-operator/internal/memgraph" +) + +// fakeMemgraph is an in-memory Memgraph HA cluster behind the +// memgraph.Connector seam. It keeps one shared SHOW INSTANCES view, applies +// registration commands to it, and — like the real thing — rejects duplicate +// registrations and second MAIN promotions, so any controller behavior that +// is not read-before-write fails the suite loudly. +type fakeMemgraph struct { + mu sync.Mutex + + // instances is the cluster view every coordinator serves. + instances []memgraph.Instance + connectAttempts int + // executed records every mutating command as ": ". + executed []string +} + +func newFakeMemgraph() *fakeMemgraph { + return &fakeMemgraph{} +} + +func (f *fakeMemgraph) Connect(_ context.Context, address string) (memgraph.Client, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.connectAttempts++ + return &fakeClient{cluster: f, address: address}, nil +} + +func (f *fakeMemgraph) connects() int { + f.mu.Lock() + defer f.mu.Unlock() + return f.connectAttempts +} + +func (f *fakeMemgraph) executedCommands() []string { + f.mu.Lock() + defer f.mu.Unlock() + return slices.Clone(f.executed) +} + +func (f *fakeMemgraph) setInstances(instances []memgraph.Instance) { + f.mu.Lock() + defer f.mu.Unlock() + f.instances = slices.Clone(instances) +} + +type fakeClient struct { + cluster *fakeMemgraph + address string + closed bool +} + +func (c *fakeClient) ShowInstances(context.Context) ([]memgraph.Instance, error) { + c.cluster.mu.Lock() + defer c.cluster.mu.Unlock() + if c.closed { + return nil, fmt.Errorf("fake memgraph: connection to %s already closed", c.address) + } + return slices.Clone(c.cluster.instances), nil +} + +func (c *fakeClient) AddCoordinator(_ context.Context, coordinator memgraph.CoordinatorSpec) error { + return c.execute(fmt.Sprintf("ADD COORDINATOR %d", coordinator.ID), func() error { + if c.cluster.hasInstance(coordinator.Name()) { + return fmt.Errorf("fake memgraph: coordinator %s already exists", coordinator.Name()) + } + c.cluster.instances = append(c.cluster.instances, memgraph.Instance{ + Name: coordinator.Name(), + BoltServer: coordinator.BoltServer, + CoordinatorServer: coordinator.CoordinatorServer, + ManagementServer: coordinator.ManagementServer, + Health: "up", + Role: memgraph.RoleFollower, + }) + return nil + }) +} + +func (c *fakeClient) RegisterInstance(_ context.Context, instance memgraph.DataInstanceSpec) error { + return c.execute("REGISTER INSTANCE "+instance.Name, func() error { + if c.cluster.hasInstance(instance.Name) { + return fmt.Errorf("fake memgraph: instance %s already registered", instance.Name) + } + c.cluster.instances = append(c.cluster.instances, memgraph.Instance{ + Name: instance.Name, + BoltServer: instance.BoltServer, + ManagementServer: instance.ManagementServer, + Health: "up", + Role: memgraph.RoleReplica, + }) + return nil + }) +} + +func (c *fakeClient) SetInstanceToMain(_ context.Context, name string) error { + return c.execute(fmt.Sprintf("SET INSTANCE %s TO MAIN", name), func() error { + for _, instance := range c.cluster.instances { + if instance.IsMain() { + return fmt.Errorf("fake memgraph: %s is already MAIN", instance.Name) + } + } + for i, instance := range c.cluster.instances { + if instance.Name == name { + c.cluster.instances[i].Role = memgraph.RoleMain + return nil + } + } + return fmt.Errorf("fake memgraph: instance %s is not registered", name) + }) +} + +func (c *fakeClient) Close(context.Context) error { + c.cluster.mu.Lock() + defer c.cluster.mu.Unlock() + c.closed = true + return nil +} + +// execute records the command and applies it to the shared cluster view. +func (c *fakeClient) execute(command string, apply func() error) error { + c.cluster.mu.Lock() + defer c.cluster.mu.Unlock() + if c.closed { + return fmt.Errorf("fake memgraph: connection to %s already closed", c.address) + } + if err := apply(); err != nil { + return err + } + c.cluster.executed = append(c.cluster.executed, c.address+": "+command) + return nil +} + +// hasInstance must be called with the cluster lock held. +func (f *fakeMemgraph) hasInstance(name string) bool { + return slices.ContainsFunc(f.instances, func(instance memgraph.Instance) bool { + return instance.Name == name + }) +} diff --git a/internal/controller/memgraphcluster_controller.go b/internal/controller/memgraphcluster_controller.go index 6de579a..fa5e011 100644 --- a/internal/controller/memgraphcluster_controller.go +++ b/internal/controller/memgraphcluster_controller.go @@ -18,18 +18,23 @@ package controller import ( "context" + "errors" "fmt" + "time" appsv1 "k8s.io/api/apps/v1" corev1 "k8s.io/api/core/v1" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime" + "k8s.io/apimachinery/pkg/types" ctrl "sigs.k8s.io/controller-runtime" "sigs.k8s.io/controller-runtime/pkg/client" "sigs.k8s.io/controller-runtime/pkg/controller/controllerutil" logf "sigs.k8s.io/controller-runtime/pkg/log" memgraphcomv1alpha1 "github.com/memgraph/kubernetes-operator/api/v1alpha1" + "github.com/memgraph/kubernetes-operator/internal/memgraph" + "github.com/memgraph/kubernetes-operator/internal/planner" "github.com/memgraph/kubernetes-operator/internal/resources" ) @@ -37,10 +42,26 @@ import ( // manager of the workload objects it provisions. const fieldOwner = "memgraph-operator" +const ( + // requeueWhilePending is how long to wait before retrying when the + // cluster cannot be registered yet — pods not ready, or coordinators not + // answering Bolt queries. Both are expected while the cluster starts up. + requeueWhilePending = 10 * time.Second + + // requeueAfterRegistration schedules the follow-up reconcile that + // verifies issued registration commands actually converged the cluster. + requeueAfterRegistration = 10 * time.Second +) + // MemgraphClusterReconciler reconciles a MemgraphCluster object type MemgraphClusterReconciler struct { client.Client Scheme *runtime.Scheme + + // Memgraph opens Bolt connections to coordinators. Tests substitute a + // fake; everything above the memgraph.Client interface never touches the + // Bolt driver. + Memgraph memgraph.Connector } // +kubebuilder:rbac:groups=memgraph.com,resources=memgraphclusters,verbs=get;list;watch;create;update;patch;delete @@ -49,11 +70,16 @@ type MemgraphClusterReconciler struct { // +kubebuilder:rbac:groups=apps,resources=statefulsets,verbs=get;list;watch;create;update;patch;delete // +kubebuilder:rbac:groups=core,resources=services,verbs=get;list;watch;create;update;patch;delete -// Reconcile drives the cluster toward the declared MemgraphCluster spec by -// server-side-applying the builders' desired objects: one StatefulSet per -// role (coordinators, data instances), each backed by a headless Service. -// Deletion needs no handling here — every object carries a controller owner -// reference, so garbage collection removes the workloads with the CR. +// Reconcile drives the cluster toward the declared MemgraphCluster spec in +// two stages. First it server-side-applies the builders' desired objects: one +// StatefulSet per role (coordinators, data instances), each backed by a +// headless Service. Then, once every pod is ready, it reconciles cluster +// registration: observe SHOW INSTANCES on the coordinator leader, diff +// against the declared topology, and issue only the missing commands. All +// interaction is read-before-write and idempotent, so an operator restart +// mid-bootstrap is harmless. Deletion needs no handling here — every object +// carries a controller owner reference, so garbage collection removes the +// workloads with the CR. func (r *MemgraphClusterReconciler) Reconcile(ctx context.Context, req ctrl.Request) (ctrl.Result, error) { log := logf.FromContext(ctx) @@ -79,7 +105,154 @@ func (r *MemgraphClusterReconciler) Reconcile(ctx context.Context, req ctrl.Requ log.Info("Applied desired workload objects for MemgraphCluster", "memgraphcluster", req.NamespacedName) - return ctrl.Result{}, nil + return r.reconcileRegistration(ctx, &cluster) +} + +// reconcileRegistration converges cluster registration once the workloads are +// ready: find the coordinator leader, plan against its SHOW INSTANCES view, +// and execute the missing commands. Unreachable coordinators are retried on a +// delay rather than surfaced as errors — Bolt endpoints lagging pod readiness +// is a normal startup phase, not a failure. +func (r *MemgraphClusterReconciler) reconcileRegistration( + ctx context.Context, + cluster *memgraphcomv1alpha1.MemgraphCluster, +) (ctrl.Result, error) { + log := logf.FromContext(ctx) + + ready, err := r.workloadsReady(ctx, cluster) + if err != nil { + return ctrl.Result{}, err + } + if !ready { + log.Info("Waited for workload pods to become ready before registration") + return ctrl.Result{RequeueAfter: requeueWhilePending}, nil + } + + topology := resources.DeclaredTopology(cluster) + leader, observed, err := r.observeCluster(ctx, topology) + if err != nil { + log.Info("Deferred registration because no coordinator answered", "reason", err.Error()) + return ctrl.Result{RequeueAfter: requeueWhilePending}, nil + } + defer func() { + if err := leader.Close(ctx); err != nil { + log.Error(err, "Failed to close coordinator connection") + } + }() + + commands := planner.Plan(topology, observed) + if len(commands) == 0 { + log.Info("Confirmed cluster registration is converged") + return ctrl.Result{}, nil + } + for _, command := range commands { + if err := command.Run(ctx, leader); err != nil { + return ctrl.Result{}, fmt.Errorf("executing registration command %q: %w", command, err) + } + log.Info("Executed registration command", "command", command.String()) + } + + // Registration was issued, not yet observed back; verify convergence on a + // follow-up reconcile instead of assuming success. + return ctrl.Result{RequeueAfter: requeueAfterRegistration}, nil +} + +// workloadsReady reports whether both role StatefulSets have all their pods +// ready. Registration waits for the full topology: coordinators cannot form a +// Raft cluster and data instances cannot be registered until every advertised +// address resolves to a running pod. +func (r *MemgraphClusterReconciler) workloadsReady( + ctx context.Context, + cluster *memgraphcomv1alpha1.MemgraphCluster, +) (bool, error) { + for _, name := range []string{resources.CoordinatorName(cluster), resources.DataName(cluster)} { + var sts appsv1.StatefulSet + if err := r.Get(ctx, types.NamespacedName{Name: name, Namespace: cluster.Namespace}, &sts); err != nil { + return false, fmt.Errorf("getting StatefulSet %s: %w", name, err) + } + if sts.Spec.Replicas == nil || sts.Status.ReadyReplicas < *sts.Spec.Replicas { + return false, nil + } + } + return true, nil +} + +// observeCluster connects to the coordinator leader and returns its client +// together with the SHOW INSTANCES view the planner diffs against. +// Coordinators are tried in ordinal order: one reporting itself leader is +// used directly, a follower redirects to the leader it reports, and when no +// leader exists yet (fresh cluster, Raft not formed) the first reachable +// coordinator is used — adding coordinators to it makes it the leader, +// mirroring the HA chart's bootstrap against its first coordinator. +func (r *MemgraphClusterReconciler) observeCluster( + ctx context.Context, + topology planner.Topology, +) (memgraph.Client, []memgraph.Instance, error) { + var errs []error + for _, coordinator := range topology.Coordinators { + leader, observed, err := r.showInstances(ctx, coordinator) + if err != nil { + errs = append(errs, err) + continue + } + + leaderName := "" + for _, instance := range observed { + if instance.IsLeader() { + leaderName = instance.Name + break + } + } + if leaderName == "" || leaderName == coordinator.Name() { + return leader, observed, nil + } + + // This coordinator is a follower; redirect to the leader it reports. + if err := leader.Close(ctx); err != nil { + errs = append(errs, err) + } + candidate, found := coordinatorByName(topology, leaderName) + if !found { + errs = append(errs, fmt.Errorf("%s reported leader %s, which is not declared", coordinator.Name(), leaderName)) + continue + } + leader, observed, err = r.showInstances(ctx, candidate) + if err != nil { + errs = append(errs, err) + continue + } + return leader, observed, nil + } + return nil, nil, fmt.Errorf("no coordinator leader reachable: %w", errors.Join(errs...)) +} + +func coordinatorByName(topology planner.Topology, name string) (memgraph.CoordinatorSpec, bool) { + for _, coordinator := range topology.Coordinators { + if coordinator.Name() == name { + return coordinator, true + } + } + return memgraph.CoordinatorSpec{}, false +} + +// showInstances connects to one coordinator and fetches its cluster view, +// closing the connection again on query failure. +func (r *MemgraphClusterReconciler) showInstances( + ctx context.Context, + coordinator memgraph.CoordinatorSpec, +) (memgraph.Client, []memgraph.Instance, error) { + c, err := r.Memgraph.Connect(ctx, coordinator.BoltServer) + if err != nil { + return nil, nil, err + } + observed, err := c.ShowInstances(ctx) + if err != nil { + if closeErr := c.Close(ctx); closeErr != nil { + return nil, nil, errors.Join(err, closeErr) + } + return nil, nil, err + } + return c, observed, nil } // apply server-side-applies a desired object built by the resource builders. diff --git a/internal/controller/memgraphcluster_controller_test.go b/internal/controller/memgraphcluster_controller_test.go index 56f39b8..344d497 100644 --- a/internal/controller/memgraphcluster_controller_test.go +++ b/internal/controller/memgraphcluster_controller_test.go @@ -31,6 +31,7 @@ import ( "sigs.k8s.io/controller-runtime/pkg/reconcile" memgraphcomv1alpha1 "github.com/memgraph/kubernetes-operator/api/v1alpha1" + "github.com/memgraph/kubernetes-operator/internal/memgraph" ) // Name suffixes of the per-role workload objects a reconcile creates. @@ -44,21 +45,27 @@ var _ = Describe("MemgraphCluster Controller", func() { ctx := context.Background() - var reconciler *MemgraphClusterReconciler + var ( + reconciler *MemgraphClusterReconciler + fake *fakeMemgraph + ) BeforeEach(func() { + fake = newFakeMemgraph() reconciler = &MemgraphClusterReconciler{ - Client: k8sClient, - Scheme: k8sClient.Scheme(), + Client: k8sClient, + Scheme: k8sClient.Scheme(), + Memgraph: fake, } }) - reconcileCluster := func(name string) { + reconcileCluster := func(name string) reconcile.Result { GinkgoHelper() - _, err := reconciler.Reconcile(ctx, reconcile.Request{ + result, err := reconciler.Reconcile(ctx, reconcile.Request{ NamespacedName: types.NamespacedName{Name: name, Namespace: resourceNamespace}, }) Expect(err).NotTo(HaveOccurred()) + return result } get := func(name string, obj client.Object) { @@ -83,6 +90,22 @@ var _ = Describe("MemgraphCluster Controller", func() { } } + // markWorkloadsReady simulates the kubelet envtest does not run: it + // reports every replica of both role StatefulSets as ready, which is what + // gates the registration flow. + markWorkloadsReady := func(clusterName string) { + GinkgoHelper() + for _, suffix := range []string{coordinatorSuffix, dataSuffix} { + sts := &appsv1.StatefulSet{} + get(clusterName+suffix, sts) + sts.Status.Replicas = *sts.Spec.Replicas + sts.Status.ReadyReplicas = *sts.Spec.Replicas + sts.Status.AvailableReplicas = *sts.Spec.Replicas + sts.Status.ObservedGeneration = sts.Generation + Expect(k8sClient.Status().Update(ctx, sts)).To(Succeed()) + } + } + expectControlledBy := func(obj client.Object, cluster *memgraphcomv1alpha1.MemgraphCluster) { GinkgoHelper() ref := metav1.GetControllerOf(obj) @@ -250,4 +273,118 @@ var _ = Describe("MemgraphCluster Controller", func() { Expect(k8sClient.Create(ctx, invalid)).NotTo(Succeed()) }) }) + + Context("when bootstrapping cluster registration", func() { + const resourceName = "mgc-bootstrap" + + coordinatorAddress := func(ordinal int) string { + return fmt.Sprintf("%s-coordinator-%d.%s-coordinator.%s.svc.cluster.local:7687", + resourceName, ordinal, resourceName, resourceNamespace) + } + + // observedCoordinator reports the coordinator with the given 1-based + // Raft ID, which runs on the pod with ordinal ID-1. + observedCoordinator := func(id int, role string) memgraph.Instance { + return memgraph.Instance{ + Name: fmt.Sprintf("coordinator_%d", id), + BoltServer: coordinatorAddress(id - 1), + Health: "up", + Role: role, + } + } + + observedDataInstance := func(i int, role string) memgraph.Instance { + return memgraph.Instance{ + Name: fmt.Sprintf("instance_%d", i), + Health: "up", + Role: role, + } + } + + BeforeEach(func() { + resource := &memgraphcomv1alpha1.MemgraphCluster{ + ObjectMeta: metav1.ObjectMeta{Name: resourceName, Namespace: resourceNamespace}, + } + Expect(k8sClient.Create(ctx, resource)).To(Succeed()) + }) + + AfterEach(func() { + cluster := &memgraphcomv1alpha1.MemgraphCluster{} + get(resourceName, cluster) + Expect(k8sClient.Delete(ctx, cluster)).To(Succeed()) + deleteOwned(resourceName) + }) + + It("should not touch Memgraph before every pod is ready", func() { + result := reconcileCluster(resourceName) + + Expect(result.RequeueAfter).To(BeNumerically(">", 0)) + Expect(fake.connects()).To(BeZero()) + }) + + It("should bootstrap a fresh cluster to fully registered with one MAIN", func() { + reconcileCluster(resourceName) + markWorkloadsReady(resourceName) + + result := reconcileCluster(resourceName) + + leader := coordinatorAddress(0) + Expect(fake.executedCommands()).To(Equal([]string{ + leader + ": ADD COORDINATOR 1", + leader + ": ADD COORDINATOR 2", + leader + ": ADD COORDINATOR 3", + leader + ": REGISTER INSTANCE instance_0", + leader + ": REGISTER INSTANCE instance_1", + leader + ": SET INSTANCE instance_0 TO MAIN", + })) + Expect(result.RequeueAfter).To(BeNumerically(">", 0), + "registration was issued, so a follow-up reconcile must verify convergence") + + result = reconcileCluster(resourceName) + Expect(fake.executedCommands()).To(HaveLen(6), + "a converged cluster must not receive further commands") + Expect(result.RequeueAfter).To(BeZero()) + }) + + It("should resume a partial bootstrap without duplicate registrations or a second MAIN", func() { + // The state a crash mid-bootstrap leaves behind: two coordinators + // formed, the first data instance registered and promoted. The + // fake rejects duplicate registrations and second promotions, so + // re-issuing anything fails this test loudly. + fake.setInstances([]memgraph.Instance{ + observedCoordinator(1, memgraph.RoleLeader), + observedCoordinator(2, memgraph.RoleFollower), + observedDataInstance(0, memgraph.RoleMain), + }) + + reconcileCluster(resourceName) + markWorkloadsReady(resourceName) + reconcileCluster(resourceName) + + leader := coordinatorAddress(0) + Expect(fake.executedCommands()).To(Equal([]string{ + leader + ": ADD COORDINATOR 3", + leader + ": REGISTER INSTANCE instance_1", + })) + }) + + It("should execute registration on the leader a follower reports", func() { + fake.setInstances([]memgraph.Instance{ + observedCoordinator(1, memgraph.RoleFollower), + observedCoordinator(2, memgraph.RoleLeader), + observedCoordinator(3, memgraph.RoleFollower), + }) + + reconcileCluster(resourceName) + markWorkloadsReady(resourceName) + reconcileCluster(resourceName) + + leader := coordinatorAddress(1) + Expect(fake.executedCommands()).To(Equal([]string{ + leader + ": REGISTER INSTANCE instance_0", + leader + ": REGISTER INSTANCE instance_1", + leader + ": SET INSTANCE instance_0 TO MAIN", + })) + }) + }) }) diff --git a/internal/memgraph/bolt.go b/internal/memgraph/bolt.go new file mode 100644 index 0000000..cbec9ef --- /dev/null +++ b/internal/memgraph/bolt.go @@ -0,0 +1,123 @@ +/* +Copyright 2026. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package memgraph + +import ( + "context" + "fmt" + + "github.com/neo4j/neo4j-go-driver/v5/neo4j" + "github.com/neo4j/neo4j-go-driver/v5/neo4j/db" +) + +// NewBoltConnector returns the production Connector dialing coordinators over +// unauthenticated Bolt (Bolt auth is out of scope for v1alpha1). +func NewBoltConnector() Connector { + return boltConnector{} +} + +type boltConnector struct{} + +func (boltConnector) Connect(ctx context.Context, address string) (Client, error) { + driver, err := neo4j.NewDriverWithContext("bolt://"+address, neo4j.NoAuth()) + if err != nil { + return nil, fmt.Errorf("creating bolt driver for %s: %w", address, err) + } + if err := driver.VerifyConnectivity(ctx); err != nil { + _ = driver.Close(ctx) + return nil, fmt.Errorf("connecting to %s: %w", address, err) + } + return &boltClient{driver: driver}, nil +} + +type boltClient struct { + driver neo4j.DriverWithContext +} + +func (c *boltClient) ShowInstances(ctx context.Context) ([]Instance, error) { + records, err := c.run(ctx, showInstancesQuery) + if err != nil { + return nil, err + } + instances := make([]Instance, 0, len(records)) + for _, record := range records { + instances = append(instances, instanceFromRecord(record)) + } + return instances, nil +} + +func (c *boltClient) AddCoordinator(ctx context.Context, coordinator CoordinatorSpec) error { + _, err := c.run(ctx, addCoordinatorQuery(coordinator)) + return err +} + +func (c *boltClient) RegisterInstance(ctx context.Context, instance DataInstanceSpec) error { + _, err := c.run(ctx, registerInstanceQuery(instance)) + return err +} + +func (c *boltClient) SetInstanceToMain(ctx context.Context, name string) error { + _, err := c.run(ctx, setInstanceToMainQuery(name)) + return err +} + +func (c *boltClient) Close(ctx context.Context) error { + return c.driver.Close(ctx) +} + +// run executes one query in an autocommit session; Memgraph's coordinator +// queries cannot run inside explicit transactions. +func (c *boltClient) run(ctx context.Context, query string) ([]*db.Record, error) { + session := c.driver.NewSession(ctx, neo4j.SessionConfig{}) + defer func() { _ = session.Close(ctx) }() + + result, err := session.Run(ctx, query, nil) + if err != nil { + return nil, fmt.Errorf("running %q: %w", query, err) + } + records, err := result.Collect(ctx) + if err != nil { + return nil, fmt.Errorf("collecting results of %q: %w", query, err) + } + return records, nil +} + +// instanceFromRecord maps one SHOW INSTANCES row to an Instance. Columns are +// looked up by name so the parsing survives added or reordered columns +// (last_succ_resp_ms is deliberately ignored). +func instanceFromRecord(record *db.Record) Instance { + return Instance{ + Name: stringColumn(record, "name"), + BoltServer: stringColumn(record, "bolt_server"), + CoordinatorServer: stringColumn(record, "coordinator_server"), + ManagementServer: stringColumn(record, "management_server"), + Health: stringColumn(record, "health"), + Role: stringColumn(record, "role"), + } +} + +func stringColumn(record *db.Record, key string) string { + value, ok := record.Get(key) + if !ok { + return "" + } + s, ok := value.(string) + if !ok { + return "" + } + return s +} diff --git a/internal/memgraph/client.go b/internal/memgraph/client.go new file mode 100644 index 0000000..f9a3fda --- /dev/null +++ b/internal/memgraph/client.go @@ -0,0 +1,98 @@ +/* +Copyright 2026. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +// Package memgraph provides the narrow client surface the operator uses to +// drive a Memgraph high-availability cluster over Bolt: show instances, add +// coordinator, register instance, set main. All higher layers depend on the +// Client and Connector interfaces, never on the Bolt driver — this package is +// the mock seam for testing and the only place the driver is referenced. +package memgraph + +import ( + "context" + "fmt" + "strings" +) + +// Roles reported in the SHOW INSTANCES role column: coordinators are +// leader/follower, data instances are main/replica. +const ( + RoleLeader = "leader" + RoleFollower = "follower" + RoleMain = "main" + RoleReplica = "replica" +) + +// Instance is one row of SHOW INSTANCES: a coordinator or data instance the +// cluster currently knows about. +type Instance struct { + Name string + BoltServer string + CoordinatorServer string + ManagementServer string + Health string + Role string +} + +// IsLeader reports whether the instance is the current coordinator leader. +func (i Instance) IsLeader() bool { + return strings.EqualFold(i.Role, RoleLeader) +} + +// IsMain reports whether the instance is the current MAIN data instance. +func (i Instance) IsMain() bool { + return strings.EqualFold(i.Role, RoleMain) +} + +// CoordinatorSpec declares one coordinator to add to the cluster. Servers are +// "host:port" addresses the rest of the cluster reaches the coordinator at. +type CoordinatorSpec struct { + // ID is the Raft coordinator ID (1-based; Memgraph treats ID 0 as unset). + ID int32 + BoltServer string + CoordinatorServer string + ManagementServer string +} + +// Name returns the instance name Memgraph derives from the coordinator ID and +// reports in SHOW INSTANCES. +func (c CoordinatorSpec) Name() string { + return fmt.Sprintf("coordinator_%d", c.ID) +} + +// DataInstanceSpec declares one data instance to register with the cluster. +type DataInstanceSpec struct { + Name string + BoltServer string + ManagementServer string + ReplicationServer string +} + +// Client is the narrow surface of a single coordinator's Bolt endpoint. Every +// method issues exactly one HA management query. +type Client interface { + ShowInstances(ctx context.Context) ([]Instance, error) + AddCoordinator(ctx context.Context, coordinator CoordinatorSpec) error + RegisterInstance(ctx context.Context, instance DataInstanceSpec) error + SetInstanceToMain(ctx context.Context, name string) error + Close(ctx context.Context) error +} + +// Connector opens a Client to a coordinator's "host:port" Bolt address. The +// controller depends on this interface so tests can substitute a fake cluster. +type Connector interface { + Connect(ctx context.Context, address string) (Client, error) +} diff --git a/internal/memgraph/queries.go b/internal/memgraph/queries.go new file mode 100644 index 0000000..f19b9a9 --- /dev/null +++ b/internal/memgraph/queries.go @@ -0,0 +1,50 @@ +/* +Copyright 2026. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package memgraph + +import "fmt" + +// The HA management query grammar, mirroring what the memgraph-high-availability +// Helm chart's registration job issues. Config values are rendered inline +// because Memgraph's coordinator queries do not accept Bolt parameters; all +// inputs are operator-derived names and "host:port" addresses, never user text. + +const showInstancesQuery = "SHOW INSTANCES" + +func addCoordinatorQuery(coordinator CoordinatorSpec) string { + return fmt.Sprintf( + `ADD COORDINATOR %d WITH CONFIG {"bolt_server": %q, "coordinator_server": %q, "management_server": %q}`, + coordinator.ID, + coordinator.BoltServer, + coordinator.CoordinatorServer, + coordinator.ManagementServer, + ) +} + +func registerInstanceQuery(instance DataInstanceSpec) string { + return fmt.Sprintf( + `REGISTER INSTANCE %s WITH CONFIG {"bolt_server": %q, "management_server": %q, "replication_server": %q}`, + instance.Name, + instance.BoltServer, + instance.ManagementServer, + instance.ReplicationServer, + ) +} + +func setInstanceToMainQuery(name string) string { + return fmt.Sprintf("SET INSTANCE %s TO MAIN", name) +} diff --git a/internal/memgraph/queries_test.go b/internal/memgraph/queries_test.go new file mode 100644 index 0000000..572d64e --- /dev/null +++ b/internal/memgraph/queries_test.go @@ -0,0 +1,98 @@ +/* +Copyright 2026. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package memgraph + +import ( + "testing" + + "github.com/google/go-cmp/cmp" + "github.com/neo4j/neo4j-go-driver/v5/neo4j/db" +) + +const testInstanceName = "instance_1" + +func TestAddCoordinatorQuery(t *testing.T) { + got := addCoordinatorQuery(CoordinatorSpec{ + ID: 2, + BoltServer: "example-coordinator-1.example-coordinator.default.svc.cluster.local:7687", + CoordinatorServer: "example-coordinator-1.example-coordinator.default.svc.cluster.local:12000", + ManagementServer: "example-coordinator-1.example-coordinator.default.svc.cluster.local:10000", + }) + want := `ADD COORDINATOR 2 WITH CONFIG {` + + `"bolt_server": "example-coordinator-1.example-coordinator.default.svc.cluster.local:7687", ` + + `"coordinator_server": "example-coordinator-1.example-coordinator.default.svc.cluster.local:12000", ` + + `"management_server": "example-coordinator-1.example-coordinator.default.svc.cluster.local:10000"}` + if got != want { + t.Errorf("addCoordinatorQuery() = %q, want %q", got, want) + } +} + +func TestRegisterInstanceQuery(t *testing.T) { + got := registerInstanceQuery(DataInstanceSpec{ + Name: testInstanceName, + BoltServer: "example-data-0.example-data.default.svc.cluster.local:7687", + ManagementServer: "example-data-0.example-data.default.svc.cluster.local:10000", + ReplicationServer: "example-data-0.example-data.default.svc.cluster.local:20000", + }) + want := `REGISTER INSTANCE instance_1 WITH CONFIG {` + + `"bolt_server": "example-data-0.example-data.default.svc.cluster.local:7687", ` + + `"management_server": "example-data-0.example-data.default.svc.cluster.local:10000", ` + + `"replication_server": "example-data-0.example-data.default.svc.cluster.local:20000"}` + if got != want { + t.Errorf("registerInstanceQuery() = %q, want %q", got, want) + } +} + +func TestSetInstanceToMainQuery(t *testing.T) { + got := setInstanceToMainQuery(testInstanceName) + if want := "SET INSTANCE instance_1 TO MAIN"; got != want { + t.Errorf("setInstanceToMainQuery() = %q, want %q", got, want) + } +} + +func TestInstanceFromRecord(t *testing.T) { + record := &db.Record{ + Keys: []string{ + "name", "bolt_server", "coordinator_server", "management_server", "health", "role", "last_succ_resp_ms", + }, + Values: []any{ + "coordinator_1", "localhost:7687", "localhost:12000", "localhost:10000", "up", "leader", int64(12), + }, + } + want := Instance{ + Name: "coordinator_1", + BoltServer: "localhost:7687", + CoordinatorServer: "localhost:12000", + ManagementServer: "localhost:10000", + Health: "up", + Role: "leader", + } + if diff := cmp.Diff(want, instanceFromRecord(record)); diff != "" { + t.Errorf("instanceFromRecord() mismatch (-want +got):\n%s", diff) + } +} + +func TestInstanceFromRecordToleratesMissingColumns(t *testing.T) { + record := &db.Record{ + Keys: []string{"name", "role"}, + Values: []any{testInstanceName, "main"}, + } + want := Instance{Name: testInstanceName, Role: "main"} + if diff := cmp.Diff(want, instanceFromRecord(record)); diff != "" { + t.Errorf("instanceFromRecord() mismatch (-want +got):\n%s", diff) + } +} diff --git a/internal/planner/planner.go b/internal/planner/planner.go new file mode 100644 index 0000000..336efbb --- /dev/null +++ b/internal/planner/planner.go @@ -0,0 +1,120 @@ +/* +Copyright 2026. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +// Package planner computes the ordered registration commands that drive an +// observed Memgraph HA cluster toward its declared topology. The logic is a +// pure diff — declared topology plus observed SHOW INSTANCES output in, +// commands out (empty when converged) — so reconciliation stays idempotent +// and read-before-write: only missing registrations are re-issued, and the +// initial MAIN promotion happens exactly once, when no MAIN exists. After +// bootstrap, failover belongs to the Raft coordinators; the planner never +// overrides an existing MAIN. +package planner + +import ( + "context" + "fmt" + + "github.com/memgraph/kubernetes-operator/internal/memgraph" +) + +// Topology is the declared cluster registration state: every coordinator and +// data instance the CR says must exist, with the addresses each advertises. +type Topology struct { + Coordinators []memgraph.CoordinatorSpec + DataInstances []memgraph.DataInstanceSpec +} + +// Command is one registration step to execute against the coordinator leader. +type Command interface { + Run(ctx context.Context, client memgraph.Client) error + fmt.Stringer +} + +// AddCoordinator adds one declared coordinator to the Raft cluster. +type AddCoordinator struct { + Coordinator memgraph.CoordinatorSpec +} + +// Run implements Command. +func (c AddCoordinator) Run(ctx context.Context, client memgraph.Client) error { + return client.AddCoordinator(ctx, c.Coordinator) +} + +func (c AddCoordinator) String() string { + return fmt.Sprintf("ADD COORDINATOR %d", c.Coordinator.ID) +} + +// RegisterInstance registers one declared data instance with the cluster. +type RegisterInstance struct { + Instance memgraph.DataInstanceSpec +} + +// Run implements Command. +func (c RegisterInstance) Run(ctx context.Context, client memgraph.Client) error { + return client.RegisterInstance(ctx, c.Instance) +} + +func (c RegisterInstance) String() string { + return "REGISTER INSTANCE " + c.Instance.Name +} + +// SetInstanceToMain promotes the named data instance to MAIN at bootstrap. +type SetInstanceToMain struct { + Name string +} + +// Run implements Command. +func (c SetInstanceToMain) Run(ctx context.Context, client memgraph.Client) error { + return client.SetInstanceToMain(ctx, c.Name) +} + +func (c SetInstanceToMain) String() string { + return fmt.Sprintf("SET INSTANCE %s TO MAIN", c.Name) +} + +// Plan diffs the declared topology against the observed instances and returns +// the commands still needed, in execution order: coordinators before data +// instances (registration requires a formed Raft cluster), the initial MAIN +// promotion last. Instances the cluster knows but the topology does not +// declare are left untouched — unregistration is out of scope for v1. +func Plan(declared Topology, observed []memgraph.Instance) []Command { + registered := make(map[string]memgraph.Instance, len(observed)) + hasMain := false + for _, instance := range observed { + registered[instance.Name] = instance + hasMain = hasMain || instance.IsMain() + } + + var commands []Command + for _, coordinator := range declared.Coordinators { + // A coordinator reports itself in SHOW INSTANCES with an empty + // bolt_server until ADD COORDINATOR is issued for its ID, so presence + // alone does not prove registration. + if observed, ok := registered[coordinator.Name()]; !ok || observed.BoltServer == "" { + commands = append(commands, AddCoordinator{Coordinator: coordinator}) + } + } + for _, instance := range declared.DataInstances { + if _, ok := registered[instance.Name]; !ok { + commands = append(commands, RegisterInstance{Instance: instance}) + } + } + if !hasMain && len(declared.DataInstances) > 0 { + commands = append(commands, SetInstanceToMain{Name: declared.DataInstances[0].Name}) + } + return commands +} diff --git a/internal/planner/planner_test.go b/internal/planner/planner_test.go new file mode 100644 index 0000000..d3e8bfe --- /dev/null +++ b/internal/planner/planner_test.go @@ -0,0 +1,209 @@ +/* +Copyright 2026. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package planner_test + +import ( + "fmt" + "testing" + + "github.com/google/go-cmp/cmp" + + "github.com/memgraph/kubernetes-operator/internal/memgraph" + "github.com/memgraph/kubernetes-operator/internal/planner" +) + +// declaredTopology is the canonical 3-coordinator, 2-data-instance fixture +// the cases below diff observed cluster states against. +func declaredTopology() planner.Topology { + topology := planner.Topology{} + for id := int32(1); id <= 3; id++ { + topology.Coordinators = append(topology.Coordinators, coordinatorSpec(id)) + } + for i := range 2 { + topology.DataInstances = append(topology.DataInstances, dataInstanceSpec(i)) + } + return topology +} + +// coordinatorSpec builds the declared coordinator with the given 1-based Raft +// ID (Memgraph treats ID 0 as unset), hosted on the pod with ordinal ID-1. +func coordinatorSpec(id int32) memgraph.CoordinatorSpec { + host := fmt.Sprintf("example-coordinator-%d.example-coordinator.default.svc.cluster.local", id-1) + return memgraph.CoordinatorSpec{ + ID: id, + BoltServer: host + ":7687", + CoordinatorServer: host + ":12000", + ManagementServer: host + ":10000", + } +} + +func dataInstanceSpec(i int) memgraph.DataInstanceSpec { + host := fmt.Sprintf("example-data-%d.example-data.default.svc.cluster.local", i) + return memgraph.DataInstanceSpec{ + Name: fmt.Sprintf("instance_%d", i), + BoltServer: host + ":7687", + ManagementServer: host + ":10000", + ReplicationServer: host + ":20000", + } +} + +func observedCoordinator(id int32, role string) memgraph.Instance { + spec := coordinatorSpec(id) + return memgraph.Instance{ + Name: spec.Name(), + BoltServer: spec.BoltServer, + CoordinatorServer: spec.CoordinatorServer, + ManagementServer: spec.ManagementServer, + Health: "up", + Role: role, + } +} + +func observedDataInstance(i int, role string) memgraph.Instance { + spec := dataInstanceSpec(i) + return memgraph.Instance{ + Name: spec.Name, + BoltServer: spec.BoltServer, + ManagementServer: spec.ManagementServer, + Health: "up", + Role: role, + } +} + +func TestPlan(t *testing.T) { + declared := declaredTopology() + + cases := []struct { + name string + observed []memgraph.Instance + want []planner.Command + }{ + { + name: "fresh cluster bootstraps everything and promotes one MAIN", + observed: nil, + want: []planner.Command{ + planner.AddCoordinator{Coordinator: coordinatorSpec(1)}, + planner.AddCoordinator{Coordinator: coordinatorSpec(2)}, + planner.AddCoordinator{Coordinator: coordinatorSpec(3)}, + planner.RegisterInstance{Instance: dataInstanceSpec(0)}, + planner.RegisterInstance{Instance: dataInstanceSpec(1)}, + planner.SetInstanceToMain{Name: "instance_0"}, + }, + }, + { + name: "partially registered cluster gets only the missing registrations", + observed: []memgraph.Instance{ + observedCoordinator(1, memgraph.RoleLeader), + observedCoordinator(3, memgraph.RoleFollower), + observedDataInstance(0, memgraph.RoleMain), + }, + want: []planner.Command{ + planner.AddCoordinator{Coordinator: coordinatorSpec(2)}, + planner.RegisterInstance{Instance: dataInstanceSpec(1)}, + }, + }, + { + name: "self-reporting coordinator with empty bolt server is still added", + observed: []memgraph.Instance{ + // The coordinator the client is connected to lists itself in + // SHOW INSTANCES with an empty bolt_server until explicitly + // added. + func() memgraph.Instance { + instance := observedCoordinator(2, memgraph.RoleLeader) + instance.BoltServer = "" + return instance + }(), + observedCoordinator(1, memgraph.RoleFollower), + observedCoordinator(3, memgraph.RoleFollower), + observedDataInstance(0, memgraph.RoleMain), + observedDataInstance(1, memgraph.RoleReplica), + }, + want: []planner.Command{ + planner.AddCoordinator{Coordinator: coordinatorSpec(2)}, + }, + }, + { + name: "fully converged cluster is a no-op", + observed: []memgraph.Instance{ + observedCoordinator(1, memgraph.RoleLeader), + observedCoordinator(2, memgraph.RoleFollower), + observedCoordinator(3, memgraph.RoleFollower), + observedDataInstance(0, memgraph.RoleMain), + observedDataInstance(1, memgraph.RoleReplica), + }, + want: nil, + }, + { + name: "an existing MAIN is never overridden, even on another instance", + observed: []memgraph.Instance{ + observedCoordinator(1, memgraph.RoleLeader), + observedCoordinator(2, memgraph.RoleFollower), + observedCoordinator(3, memgraph.RoleFollower), + observedDataInstance(0, memgraph.RoleReplica), + observedDataInstance(1, memgraph.RoleMain), + }, + want: nil, + }, + { + name: "registered but leaderless data plane still gets the one MAIN promotion", + observed: []memgraph.Instance{ + observedCoordinator(1, memgraph.RoleLeader), + observedCoordinator(2, memgraph.RoleFollower), + observedCoordinator(3, memgraph.RoleFollower), + observedDataInstance(0, memgraph.RoleReplica), + observedDataInstance(1, memgraph.RoleReplica), + }, + want: []planner.Command{ + planner.SetInstanceToMain{Name: "instance_0"}, + }, + }, + { + name: "missing instance registers without MAIN promotion when a MAIN exists", + observed: []memgraph.Instance{ + observedCoordinator(1, memgraph.RoleLeader), + observedCoordinator(2, memgraph.RoleFollower), + observedCoordinator(3, memgraph.RoleFollower), + observedDataInstance(1, memgraph.RoleMain), + }, + want: []planner.Command{ + planner.RegisterInstance{Instance: dataInstanceSpec(0)}, + }, + }, + { + name: "instances the topology does not declare are left untouched", + observed: []memgraph.Instance{ + observedCoordinator(1, memgraph.RoleLeader), + observedCoordinator(2, memgraph.RoleFollower), + observedCoordinator(3, memgraph.RoleFollower), + observedCoordinator(4, memgraph.RoleFollower), + observedDataInstance(0, memgraph.RoleMain), + observedDataInstance(1, memgraph.RoleReplica), + observedDataInstance(2, memgraph.RoleReplica), + }, + want: nil, + }, + } + + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + got := planner.Plan(declared, tc.observed) + if diff := cmp.Diff(tc.want, got); diff != "" { + t.Errorf("Plan() mismatch (-want +got):\n%s", diff) + } + }) + } +} diff --git a/internal/resources/statefulset.go b/internal/resources/statefulset.go index 8e713fb..c784c2a 100644 --- a/internal/resources/statefulset.go +++ b/internal/resources/statefulset.go @@ -91,10 +91,11 @@ func DataStatefulSet(cluster *memgraphcomv1alpha1.MemgraphCluster) *appsv1.State // coordinatorStartScript derives the coordinator's identity from its pod // ordinal (the numeric suffix of the pod name): ordinal N becomes coordinator -// ID N+1 (Raft IDs start at 1) advertised at the pod's stable DNS name within -// the headless Service. +// ID N+1 (Memgraph treats coordinator ID 0 as unset and refuses to start, so +// IDs stay 1-based) advertised at the pod's stable DNS name within the +// headless Service. func coordinatorStartScript(cluster *memgraphcomv1alpha1.MemgraphCluster) string { - fqdnSuffix := fmt.Sprintf("%s.%s.svc.%s", CoordinatorName(cluster), cluster.Namespace, clusterDomain) + fqdnSuffix := podFQDNSuffix(cluster, CoordinatorName(cluster)) return fmt.Sprintf(`ordinal="${POD_NAME##*-}" exec %s \ --coordinator-id="$((ordinal + 1))" \ diff --git a/internal/resources/topology.go b/internal/resources/topology.go new file mode 100644 index 0000000..9c5370b --- /dev/null +++ b/internal/resources/topology.go @@ -0,0 +1,75 @@ +/* +Copyright 2026. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package resources + +import ( + "fmt" + + memgraphcomv1alpha1 "github.com/memgraph/kubernetes-operator/api/v1alpha1" + "github.com/memgraph/kubernetes-operator/internal/memgraph" + "github.com/memgraph/kubernetes-operator/internal/planner" +) + +// DeclaredTopology derives the registration topology the planner drives the +// cluster toward. Identity follows the pod ordinal exactly as the workload +// pods advertise it: coordinator ordinal N is Raft coordinator N+1 (Memgraph +// treats coordinator ID 0 as unset, so IDs stay 1-based), data ordinal N +// registers as instance_N, and every address is the pod's stable DNS name +// within its headless Service. +func DeclaredTopology(cluster *memgraphcomv1alpha1.MemgraphCluster) planner.Topology { + spec := normalize(cluster.Spec) + + topology := planner.Topology{ + Coordinators: make([]memgraph.CoordinatorSpec, 0, spec.coordinators), + DataInstances: make([]memgraph.DataInstanceSpec, 0, spec.dataInstances), + } + for ordinal := range spec.coordinators { + fqdn := podFQDN(cluster, CoordinatorName(cluster), ordinal) + topology.Coordinators = append(topology.Coordinators, memgraph.CoordinatorSpec{ + ID: ordinal + 1, + BoltServer: hostPort(fqdn, BoltPort), + CoordinatorServer: hostPort(fqdn, CoordinatorPort), + ManagementServer: hostPort(fqdn, ManagementPort), + }) + } + for ordinal := range spec.dataInstances { + fqdn := podFQDN(cluster, DataName(cluster), ordinal) + topology.DataInstances = append(topology.DataInstances, memgraph.DataInstanceSpec{ + Name: fmt.Sprintf("instance_%d", ordinal), + BoltServer: hostPort(fqdn, BoltPort), + ManagementServer: hostPort(fqdn, ManagementPort), + ReplicationServer: hostPort(fqdn, ReplicationPort), + }) + } + return topology +} + +// podFQDNSuffix returns the DNS suffix a pod name is appended to for pods of +// the given headless Service: "..svc.". +func podFQDNSuffix(cluster *memgraphcomv1alpha1.MemgraphCluster, serviceName string) string { + return fmt.Sprintf("%s.%s.svc.%s", serviceName, cluster.Namespace, clusterDomain) +} + +// podFQDN returns the stable DNS name of the pod with the given ordinal in +// the StatefulSet backed by the given headless Service (both share one name). +func podFQDN(cluster *memgraphcomv1alpha1.MemgraphCluster, serviceName string, ordinal int32) string { + return fmt.Sprintf("%s-%d.%s", serviceName, ordinal, podFQDNSuffix(cluster, serviceName)) +} + +func hostPort(host string, port int32) string { + return fmt.Sprintf("%s:%d", host, port) +} diff --git a/internal/resources/topology_test.go b/internal/resources/topology_test.go new file mode 100644 index 0000000..0ee9d6a --- /dev/null +++ b/internal/resources/topology_test.go @@ -0,0 +1,115 @@ +/* +Copyright 2026. + +Licensed under the Apache License, Version 2.0 (the "License"); +you may not use this file except in compliance with the License. +You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + +Unless required by applicable law or agreed to in writing, software +distributed under the License is distributed on an "AS IS" BASIS, +WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +See the License for the specific language governing permissions and +limitations under the License. +*/ + +package resources_test + +import ( + "fmt" + "strings" + "testing" + + "github.com/google/go-cmp/cmp" + + "github.com/memgraph/kubernetes-operator/internal/memgraph" + "github.com/memgraph/kubernetes-operator/internal/planner" + "github.com/memgraph/kubernetes-operator/internal/resources" +) + +func TestDeclaredTopologyDefaults(t *testing.T) { + got := resources.DeclaredTopology(minimalCluster()) + + coordinatorFQDN := func(ordinal int) string { + return fmt.Sprintf("%s-%d.%s.%s.svc.cluster.local", coordinatorName, ordinal, coordinatorName, testNamespace) + } + dataFQDN := func(ordinal int) string { + return fmt.Sprintf("%s-%d.%s.%s.svc.cluster.local", dataName, ordinal, dataName, testNamespace) + } + + want := planner.Topology{ + Coordinators: []memgraph.CoordinatorSpec{ + { + ID: 1, + BoltServer: coordinatorFQDN(0) + ":7687", + CoordinatorServer: coordinatorFQDN(0) + ":12000", + ManagementServer: coordinatorFQDN(0) + ":10000", + }, + { + ID: 2, + BoltServer: coordinatorFQDN(1) + ":7687", + CoordinatorServer: coordinatorFQDN(1) + ":12000", + ManagementServer: coordinatorFQDN(1) + ":10000", + }, + { + ID: 3, + BoltServer: coordinatorFQDN(2) + ":7687", + CoordinatorServer: coordinatorFQDN(2) + ":12000", + ManagementServer: coordinatorFQDN(2) + ":10000", + }, + }, + DataInstances: []memgraph.DataInstanceSpec{ + { + Name: "instance_0", + BoltServer: dataFQDN(0) + ":7687", + ManagementServer: dataFQDN(0) + ":10000", + ReplicationServer: dataFQDN(0) + ":20000", + }, + { + Name: "instance_1", + BoltServer: dataFQDN(1) + ":7687", + ManagementServer: dataFQDN(1) + ":10000", + ReplicationServer: dataFQDN(1) + ":20000", + }, + }, + } + + if diff := cmp.Diff(want, got); diff != "" { + t.Errorf("DeclaredTopology() mismatch (-want +got):\n%s", diff) + } +} + +func TestDeclaredTopologyFollowsReplicaCounts(t *testing.T) { + got := resources.DeclaredTopology(specifiedCluster()) + + if len(got.Coordinators) != 5 { + t.Errorf("DeclaredTopology() declared %d coordinators, want 5", len(got.Coordinators)) + } + if len(got.DataInstances) != 3 { + t.Errorf("DeclaredTopology() declared %d data instances, want 3", len(got.DataInstances)) + } +} + +// The registration topology must advertise exactly the identity the +// coordinator pods derive for themselves at startup, otherwise the Raft +// cluster and the registrations disagree about who is who. +func TestDeclaredTopologyMatchesCoordinatorStartScript(t *testing.T) { + cluster := minimalCluster() + topology := resources.DeclaredTopology(cluster) + sts := resources.CoordinatorStatefulSet(cluster) + script := strings.Join(sts.Spec.Template.Spec.Containers[0].Command, "\n") + + // The script derives '.' from POD_NAME; every declared + // coordinator_server must be a pod FQDN under that same suffix. + suffix := fmt.Sprintf("%s.%s.svc.cluster.local", coordinatorName, testNamespace) + if !strings.Contains(script, `--coordinator-hostname="${POD_NAME}.`+suffix+`"`) { + t.Errorf("coordinator start script does not advertise the headless-service pod FQDN:\n%s", script) + } + for i, coordinator := range topology.Coordinators { + wantHost := fmt.Sprintf("%s-%d.%s:12000", coordinatorName, i, suffix) + if coordinator.CoordinatorServer != wantHost { + t.Errorf("coordinator %d advertises %q, want %q", coordinator.ID, coordinator.CoordinatorServer, wantHost) + } + } +}