From ce9a4826b71482bf251d13790db4fdbf82d927f0 Mon Sep 17 00:00:00 2001 From: Assaf Giladi Date: Sun, 6 Sep 2026 14:59:15 +0300 Subject: [PATCH] fix: support new wekactl interface group response --- .../csi/daemonset_placement_test.go | 2 +- internal/pkg/domain/feature_flags.go | 4 + internal/services/weka.go | 108 +++++++++++---- .../services/weka_nfs_interface_group_test.go | 129 ++++++++++++++++++ pkg/util/hashes.go | 2 +- 5 files changed, 213 insertions(+), 32 deletions(-) create mode 100644 internal/services/weka_nfs_interface_group_test.go diff --git a/internal/controllers/operations/csi/daemonset_placement_test.go b/internal/controllers/operations/csi/daemonset_placement_test.go index 5147b78af..0629a8fbb 100644 --- a/internal/controllers/operations/csi/daemonset_placement_test.go +++ b/internal/controllers/operations/csi/daemonset_placement_test.go @@ -127,7 +127,7 @@ func TestGetCsiNodeDaemonSetHash_ChangesWithSelector(t *testing.T) { } } -// Pod NodeSelector and nodeAffinity are ANDed. Leaving NodeSelector populated would make the retain +// Pod NodeSelector and nodeAffinity are combined with AND. Leaving NodeSelector populated would make the retain // term dead code and reproduce the deadlock exactly, so assert placement lives only in affinity. func TestNewCsiNodeDaemonSet_PlacementOnlyInAffinity(t *testing.T) { config.Config.Csi.WekafsImage = "test-csi-image" diff --git a/internal/pkg/domain/feature_flags.go b/internal/pkg/domain/feature_flags.go index f0136f535..64941c738 100644 --- a/internal/pkg/domain/feature_flags.go +++ b/internal/pkg/domain/feature_flags.go @@ -16,4 +16,8 @@ type FeatureFlags struct { SsdProxyIncludesDpdkMemory bool `json:"ssd_proxy_includes_dpdk_memory"` // 9 // flags 10, 11 are not used by the operator WekaManagesNonIonodeAffinity bool `json:"weka_manages_non_ionode_affinity"` // 12 + // flag 13 is not used by the operator + // WekactlAsDefault means the go CLI (wekactl) backs the `weka` command. It changes some JSON + // schemas, notably `weka nfs interface-group` port identity — see NfsInterfaceGroupPort. + WekactlAsDefault bool `json:"wekactl_as_default"` // 14 } diff --git a/internal/services/weka.go b/internal/services/weka.go index 215f87b86..e00b35ee9 100644 --- a/internal/services/weka.go +++ b/internal/services/weka.go @@ -244,11 +244,35 @@ type SmbwCluster struct { Active bool `json:"active"` } +// NfsInterfaceGroupPort mirrors one entry of the "ports" array of `weka nfs interface-group --json`. +// The two CLI generations identify the owning container differently, and both are accepted: +// +// wekactl (domain.FeatureFlags.WekactlAsDefault): "container": 13, "container_uid": "..." +// legacy python CLI: "host_id": "HostId<13>", "host_uid": "..." +// +// Container is a pointer so that a genuine container 13 is distinguishable from an absent field, +// which matters because container 0 is a valid id. type NfsInterfaceGroupPort struct { - HostId string `json:"host_id"` - HostUid string `json:"host_uid"` - Port string `json:"port"` - Status string `json:"status"` + Container *int `json:"container"` + ContainerUid string `json:"container_uid"` + HostId string `json:"host_id"` + HostUid string `json:"host_uid"` + Port string `json:"port"` + Status string `json:"status"` +} + +// OwnerContainerId resolves the weka container owning this port under either CLI schema. +// It errors rather than defaulting, so an unrecognized schema cannot be silently read as +// "this port belongs to nobody" — that reading previously let port removal no-op and wedge +// container deletion behind a failing deactivate. +func (p *NfsInterfaceGroupPort) OwnerContainerId() (int, error) { + if p.Container != nil { + return *p.Container, nil + } + if p.HostId != "" { + return resources.HostIdToContainerId(p.HostId) + } + return 0, errors.Errorf("interface group port %q has neither a 'container' nor a 'host_id' field", p.Port) } type NfsInterfaceGroup struct { @@ -258,11 +282,28 @@ type NfsInterfaceGroup struct { Name string `json:"name"` Ports []NfsInterfaceGroupPort `json:"ports"` Status string `json:"status"` - SubnetMask string `json:"subnet_mask"` Type string `json:"type"` Uid string `json:"uid"` } +// ContainerPorts returns the port device names the given weka container currently holds in the group. +// Any port whose owner cannot be resolved fails the call: an unattributable port may well be the +// caller's own, and treating it as somebody else's is what turns a missed removal into a wedge. +func (g *NfsInterfaceGroup) ContainerPorts(containerId int) ([]string, error) { + var ports []string + for i := range g.Ports { + port := &g.Ports[i] + owner, err := port.OwnerContainerId() + if err != nil { + return nil, errors.Wrapf(err, "cannot determine port ownership in interface group %s", g.Name) + } + if owner == containerId { + ports = append(ports, port.Port) + } + } + return ports, nil +} + type WekaUserResponse struct { // OrgId int `json:"org_id"` // PosixGid string `json:"posix_gid"` @@ -1050,47 +1091,54 @@ func (c *CliWekaService) RemoveFromSmbwCluster(ctx context.Context, containerId return nil } -// EnsureNfsInterfaceGroupPorts ensures the NFS interface group has the specified ports for a container. -// It fetches current state and reconciles to desired state by adding missing ports and removing extra ones. -func (c *CliWekaService) EnsureNfsInterfaceGroupPorts(ctx context.Context, interfaceGroupName string, containerId int, targetInterfaces []string) error { - ctx, logger := instrumentation.CreateLogSpan(ctx, "EnsureNfsInterfaceGroupPorts") - defer logger.End() - - executor, err := c.getExecutor(ctx) - if err != nil { - return err - } - - containerIdStr := strconv.Itoa(containerId) - hostIdStr := fmt.Sprintf("HostId<%d>", containerId) +func (c *CliWekaService) getNfsInterfaceGroup(ctx context.Context, executor podexec.Exec, interfaceGroupName string) (*NfsInterfaceGroup, error) { + logger := instrumentation.CurrentSpanLogger(ctx) - // Fetch current interface group configuration cmd := []string{ "weka", "nfs", "interface-group", "--name", interfaceGroupName, "--json", } stdout, stderr, err := executor.ExecNamed(ctx, "GetNfsInterfaceGroup", cmd) if err != nil { logger.SetError(err, "Failed to get NFS interface group", "interfaceGroup", interfaceGroupName, "stderr", stderr.String()) - return err + return nil, err } - // Parse the response var interfaceGroups []NfsInterfaceGroup if parseErr := json.Unmarshal(stdout.Bytes(), &interfaceGroups); parseErr != nil { logger.SetError(parseErr, "Failed to parse NFS interface group JSON", "stdout", stdout.String()) - return parseErr + return nil, parseErr } if len(interfaceGroups) == 0 { - return errors.Errorf("NFS interface group %s not found", interfaceGroupName) + return nil, errors.Errorf("NFS interface group %s not found", interfaceGroupName) + } + + return &interfaceGroups[0], nil +} + +// EnsureNfsInterfaceGroupPorts ensures the NFS interface group has the specified ports for a container. +// It fetches current state and reconciles to desired state by adding missing ports and removing extra ones. +func (c *CliWekaService) EnsureNfsInterfaceGroupPorts(ctx context.Context, interfaceGroupName string, containerId int, targetInterfaces []string) error { + ctx, logger := instrumentation.CreateLogSpan(ctx, "EnsureNfsInterfaceGroupPorts") + defer logger.End() + + executor, err := c.getExecutor(ctx) + if err != nil { + return err + } + + containerIdStr := strconv.Itoa(containerId) + + group, err := c.getNfsInterfaceGroup(ctx, executor, interfaceGroupName) + if err != nil { + return err } // Get current ports for this container - var currentInterfaces []string - for _, port := range interfaceGroups[0].Ports { - if port.HostId == hostIdStr { - currentInterfaces = append(currentInterfaces, port.Port) - } + currentInterfaces, err := group.ContainerPorts(containerId) + if err != nil { + logger.SetError(err, "Failed to read current NFS interface group ports", "containerId", containerId) + return err } // Calculate what needs to be added and removed @@ -1106,7 +1154,7 @@ func (c *CliWekaService) EnsureNfsInterfaceGroupPorts(ctx context.Context, inter // Add new interfaces first (before removing old ones to avoid service interruption) for _, interfaceName := range toAdd { - cmd = []string{ + cmd := []string{ "wekaauthcli", "nfs", "interface-group", "port", "add", interfaceGroupName, containerIdStr, interfaceName, } _, stderr, err := executor.ExecNamed(ctx, "AddNfsInterfaceGroupPort", cmd) @@ -1147,7 +1195,7 @@ func (c *CliWekaService) EnsureNfsInterfaceGroupPorts(ctx context.Context, inter // Remove interfaces that shouldn't be there for _, interfaceName := range toRemove { - cmd = []string{ + cmd := []string{ "wekaauthcli", "nfs", "interface-group", "port", "delete", "-f", interfaceGroupName, containerIdStr, interfaceName, } _, stderr, err := executor.ExecNamed(ctx, "RemoveNfsInterfaceGroupPort", cmd) diff --git a/internal/services/weka_nfs_interface_group_test.go b/internal/services/weka_nfs_interface_group_test.go new file mode 100644 index 000000000..ee2585247 --- /dev/null +++ b/internal/services/weka_nfs_interface_group_test.go @@ -0,0 +1,129 @@ +package services + +import ( + "encoding/json" + "testing" +) + +// wekactlInterfaceGroupJSON was captured verbatim from `weka nfs interface-group --name +// MgmtInterfaceGroup --json` on a live 6.0.0.6363 cluster (feature flag wekactl_as_default=true). +// Container 13 is mid-deletion, hence UNREACHABLE. +const wekactlInterfaceGroupJSON = `[ + { + "allow_manage_gids": true, + "gateway": "255.255.255.255", + "ips": [], + "name": "MgmtInterfaceGroup", + "netmask": 32, + "ports": [ + {"container_uid": "b4b129f6-88fa-7050-d2bd-6d7f7d5c7e9f", "container": 12, "port": "enp99s0f0np0", "status": "OK"}, + {"container_uid": "8026f2bb-10f3-9e82-2882-97d0f26db15a", "container": 15, "port": "enp99s0f0np0", "status": "OK"}, + {"container_uid": "fee78013-db99-4a9b-844e-5d3c5c2f67bc", "container": 13, "port": "enp99s0f0np0", "status": "UNREACHABLE"}, + {"container_uid": "eef2d3e6-195a-81fb-db0d-2a96872dcbfe", "container": 14, "port": "enp99s0f0np0", "status": "OK"} + ], + "status": "OK", + "tenant_ids": [], + "type": "NFS", + "uid": "dbab1da8-db5c-bc93-0f63-1460c1f08450" + } +]` + +// legacyInterfaceGroupJSON is the pre-wekactl python CLI shape, which identifies the owner with a +// "HostId" string instead of a numeric container. +const legacyInterfaceGroupJSON = `[ + { + "name": "MgmtInterfaceGroup", + "ports": [ + {"host_uid": "b4b129f6-88fa-7050-d2bd-6d7f7d5c7e9f", "host_id": "HostId<12>", "port": "enp99s0f0np0", "status": "OK"}, + {"host_uid": "fee78013-db99-4a9b-844e-5d3c5c2f67bc", "host_id": "HostId<13>", "port": "enp99s0f0np0", "status": "UNREACHABLE"} + ], + "status": "OK", + "type": "NFS" + } +]` + +func parseGroup(t *testing.T, payload string) NfsInterfaceGroup { + t.Helper() + var groups []NfsInterfaceGroup + if err := json.Unmarshal([]byte(payload), &groups); err != nil { + t.Fatalf("unmarshal: %v", err) + } + if len(groups) != 1 { + t.Fatalf("expected 1 interface group, got %d", len(groups)) + } + return groups[0] +} + +// TestContainerPortsAcrossCliGenerations pins the port-identity field of both CLI generations. +// A mismatch here does not fail loudly in production: it makes ContainerPorts return nothing, so +// RemoveFromNfs removes nothing yet still reports success, and container deletion then wedges +// forever on "HostId is still part of an interface group" from the deactivate step. +func TestContainerPortsAcrossCliGenerations(t *testing.T) { + for _, tc := range []struct { + name string + payload string + }{ + {"wekactl", wekactlInterfaceGroupJSON}, + {"legacy python cli", legacyInterfaceGroupJSON}, + } { + t.Run(tc.name, func(t *testing.T) { + group := parseGroup(t, tc.payload) + + ports, err := group.ContainerPorts(13) + if err != nil { + t.Fatalf("ContainerPorts(13): %v", err) + } + if len(ports) != 1 || ports[0] != "enp99s0f0np0" { + t.Fatalf("container 13 must own enp99s0f0np0, got %v", ports) + } + + // The port must be attributed to exactly one container, or removing it from 13 would + // also be attempted for its neighbours. + other, err := group.ContainerPorts(12) + if err != nil { + t.Fatalf("ContainerPorts(12): %v", err) + } + if len(other) != 1 { + t.Fatalf("container 12 must own exactly one port, got %v", other) + } + + absent, err := group.ContainerPorts(99) + if err != nil { + t.Fatalf("ContainerPorts(99): %v", err) + } + if len(absent) != 0 { + t.Fatalf("container 99 owns no ports, got %v", absent) + } + }) + } +} + +// TestContainerPortsRejectsUnknownSchema locks in loud failure on a third schema. Returning an empty +// slice instead would reproduce the original wedge: nothing to remove, success reported, deletion +// stuck behind a deactivate that keeps failing. +func TestContainerPortsRejectsUnknownSchema(t *testing.T) { + group := parseGroup(t, `[{"name": "MgmtInterfaceGroup", "ports": [{"owner": 13, "port": "enp99s0f0np0"}]}]`) + + if _, err := group.ContainerPorts(13); err == nil { + t.Fatal("a port with no recognizable owner field must fail, not read as unowned") + } +} + +// TestContainerZeroIsDistinguishableFromAbsent guards the pointer on NfsInterfaceGroupPort.Container. +// With a plain int, container 0 and "field missing" are the same value, so a malformed payload +// would silently claim to own container 0's ports. +func TestContainerZeroIsDistinguishableFromAbsent(t *testing.T) { + group := parseGroup(t, `[{"name": "g", "ports": [{"container": 0, "port": "eth0"}]}]`) + ports, err := group.ContainerPorts(0) + if err != nil { + t.Fatalf("ContainerPorts(0): %v", err) + } + if len(ports) != 1 { + t.Fatalf("container 0 must own eth0, got %v", ports) + } + + missing := parseGroup(t, `[{"name": "g", "ports": [{"port": "eth0"}]}]`) + if _, err := missing.ContainerPorts(0); err == nil { + t.Fatal("an absent container field must not be read as container 0") + } +} diff --git a/pkg/util/hashes.go b/pkg/util/hashes.go index c88b77b11..79fe98648 100644 --- a/pkg/util/hashes.go +++ b/pkg/util/hashes.go @@ -78,7 +78,7 @@ func checkForMaps(v reflect.Value) error { } } } - case reflect.Ptr: + case reflect.Pointer: if !v.IsNil() { if err := checkForMaps(v.Elem()); err != nil { return err