diff --git a/weed/command/server.go b/weed/command/server.go index 52a30b3d1..9bcc39200 100644 --- a/weed/command/server.go +++ b/weed/command/server.go @@ -153,6 +153,7 @@ func init() { serverOptions.v.inflightDownloadDataTimeout = cmdServer.Flag.Duration("volume.inflightDownloadDataTimeout", 60*time.Second, "inflight download data wait timeout of volume servers") serverOptions.v.udsListen = cmdServer.Flag.String("volume.uds.listen", "", "Unix domain socket path for RDMA sidecar locate API (e.g., /tmp/sra-volume.sock)") + serverOptions.v.udsTransport = cmdServer.Flag.String("volume.uds.transport", "", "Unix domain socket path for outbound RDMA replication (e.g., /tmp/sra-transport.sock)") serverOptions.v.hasSlowRead = cmdServer.Flag.Bool("volume.hasSlowRead", true, " if true, this prevents slow reads from blocking other requests, but large file read P99 latency will increase.") serverOptions.v.readBufferSizeMB = cmdServer.Flag.Int("volume.readBufferSizeMB", 4, " larger values can optimize query performance but will increase some memory usage,Use with hasSlowRead normally") diff --git a/weed/command/volume.go b/weed/command/volume.go index ef0aa2d30..1b021deef 100644 --- a/weed/command/volume.go +++ b/weed/command/volume.go @@ -75,6 +75,7 @@ type VolumeServerOptions struct { debug *bool debugPort *int udsListen *string // UDS socket path for RDMA sidecar integration + udsTransport *string // UDS socket path for outbound RDMA replication } func init() { @@ -116,6 +117,7 @@ func init() { 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)") + v.udsTransport = cmdVolume.Flag.String("uds.transport", "", "Unix domain socket path for outbound RDMA replication (e.g., /tmp/sra-transport.sock)") } var cmdVolume = &Command{ @@ -302,6 +304,11 @@ func (v VolumeServerOptions) startVolumeServer(volumeFolders, maxVolumeCounts, v udsServer.Start() } + // set up outbound RDMA replication transport + if *v.udsTransport != "" { + volumeServer.SetSraTransport(storage.NewSraTransport(*v.udsTransport)) + } + // starting public http server var publicHttpDown httpdown.Server if v.isSeparatedPublicPort() { diff --git a/weed/server/volume_server.go b/weed/server/volume_server.go index 6bc3a6898..c840175e8 100644 --- a/weed/server/volume_server.go +++ b/weed/server/volume_server.go @@ -159,8 +159,16 @@ func (vs *VolumeServer) LoadNewVolumes() { vs.store.LoadNewVolumes() } +// SetSraTransport configures the outbound RDMA replication transport on the volume store. +func (vs *VolumeServer) SetSraTransport(t *storage.SraTransport) { + vs.store.SraTransport = t +} + func (vs *VolumeServer) Shutdown() { glog.V(0).Infoln("Shutting down volume server...") + if vs.store.SraTransport != nil { + vs.store.SraTransport.Close() + } vs.store.Close() glog.V(0).Infoln("Shut down successfully!") } diff --git a/weed/server/volume_server_uds.go b/weed/server/volume_server_uds.go index 46dec568c..4ba68d684 100644 --- a/weed/server/volume_server_uds.go +++ b/weed/server/volume_server_uds.go @@ -11,13 +11,28 @@ import ( "github.com/seaweedfs/seaweedfs/weed/glog" "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/types" ) // UDS protocol constants const ( - UdsRequestSize = 24 // fid(16) + version(4) + flags(4) + UdsRequestSize = 24 // opcode(1) + pad(3) + request_id(4) + fid(16) UdsResponseSize = 32 // status(1) + pad(3) + volume_id(4) + offset(8) + length(8) + dat_path_len(2) + reserved(6) UdsMaxDatPathLen = 256 // max .dat file path length + + // Store request: opcode(1) + pad(3) + request_id(4) + volume_id(4) + needle_version(1) + pad(3) + raw_bytes_len(4) = 20 bytes header + UdsStoreHeaderSize = 20 + // Store response: status(1) + pad(3) + request_id(4) = 8 bytes + UdsStoreResponseSize = 8 + + // Maximum raw needle size for Store (256 MB) + UdsMaxRawBytesLen = 256 * 1024 * 1024 +) + +// UDS opcodes +const ( + UdsOpcodeLocate uint8 = 0x01 + UdsOpcodeStore uint8 = 0x02 ) // UDS status codes @@ -57,6 +72,28 @@ type LocateResponse struct { DatPath string // variable-length .dat file path (not in wire header) } +// StoreRequest represents a UDS store request for volume replication. +// Wire format: +// +// opcode(1) + pad(3) + request_id(4) + volume_id(4) + +// needle_version(1) + pad(3) + raw_bytes_len(4) = 20 bytes header +// + raw_bytes(variable) +type StoreRequest struct { + Opcode uint8 + RequestId uint32 + VolumeId uint32 + NeedleVersion uint8 + RawBytesLen uint32 + RawBytes []byte +} + +// StoreResponse represents a UDS store response. +// Wire format: status(1) + pad(3) + request_id(4) = 8 bytes +type StoreResponse struct { + Status uint8 + RequestId uint32 +} + // NewUdsServer creates a new UDS server for the volume server func NewUdsServer(vs *VolumeServer, socketPath string) (*UdsServer, error) { // Remove existing socket file if exists @@ -126,12 +163,16 @@ func (u *UdsServer) acceptLoop() { func (u *UdsServer) handleConnection(conn net.Conn) { defer conn.Close() - reqBuf := make([]byte, UdsRequestSize) - respBuf := make([]byte, UdsResponseSize) + // Common prefix: opcode(1) + pad(3) + request_id(4) = 8 bytes + commonBuf := make([]byte, 8) + locateRestBuf := make([]byte, UdsRequestSize-8) // remaining 16 bytes for Locate + locateRespBuf := make([]byte, UdsResponseSize) + storeRestBuf := make([]byte, UdsStoreHeaderSize-8) // remaining 12 bytes for Store header + storeRespBuf := make([]byte, UdsStoreResponseSize) for { - // Read request - _, err := io.ReadFull(conn, reqBuf) + // Read common prefix (8 bytes) + _, err := io.ReadFull(conn, commonBuf) if err != nil { if err != io.EOF { glog.V(2).Infof("UDS read error: %v", err) @@ -139,43 +180,103 @@ func (u *UdsServer) handleConnection(conn net.Conn) { return } - // Parse request (matches sra-common::uds_proto::LocateRequest repr(C)) - // opcode(1) + pad(3) + request_id(4) + fid(16) = 24 bytes - var req LocateRequest - req.Opcode = reqBuf[0] - req.RequestId = binary.LittleEndian.Uint32(reqBuf[4:8]) - copy(req.Fid[:], reqBuf[8:24]) + opcode := commonBuf[0] + requestId := binary.LittleEndian.Uint32(commonBuf[4:8]) - // Handle request - resp := u.handleLocate(&req) - - // Serialize response header (32 bytes) - 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) - binary.LittleEndian.PutUint16(respBuf[24:26], resp.DatPathLen) - // reserved bytes 26-32 are zero - - // Write fixed header - if _, err := conn.Write(respBuf); err != nil { - glog.V(2).Infof("UDS write error: %v", err) - return - } - - // Write variable-length .dat path if present - if resp.DatPathLen > 0 { - if _, err := conn.Write([]byte(resp.DatPath)); err != nil { - glog.V(2).Infof("UDS write dat path error: %v", err) + switch opcode { + case UdsOpcodeLocate: + // Read remaining 16 bytes (fid) + if _, err := io.ReadFull(conn, locateRestBuf); err != nil { + glog.V(2).Infof("UDS read locate body: %v", err) return } + + var req LocateRequest + req.Opcode = opcode + req.RequestId = requestId + copy(req.Fid[:], locateRestBuf) + + resp := u.handleLocate(&req) + + // Serialize Locate response (32 bytes) + locateRespBuf[0] = resp.Status + locateRespBuf[1] = 0 + locateRespBuf[2] = 0 + locateRespBuf[3] = 0 + binary.LittleEndian.PutUint32(locateRespBuf[4:8], resp.VolumeId) + binary.LittleEndian.PutUint64(locateRespBuf[8:16], resp.Offset) + binary.LittleEndian.PutUint64(locateRespBuf[16:24], resp.Length) + binary.LittleEndian.PutUint16(locateRespBuf[24:26], resp.DatPathLen) + // reserved bytes 26-32 are zero + for i := 26; i < 32; i++ { + locateRespBuf[i] = 0 + } + + if _, err := conn.Write(locateRespBuf); err != nil { + glog.V(2).Infof("UDS write error: %v", err) + return + } + if resp.DatPathLen > 0 { + if _, err := conn.Write([]byte(resp.DatPath)); err != nil { + glog.V(2).Infof("UDS write dat path error: %v", err) + return + } + } + + case UdsOpcodeStore: + // Read remaining 12 bytes of Store header + if _, err := io.ReadFull(conn, storeRestBuf); err != nil { + glog.V(2).Infof("UDS read store header: %v", err) + return + } + + req := StoreRequest{ + Opcode: opcode, + RequestId: requestId, + VolumeId: binary.LittleEndian.Uint32(storeRestBuf[0:4]), + NeedleVersion: storeRestBuf[4], + RawBytesLen: binary.LittleEndian.Uint32(storeRestBuf[8:12]), + } + + // Validate raw_bytes_len + if req.RawBytesLen == 0 || req.RawBytesLen > UdsMaxRawBytesLen { + glog.V(2).Infof("UDS: invalid raw_bytes_len %d", req.RawBytesLen) + u.writeStoreResponse(conn, storeRespBuf, requestId, UdsStatusError) + return + } + + // Read raw needle bytes + req.RawBytes = make([]byte, req.RawBytesLen) + if _, err := io.ReadFull(conn, req.RawBytes); err != nil { + glog.V(2).Infof("UDS read store payload: %v", err) + return + } + + resp := u.handleStore(&req) + if err := u.writeStoreResponse(conn, storeRespBuf, resp.RequestId, resp.Status); err != nil { + return + } + + default: + glog.V(0).Infof("UDS: unknown opcode 0x%02x from connection", opcode) + return } } } +func (u *UdsServer) writeStoreResponse(conn net.Conn, buf []byte, requestId uint32, status uint8) error { + buf[0] = status + buf[1] = 0 + buf[2] = 0 + buf[3] = 0 + binary.LittleEndian.PutUint32(buf[4:8], requestId) + if _, err := conn.Write(buf); err != nil { + glog.V(2).Infof("UDS write store response error: %v", err) + return err + } + return nil +} + func (u *UdsServer) handleLocate(req *LocateRequest) *LocateResponse { resp := &LocateResponse{} @@ -255,3 +356,36 @@ func (u *UdsServer) handleLocate(req *LocateRequest) *LocateResponse { glog.V(3).Infof("UDS: located %s -> vol=%d offset=%d size=%d dat=%s", fid, volumeId, resp.Offset, resp.Length, datPath) return resp } + +func (u *UdsServer) handleStore(req *StoreRequest) *StoreResponse { + resp := &StoreResponse{RequestId: req.RequestId} + + if u.vs == nil { + resp.Status = UdsStatusError + return resp + } + + // Validate raw bytes contain at least a needle header (Cookie + NeedleId + Size = 16 bytes) + if len(req.RawBytes) < types.NeedleHeaderSize { + glog.V(2).Infof("UDS store: raw_bytes too short (%d < %d)", len(req.RawBytes), types.NeedleHeaderSize) + resp.Status = UdsStatusError + return resp + } + + // Parse needle header to extract needle ID and size + var n needle.Needle + n.ParseNeedleHeader(req.RawBytes[:types.NeedleHeaderSize]) + + vid := needle.VolumeId(req.VolumeId) + + // Use the store's WriteNeedleBlob to append raw bytes and update the needle map + if err := u.vs.store.WriteVolumeNeedleBlob(vid, n.Id, req.RawBytes, n.Size); err != nil { + glog.V(2).Infof("UDS store: write failed for vol=%d needle=%v: %v", req.VolumeId, n.Id, err) + resp.Status = UdsStatusError + return resp + } + + glog.V(3).Infof("UDS: stored vol=%d needle=%v size=%d raw_len=%d", req.VolumeId, n.Id, n.Size, len(req.RawBytes)) + resp.Status = UdsStatusOk + return resp +} diff --git a/weed/server/volume_server_uds_integration_test.go b/weed/server/volume_server_uds_integration_test.go new file mode 100644 index 000000000..4f240230a --- /dev/null +++ b/weed/server/volume_server_uds_integration_test.go @@ -0,0 +1,297 @@ +package weed_server + +import ( + "encoding/binary" + "fmt" + "io" + "net" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/seaweedfs/seaweedfs/weed/storage" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" + "github.com/seaweedfs/seaweedfs/weed/storage/types" + "github.com/seaweedfs/seaweedfs/weed/util" +) + +// newTestStoreForUds creates a minimal storage.Store for UDS integration tests. +// Caller must close(done) when finished to stop the channel drain goroutine. +func newTestStoreForUds(t *testing.T) (s *storage.Store, done chan struct{}) { + t.Helper() + tmpDir := t.TempDir() + volDir := filepath.Join(tmpDir, "vol") + if err := os.MkdirAll(volDir, 0755); err != nil { + t.Fatalf("MkdirAll: %v", err) + } + + s = storage.NewStore(nil, "localhost", 8080, 18080, "http://localhost:8080", "", + []string{volDir}, []int32{10}, []util.MinFreeSpace{{}}, "", + storage.NeedleMapInMemory, []types.DiskType{types.HardDriveType}, 3) + + // Drain NewVolumesChan to prevent AddVolume from blocking + done = make(chan struct{}) + go func() { + for { + select { + case <-s.NewVolumesChan: + case <-done: + return + } + } + }() + + return s, done +} + +// TestUdsIntegration_LocateWrittenNeedle is an end-to-end test: +// 1. Create a volume store and write a needle +// 2. Start a UDS server backed by a VolumeServer +// 3. Send a Locate request over UDS +// 4. Verify the response contains correct offset, length, and dat_path +func TestUdsIntegration_LocateWrittenNeedle(t *testing.T) { + store, done := newTestStoreForUds(t) + defer close(done) + + vid := needle.VolumeId(1) + err := store.AddVolume(vid, "", storage.NeedleMapInMemory, "000", "", + 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 3) + if err != nil { + t.Fatalf("AddVolume: %v", err) + } + + // Write a needle + n := new(needle.Needle) + n.Id = types.Uint64ToNeedleId(0x01) + n.Cookie = types.Uint32ToCookie(0x12345678) + n.Data = []byte("hello world integration test data") + n.Checksum = needle.NewCRC(n.Data) + + _, err = store.WriteVolumeNeedle(vid, n, false, false) + if err != nil { + t.Fatalf("WriteVolumeNeedle: %v", err) + } + + // Construct the fid string the same way SeaweedFS does + fileId := needle.NewFileIdFromNeedle(vid, n) + fid := fileId.String() // e.g. "1,0112345678" + + // Create a minimal VolumeServer with just the store field + vs := &VolumeServer{ + store: store, + } + + socketPath := filepath.Join(t.TempDir(), "test.sock") + uds, err := NewUdsServer(vs, socketPath) + if err != nil { + t.Fatalf("NewUdsServer: %v", err) + } + defer uds.Stop() + uds.Start() + time.Sleep(10 * time.Millisecond) + + conn, err := net.Dial("unix", socketPath) + if err != nil { + t.Fatalf("Dial: %v", err) + } + defer conn.Close() + + // Send Locate request + reqBuf := make([]byte, UdsRequestSize) + reqBuf[0] = 1 // opcode = Locate + binary.LittleEndian.PutUint32(reqBuf[4:8], 42) + + // Copy fid into the 16-byte field (strip "vol," prefix — keep just keyCookie part? No — the UDS handler parses the full "vol,keyCookie" fid) + if len(fid) > 16 { + t.Fatalf("fid %q too long for 16-byte field", fid) + } + copy(reqBuf[8:24], fid) + + if _, err := conn.Write(reqBuf); err != nil { + t.Fatalf("Write request: %v", err) + } + + // Read 32-byte response header + respBuf := make([]byte, UdsResponseSize) + if _, err := io.ReadFull(conn, respBuf); err != nil { + t.Fatalf("Read response: %v", err) + } + + status := respBuf[0] + if status != UdsStatusOk { + t.Fatalf("Expected UdsStatusOk (0), got %d", status) + } + + volumeId := binary.LittleEndian.Uint32(respBuf[4:8]) + if volumeId != uint32(vid) { + t.Errorf("volumeId: expected %d, got %d", vid, volumeId) + } + + offset := binary.LittleEndian.Uint64(respBuf[8:16]) + if offset == 0 { + t.Error("Expected non-zero offset (super block occupies the first bytes)") + } + + length := binary.LittleEndian.Uint64(respBuf[16:24]) + if length == 0 { + t.Error("Expected non-zero length") + } + + datPathLen := binary.LittleEndian.Uint16(respBuf[24:26]) + if datPathLen == 0 { + t.Fatal("Expected non-zero dat_path_len") + } + + // Read variable-length dat_path + datPathBuf := make([]byte, datPathLen) + if _, err := io.ReadFull(conn, datPathBuf); err != nil { + t.Fatalf("Read dat_path: %v", err) + } + datPath := string(datPathBuf) + + if !strings.HasSuffix(datPath, ".dat") { + t.Errorf("dat_path should end with .dat, got %q", datPath) + } + + t.Logf("Locate %s -> offset=%d length=%d dat_path=%s", fid, offset, length, datPath) +} + +// TestUdsIntegration_LocateNotFound verifies that looking up a non-existent needle returns NotFound. +func TestUdsIntegration_LocateNotFound(t *testing.T) { + store, done := newTestStoreForUds(t) + defer close(done) + + vid := needle.VolumeId(1) + err := store.AddVolume(vid, "", storage.NeedleMapInMemory, "000", "", + 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 3) + if err != nil { + t.Fatalf("AddVolume: %v", err) + } + + vs := &VolumeServer{ + store: store, + } + + socketPath := filepath.Join(t.TempDir(), "test.sock") + uds, err := NewUdsServer(vs, socketPath) + if err != nil { + t.Fatalf("NewUdsServer: %v", err) + } + defer uds.Stop() + uds.Start() + time.Sleep(10 * time.Millisecond) + + conn, err := net.Dial("unix", socketPath) + if err != nil { + t.Fatalf("Dial: %v", err) + } + defer conn.Close() + + // Locate a needle that was never written + reqBuf := make([]byte, UdsRequestSize) + reqBuf[0] = 1 + binary.LittleEndian.PutUint32(reqBuf[4:8], 1) + copy(reqBuf[8:24], "1,ffaabbccddee") + + if _, err := conn.Write(reqBuf); err != nil { + t.Fatalf("Write: %v", err) + } + + respBuf := make([]byte, UdsResponseSize) + if _, err := io.ReadFull(conn, respBuf); err != nil { + t.Fatalf("Read: %v", err) + } + + status := respBuf[0] + if status != UdsStatusNotFound { + t.Errorf("Expected UdsStatusNotFound (%d), got %d", UdsStatusNotFound, status) + } +} + +// TestUdsIntegration_MultipleRequests verifies that multiple sequential requests on the same connection work. +func TestUdsIntegration_MultipleRequests(t *testing.T) { + store, done := newTestStoreForUds(t) + defer close(done) + + vid := needle.VolumeId(1) + err := store.AddVolume(vid, "", storage.NeedleMapInMemory, "000", "", + 0, needle.GetCurrentVersion(), 0, types.HardDriveType, 3) + if err != nil { + t.Fatalf("AddVolume: %v", err) + } + + // Write 3 needles + type needleInfo struct { + n *needle.Needle + fid string + } + needles := make([]needleInfo, 3) + for i := 0; i < 3; i++ { + n := new(needle.Needle) + n.Id = types.Uint64ToNeedleId(uint64(i + 1)) + n.Cookie = types.Uint32ToCookie(uint32(0xAABBCC00 + i)) + n.Data = []byte(fmt.Sprintf("needle data %d", i)) + n.Checksum = needle.NewCRC(n.Data) + + if _, err := store.WriteVolumeNeedle(vid, n, false, false); err != nil { + t.Fatalf("WriteVolumeNeedle[%d]: %v", i, err) + } + fileId := needle.NewFileIdFromNeedle(vid, n) + needles[i] = needleInfo{n: n, fid: fileId.String()} + } + + vs := &VolumeServer{store: store} + socketPath := filepath.Join(t.TempDir(), "test.sock") + uds, err := NewUdsServer(vs, socketPath) + if err != nil { + t.Fatalf("NewUdsServer: %v", err) + } + defer uds.Stop() + uds.Start() + time.Sleep(10 * time.Millisecond) + + conn, err := net.Dial("unix", socketPath) + if err != nil { + t.Fatalf("Dial: %v", err) + } + defer conn.Close() + + // Send 3 Locate requests sequentially on the same connection + for i, ni := range needles { + reqBuf := make([]byte, UdsRequestSize) + reqBuf[0] = 1 + binary.LittleEndian.PutUint32(reqBuf[4:8], uint32(i+1)) + copy(reqBuf[8:24], ni.fid) + + if _, err := conn.Write(reqBuf); err != nil { + t.Fatalf("Write[%d]: %v", i, err) + } + + respBuf := make([]byte, UdsResponseSize) + if _, err := io.ReadFull(conn, respBuf); err != nil { + t.Fatalf("Read[%d]: %v", i, err) + } + + status := respBuf[0] + if status != UdsStatusOk { + t.Errorf("needle[%d] fid=%s: expected UdsStatusOk, got %d", i, ni.fid, status) + continue + } + + length := binary.LittleEndian.Uint64(respBuf[16:24]) + if length == 0 { + t.Errorf("needle[%d]: expected non-zero length", i) + } + + // Consume the dat_path bytes + datPathLen := binary.LittleEndian.Uint16(respBuf[24:26]) + if datPathLen > 0 { + datPathBuf := make([]byte, datPathLen) + if _, err := io.ReadFull(conn, datPathBuf); err != nil { + t.Fatalf("Read dat_path[%d]: %v", i, err) + } + } + } +} diff --git a/weed/server/volume_server_uds_test.go b/weed/server/volume_server_uds_test.go index f984af753..52c5000c3 100644 --- a/weed/server/volume_server_uds_test.go +++ b/weed/server/volume_server_uds_test.go @@ -279,6 +279,114 @@ func TestNeedleHeaderIsBigEndian(t *testing.T) { } } +// ============================================================ +// Golden wire tests for Store opcode (0x02) +// ============================================================ + +func TestGoldenStoreRequest(t *testing.T) { + // Store request header: opcode(1) + pad(3) + request_id(4) + volume_id(4) + + // needle_version(1) + pad(3) + raw_bytes_len(4) = 20 bytes + // Values: opcode=0x02, request_id=99, volume_id=5, needle_version=2, raw_bytes_len=1024 + golden := [UdsStoreHeaderSize]byte{ + 0x02, // opcode = Store + 0x00, 0x00, 0x00, // padding + 0x63, 0x00, 0x00, 0x00, // request_id = 99 (LE) + 0x05, 0x00, 0x00, 0x00, // volume_id = 5 (LE) + 0x02, // needle_version = 2 + 0x00, 0x00, 0x00, // padding + 0x00, 0x04, 0x00, 0x00, // raw_bytes_len = 1024 (LE) + } + + // Serialize from Go + buf := make([]byte, UdsStoreHeaderSize) + buf[0] = UdsOpcodeStore + binary.LittleEndian.PutUint32(buf[4:8], 99) + binary.LittleEndian.PutUint32(buf[8:12], 5) + buf[12] = 2 // needle_version + binary.LittleEndian.PutUint32(buf[16:20], 1024) + + for i := 0; i < UdsStoreHeaderSize; i++ { + if buf[i] != golden[i] { + t.Errorf("Store request byte[%d] = 0x%02X, golden = 0x%02X", i, buf[i], golden[i]) + } + } +} + +func TestGoldenStoreResponse(t *testing.T) { + // Store response: status(1) + pad(3) + request_id(4) = 8 bytes + golden := [UdsStoreResponseSize]byte{ + 0x00, // status = OK + 0x00, 0x00, 0x00, // padding + 0x63, 0x00, 0x00, 0x00, // request_id = 99 (LE) + } + + buf := make([]byte, UdsStoreResponseSize) + buf[0] = UdsStatusOk + binary.LittleEndian.PutUint32(buf[4:8], 99) + + for i := 0; i < UdsStoreResponseSize; i++ { + if buf[i] != golden[i] { + t.Errorf("Store response byte[%d] = 0x%02X, golden = 0x%02X", i, buf[i], golden[i]) + } + } +} + +// ============================================================ +// Golden wire tests for Replicate opcode (0x03) +// These match the Rust transport_listener wire format. +// ============================================================ + +func TestGoldenReplicateRequestHeader(t *testing.T) { + // Replicate request header: opcode(1) + pad(3) + request_id(4) + volume_id(4) + target_host_len(4) = 16 bytes + // Values: opcode=0x03, request_id=7, volume_id=3, target_host_len=15 ("10.0.0.3:18515") + golden := [16]byte{ + 0x03, // opcode = Replicate + 0x00, 0x00, 0x00, // padding + 0x07, 0x00, 0x00, 0x00, // request_id = 7 (LE) + 0x03, 0x00, 0x00, 0x00, // volume_id = 3 (LE) + 0x0F, 0x00, 0x00, 0x00, // target_host_len = 15 (LE) + } + + buf := make([]byte, 16) + buf[0] = 0x03 // opcode + binary.LittleEndian.PutUint32(buf[4:8], 7) + binary.LittleEndian.PutUint32(buf[8:12], 3) + binary.LittleEndian.PutUint32(buf[12:16], 15) + + for i := 0; i < 16; i++ { + if buf[i] != golden[i] { + t.Errorf("Replicate header byte[%d] = 0x%02X, golden = 0x%02X", i, buf[i], golden[i]) + } + } + + // Verify target_host string + targetHost := "10.0.0.3:18515" + if len(targetHost) != 14 { + // Note: "10.0.0.3:18515" is 14 chars. We use 15 in golden to test the len field. + // In real usage it would be len(targetHost). + } +} + +func TestGoldenReplicateResponse(t *testing.T) { + // Replicate response: status(1) + pad(3) + request_id(4) = 8 bytes + // Same format as Store response. + golden := [8]byte{ + 0xFF, // status = ERROR (stub always returns error) + 0x00, 0x00, 0x00, // padding + 0x07, 0x00, 0x00, 0x00, // request_id = 7 (LE) + } + + buf := make([]byte, 8) + buf[0] = 0xFF // status = ERROR + binary.LittleEndian.PutUint32(buf[4:8], 7) + + for i := 0; i < 8; i++ { + if buf[i] != golden[i] { + t.Errorf("Replicate response byte[%d] = 0x%02X, golden = 0x%02X", i, buf[i], golden[i]) + } + } +} + func TestFidParsing(t *testing.T) { tests := []struct { fid string diff --git a/weed/storage/needle/needle_write.go b/weed/storage/needle/needle_write.go index 009bf393e..fe7590f8a 100644 --- a/weed/storage/needle/needle_write.go +++ b/weed/storage/needle/needle_write.go @@ -45,6 +45,48 @@ func (n *Needle) Append(w backend.BackendStorageFile, version Version) (offset u return offset, size, actualSize, err } +// AppendGetBytes is like Append but also returns a copy of the raw serialized bytes +// that were written to disk. This enables RDMA replication to send the exact .dat bytes +// to a remote volume without re-serialization. +func (n *Needle) AppendGetBytes(w backend.BackendStorageFile, version Version) (offset uint64, size Size, actualSize int64, rawBytes []byte, err error) { + end, _, e := w.GetStat() + if e != nil { + err = fmt.Errorf("Cannot Read Current Volume Position: %w", e) + return + } + offset = uint64(end) + if offset >= MaxPossibleVolumeSize && len(n.Data) != 0 { + err = fmt.Errorf("Volume Size %d Exceeded %d", offset, MaxPossibleVolumeSize) + return + } + bytesBuffer := buffer_pool.SyncPoolGetBuffer() + defer func() { + if err != nil { + if te := w.Truncate(end); te != nil { + // handle error or log + } + } + buffer_pool.SyncPoolPutBuffer(bytesBuffer) + }() + + size, actualSize, err = writeNeedleByVersion(version, n, offset, bytesBuffer) + if err != nil { + return + } + + // Copy serialized bytes before they are written to disk and the buffer is returned to pool + src := bytesBuffer.Bytes() + rawBytes = make([]byte, len(src)) + copy(rawBytes, src) + + _, err = w.WriteAt(src, int64(offset)) + if err != nil { + err = fmt.Errorf("failed to write %d bytes to %s at offset %d: %w", actualSize, w.Name(), offset, err) + } + + return offset, size, actualSize, rawBytes, err +} + func WriteNeedleBlob(w backend.BackendStorageFile, dataSlice []byte, size Size, appendAtNs uint64, version Version) (offset uint64, err error) { if end, _, e := w.GetStat(); e == nil { diff --git a/weed/storage/sra_transport.go b/weed/storage/sra_transport.go new file mode 100644 index 000000000..74ddaef11 --- /dev/null +++ b/weed/storage/sra_transport.go @@ -0,0 +1,157 @@ +package storage + +import ( + "context" + "encoding/binary" + "fmt" + "io" + "net" + "sync" + "time" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/storage/needle" +) + +// SraTransport provides outbound RDMA replication via the sra-volume sidecar. +// It connects to the sidecar's transport UDS socket and sends Replicate requests. +// Nil-safe: callers should check Store.SraTransport != nil before using. +type SraTransport struct { + socketPath string + mu sync.Mutex + conn net.Conn +} + +// Replicate request wire format (opcode 0x03): +// +// opcode(1) + pad(3) + request_id(4) + volume_id(4) + +// target_host_len(4) + target_host(N) + raw_bytes_len(4) + raw_bytes(M) +// +// Response: status(1) + pad(3) + request_id(4) = 8 bytes +const ( + sraOpcodeReplicate uint8 = 0x03 + sraReplicateHeaderSize = 16 // opcode(1)+pad(3)+request_id(4)+volume_id(4)+target_host_len(4) + sraReplicateResponseSize = 8 + sraStatusOk uint8 = 0x00 + sraStatusError uint8 = 0xFF +) + +var nextRequestId uint32 +var nextRequestIdMu sync.Mutex + +func nextReqId() uint32 { + nextRequestIdMu.Lock() + defer nextRequestIdMu.Unlock() + nextRequestId++ + return nextRequestId +} + +// NewSraTransport creates a new outbound transport client. +// Returns nil if socketPath is empty (RDMA replication disabled). +func NewSraTransport(socketPath string) *SraTransport { + if socketPath == "" { + return nil + } + return &SraTransport{ + socketPath: socketPath, + } +} + +// Replicate sends raw needle bytes to a remote volume server via the sidecar's RDMA transport. +// targetHost is the RDMA endpoint of the destination (e.g., "10.0.0.3:18515"). +// volumeId is the destination volume. rawBytes is the serialized needle data. +func (t *SraTransport) Replicate(ctx context.Context, targetHost string, volumeId needle.VolumeId, rawBytes []byte) error { + t.mu.Lock() + defer t.mu.Unlock() + + conn, err := t.getConn() + if err != nil { + return fmt.Errorf("sra transport connect: %w", err) + } + + reqId := nextReqId() + + // Build request + targetHostBytes := []byte(targetHost) + headerBuf := make([]byte, sraReplicateHeaderSize) + headerBuf[0] = sraOpcodeReplicate + // pad[1..3] = 0 + binary.LittleEndian.PutUint32(headerBuf[4:8], reqId) + binary.LittleEndian.PutUint32(headerBuf[8:12], uint32(volumeId)) + binary.LittleEndian.PutUint32(headerBuf[12:16], uint32(len(targetHostBytes))) + + rawBytesLenBuf := make([]byte, 4) + binary.LittleEndian.PutUint32(rawBytesLenBuf, uint32(len(rawBytes))) + + // Set deadline from context or default 30s + deadline, ok := ctx.Deadline() + if !ok { + deadline = time.Now().Add(30 * time.Second) + } + conn.SetDeadline(deadline) + + // Write: header + target_host + raw_bytes_len + raw_bytes + if _, err := conn.Write(headerBuf); err != nil { + t.closeConn() + return fmt.Errorf("sra transport write header: %w", err) + } + if _, err := conn.Write(targetHostBytes); err != nil { + t.closeConn() + return fmt.Errorf("sra transport write target_host: %w", err) + } + if _, err := conn.Write(rawBytesLenBuf); err != nil { + t.closeConn() + return fmt.Errorf("sra transport write raw_bytes_len: %w", err) + } + if _, err := conn.Write(rawBytes); err != nil { + t.closeConn() + return fmt.Errorf("sra transport write raw_bytes: %w", err) + } + + // Read response (8 bytes) + respBuf := make([]byte, sraReplicateResponseSize) + if _, err := io.ReadFull(conn, respBuf); err != nil { + t.closeConn() + return fmt.Errorf("sra transport read response: %w", err) + } + + status := respBuf[0] + respReqId := binary.LittleEndian.Uint32(respBuf[4:8]) + if respReqId != reqId { + t.closeConn() + return fmt.Errorf("sra transport request_id mismatch: sent %d, got %d", reqId, respReqId) + } + + if status != sraStatusOk { + return fmt.Errorf("sra transport replicate failed: status=0x%02x", status) + } + + glog.V(3).Infof("sra transport: replicated %d bytes to %s vol=%d", len(rawBytes), targetHost, volumeId) + return nil +} + +// Close closes the transport connection. +func (t *SraTransport) Close() { + t.mu.Lock() + defer t.mu.Unlock() + t.closeConn() +} + +func (t *SraTransport) getConn() (net.Conn, error) { + if t.conn != nil { + return t.conn, nil + } + conn, err := net.DialTimeout("unix", t.socketPath, 5*time.Second) + if err != nil { + return nil, err + } + t.conn = conn + return conn, nil +} + +func (t *SraTransport) closeConn() { + if t.conn != nil { + t.conn.Close() + t.conn = nil + } +} diff --git a/weed/storage/store.go b/weed/storage/store.go index 3e5bcd3cf..4b0159c38 100644 --- a/weed/storage/store.go +++ b/weed/storage/store.go @@ -78,6 +78,7 @@ type Store struct { NewEcShardsChan chan master_pb.VolumeEcShardInformationMessage DeletedEcShardsChan chan master_pb.VolumeEcShardInformationMessage isStopping bool + SraTransport *SraTransport // outbound RDMA replication transport (nil = HTTP-only) } func (s *Store) String() (str string) { @@ -578,6 +579,19 @@ func (s *Store) WriteVolumeNeedle(i needle.VolumeId, n *needle.Needle, checkCook return } +// WriteVolumeNeedleBlob appends raw needle bytes to a volume's .dat file and updates +// the needle map. Used by the UDS Store opcode for RDMA replication where the sender +// provides already-serialized needle bytes. +func (s *Store) WriteVolumeNeedleBlob(i needle.VolumeId, needleId NeedleId, needleBlob []byte, size Size) error { + if v := s.findVolume(i); v != nil { + if v.IsReadOnly() { + return fmt.Errorf("volume %d is read only", i) + } + return v.WriteNeedleBlob(needleId, needleBlob, size) + } + return fmt.Errorf("volume %d not found on %s:%d", i, s.Ip, s.Port) +} + func (s *Store) DeleteVolumeNeedle(i needle.VolumeId, n *needle.Needle) (Size, error) { if v := s.findVolume(i); v != nil { if v.noWriteOrDelete {