diff --git a/.github/workflows/helm_ci.yml b/.github/workflows/helm_ci.yml index b8fad9069..0bc8318e3 100644 --- a/.github/workflows/helm_ci.yml +++ b/.github/workflows/helm_ci.yml @@ -217,6 +217,106 @@ jobs: print("✓ No blank lines in security+S3 command blocks") PYEOF + echo "" + echo "=== Testing security+S3: -cert.file/-key.file gated on httpsPort (issue #9202) ===" + # Regression test: when enableSecurity=true but *.httpsPort is 0 (the default), + # the chart must NOT emit -cert.file / -key.file to the S3 frontend. Passing + # them promotes weed s3's main -port to HTTPS (see weed/command/s3.go), which + # makes the HTTP readinessProbe spam "TLS handshake error ... client sent an + # HTTP request to an HTTPS server" into the pod log. + # + # When *.httpsPort > 0, both -port.https and cert/key args MUST be emitted + # together so the opt-in HTTPS listener actually has credentials. + python3 - "$CHART_DIR" <<'PYEOF' + import subprocess, sys, yaml + chart = sys.argv[1] + + def render(values): + args = ["helm", "template", "test", chart] + for k, v in values.items(): + args += ["--set", f"{k}={v}"] + return subprocess.check_output(args, text=True) + + def script_of(manifest, kind_name): + for doc in yaml.safe_load_all(manifest): + if not doc or doc.get("kind") not in ("Deployment", "StatefulSet"): + continue + if doc["metadata"]["name"] != kind_name: + continue + for c in doc["spec"]["template"]["spec"]["containers"]: + cmd = c.get("command", []) + if len(cmd) >= 3 and cmd[0] == "/bin/sh" and cmd[1] == "-ec": + return cmd[2] + raise AssertionError(f"no container script for {kind_name}") + + cases = [ + # (values, workload-name, httpsPort-set?, arg-prefix) + ({"global.seaweedfs.enableSecurity": "true", + "s3.enabled": "true"}, + "test-seaweedfs-s3", False, ""), + ({"global.seaweedfs.enableSecurity": "true", + "s3.enabled": "true", + "s3.httpsPort": "8443"}, + "test-seaweedfs-s3", True, ""), + ({"global.seaweedfs.enableSecurity": "true", + "filer.s3.enabled": "true"}, + "test-seaweedfs-filer", False, "s3."), + ({"global.seaweedfs.enableSecurity": "true", + "filer.s3.enabled": "true", + "filer.s3.httpsPort": "8444"}, + "test-seaweedfs-filer", True, "s3."), + ({"global.seaweedfs.enableSecurity": "true", + "allInOne.enabled": "true", + "allInOne.s3.enabled": "true"}, + "test-seaweedfs-all-in-one", False, "s3."), + ({"global.seaweedfs.enableSecurity": "true", + "allInOne.enabled": "true", + "allInOne.s3.enabled": "true", + "allInOne.s3.httpsPort": "8445"}, + "test-seaweedfs-all-in-one", True, "s3."), + ] + + failed = False + for values, name, https_on, prefix in cases: + script = script_of(render(values), name) + cert_flag = f"-{prefix}cert.file=" + key_flag = f"-{prefix}key.file=" + https_flag = f"-{prefix}port.https=" + has_cert = cert_flag in script + has_key = key_flag in script + has_https = https_flag in script + label = f"{name} (httpsPort {'set' if https_on else 'unset'})" + if https_on: + if not (has_cert and has_key and has_https): + print(f"FAIL: {label}: expected {cert_flag}, {key_flag}, {https_flag} all present " + f"(got cert={has_cert} key={has_key} https={has_https})", file=sys.stderr) + failed = True + else: + print(f"✓ {label}: cert/key/https args emitted together") + else: + if has_cert or has_key or has_https: + print(f"FAIL: {label}: expected none of {cert_flag}/{key_flag}/{https_flag}; " + f"main S3 -port would silently become HTTPS and break HTTP probes " + f"(got cert={has_cert} key={has_key} https={has_https})", file=sys.stderr) + failed = True + else: + print(f"✓ {label}: no TLS args emitted, main -port stays HTTP") + + # bash -n: pin down that the rendered script parses. Guards against + # a future helper change that leaves a dangling `\` with nothing + # after it (every current caller already exits cleanly because + # bash treats trailing `\` as line-continuation to + # an empty line — but keep the contract explicit). + parse = subprocess.run(["bash", "-n"], input=script, text=True, + capture_output=True) + if parse.returncode != 0: + print(f"FAIL: {label}: bash -n rejected rendered script: {parse.stderr.strip()}", + file=sys.stderr) + failed = True + + sys.exit(1 if failed else 0) + PYEOF + echo "✅ All template rendering tests passed!" - name: Create kind cluster diff --git a/go.mod b/go.mod index 0f8fdddde..025633139 100644 --- a/go.mod +++ b/go.mod @@ -303,7 +303,7 @@ require ( github.com/Azure/azure-sdk-for-go/sdk/internal v1.11.2 // indirect github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.6.4 github.com/Azure/azure-sdk-for-go/sdk/storage/azfile v1.5.3 // indirect - github.com/Azure/go-ntlmssp v0.1.0 // indirect + github.com/Azure/go-ntlmssp v0.1.1 // indirect github.com/AzureAD/microsoft-authentication-library-for-go v1.6.0 // indirect github.com/Files-com/files-sdk-go/v3 v3.2.264 // indirect github.com/GoogleCloudPlatform/opentelemetry-operations-go/detectors/gcp v1.30.0 // indirect diff --git a/go.sum b/go.sum index 1c30d438c..e26d805fc 100644 --- a/go.sum +++ b/go.sum @@ -577,8 +577,8 @@ github.com/Azure/go-autorest v14.2.0+incompatible h1:V5VMDjClD3GiElqLWO7mz2MxNAK github.com/Azure/go-autorest v14.2.0+incompatible/go.mod h1:r+4oMnoxhatjLLJ6zxSWATqVooLgysK6ZNox3g/xq24= github.com/Azure/go-autorest/autorest/to v0.4.1 h1:CxNHBqdzTr7rLtdrtb5CMjJcDut+WNGCVv7OmS5+lTc= github.com/Azure/go-autorest/autorest/to v0.4.1/go.mod h1:EtaofgU4zmtvn1zT2ARsjRFdq9vXx0YWtmElwL+GZ9M= -github.com/Azure/go-ntlmssp v0.1.0 h1:DjFo6YtWzNqNvQdrwEyr/e4nhU3vRiwenz5QX7sFz+A= -github.com/Azure/go-ntlmssp v0.1.0/go.mod h1:NYqdhxd/8aAct/s4qSYZEerdPuH1liG2/X9DiVTbhpk= +github.com/Azure/go-ntlmssp v0.1.1 h1:l+FM/EEMb0U9QZE7mKNEDw5Mu3mFiaa2GKOoTSsNDPw= +github.com/Azure/go-ntlmssp v0.1.1/go.mod h1:NYqdhxd/8aAct/s4qSYZEerdPuH1liG2/X9DiVTbhpk= github.com/AzureAD/microsoft-authentication-extensions-for-go/cache v0.1.1 h1:WJTmL004Abzc5wDB5VtZG2PJk5ndYDgVacGqfirKxjM= github.com/AzureAD/microsoft-authentication-extensions-for-go/cache v0.1.1/go.mod h1:tCcJZ0uHAmvjsVYzEFivsRTN00oz5BEsRgQHu5JZ9WE= github.com/AzureAD/microsoft-authentication-library-for-go v1.6.0 h1:XRzhVemXdgvJqCH0sFfrBUTnUJSBrBf7++ypk+twtRs= diff --git a/k8s/charts/seaweedfs/templates/all-in-one/all-in-one-deployment.yaml b/k8s/charts/seaweedfs/templates/all-in-one/all-in-one-deployment.yaml index a7b116153..3b4d7a405 100644 --- a/k8s/charts/seaweedfs/templates/all-in-one/all-in-one-deployment.yaml +++ b/k8s/charts/seaweedfs/templates/all-in-one/all-in-one-deployment.yaml @@ -247,9 +247,9 @@ spec: {{- $httpsPort := .Values.allInOne.s3.httpsPort | default .Values.s3.httpsPort }} {{- if $httpsPort }} -s3.port.https={{ $httpsPort }} \ - {{- end }} {{- include "seaweedfs.s3.tlsArgs" (dict "root" . "prefix" "s3.") | nindent 14 }} {{- end }} + {{- end }} {{- if or .Values.allInOne.s3.enableAuth .Values.s3.enableAuth .Values.filer.s3.enableAuth }} -s3.config=/etc/sw/s3/seaweedfs_s3_config \ {{- end }} diff --git a/k8s/charts/seaweedfs/templates/filer/filer-statefulset.yaml b/k8s/charts/seaweedfs/templates/filer/filer-statefulset.yaml index 0574b2c3e..89e8666c3 100644 --- a/k8s/charts/seaweedfs/templates/filer/filer-statefulset.yaml +++ b/k8s/charts/seaweedfs/templates/filer/filer-statefulset.yaml @@ -200,9 +200,9 @@ spec: {{- if .Values.global.seaweedfs.enableSecurity }} {{- if .Values.filer.s3.httpsPort }} -s3.port.https={{ .Values.filer.s3.httpsPort }} \ - {{- end }} {{- include "seaweedfs.s3.tlsArgs" (dict "root" . "prefix" "s3.") | nindent 14 }} {{- end }} + {{- end }} {{- if .Values.filer.s3.enableAuth }} -s3.config=/etc/sw/seaweedfs_s3_config \ {{- end }} diff --git a/k8s/charts/seaweedfs/templates/s3/s3-deployment.yaml b/k8s/charts/seaweedfs/templates/s3/s3-deployment.yaml index 4ee10e832..0fe5374b3 100644 --- a/k8s/charts/seaweedfs/templates/s3/s3-deployment.yaml +++ b/k8s/charts/seaweedfs/templates/s3/s3-deployment.yaml @@ -127,9 +127,9 @@ spec: {{- if .Values.global.seaweedfs.enableSecurity }} {{- if .Values.s3.httpsPort }} -port.https={{ .Values.s3.httpsPort }} \ - {{- end }} {{- include "seaweedfs.s3.tlsArgs" (dict "root" . "prefix" "") | nindent 14 }} {{- end }} + {{- end }} {{- if .Values.s3.domainName }} -domainName={{ .Values.s3.domainName }} \ {{- end }} diff --git a/test/kafka/go.mod b/test/kafka/go.mod index cb81a6478..f60ea68ff 100644 --- a/test/kafka/go.mod +++ b/test/kafka/go.mod @@ -22,7 +22,7 @@ require ( github.com/Azure/azure-sdk-for-go/sdk/internal v1.11.2 // indirect github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.6.4 // indirect github.com/Azure/azure-sdk-for-go/sdk/storage/azfile v1.5.3 // indirect - github.com/Azure/go-ntlmssp v0.1.0 // indirect + github.com/Azure/go-ntlmssp v0.1.1 // indirect github.com/AzureAD/microsoft-authentication-library-for-go v1.6.0 // indirect github.com/FilenCloudDienste/filen-sdk-go v0.0.38 // indirect github.com/Files-com/files-sdk-go/v3 v3.2.264 // indirect diff --git a/test/kafka/go.sum b/test/kafka/go.sum index df6d3eb3d..e8e944fb5 100644 --- a/test/kafka/go.sum +++ b/test/kafka/go.sum @@ -51,8 +51,8 @@ github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.6.4 h1:jWQK1GI+LeGGUKBAD github.com/Azure/azure-sdk-for-go/sdk/storage/azblob v1.6.4/go.mod h1:8mwH4klAm9DUgR2EEHyEEAQlRDvLPyg5fQry3y+cDew= github.com/Azure/azure-sdk-for-go/sdk/storage/azfile v1.5.3 h1:sxgSqOB9CDToiaVFpxuvb5wGgGqWa3lCShcm5o0n3bE= github.com/Azure/azure-sdk-for-go/sdk/storage/azfile v1.5.3/go.mod h1:XdED8i399lEVblYHTZM8eXaP07gv4Z58IL6ueMlVlrg= -github.com/Azure/go-ntlmssp v0.1.0 h1:DjFo6YtWzNqNvQdrwEyr/e4nhU3vRiwenz5QX7sFz+A= -github.com/Azure/go-ntlmssp v0.1.0/go.mod h1:NYqdhxd/8aAct/s4qSYZEerdPuH1liG2/X9DiVTbhpk= +github.com/Azure/go-ntlmssp v0.1.1 h1:l+FM/EEMb0U9QZE7mKNEDw5Mu3mFiaa2GKOoTSsNDPw= +github.com/Azure/go-ntlmssp v0.1.1/go.mod h1:NYqdhxd/8aAct/s4qSYZEerdPuH1liG2/X9DiVTbhpk= github.com/AzureAD/microsoft-authentication-extensions-for-go/cache v0.1.1 h1:WJTmL004Abzc5wDB5VtZG2PJk5ndYDgVacGqfirKxjM= github.com/AzureAD/microsoft-authentication-extensions-for-go/cache v0.1.1/go.mod h1:tCcJZ0uHAmvjsVYzEFivsRTN00oz5BEsRgQHu5JZ9WE= github.com/AzureAD/microsoft-authentication-library-for-go v1.6.0 h1:XRzhVemXdgvJqCH0sFfrBUTnUJSBrBf7++ypk+twtRs= diff --git a/weed/command/nfs.go b/weed/command/nfs.go index d1f01489d..b064a0fe8 100644 --- a/weed/command/nfs.go +++ b/weed/command/nfs.go @@ -23,6 +23,7 @@ type NfsOptions struct { readOnly *bool allowedClients *string volumeServerAccess *string + portmapBind *string } func init() { @@ -34,6 +35,7 @@ func init() { nfsStandaloneOptions.readOnly = cmdNfs.Flag.Bool("readOnly", false, "export the filer path as read only") nfsStandaloneOptions.allowedClients = cmdNfs.Flag.String("allowedClients", "", "comma-separated client IPs, hostnames, or CIDRs allowed to connect") nfsStandaloneOptions.volumeServerAccess = cmdNfs.Flag.String("volumeServerAccess", "direct", "access volume servers by [direct|publicUrl|filerProxy]") + nfsStandaloneOptions.portmapBind = cmdNfs.Flag.String("portmap.bind", "", "when set, bind a built-in portmap v2 responder on :111 so plain `mount -t nfs` works without client-side portmap bypass. Empty disables it. Binding port 111 requires root or CAP_NET_BIND_SERVICE and must not conflict with a system rpcbind.") } var cmdNfs = &Command{ @@ -53,6 +55,26 @@ Safer defaults (since export ACLs are still not implemented): Override -ip.bind to a routable address only after you have reviewed -allowedClients and the readiness of the rest of your deployment. + +Mounting from a Linux client +---------------------------- +The server does not run portmap/rpcbind by default. That means Linux +mount.nfs, which queries portmap on port 111 first, will fail with +"portmap query failed" against the plain form: + + mount -t nfs -o nfsvers=3,nolock :/export /mnt + +Either tell the client to bypass portmap: + + mount -t nfs -o nfsvers=3,nolock,port=2049,mountport=2049,\ + proto=tcp,mountproto=tcp :/export /mnt + +or enable the built-in portmap responder on the server: + + weed nfs ... -portmap.bind=0.0.0.0 + +Binding port 111 requires root or CAP_NET_BIND_SERVICE and must not +collide with a system rpcbind. `, } @@ -85,6 +107,7 @@ func runNfs(cmd *Command, args []string) bool { AllowedClients: util.StringSplit(*nfsStandaloneOptions.allowedClients, ","), VolumeServerAccess: *nfsStandaloneOptions.volumeServerAccess, GrpcDialOption: grpcDialOption, + PortmapBind: *nfsStandaloneOptions.portmapBind, }) if err != nil { glog.Errorf("NFS Server startup error: %v", err) diff --git a/weed/server/nfs/portmap.go b/weed/server/nfs/portmap.go new file mode 100644 index 000000000..899100dff --- /dev/null +++ b/weed/server/nfs/portmap.go @@ -0,0 +1,443 @@ +package nfs + +import ( + "encoding/binary" + "errors" + "fmt" + "io" + "net" + "sync" + "time" + + "github.com/seaweedfs/seaweedfs/weed/glog" +) + +// Minimal PORTMAP v2 responder. +// +// The upstream willscott/go-nfs library serves NFSv3 and MOUNT on a single TCP +// port and deliberately does not register with portmap (RPC program 100000). +// Linux mount.nfs, however, queries portmap on port 111 before sending the +// MOUNT RPC, so the plain `mount -t nfs host:/export /mnt` command fails +// against a default `weed nfs` deployment. +// +// When enabled, this responder binds the privileged port 111 (RFC 1833) on +// both TCP and UDP and answers the subset of PORTMAP v2 calls that standard +// Linux clients make: PMAP_NULL, PMAP_GETPORT and PMAP_DUMP. It refuses +// registration from third parties (PMAP_SET / PMAP_UNSET return false) and +// only exposes the programs that weed itself serves. +// +// References: RFC 1833 (Portmap v2), RFC 5531 (RPC). +const ( + portmapProgram = 100000 + portmapVersion = 2 + portmapPort = 111 + + pmapProcNull = 0 + pmapProcSet = 1 + pmapProcUnset = 2 + pmapProcGetPort = 3 + pmapProcDump = 4 + + ipProtoTCP = 6 + ipProtoUDP = 17 + + nfsProgram = 100003 + mountProgram = 100005 + + // RPC + rpcMsgCall = 0 + rpcMsgReply = 1 + + rpcMsgAccepted = 0 + + rpcAcceptSuccess = 0 + rpcAcceptProgUnavail = 1 + rpcAcceptProgMismatch = 2 + rpcAcceptProcUnavail = 3 + rpcAcceptGarbageArgs = 4 + + rpcAuthNone = 0 + + // Defensive limits. Portmap messages are tiny in practice; these caps + // protect the responder from large or slow reads. + portmapMaxRecord = 64 * 1024 + + // Per-connection read/write deadlines on the TCP listener. The idle + // timeout bounds how long we wait for the next request on an otherwise + // quiet connection; the IO timeout bounds a single read or write once + // one is in flight. Both guard against slowloris-style stalls on the + // privileged port 111. + portmapTCPIdleTimeout = 30 * time.Second + portmapTCPIOTimeout = 10 * time.Second + + // Back-off applied before retrying after a non-fatal listener error + // (e.g. EMFILE on TCP Accept, or a transient UDP read failure) so we + // don't busy-loop when the host is under pressure. + portmapRetryBackoff = 50 * time.Millisecond +) + +type portmapEntry struct { + Program uint32 + Version uint32 + Protocol uint32 + Port uint32 +} + +type portmapServer struct { + bindIP string + port int + entries []portmapEntry + + tcpListener net.Listener + udpConn *net.UDPConn + + // mu guards closed and conns. It is held only for bookkeeping, never + // across network IO. + mu sync.Mutex + closed bool + conns map[net.Conn]struct{} + // done is closed exactly once by Close() so that background loops can + // interrupt a retry-backoff sleep instead of waiting it out. + done chan struct{} + wg sync.WaitGroup +} + +// newPortmapServer builds a responder advertising the NFS services the caller +// runs on nfsTCPPort. We expose NFS v3 TCP and MOUNT v3 TCP only: the +// underlying library does not handle UDP or older MOUNT versions, so it would +// be misleading to advertise them. +func newPortmapServer(bindIP string, port int, nfsTCPPort uint32) *portmapServer { + if port <= 0 { + port = portmapPort + } + return &portmapServer{ + bindIP: bindIP, + port: port, + done: make(chan struct{}), + entries: []portmapEntry{ + {Program: nfsProgram, Version: 3, Protocol: ipProtoTCP, Port: nfsTCPPort}, + {Program: mountProgram, Version: 3, Protocol: ipProtoTCP, Port: nfsTCPPort}, + }, + } +} + +func (ps *portmapServer) Start() error { + addr := net.JoinHostPort(ps.bindIP, fmt.Sprintf("%d", ps.port)) + + tcpLn, err := net.Listen("tcp", addr) + if err != nil { + return fmt.Errorf("portmap tcp listen %s: %w", addr, err) + } + udpAddr, err := net.ResolveUDPAddr("udp", addr) + if err != nil { + _ = tcpLn.Close() + return fmt.Errorf("portmap udp resolve %s: %w", addr, err) + } + udpConn, err := net.ListenUDP("udp", udpAddr) + if err != nil { + _ = tcpLn.Close() + return fmt.Errorf("portmap udp listen %s: %w", addr, err) + } + ps.tcpListener = tcpLn + ps.udpConn = udpConn + + ps.wg.Add(2) + go func() { + defer ps.wg.Done() + ps.serveTCP() + }() + go func() { + defer ps.wg.Done() + ps.serveUDP() + }() + return nil +} + +func (ps *portmapServer) Close() error { + ps.mu.Lock() + if ps.closed { + ps.mu.Unlock() + return nil + } + ps.closed = true + conns := ps.conns + ps.conns = nil + close(ps.done) + ps.mu.Unlock() + + var first error + if ps.tcpListener != nil { + if err := ps.tcpListener.Close(); err != nil { + first = err + } + } + if ps.udpConn != nil { + if err := ps.udpConn.Close(); err != nil && first == nil { + first = err + } + } + // Evict in-flight TCP handlers so Close() does not block on idle + // clients; their read goroutines will unwind on the closed conn. + for c := range conns { + _ = c.Close() + } + ps.wg.Wait() + return first +} + +func (ps *portmapServer) isClosed() bool { + ps.mu.Lock() + defer ps.mu.Unlock() + return ps.closed +} + +// addConn registers c for shutdown eviction. It returns false (and the +// caller must drop c) if the server has already started shutting down. +func (ps *portmapServer) addConn(c net.Conn) bool { + ps.mu.Lock() + defer ps.mu.Unlock() + if ps.closed { + return false + } + if ps.conns == nil { + ps.conns = make(map[net.Conn]struct{}) + } + ps.conns[c] = struct{}{} + return true +} + +func (ps *portmapServer) removeConn(c net.Conn) { + ps.mu.Lock() + defer ps.mu.Unlock() + delete(ps.conns, c) +} + +func (ps *portmapServer) serveTCP() { + for { + conn, err := ps.tcpListener.Accept() + if err != nil { + if ps.isClosed() { + return + } + // Non-fatal (e.g. EMFILE, EINTR): log and back off rather + // than tear the listener down on a transient resource blip. + // Wake early if Close() fires during the sleep. + glog.V(1).Infof("portmap tcp accept: %v", err) + select { + case <-ps.done: + return + case <-time.After(portmapRetryBackoff): + continue + } + } + if !ps.addConn(conn) { + _ = conn.Close() + continue + } + ps.wg.Add(1) + go func(c net.Conn) { + defer ps.wg.Done() + defer ps.removeConn(c) + ps.handleTCPConn(c) + }(conn) + } +} + +func (ps *portmapServer) handleTCPConn(conn net.Conn) { + defer conn.Close() + hdr := make([]byte, 4) + for { + _ = conn.SetReadDeadline(time.Now().Add(portmapTCPIdleTimeout)) + if _, err := io.ReadFull(conn, hdr); err != nil { + return + } + mark := binary.BigEndian.Uint32(hdr) + // Bit 31: last-fragment flag. Portmap messages are always single + // fragment in practice; drop the connection if we see otherwise. + if mark&(1<<31) == 0 { + return + } + recLen := mark &^ (1 << 31) + if recLen == 0 || recLen > portmapMaxRecord { + return + } + buf := make([]byte, recLen) + _ = conn.SetReadDeadline(time.Now().Add(portmapTCPIOTimeout)) + if _, err := io.ReadFull(conn, buf); err != nil { + return + } + reply := ps.handleCall(buf) + if reply == nil { + continue + } + out := make([]byte, 4+len(reply)) + binary.BigEndian.PutUint32(out[0:4], uint32(len(reply))|(1<<31)) + copy(out[4:], reply) + _ = conn.SetWriteDeadline(time.Now().Add(portmapTCPIOTimeout)) + if _, err := conn.Write(out); err != nil { + return + } + } +} + +func (ps *portmapServer) serveUDP() { + buf := make([]byte, portmapMaxRecord) + for { + n, addr, err := ps.udpConn.ReadFromUDP(buf) + if err != nil { + if ps.isClosed() { + return + } + // Transient read failure: log, back off, and keep the + // responder alive instead of taking UDP portmap down. + // Wake early if Close() fires during the sleep. + glog.V(1).Infof("portmap udp read: %v", err) + select { + case <-ps.done: + return + case <-time.After(portmapRetryBackoff): + continue + } + } + reply := ps.handleCall(buf[:n]) + if reply == nil { + continue + } + if _, err := ps.udpConn.WriteToUDP(reply, addr); err != nil { + glog.V(1).Infof("portmap udp write to %s: %v", addr, err) + } + } +} + +// handleCall parses one RPC CALL message and returns the encoded reply, or nil +// if the call is malformed enough that we should drop it silently. +func (ps *portmapServer) handleCall(callBuf []byte) []byte { + xid, prog, vers, proc, args, err := parseRPCCall(callBuf) + if err != nil { + return nil + } + if prog != portmapProgram { + return encodeAcceptedReply(xid, rpcAcceptProgUnavail, nil) + } + if vers != portmapVersion { + // Program-version mismatch: RFC 5531 says we should return the + // accepted range; keep it simple and report 2..2. + body := make([]byte, 8) + binary.BigEndian.PutUint32(body[0:4], portmapVersion) + binary.BigEndian.PutUint32(body[4:8], portmapVersion) + return encodeAcceptedReply(xid, rpcAcceptProgMismatch, body) + } + switch proc { + case pmapProcNull: + return encodeAcceptedReply(xid, rpcAcceptSuccess, nil) + case pmapProcGetPort: + if len(args) < 16 { + return encodeAcceptedReply(xid, rpcAcceptGarbageArgs, nil) + } + q := portmapEntry{ + Program: binary.BigEndian.Uint32(args[0:4]), + Version: binary.BigEndian.Uint32(args[4:8]), + Protocol: binary.BigEndian.Uint32(args[8:12]), + } + port := uint32(0) + for _, e := range ps.entries { + if e.Program == q.Program && e.Version == q.Version && e.Protocol == q.Protocol { + port = e.Port + break + } + } + body := make([]byte, 4) + binary.BigEndian.PutUint32(body, port) + return encodeAcceptedReply(xid, rpcAcceptSuccess, body) + case pmapProcDump: + // Each entry is 4-byte value_follows + 16-byte mapping = 20 bytes, + // plus a 4-byte terminator value_follows=FALSE. + body := make([]byte, 0, 20*len(ps.entries)+4) + for _, e := range ps.entries { + chunk := make([]byte, 20) + binary.BigEndian.PutUint32(chunk[0:4], 1) // value_follows = TRUE + binary.BigEndian.PutUint32(chunk[4:8], e.Program) + binary.BigEndian.PutUint32(chunk[8:12], e.Version) + binary.BigEndian.PutUint32(chunk[12:16], e.Protocol) + binary.BigEndian.PutUint32(chunk[16:20], e.Port) + body = append(body, chunk...) + } + end := make([]byte, 4) // value_follows = FALSE + body = append(body, end...) + return encodeAcceptedReply(xid, rpcAcceptSuccess, body) + case pmapProcSet, pmapProcUnset: + // Don't accept third-party registrations. bool=FALSE. + body := make([]byte, 4) + return encodeAcceptedReply(xid, rpcAcceptSuccess, body) + default: + return encodeAcceptedReply(xid, rpcAcceptProcUnavail, nil) + } +} + +// parseRPCCall parses the fixed portion of an RPC CALL header and returns the +// remaining procedure arguments. It skips both opaque_auth fields (cred and +// verf) so callers get a buffer starting at the procedure arguments. +func parseRPCCall(buf []byte) (xid, prog, vers, proc uint32, args []byte, err error) { + // Minimum header: xid + msg_type + rpcvers + prog + vers + proc + 2x + // (flavor + len) = 6*4 + 2*8 = 40 bytes. + const minHeader = 40 + if len(buf) < minHeader { + err = fmt.Errorf("rpc call too short: %d bytes", len(buf)) + return + } + xid = binary.BigEndian.Uint32(buf[0:4]) + if msgType := binary.BigEndian.Uint32(buf[4:8]); msgType != rpcMsgCall { + err = fmt.Errorf("not an rpc call: msg_type=%d", msgType) + return + } + if rpcvers := binary.BigEndian.Uint32(buf[8:12]); rpcvers != 2 { + err = fmt.Errorf("unsupported rpc version %d", rpcvers) + return + } + prog = binary.BigEndian.Uint32(buf[12:16]) + vers = binary.BigEndian.Uint32(buf[16:20]) + proc = binary.BigEndian.Uint32(buf[20:24]) + + p := 24 + for i := 0; i < 2; i++ { + if len(buf) < p+8 { + err = fmt.Errorf("truncated opaque_auth at offset %d", p) + return + } + authLen := binary.BigEndian.Uint32(buf[p+4 : p+8]) + // Validate before applying the XDR 4-byte padding so that + // lengths near uint32 max can't wrap to a tiny padded value. + if authLen > uint32(portmapMaxRecord) { + err = errors.New("opaque_auth length exceeds limit") + return + } + padded := (authLen + 3) &^ 3 + end := uint64(p) + 8 + uint64(padded) + if end > uint64(len(buf)) { + err = fmt.Errorf("truncated opaque_auth body at offset %d (len=%d)", p, authLen) + return + } + p = int(end) + } + args = buf[p:] + return +} + +// encodeAcceptedReply builds a MSG_ACCEPTED reply with the given accept_stat. +// body is the already-XDR-encoded data that follows accept_stat in the reply. +// For SUCCESS it is the procedure result; it is nil for most error +// accept_stat values (PROG_UNAVAIL, PROC_UNAVAIL, GARBAGE_ARGS) but is +// non-nil for PROG_MISMATCH, which carries a struct { uint32 low; uint32 +// high; } mismatch_info range per RFC 5531 §9. +func encodeAcceptedReply(xid, acceptStat uint32, body []byte) []byte { + out := make([]byte, 24+len(body)) + binary.BigEndian.PutUint32(out[0:4], xid) + binary.BigEndian.PutUint32(out[4:8], rpcMsgReply) + binary.BigEndian.PutUint32(out[8:12], rpcMsgAccepted) + // verf: AUTH_NONE, zero-length opaque + binary.BigEndian.PutUint32(out[12:16], rpcAuthNone) + binary.BigEndian.PutUint32(out[16:20], 0) + binary.BigEndian.PutUint32(out[20:24], acceptStat) + copy(out[24:], body) + return out +} diff --git a/weed/server/nfs/portmap_test.go b/weed/server/nfs/portmap_test.go new file mode 100644 index 000000000..ed470959b --- /dev/null +++ b/weed/server/nfs/portmap_test.go @@ -0,0 +1,416 @@ +package nfs + +import ( + "bytes" + "encoding/binary" + "io" + "net" + "strconv" + "testing" + "time" +) + +func buildRPCCall(t *testing.T, xid, prog, vers, proc uint32, credBody, verfBody, args []byte) []byte { + t.Helper() + pad := func(b []byte) []byte { + r := len(b) % 4 + if r == 0 { + return b + } + out := make([]byte, len(b)+4-r) + copy(out, b) + return out + } + buf := new(bytes.Buffer) + write := func(v uint32) { + var b [4]byte + binary.BigEndian.PutUint32(b[:], v) + buf.Write(b[:]) + } + write(xid) + write(rpcMsgCall) + write(2) // rpcvers + write(prog) + write(vers) + write(proc) + // cred + write(rpcAuthNone) + write(uint32(len(credBody))) + buf.Write(pad(credBody)) + // verf + write(rpcAuthNone) + write(uint32(len(verfBody))) + buf.Write(pad(verfBody)) + buf.Write(args) + return buf.Bytes() +} + +func parseAcceptedReply(t *testing.T, reply []byte) (xid, acceptStat uint32, body []byte) { + t.Helper() + if len(reply) < 24 { + t.Fatalf("reply too short: %d bytes", len(reply)) + } + xid = binary.BigEndian.Uint32(reply[0:4]) + if mt := binary.BigEndian.Uint32(reply[4:8]); mt != rpcMsgReply { + t.Fatalf("msg_type=%d, want REPLY", mt) + } + if rs := binary.BigEndian.Uint32(reply[8:12]); rs != rpcMsgAccepted { + t.Fatalf("reply_stat=%d, want ACCEPTED", rs) + } + // verf + verfLen := binary.BigEndian.Uint32(reply[16:20]) + if verfLen != 0 { + t.Fatalf("unexpected verf length %d", verfLen) + } + acceptStat = binary.BigEndian.Uint32(reply[20:24]) + body = reply[24:] + return +} + +func newTestPortmap() *portmapServer { + return newPortmapServer("127.0.0.1", portmapPort, 2049) +} + +func TestParseRPCCall_SkipsAuth(t *testing.T) { + cred := []byte("hello") // 5 bytes -> padded to 8 + verf := []byte{} + args := []byte{0x01, 0x02, 0x03, 0x04} + msg := buildRPCCall(t, 42, portmapProgram, portmapVersion, pmapProcNull, cred, verf, args) + + xid, prog, vers, proc, gotArgs, err := parseRPCCall(msg) + if err != nil { + t.Fatalf("parseRPCCall: %v", err) + } + if xid != 42 || prog != portmapProgram || vers != portmapVersion || proc != pmapProcNull { + t.Fatalf("header mismatch: xid=%d prog=%d vers=%d proc=%d", xid, prog, vers, proc) + } + if !bytes.Equal(gotArgs, args) { + t.Fatalf("args mismatch: got %x want %x", gotArgs, args) + } +} + +func TestParseRPCCall_RejectsReply(t *testing.T) { + buf := make([]byte, 40) + binary.BigEndian.PutUint32(buf[4:8], rpcMsgReply) + if _, _, _, _, _, err := parseRPCCall(buf); err == nil { + t.Fatal("expected error on reply-typed message") + } +} + +func TestParseRPCCall_TruncatedAuth(t *testing.T) { + // Claim huge cred length but provide no body. + buf := make([]byte, 40) + binary.BigEndian.PutUint32(buf[4:8], rpcMsgCall) + binary.BigEndian.PutUint32(buf[8:12], 2) + binary.BigEndian.PutUint32(buf[28:32], 1000) // cred len + if _, _, _, _, _, err := parseRPCCall(buf); err == nil { + t.Fatal("expected error on truncated auth") + } +} + +func TestHandleCall_Null(t *testing.T) { + ps := newTestPortmap() + msg := buildRPCCall(t, 7, portmapProgram, portmapVersion, pmapProcNull, nil, nil, nil) + reply := ps.handleCall(msg) + xid, acc, body := parseAcceptedReply(t, reply) + if xid != 7 || acc != rpcAcceptSuccess || len(body) != 0 { + t.Fatalf("null reply xid=%d acc=%d body=%x", xid, acc, body) + } +} + +func TestHandleCall_GetPort_HitAndMiss(t *testing.T) { + ps := newTestPortmap() + + buildQuery := func(prog, vers, prot uint32) []byte { + args := make([]byte, 16) + binary.BigEndian.PutUint32(args[0:4], prog) + binary.BigEndian.PutUint32(args[4:8], vers) + binary.BigEndian.PutUint32(args[8:12], prot) + // port field is ignored by the server; leave zero + return args + } + + cases := []struct { + name string + prog, vers, prot uint32 + wantPort uint32 + }{ + {"nfs-v3-tcp-hit", nfsProgram, 3, ipProtoTCP, 2049}, + {"mount-v3-tcp-hit", mountProgram, 3, ipProtoTCP, 2049}, + {"mount-v1-tcp-miss", mountProgram, 1, ipProtoTCP, 0}, + {"nfs-v3-udp-miss", nfsProgram, 3, ipProtoUDP, 0}, + {"nlm-miss", 100021, 4, ipProtoTCP, 0}, + } + for _, tc := range cases { + t.Run(tc.name, func(t *testing.T) { + msg := buildRPCCall(t, 11, portmapProgram, portmapVersion, pmapProcGetPort, nil, nil, buildQuery(tc.prog, tc.vers, tc.prot)) + reply := ps.handleCall(msg) + xid, acc, body := parseAcceptedReply(t, reply) + if xid != 11 { + t.Fatalf("xid=%d want 11", xid) + } + if acc != rpcAcceptSuccess { + t.Fatalf("acc=%d want SUCCESS", acc) + } + if len(body) != 4 { + t.Fatalf("getport body len=%d want 4", len(body)) + } + got := binary.BigEndian.Uint32(body) + if got != tc.wantPort { + t.Fatalf("port=%d want %d", got, tc.wantPort) + } + }) + } +} + +func TestHandleCall_Dump(t *testing.T) { + ps := newTestPortmap() + msg := buildRPCCall(t, 13, portmapProgram, portmapVersion, pmapProcDump, nil, nil, nil) + reply := ps.handleCall(msg) + _, acc, body := parseAcceptedReply(t, reply) + if acc != rpcAcceptSuccess { + t.Fatalf("acc=%d", acc) + } + var entries []portmapEntry + p := 0 + for p+4 <= len(body) { + vf := binary.BigEndian.Uint32(body[p : p+4]) + p += 4 + if vf == 0 { + break + } + if p+16 > len(body) { + t.Fatalf("truncated entry at %d", p) + } + entries = append(entries, portmapEntry{ + Program: binary.BigEndian.Uint32(body[p : p+4]), + Version: binary.BigEndian.Uint32(body[p+4 : p+8]), + Protocol: binary.BigEndian.Uint32(body[p+8 : p+12]), + Port: binary.BigEndian.Uint32(body[p+12 : p+16]), + }) + p += 16 + } + if len(entries) != 2 { + t.Fatalf("got %d dump entries, want 2: %+v", len(entries), entries) + } + wantSet := map[portmapEntry]bool{ + {Program: nfsProgram, Version: 3, Protocol: ipProtoTCP, Port: 2049}: false, + {Program: mountProgram, Version: 3, Protocol: ipProtoTCP, Port: 2049}: false, + } + for _, e := range entries { + if _, ok := wantSet[e]; !ok { + t.Fatalf("unexpected dump entry %+v", e) + } + wantSet[e] = true + } + for e, seen := range wantSet { + if !seen { + t.Fatalf("missing dump entry %+v", e) + } + } +} + +func TestHandleCall_UnknownProg(t *testing.T) { + ps := newTestPortmap() + msg := buildRPCCall(t, 1, 999999, 1, 0, nil, nil, nil) + reply := ps.handleCall(msg) + _, acc, _ := parseAcceptedReply(t, reply) + if acc != rpcAcceptProgUnavail { + t.Fatalf("acc=%d want PROG_UNAVAIL", acc) + } +} + +func TestHandleCall_VersionMismatch(t *testing.T) { + ps := newTestPortmap() + msg := buildRPCCall(t, 1, portmapProgram, 42, pmapProcNull, nil, nil, nil) + reply := ps.handleCall(msg) + _, acc, body := parseAcceptedReply(t, reply) + if acc != rpcAcceptProgMismatch { + t.Fatalf("acc=%d want PROG_MISMATCH", acc) + } + if len(body) != 8 { + t.Fatalf("mismatch body len=%d want 8", len(body)) + } + lo := binary.BigEndian.Uint32(body[0:4]) + hi := binary.BigEndian.Uint32(body[4:8]) + if lo != portmapVersion || hi != portmapVersion { + t.Fatalf("mismatch range lo=%d hi=%d", lo, hi) + } +} + +func TestHandleCall_UnknownProc(t *testing.T) { + ps := newTestPortmap() + msg := buildRPCCall(t, 1, portmapProgram, portmapVersion, 42, nil, nil, nil) + reply := ps.handleCall(msg) + _, acc, _ := parseAcceptedReply(t, reply) + if acc != rpcAcceptProcUnavail { + t.Fatalf("acc=%d want PROC_UNAVAIL", acc) + } +} + +func TestHandleCall_SetRefused(t *testing.T) { + ps := newTestPortmap() + args := make([]byte, 16) // mapping struct + msg := buildRPCCall(t, 1, portmapProgram, portmapVersion, pmapProcSet, nil, nil, args) + reply := ps.handleCall(msg) + _, acc, body := parseAcceptedReply(t, reply) + if acc != rpcAcceptSuccess { + t.Fatalf("acc=%d", acc) + } + if len(body) != 4 || binary.BigEndian.Uint32(body) != 0 { + t.Fatalf("PMAP_SET must return FALSE, got %x", body) + } +} + +// pickFreePort asks the OS for an unused high port by opening and closing a +// listener on it. Used so the end-to-end tests can run in parallel without +// stepping on the privileged default port 111. +func pickFreePort(t *testing.T) int { + t.Helper() + ln, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatalf("listen: %v", err) + } + defer ln.Close() + return ln.Addr().(*net.TCPAddr).Port +} + +func TestPortmapServer_UDPGetPort(t *testing.T) { + port := pickFreePort(t) + ps := newPortmapServer("127.0.0.1", port, 2049) + if err := ps.Start(); err != nil { + t.Fatalf("start: %v", err) + } + t.Cleanup(func() { _ = ps.Close() }) + + args := make([]byte, 16) + binary.BigEndian.PutUint32(args[0:4], nfsProgram) + binary.BigEndian.PutUint32(args[4:8], 3) + binary.BigEndian.PutUint32(args[8:12], ipProtoTCP) + msg := buildRPCCall(t, 99, portmapProgram, portmapVersion, pmapProcGetPort, nil, nil, args) + + conn, err := net.Dial("udp", net.JoinHostPort("127.0.0.1", strconv.Itoa(port))) + if err != nil { + t.Fatalf("dial udp: %v", err) + } + defer conn.Close() + _ = conn.SetDeadline(time.Now().Add(2 * time.Second)) + if _, err := conn.Write(msg); err != nil { + t.Fatalf("write: %v", err) + } + buf := make([]byte, 4096) + n, err := conn.Read(buf) + if err != nil { + t.Fatalf("read: %v", err) + } + xid, acc, body := parseAcceptedReply(t, buf[:n]) + if xid != 99 || acc != rpcAcceptSuccess || len(body) != 4 { + t.Fatalf("bad reply xid=%d acc=%d body=%x", xid, acc, body) + } + if got := binary.BigEndian.Uint32(body); got != 2049 { + t.Fatalf("udp getport port=%d want 2049", got) + } +} + +func TestPortmapServer_CloseEvictsIdleTCPConn(t *testing.T) { + port := pickFreePort(t) + ps := newPortmapServer("127.0.0.1", port, 2049) + if err := ps.Start(); err != nil { + t.Fatalf("start: %v", err) + } + + conn, err := net.Dial("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(port))) + if err != nil { + _ = ps.Close() + t.Fatalf("dial: %v", err) + } + defer conn.Close() + + // Issue one call and read its reply so the server-side connection is + // definitely registered before we trigger shutdown. + msg := buildRPCCall(t, 1, portmapProgram, portmapVersion, pmapProcNull, nil, nil, nil) + var mark [4]byte + binary.BigEndian.PutUint32(mark[:], uint32(len(msg))|(1<<31)) + _ = conn.SetDeadline(time.Now().Add(2 * time.Second)) + if _, err := conn.Write(mark[:]); err != nil { + t.Fatalf("write mark: %v", err) + } + if _, err := conn.Write(msg); err != nil { + t.Fatalf("write msg: %v", err) + } + if _, err := io.ReadFull(conn, mark[:]); err != nil { + t.Fatalf("read mark: %v", err) + } + rlen := binary.BigEndian.Uint32(mark[:]) &^ (1 << 31) + if _, err := io.ReadFull(conn, make([]byte, rlen)); err != nil { + t.Fatalf("read body: %v", err) + } + + // Close must return long before the TCP idle deadline (30s) — in + // other words, the server must actively close the idle conn rather + // than wait for the deadline or for the client to disconnect. + done := make(chan error, 1) + go func() { done <- ps.Close() }() + select { + case err := <-done: + if err != nil { + t.Fatalf("Close: %v", err) + } + case <-time.After(2 * time.Second): + t.Fatal("Close did not return within 2s; in-flight conn not evicted") + } + + _ = conn.SetReadDeadline(time.Now().Add(1 * time.Second)) + if _, err := conn.Read(make([]byte, 4)); err == nil { + t.Fatal("expected read error on client conn after server Close") + } +} + +func TestPortmapServer_TCPGetPort(t *testing.T) { + port := pickFreePort(t) + ps := newPortmapServer("127.0.0.1", port, 2049) + if err := ps.Start(); err != nil { + t.Fatalf("start: %v", err) + } + t.Cleanup(func() { _ = ps.Close() }) + + args := make([]byte, 16) + binary.BigEndian.PutUint32(args[0:4], mountProgram) + binary.BigEndian.PutUint32(args[4:8], 3) + binary.BigEndian.PutUint32(args[8:12], ipProtoTCP) + msg := buildRPCCall(t, 123, portmapProgram, portmapVersion, pmapProcGetPort, nil, nil, args) + + conn, err := net.Dial("tcp", net.JoinHostPort("127.0.0.1", strconv.Itoa(port))) + if err != nil { + t.Fatalf("dial tcp: %v", err) + } + defer conn.Close() + _ = conn.SetDeadline(time.Now().Add(2 * time.Second)) + + // record mark: last-fragment bit + length + var mark [4]byte + binary.BigEndian.PutUint32(mark[:], uint32(len(msg))|(1<<31)) + if _, err := conn.Write(mark[:]); err != nil { + t.Fatalf("write mark: %v", err) + } + if _, err := conn.Write(msg); err != nil { + t.Fatalf("write msg: %v", err) + } + + var rmark [4]byte + if _, err := io.ReadFull(conn, rmark[:]); err != nil { + t.Fatalf("read mark: %v", err) + } + rlen := binary.BigEndian.Uint32(rmark[:]) &^ (1 << 31) + buf := make([]byte, rlen) + if _, err := io.ReadFull(conn, buf); err != nil { + t.Fatalf("read body: %v", err) + } + xid, acc, body := parseAcceptedReply(t, buf) + if xid != 123 || acc != rpcAcceptSuccess || len(body) != 4 { + t.Fatalf("bad reply xid=%d acc=%d body=%x", xid, acc, body) + } + if got := binary.BigEndian.Uint32(body); got != 2049 { + t.Fatalf("tcp getport port=%d want 2049", got) + } +} diff --git a/weed/server/nfs/server.go b/weed/server/nfs/server.go index 8cc710275..3d5bd3705 100644 --- a/weed/server/nfs/server.go +++ b/weed/server/nfs/server.go @@ -26,6 +26,11 @@ type Option struct { AllowedClients []string VolumeServerAccess string GrpcDialOption grpc.DialOption + // PortmapBind, when non-empty, enables a built-in portmap v2 responder + // on :111 advertising the NFS v3 and MOUNT v3 services at + // Port. Empty (the default) disables portmap; clients must then bypass + // portmap with mount -o port=,mountport=,proto=tcp,mountproto=tcp. + PortmapBind string } type Server struct { @@ -93,9 +98,41 @@ func (s *Server) Start() error { return fmt.Errorf("listen nfs on %s:%d: %w", s.option.BindIp, s.option.Port, err) } + var portmap *portmapServer + if s.option.PortmapBind != "" { + portmap = newPortmapServer(s.option.PortmapBind, portmapPort, uint32(s.option.Port)) + if pmErr := portmap.Start(); pmErr != nil { + _ = listener.Close() + return fmt.Errorf("start portmap: %w", pmErr) + } + glog.V(0).Infof("NFS portmap responder listening on %s:%d (NFS v3 tcp=%d, MOUNT v3 tcp=%d)", + s.option.PortmapBind, portmapPort, s.option.Port, s.option.Port) + defer func() { + if portmap != nil { + _ = portmap.Close() + } + }() + } + + s.logMountHint() return s.serve(listener) } +// logMountHint prints a copy-pasteable Linux mount command so operators can +// see at startup how to mount the export from a client. The go-nfs library +// does not run portmap, so without -portmap.bind the client must bypass +// portmap via -o port=,mountport=,proto=tcp,mountproto=tcp. +func (s *Server) logMountHint() { + exportPath := string(s.exportRoot) + if s.option.PortmapBind != "" { + glog.V(0).Infof("mount example: mount -t nfs -o nfsvers=3,nolock :%s ", exportPath) + return + } + glog.V(0).Infof("mount example (bypasses portmap): mount -t nfs -o nfsvers=3,nolock,noacl,port=%d,mountport=%d,proto=tcp,mountproto=tcp :%s ", + s.option.Port, s.option.Port, exportPath) + glog.V(0).Infof("tip: pass -portmap.bind to enable the built-in portmap responder on port 111 so plain `mount -t nfs host:%s /mnt` works.", exportPath) +} + func (s *Server) serve(listener net.Listener) error { if s.filerClient != nil { defer s.filerClient.Close() diff --git a/weed/storage/store.go b/weed/storage/store.go index 09bc6dd65..205aad7b8 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -807,7 +807,7 @@ func (s *Store) MaybeAdjustVolumeMax() (hasChanges bool) { } volCount := diskLocation.VolumesLen() ecShardCount := diskLocation.EcShardCount() - maxVolumeCount := int32(volCount) + int32((ecShardCount+erasure_coding.DataShardsCount)/erasure_coding.DataShardsCount) + maxVolumeCount := int32(volCount) + int32((ecShardCount+erasure_coding.DataShardsCount-1)/erasure_coding.DataShardsCount) if unclaimedSpaces > int64(volumeSizeLimit) { maxVolumeCount += int32(uint64(unclaimedSpaces)/volumeSizeLimit) - 1 } diff --git a/weed/topology/disk.go b/weed/topology/disk.go index 3616ff928..b5ce0b60c 100644 --- a/weed/topology/disk.go +++ b/weed/topology/disk.go @@ -23,6 +23,12 @@ type Disk struct { ecShardsLock sync.RWMutex } +// ecShardSlots returns the number of volume slots consumed by the given +// number of EC shards, rounded up to whole-volume equivalents. +func ecShardSlots(ecShardCount int64) int64 { + return (ecShardCount + erasure_coding.DataShardsCount - 1) / erasure_coding.DataShardsCount +} + func NewDisk(diskType string) *Disk { s := &Disk{} s.id = NodeId(diskType) @@ -67,7 +73,7 @@ func (d *DiskUsages) ToDiskInfo() map[string]*master_pb.DiskInfo { m := &master_pb.DiskInfo{ VolumeCount: diskUsageCounts.volumeCount, MaxVolumeCount: diskUsageCounts.maxVolumeCount, - FreeVolumeCount: diskUsageCounts.maxVolumeCount - (diskUsageCounts.volumeCount - diskUsageCounts.remoteVolumeCount) - (diskUsageCounts.ecShardCount+1)/erasure_coding.DataShardsCount, + FreeVolumeCount: diskUsageCounts.maxVolumeCount - (diskUsageCounts.volumeCount - diskUsageCounts.remoteVolumeCount) - ecShardSlots(diskUsageCounts.ecShardCount), ActiveVolumeCount: diskUsageCounts.activeVolumeCount, RemoteVolumeCount: diskUsageCounts.remoteVolumeCount, } @@ -111,11 +117,7 @@ func (a *DiskUsageCounts) addDiskUsageCounts(b *DiskUsageCounts) { } func (a *DiskUsageCounts) FreeSpace() int64 { - freeVolumeSlotCount := a.maxVolumeCount + a.remoteVolumeCount - a.volumeCount - if a.ecShardCount > 0 { - freeVolumeSlotCount = freeVolumeSlotCount - a.ecShardCount/erasure_coding.DataShardsCount - 1 - } - return freeVolumeSlotCount + return a.maxVolumeCount + a.remoteVolumeCount - a.volumeCount - ecShardSlots(a.ecShardCount) } func (du *DiskUsages) getOrCreateDisk(diskType types.DiskType) *DiskUsageCounts { @@ -265,7 +267,7 @@ func (d *Disk) ToDiskInfo() *master_pb.DiskInfo { Type: string(d.Id()), VolumeCount: diskUsage.volumeCount, MaxVolumeCount: diskUsage.maxVolumeCount, - FreeVolumeCount: diskUsage.maxVolumeCount - (diskUsage.volumeCount - diskUsage.remoteVolumeCount) - (diskUsage.ecShardCount+1)/erasure_coding.DataShardsCount, + FreeVolumeCount: diskUsage.maxVolumeCount - (diskUsage.volumeCount - diskUsage.remoteVolumeCount) - ecShardSlots(diskUsage.ecShardCount), ActiveVolumeCount: diskUsage.activeVolumeCount, RemoteVolumeCount: diskUsage.remoteVolumeCount, DiskId: diskId, diff --git a/weed/topology/node.go b/weed/topology/node.go index 6b82dd223..3a11d0433 100644 --- a/weed/topology/node.go +++ b/weed/topology/node.go @@ -11,7 +11,6 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/stats" - "github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding" "github.com/seaweedfs/seaweedfs/weed/storage/needle" "github.com/seaweedfs/seaweedfs/weed/storage/types" ) @@ -272,10 +271,7 @@ func (n *NodeImpl) getOrCreateDisk(diskType types.DiskType) *DiskUsageCounts { func (n *NodeImpl) AvailableSpaceFor(option *VolumeGrowOption) int64 { t := n.getOrCreateDisk(option.DiskType) freeVolumeSlotCount := atomic.LoadInt64(&t.maxVolumeCount) + atomic.LoadInt64(&t.remoteVolumeCount) - atomic.LoadInt64(&t.volumeCount) - ecShardCount := atomic.LoadInt64(&t.ecShardCount) - if ecShardCount > 0 { - freeVolumeSlotCount = freeVolumeSlotCount - ecShardCount/erasure_coding.DataShardsCount - 1 - } + freeVolumeSlotCount -= ecShardSlots(atomic.LoadInt64(&t.ecShardCount)) return freeVolumeSlotCount }