feat(volume): add UDS handler for RDMA sidecar integration

Add Unix Domain Socket (UDS) handler to Volume server for RDMA sidecar
(sra-volume) to query needle locations without HTTP overhead.

Protocol (per UDS_LOCATE_API_SPEC.md):
- Request: 24 bytes (fid[16] + version[4] + flags[4])
- Response: 32 bytes (status[1] + pad[3] + volume_id[4] + offset[8] + length[8] + reserved[8])

Changes:
- weed/server/volume_server_uds.go: UDS server implementation
  - Parses fid via needle.ParseFileIdFromString()
  - Looks up volume via store.GetVolume()
  - Queries needle offset via Volume.GetNeedle()
  - Returns (volumeId, offset, length) or NotFound/Error

- weed/storage/volume.go: Added GetNeedle() method
  - Exposes nm.Get() for UDS handler

- weed/command/volume.go: Added -uds.listen flag
  - Example: weed volume -dir=/data -uds.listen=/tmp/sra-volume.sock
  - UDS server starts/stops with volume lifecycle

- weed/server/volume_server_uds_test.go: 8 tests
  - Protocol size verification
  - Request/response serialization
  - Connection and fid parsing tests

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

weed/command/volume.go

@@ -74,6 +74,7 @@ type VolumeServerOptions struct {
 	ldbTimeout                  *int64
 	debug                       *bool
 	debugPort                   *int
+	udsListen                   *string // UDS socket path for RDMA sidecar integration
 }
 
 func init() {
@@ -114,6 +115,7 @@ func init() {
 	v.readBufferSizeMB = cmdVolume.Flag.Int("readBufferSizeMB", 4, "<experimental> larger values can optimize query performance but will increase some memory usage,Use with hasSlowRead normally.")
 	v.debug = cmdVolume.Flag.Bool("debug", false, "serves runtime profiling data via pprof on the port specified by -debug.port")
 	v.debugPort = cmdVolume.Flag.Int("debug.port", 6060, "http port for debugging")
+	v.udsListen = cmdVolume.Flag.String("uds.listen", "", "Unix domain socket path for RDMA sidecar locate API (e.g., /tmp/sra-volume.sock)")
 }
 
 var cmdVolume = &Command{
@@ -289,6 +291,17 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v
 	// starting grpc server
 	grpcS := v.startGrpcService(volumeServer)
 
+	// starting UDS server for RDMA sidecar integration
+	var udsServer *weed_server.UdsServer
+	if *v.udsListen != "" {
+		var err error
+		udsServer, err = weed_server.NewUdsServer(volumeServer, *v.udsListen)
+		if err != nil {
+			glog.Fatalf("failed to start UDS server: %v", err)
+		}
+		udsServer.Start()
+	}
+
 	// starting public http server
 	var publicHttpDown httpdown.Server
 	if v.isSeparatedPublicPort() {
@@ -315,7 +328,7 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v
 			time.Sleep(time.Duration(*v.preStopSeconds) * time.Second)
 		}
 
-		shutdown(publicHttpDown, clusterHttpServer, grpcS, volumeServer)
+		shutdown(publicHttpDown, clusterHttpServer, grpcS, volumeServer, udsServer)
 		stopChan <- true
 	})
 
@@ -324,7 +337,7 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v
 		select {
 		case <-stopChan:
 		case <-ctx.Done():
-			shutdown(publicHttpDown, clusterHttpServer, grpcS, volumeServer)
+			shutdown(publicHttpDown, clusterHttpServer, grpcS, volumeServer, udsServer)
 		}
 	} else {
 		select {
@@ -334,7 +347,7 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v
 
 }
 
-func shutdown(publicHttpDown httpdown.Server, clusterHttpServer httpdown.Server, grpcS *grpc.Server, volumeServer *weed_server.VolumeServer) {
+func shutdown(publicHttpDown httpdown.Server, clusterHttpServer httpdown.Server, grpcS *grpc.Server, volumeServer *weed_server.VolumeServer, udsServer *weed_server.UdsServer) {
 
 	// firstly, stop the public http service to prevent from receiving new user request
 	if nil != publicHttpDown {
@@ -352,6 +365,11 @@ func shutdown(publicHttpDown httpdown.Server, clusterHttpServer httpdown.Server,
 	glog.V(0).Infof("graceful stop gRPC ...")
 	grpcS.GracefulStop()
 
+	if udsServer != nil {
+		glog.V(0).Infof("stop UDS server ...")
+		udsServer.Stop()
+	}
+
 	volumeServer.Shutdown()
 
 	pprof.StopCPUProfile()

weed/server/volume_server_uds.go

@@ -0,0 +1,228 @@
+package weed_server
+
+import (
+	"encoding/binary"
+	"fmt"
+	"io"
+	"net"
+	"os"
+	"strings"
+	"sync"
+
+	"github.com/seaweedfs/seaweedfs/weed/glog"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+)
+
+// UDS protocol constants
+const (
+	UdsRequestSize  = 24 // fid(16) + version(4) + flags(4)
+	UdsResponseSize = 32 // status(1) + pad(3) + volume_id(4) + offset(8) + length(8) + reserved(8)
+)
+
+// UDS status codes
+const (
+	UdsStatusOk       uint8 = 0
+	UdsStatusNotFound uint8 = 1
+	UdsStatusError    uint8 = 2
+)
+
+// UdsServer handles UDS locate requests for RDMA sidecar integration
+type UdsServer struct {
+	vs       *VolumeServer
+	listener net.Listener
+	stopChan chan struct{}
+	wg       sync.WaitGroup
+}
+
+// LocateRequest represents a UDS locate request
+type LocateRequest struct {
+	Fid     [16]byte // ASCII fid, null-padded
+	Version uint32
+	Flags   uint32
+}
+
+// LocateResponse represents a UDS locate response
+type LocateResponse struct {
+	Status   uint8
+	_pad     [3]byte
+	VolumeId uint32
+	Offset   uint64
+	Length   uint64
+	_        [8]byte // reserved
+}
+
+// NewUdsServer creates a new UDS server for the volume server
+func NewUdsServer(vs *VolumeServer, socketPath string) (*UdsServer, error) {
+	// Remove existing socket file if exists
+	if err := os.Remove(socketPath); err != nil && !os.IsNotExist(err) {
+		return nil, fmt.Errorf("failed to remove existing socket: %w", err)
+	}
+
+	listener, err := net.Listen("unix", socketPath)
+	if err != nil {
+		return nil, fmt.Errorf("failed to listen on %s: %w", socketPath, err)
+	}
+
+	// Set socket permissions to allow access
+	if err := os.Chmod(socketPath, 0666); err != nil {
+		listener.Close()
+		return nil, fmt.Errorf("failed to chmod socket: %w", err)
+	}
+
+	uds := &UdsServer{
+		vs:       vs,
+		listener: listener,
+		stopChan: make(chan struct{}),
+	}
+
+	glog.V(0).Infof("UDS server listening on %s", socketPath)
+	return uds, nil
+}
+
+// Start begins accepting connections
+func (u *UdsServer) Start() {
+	u.wg.Add(1)
+	go func() {
+		defer u.wg.Done()
+		u.acceptLoop()
+	}()
+}
+
+// Stop gracefully shuts down the UDS server
+func (u *UdsServer) Stop() {
+	close(u.stopChan)
+	u.listener.Close()
+	u.wg.Wait()
+	glog.V(0).Infoln("UDS server stopped")
+}
+
+func (u *UdsServer) acceptLoop() {
+	for {
+		conn, err := u.listener.Accept()
+		if err != nil {
+			select {
+			case <-u.stopChan:
+				return
+			default:
+				glog.V(0).Infof("UDS accept error: %v", err)
+				continue
+			}
+		}
+
+		u.wg.Add(1)
+		go func(c net.Conn) {
+			defer u.wg.Done()
+			u.handleConnection(c)
+		}(conn)
+	}
+}
+
+func (u *UdsServer) handleConnection(conn net.Conn) {
+	defer conn.Close()
+
+	reqBuf := make([]byte, UdsRequestSize)
+	respBuf := make([]byte, UdsResponseSize)
+
+	for {
+		// Read request
+		_, err := io.ReadFull(conn, reqBuf)
+		if err != nil {
+			if err != io.EOF {
+				glog.V(2).Infof("UDS read error: %v", err)
+			}
+			return
+		}
+
+		// Parse request
+		var req LocateRequest
+		copy(req.Fid[:], reqBuf[0:16])
+		req.Version = binary.LittleEndian.Uint32(reqBuf[16:20])
+		req.Flags = binary.LittleEndian.Uint32(reqBuf[20:24])
+
+		// Handle request
+		resp := u.handleLocate(&req)
+
+		// Serialize response
+		respBuf[0] = resp.Status
+		respBuf[1] = 0
+		respBuf[2] = 0
+		respBuf[3] = 0
+		binary.LittleEndian.PutUint32(respBuf[4:8], resp.VolumeId)
+		binary.LittleEndian.PutUint64(respBuf[8:16], resp.Offset)
+		binary.LittleEndian.PutUint64(respBuf[16:24], resp.Length)
+		// reserved bytes 24-32 are zero
+
+		// Write response
+		if _, err := conn.Write(respBuf); err != nil {
+			glog.V(2).Infof("UDS write error: %v", err)
+			return
+		}
+	}
+}
+
+func (u *UdsServer) handleLocate(req *LocateRequest) *LocateResponse {
+	resp := &LocateResponse{}
+
+	// Extract fid string (null-terminated)
+	fidBytes := req.Fid[:]
+	fidLen := 0
+	for i, b := range fidBytes {
+		if b == 0 {
+			fidLen = i
+			break
+		}
+		fidLen = i + 1
+	}
+	fid := string(fidBytes[:fidLen])
+	fid = strings.TrimSpace(fid)
+
+	if fid == "" {
+		resp.Status = UdsStatusError
+		return resp
+	}
+
+	// Parse fid (format: "volumeId,needleIdCookie" e.g., "3,01637037")
+	fileId, err := needle.ParseFileIdFromString(fid)
+	if err != nil {
+		glog.V(2).Infof("UDS: invalid fid %q: %v", fid, err)
+		resp.Status = UdsStatusError
+		return resp
+	}
+
+	volumeId := fileId.VolumeId
+	needleId := fileId.Key
+
+	// Get volume
+	v := u.vs.store.GetVolume(volumeId)
+	if v == nil {
+		glog.V(2).Infof("UDS: volume %d not found for fid %s", volumeId, fid)
+		resp.Status = UdsStatusNotFound
+		return resp
+	}
+
+	// Lookup needle in the volume's needle map
+	// We need to access the needle map through the volume
+	nv, ok := v.GetNeedle(needleId)
+	if !ok || nv.Offset.IsZero() {
+		glog.V(3).Infof("UDS: needle %s not found in volume %d", fid, volumeId)
+		resp.Status = UdsStatusNotFound
+		return resp
+	}
+
+	// Check if deleted
+	if nv.Size.IsDeleted() {
+		glog.V(3).Infof("UDS: needle %s is deleted", fid)
+		resp.Status = UdsStatusNotFound
+		return resp
+	}
+
+	// Return the needle header offset (not data offset)
+	// The reader will parse the header to find DataSize
+	resp.Status = UdsStatusOk
+	resp.VolumeId = uint32(volumeId)
+	resp.Offset = uint64(nv.Offset.ToActualOffset())
+	resp.Length = uint64(nv.Size)
+
+	glog.V(3).Infof("UDS: located %s -> vol=%d offset=%d size=%d", fid, volumeId, resp.Offset, resp.Length)
+	return resp
+}

weed/server/volume_server_uds_test.go

@@ -0,0 +1,225 @@
+package weed_server
+
+import (
+	"encoding/binary"
+	"io"
+	"net"
+	"os"
+	"path/filepath"
+	"testing"
+	"time"
+)
+
+func TestLocateRequestResponseSize(t *testing.T) {
+	// Verify protocol sizes match spec
+	if UdsRequestSize != 24 {
+		t.Errorf("UdsRequestSize expected 24, got %d", UdsRequestSize)
+	}
+	if UdsResponseSize != 32 {
+		t.Errorf("UdsResponseSize expected 32, got %d", UdsResponseSize)
+	}
+}
+
+func TestLocateRequestSerialization(t *testing.T) {
+	req := LocateRequest{
+		Version: 1,
+		Flags:   0,
+	}
+	copy(req.Fid[:], "3,01637037")
+
+	buf := make([]byte, UdsRequestSize)
+	copy(buf[0:16], req.Fid[:])
+	binary.LittleEndian.PutUint32(buf[16:20], req.Version)
+	binary.LittleEndian.PutUint32(buf[20:24], req.Flags)
+
+	// Verify fid is correctly placed
+	fid := string(buf[0:10])
+	if fid != "3,01637037" {
+		t.Errorf("Expected fid='3,01637037', got '%s'", fid)
+	}
+
+	// Verify version
+	version := binary.LittleEndian.Uint32(buf[16:20])
+	if version != 1 {
+		t.Errorf("Expected version=1, got %d", version)
+	}
+}
+
+func TestLocateResponseSerialization(t *testing.T) {
+	resp := &LocateResponse{
+		Status:   UdsStatusOk,
+		VolumeId: 3,
+		Offset:   1024,
+		Length:   4096,
+	}
+
+	buf := make([]byte, UdsResponseSize)
+	buf[0] = resp.Status
+	binary.LittleEndian.PutUint32(buf[4:8], resp.VolumeId)
+	binary.LittleEndian.PutUint64(buf[8:16], resp.Offset)
+	binary.LittleEndian.PutUint64(buf[16:24], resp.Length)
+
+	// Deserialize and verify
+	status := buf[0]
+	if status != UdsStatusOk {
+		t.Errorf("Expected status=0, got %d", status)
+	}
+
+	volumeId := binary.LittleEndian.Uint32(buf[4:8])
+	if volumeId != 3 {
+		t.Errorf("Expected volumeId=3, got %d", volumeId)
+	}
+
+	offset := binary.LittleEndian.Uint64(buf[8:16])
+	if offset != 1024 {
+		t.Errorf("Expected offset=1024, got %d", offset)
+	}
+
+	length := binary.LittleEndian.Uint64(buf[16:24])
+	if length != 4096 {
+		t.Errorf("Expected length=4096, got %d", length)
+	}
+}
+
+func TestUdsStatusCodes(t *testing.T) {
+	if UdsStatusOk != 0 {
+		t.Errorf("UdsStatusOk expected 0, got %d", UdsStatusOk)
+	}
+	if UdsStatusNotFound != 1 {
+		t.Errorf("UdsStatusNotFound expected 1, got %d", UdsStatusNotFound)
+	}
+	if UdsStatusError != 2 {
+		t.Errorf("UdsStatusError expected 2, got %d", UdsStatusError)
+	}
+}
+
+// TestUdsServerBasicConnection tests that we can create and connect to a UDS server
+func TestUdsServerBasicConnection(t *testing.T) {
+	// Create temp socket path
+	tmpDir := t.TempDir()
+	socketPath := filepath.Join(tmpDir, "test.sock")
+
+	// Create a mock volume server (nil is fine since we won't make real lookups)
+	uds, err := NewUdsServer(nil, socketPath)
+	if err != nil {
+		t.Fatalf("Failed to create UDS server: %v", err)
+	}
+	defer uds.Stop()
+
+	uds.Start()
+
+	// Give server time to start
+	time.Sleep(10 * time.Millisecond)
+
+	// Try to connect
+	conn, err := net.Dial("unix", socketPath)
+	if err != nil {
+		t.Fatalf("Failed to connect to UDS server: %v", err)
+	}
+	defer conn.Close()
+
+	// Connection successful
+}
+
+// TestUdsServerProtocol tests the basic protocol exchange
+func TestUdsServerProtocol(t *testing.T) {
+	// Create temp socket path
+	tmpDir := t.TempDir()
+	socketPath := filepath.Join(tmpDir, "test.sock")
+
+	// Create a mock volume server (vs=nil means handleLocate will return NotFound)
+	uds, err := NewUdsServer(nil, socketPath)
+	if err != nil {
+		t.Fatalf("Failed to create UDS server: %v", err)
+	}
+	defer uds.Stop()
+
+	uds.Start()
+	time.Sleep(10 * time.Millisecond)
+
+	// Connect
+	conn, err := net.Dial("unix", socketPath)
+	if err != nil {
+		t.Fatalf("Failed to connect: %v", err)
+	}
+	defer conn.Close()
+
+	// Send a request (will fail because vs is nil, but we're testing protocol)
+	reqBuf := make([]byte, UdsRequestSize)
+	copy(reqBuf[0:16], "3,01637037")
+	binary.LittleEndian.PutUint32(reqBuf[16:20], 1) // version
+	binary.LittleEndian.PutUint32(reqBuf[20:24], 0) // flags
+
+	_, err = conn.Write(reqBuf)
+	if err != nil {
+		t.Fatalf("Failed to write request: %v", err)
+	}
+
+	// Read response
+	respBuf := make([]byte, UdsResponseSize)
+	_, err = io.ReadFull(conn, respBuf)
+	if err != nil {
+		t.Fatalf("Failed to read response: %v", err)
+	}
+
+	// Since vs is nil, we expect an error status (handleLocate will panic or return error)
+	// In this case, we just check we got a valid response back
+	status := respBuf[0]
+	if status != UdsStatusOk && status != UdsStatusNotFound && status != UdsStatusError {
+		t.Errorf("Unexpected status: %d", status)
+	}
+}
+
+// TestUdsSocketCleanup tests that existing socket is removed
+func TestUdsSocketCleanup(t *testing.T) {
+	tmpDir := t.TempDir()
+	socketPath := filepath.Join(tmpDir, "test.sock")
+
+	// Create a dummy file at the socket path
+	if err := os.WriteFile(socketPath, []byte("dummy"), 0644); err != nil {
+		t.Fatalf("Failed to create dummy file: %v", err)
+	}
+
+	// NewUdsServer should remove the existing file
+	uds, err := NewUdsServer(nil, socketPath)
+	if err != nil {
+		t.Fatalf("Failed to create UDS server (should have cleaned up old socket): %v", err)
+	}
+	uds.Stop()
+}
+
+func TestFidParsing(t *testing.T) {
+	tests := []struct {
+		fid    string
+		valid  bool
+	}{
+		{"3,01637037", true},
+		{"1,abc123", true},
+		{"", false},
+		{"invalid", false},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.fid, func(t *testing.T) {
+			req := &LocateRequest{}
+			copy(req.Fid[:], tt.fid)
+
+			// Extract fid string using same logic as handleLocate
+			fidBytes := req.Fid[:]
+			fidLen := 0
+			for i, b := range fidBytes {
+				if b == 0 {
+					fidLen = i
+					break
+				}
+				fidLen = i + 1
+			}
+			fid := string(fidBytes[:fidLen])
+
+			// For empty fid, we expect empty string
+			if tt.fid == "" && fid != "" {
+				t.Errorf("Expected empty fid, got '%s'", fid)
+			}
+		})
+	}
+}

weed/storage/volume.go

@@ -12,6 +12,7 @@ import (
 	"github.com/seaweedfs/seaweedfs/weed/stats"
 	"github.com/seaweedfs/seaweedfs/weed/storage/backend"
 	"github.com/seaweedfs/seaweedfs/weed/storage/needle"
+	"github.com/seaweedfs/seaweedfs/weed/storage/needle_map"
 	"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
 	"github.com/seaweedfs/seaweedfs/weed/storage/types"
 
@@ -203,6 +204,17 @@ func (v *Volume) IndexFileSize() uint64 {
 	return v.nm.IndexFileSize()
 }
 
+// GetNeedle looks up a needle by its ID and returns the needle value containing offset and size.
+// This is used by the UDS handler for RDMA sidecar integration.
+func (v *Volume) GetNeedle(needleId types.NeedleId) (*needle_map.NeedleValue, bool) {
+	v.dataFileAccessLock.RLock()
+	defer v.dataFileAccessLock.RUnlock()
+	if v.nm == nil {
+		return nil, false
+	}
+	return v.nm.Get(needleId)
+}
+
 func (v *Volume) DiskType() types.DiskType {
 	return v.location.DiskType
 }