From 6eb574dc1a97778cf6b89016a60aa6d4c5b7ca11 Mon Sep 17 00:00:00 2001 From: Jihyeon Gim Date: Sat, 5 Sep 2026 23:58:18 +0900 Subject: [PATCH] rdma: expose live RC sessions on the admin server Add a SessionsSnapshot view over the new C ABI entry point and serve it from the admin server as GET /rc-sessions. The admin server gains a WithAdminRoute option so an embedding binary can register extra admin routes that run with the same signature verification and admin checks as the built-in endpoints; vgwrdma registers the snapshot there when the RC feature is enabled. The route replies with the usual XML error surface so unsigned or non-admin requests get a 403 rather than a generic 500. Stub builds return a not supported error, keeping the build matrix unchanged. --- cmd/vgwrdma/main.go | 34 +++++++++++ embedgw/embedgw.go | 6 ++ rdma/rcroutes/admin_linux.go | 69 ++++++++++++++++++++++ rdma/rcroutes/admin_stub.go | 33 +++++++++++ rdma/rcserver/rcserver_cgo_linux_test.go | 15 +++++ rdma/rcserver/rcserver_linux.go | 73 ++++++++++++++++++++++++ rdma/rcserver/rcserver_stub.go | 26 +++++++++ s3api/admin-server.go | 32 +++++++++++ 8 files changed, 288 insertions(+) create mode 100644 rdma/rcroutes/admin_linux.go create mode 100644 rdma/rcroutes/admin_stub.go diff --git a/cmd/vgwrdma/main.go b/cmd/vgwrdma/main.go index 1779c312..e5d3bf4d 100644 --- a/cmd/vgwrdma/main.go +++ b/cmd/vgwrdma/main.go @@ -35,11 +35,14 @@ import ( "github.com/versity/versitygw/embedgw" "github.com/versity/versitygw/internal/netutil" "github.com/versity/versitygw/internal/rdmamode" + "github.com/versity/versitygw/metrics" "github.com/versity/versitygw/rdma" "github.com/versity/versitygw/rdma/rcroutes" "github.com/versity/versitygw/rdma/rcserver" "github.com/versity/versitygw/s3api" "github.com/versity/versitygw/s3api/middlewares" + "github.com/versity/versitygw/s3api/utils" + "github.com/versity/versitygw/s3err" ) var ( @@ -1296,6 +1299,37 @@ func runGateway(ctx context.Context, be backend.Backend) error { s3api.WithRoute("POST", "/.hipobj-rc/ready", rcAuth, rcH.Ready), s3api.WithRoute("POST", "/.hipobj-rc/cancel", rcAuth, rcH.Cancel), ) + // The RC session snapshot rides the standalone admin server + // (started when --admin-port is given) behind the same + // signature verification and admin role check as the other + // admin endpoints. + rcAdminAuth := func(ctx fiber.Ctx) error { + if err := rcVerify(ctx); err != nil { + return err + } + if err := middlewares.IsAdmin(metrics.ActionAdminListBuckets)(ctx); err != nil { + return err + } + return nil + } + // Admin routes answer with the same XML error surface as the + // built-in admin endpoints; the default fiber error handler + // would turn auth failures into a bare 500. + rcAdminRoute := func(ctx fiber.Ctx) error { + err := rcAdminAuth(ctx) + if err == nil { + err = rcH.AdminSnapshot(ctx) + } + if serr, ok := err.(s3err.S3Error); ok { + requestID, hostID := utils.EnsureRequestIDs(ctx) + return ctx.Status(serr.StatusCode()).Send( + serr.XMLBody(requestID, hostID)) + } + return err + } + cfg.AdminOptions = append(cfg.AdminOptions, + s3api.WithAdminRoute("GET", "/rc-sessions", rcAdminRoute, rcH.AdminSnapshot), + ) } else { cfg.S3Options = s3Opts } diff --git a/embedgw/embedgw.go b/embedgw/embedgw.go index cd227560..f7c6daa9 100644 --- a/embedgw/embedgw.go +++ b/embedgw/embedgw.go @@ -75,6 +75,11 @@ type Config struct { // control over the admin endpoint with optionally separate TLS certs. AdminPorts []string + // AdminOptions carries extra standalone-admin-server options from + // the embedding binary (e.g. additional admin routes). Only used + // when AdminPorts is non-empty. + AdminOptions []s3api.AdminOpt + // MaxConnections is the maximum number of concurrent TCP connections // accepted by the S3 API server. MaxConnections int @@ -873,6 +878,7 @@ func RunVersityGW(ctx context.Context, be backend.Backend, cfg *Config) error { admOpts := []s3api.AdminOpt{ s3api.WithAdminConcurrencyLimiter(cfg.AdminMaxConnections, cfg.AdminMaxRequests), } + admOpts = append(admOpts, cfg.AdminOptions...) if corsAllowOrigin != "" { admOpts = append(admOpts, s3api.WithAdminCORSAllowOrigin(corsAllowOrigin)) diff --git a/rdma/rcroutes/admin_linux.go b/rdma/rcroutes/admin_linux.go new file mode 100644 index 00000000..eb60f23e --- /dev/null +++ b/rdma/rcroutes/admin_linux.go @@ -0,0 +1,69 @@ +// Copyright 2026 Versity Software +// Copyright 2026 Gluesys Inc. and Jihyeon Gim +// This file is licensed under the Apache License, Version 2.0 +// (the "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Admin session snapshot route for the RC data plane. +//go:build linux && amd64 && cgo + +package rcroutes + +import ( + "github.com/gofiber/fiber/v3" + + "github.com/versity/versitygw/rdma/rcserver" +) + +// AdminSnapshot serves the RC session snapshot on the admin surface. +// The route is registered only when the RC plane is enabled and the +// standalone admin server runs; middleware before it has already +// verified the admin signature and role. +func (h *Handler) AdminSnapshot(ctx fiber.Ctx) error { + sessions, err := h.svc.SessionsSnapshot() + if err != nil { + return WriteRouteError(ctx, mapRcError(err)) + } + type sessionRecord struct { + SessionID string `json:"session_id"` + Op string `json:"op"` + Target string `json:"target"` + State string `json:"state"` + ReapPending bool `json:"reap_pending"` + AgeMs uint64 `json:"age_ms"` + StagingBytes uint64 `json:"staging_bytes"` + } + stateNames := map[uint8]string{ + rcserver.SnapshotStatePrepared: "prepared", + rcserver.SnapshotStatePublishing: "publishing", + rcserver.SnapshotStateTransferring: "transferring", + rcserver.SnapshotStateCompleting: "completing", + rcserver.SnapshotStateReaping: "reaping", + } + out := make([]sessionRecord, 0, len(sessions)) + for _, s := range sessions { + name, ok := stateNames[s.State] + if !ok { + name = "unknown" + } + out = append(out, sessionRecord{ + SessionID: s.SessionID, + Op: s.Op, + Target: s.Target, + State: name, + ReapPending: s.ReapPending, + AgeMs: s.AgeMs, + StagingBytes: s.StagingBytes, + }) + } + return ctx.JSON(fiber.Map{"sessions": out}) +} diff --git a/rdma/rcroutes/admin_stub.go b/rdma/rcroutes/admin_stub.go new file mode 100644 index 00000000..b6d2f742 --- /dev/null +++ b/rdma/rcroutes/admin_stub.go @@ -0,0 +1,33 @@ +// Copyright 2026 Versity Software +// Copyright 2026 Gluesys Inc. and Jihyeon Gim +// This file is licensed under the Apache License, Version 2.0 +// (the "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Admin session snapshot route (stub for platforms without RDMA +// support). +//go:build !(linux && amd64 && cgo) + +package rcroutes + +import ( + "errors" + + "github.com/gofiber/fiber/v3" +) + +// AdminSnapshot is a stub; the route is never registered without the +// RC data plane. +func (h *Handler) AdminSnapshot(ctx fiber.Ctx) error { + return fiber.NewError(fiber.StatusNotFound, + errors.New("rdma rc data plane not available").Error()) +} diff --git a/rdma/rcserver/rcserver_cgo_linux_test.go b/rdma/rcserver/rcserver_cgo_linux_test.go index 6df4e6a9..7a0c3c52 100644 --- a/rdma/rcserver/rcserver_cgo_linux_test.go +++ b/rdma/rcserver/rcserver_cgo_linux_test.go @@ -1,3 +1,18 @@ +// Copyright 2026 Versity Software +// Copyright 2026 Gluesys Inc. and Jihyeon Gim +// This file is licensed under the Apache License, Version 2.0 +// (the "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + //go:build linux && cgo package rcserver diff --git a/rdma/rcserver/rcserver_linux.go b/rdma/rcserver/rcserver_linux.go index 481170e6..2713c1c7 100644 --- a/rdma/rcserver/rcserver_linux.go +++ b/rdma/rcserver/rcserver_linux.go @@ -33,6 +33,7 @@ package rcserver // level/file/line/msg are marshalled through this fixed trampoline. extern void rcgo_log_sink(void *ctx, int level, char *msg, char *file, int line); +extern void rcgo_snapshot_cb(rc_session_snapshot *rec, void *ctx); */ import "C" @@ -517,6 +518,78 @@ func (s *RCSvc) SessionInfo(sessionID string, who PrincipalID) (*SessionInfo, er }, nil } +// SessionSnapshot describes a live RC session for observability. +type SessionSnapshot struct { + SessionID string + Op string + Target string + State uint8 // SessState value, may be ORed with ReapPending + ReapPending bool + AgeMs uint64 + StagingBytes uint64 +} + +// SnapshotState constants mirrored from the C ABI. +const ( + SnapshotStatePrepared = 0 + SnapshotStatePublishing = 1 + SnapshotStateTransferring = 2 + SnapshotStateCompleting = 3 + SnapshotStateReaping = 4 + SnapshotReapPending = 0x80 +) + +// snapshotReceiver is set by SessionsSnapshot for the duration of the +// C call; the fixed trampoline copies each record into it. The mutex +// covers the case of concurrent snapshot calls. +var ( + snapshotMu sync.Mutex + snapshotSink *[]SessionSnapshot +) + +//export rcgo_snapshot_cb +func rcgo_snapshot_cb(rec *C.rc_session_snapshot, _ unsafe.Pointer) { + sink := snapshotSink + if sink == nil || rec == nil { + return + } + st := uint8(rec.state) + *sink = append(*sink, SessionSnapshot{ + SessionID: C.GoString(&rec.session_id[0]), + Op: C.GoString(&rec.op[0]), + Target: C.GoStringN(&rec.target[0], C.int(rec.target_len)), + State: st &^ SnapshotReapPending, + ReapPending: st&SnapshotReapPending != 0, + AgeMs: uint64(rec.age_ms), + StagingBytes: uint64(rec.staging_bytes), + }) +} + +// SessionsSnapshot copies every live session into Go-owned records. +// The C side snapshots under its map lock and delivers the copies +// outside it, so the callback cannot block the data plane. +func (s *RCSvc) SessionsSnapshot() ([]SessionSnapshot, error) { + if s.srv == nil { + return nil, errors.New("rcserver: service closed") + } + if !s.TryEnter() { + return nil, ErrInternal + } + defer s.Leave() + + snapshotMu.Lock() + defer snapshotMu.Unlock() + var out []SessionSnapshot + snapshotSink = &out + defer func() { snapshotSink = nil }() + rc := C.rc_server_sessions_snapshot(s.srv, + (*[0]byte)(C.rcgo_snapshot_cb), nil) + if rc != C.RC_OK { + return nil, statusError(rc) + } + return out, nil +} + // ReadyTransfer runs the data phase (READY). func (s *RCSvc) ReadyTransfer(req ReadyRequest) (*ReadyResponse, error) { if s.srv == nil { diff --git a/rdma/rcserver/rcserver_stub.go b/rdma/rcserver/rcserver_stub.go index 820858ca..99e6b220 100644 --- a/rdma/rcserver/rcserver_stub.go +++ b/rdma/rcserver/rcserver_stub.go @@ -111,6 +111,27 @@ type SessionInfo struct { Target string } +// SessionSnapshot describes a live RC session for observability. +type SessionSnapshot struct { + SessionID string + Op string + Target string + State uint8 + ReapPending bool + AgeMs uint64 + StagingBytes uint64 +} + +// Snapshot state values mirrored from the cgo ABI. +const ( + SnapshotStatePrepared = 0 + SnapshotStatePublishing = 1 + SnapshotStateTransferring = 2 + SnapshotStateCompleting = 3 + SnapshotStateReaping = 4 + SnapshotReapPending = 0x80 +) + // RCSvc owns the RC session server (stub). type RCSvc struct{} @@ -152,6 +173,11 @@ func (s *RCSvc) SessionInfo(sessionID string, who PrincipalID) (*SessionInfo, er return nil, errNotSupported } +// SessionsSnapshot is a stub. +func (s *RCSvc) SessionsSnapshot() ([]SessionSnapshot, error) { + return nil, errNotSupported +} + // ReadyTransfer is a stub. func (s *RCSvc) ReadyTransfer(req ReadyRequest) (*ReadyResponse, error) { return nil, errNotSupported diff --git a/s3api/admin-server.go b/s3api/admin-server.go index d06b98e5..3be14529 100644 --- a/s3api/admin-server.go +++ b/s3api/admin-server.go @@ -42,6 +42,15 @@ type S3AdminServer struct { maxConnections int maxRequests int socketPerm os.FileMode + extraRoutes []adminRouteMount +} + +// adminRouteMount is a route registered on the admin app after the +// built-in admin router ran. +type adminRouteMount struct { + method string + path string + handlers []fiber.Handler } func NewAdminServer(be backend.Backend, root middlewares.RootUserConfig, region string, iam auth.IAMService, l s3log.AuditLogger, ctrl controllers.S3ApiController, opts ...AdminOpt) *S3AdminServer { @@ -95,6 +104,14 @@ func NewAdminServer(be backend.Backend, root middlewares.RootUserConfig, region server.router.Init(app, be, iam, l, root, region, server.debug, server.corsAllowOrigin) + for _, r := range server.extraRoutes { + args := make([]any, 0, len(r.handlers)) + for _, h := range r.handlers { + args = append(args, h) + } + app.Add([]string{r.method}, r.path, args[0], args[1:]...) + } + return server } @@ -136,6 +153,21 @@ func WithAdminSocketPerm(perm os.FileMode) AdminOpt { return func(s *S3AdminServer) { s.socketPerm = perm } } +// WithAdminRoute registers a route on the standalone admin server, +// after the built-in admin routes and their middleware chain. Use it +// for admin-surface endpoints that do not fit the S3 admin controller +// shape. +func WithAdminRoute(method, path string, handlers ...fiber.Handler) AdminOpt { + return func(s *S3AdminServer) { + copied := append([]fiber.Handler(nil), handlers...) + s.extraRoutes = append(s.extraRoutes, adminRouteMount{ + method: method, + path: path, + handlers: copied, + }) + } +} + // ServeMultiPort creates listeners for multiple port specifications and serves // on all of them simultaneously. This supports listening on multiple ports and/or // addresses (e.g., [":8080", "localhost:8081"]).