diff --git a/cmd/vgwrdma/main.go b/cmd/vgwrdma/main.go index 33306dd3..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 ( @@ -120,6 +123,15 @@ var ( rdmaCQDepth uint rdmaRetryCount uint rdmaTunablesSet bool + rcMaxSessions uint64 + rcMaxUserSessions uint64 + rcMaxStagingBytes uint64 + rcMaxUserStagingBytes uint64 + rcMaxQPs uint64 + rcMaxUserQPs uint64 + rcMaxReadySlots uint64 + rcPrepTimeoutMs uint64 + rcExecTimeoutMs uint64 ) var ( @@ -898,6 +910,69 @@ func initFlags() []cli.Flag { EnvVars: []string{"VGW_RDMA_RC_ENABLE"}, Destination: &rdmaRCEnable, }, + &cli.Uint64Flag{ + Name: "rdma-rc-max-sessions", + Usage: "maximum concurrent hipobj-rc-v2 sessions (default 1024)", + EnvVars: []string{"VGW_RDMA_RC_MAX_SESSIONS"}, + Value: 1024, + Destination: &rcMaxSessions, + }, + &cli.Uint64Flag{ + Name: "rdma-rc-max-user-sessions", + Usage: "per-principal session limit for the hipobj-rc-v2 data plane (default 64)", + EnvVars: []string{"VGW_RDMA_RC_MAX_USER_SESSIONS"}, + Value: 64, + Destination: &rcMaxUserSessions, + }, + &cli.Uint64Flag{ + Name: "rdma-rc-max-staging-bytes", + Usage: "total staging buffer budget for hipobj-rc-v2 sessions in bytes (default 4294967296)", + EnvVars: []string{"VGW_RDMA_RC_MAX_STAGING_BYTES"}, + Value: 4 << 30, + Destination: &rcMaxStagingBytes, + }, + &cli.Uint64Flag{ + Name: "rdma-rc-max-user-staging-bytes", + Usage: "per-principal staging buffer budget for the hipobj-rc-v2 data plane in bytes (default 1073741824)", + EnvVars: []string{"VGW_RDMA_RC_MAX_USER_STAGING_BYTES"}, + Value: 1 << 30, + Destination: &rcMaxUserStagingBytes, + }, + &cli.Uint64Flag{ + Name: "rdma-rc-max-qps", + Usage: "maximum queue pairs for the hipobj-rc-v2 data plane (default 1024)", + EnvVars: []string{"VGW_RDMA_RC_MAX_QPS"}, + Value: 1024, + Destination: &rcMaxQPs, + }, + &cli.Uint64Flag{ + Name: "rdma-rc-max-user-qps", + Usage: "per-principal queue pair limit for the hipobj-rc-v2 data plane (default 16)", + EnvVars: []string{"VGW_RDMA_RC_MAX_USER_QPS"}, + Value: 16, + Destination: &rcMaxUserQPs, + }, + &cli.Uint64Flag{ + Name: "rdma-rc-max-ready-slots", + Usage: "concurrent READY transfers admitted by the hipobj-rc-v2 data plane (default 64)", + EnvVars: []string{"VGW_RDMA_RC_MAX_READY_SLOTS"}, + Value: 64, + Destination: &rcMaxReadySlots, + }, + &cli.Uint64Flag{ + Name: "rdma-rc-prep-timeout-ms", + Usage: "milliseconds a hipobj-rc-v2 session may wait for READY after PREPARE (default 100000)", + EnvVars: []string{"VGW_RDMA_RC_PREP_TIMEOUT_MS"}, + Value: 100000, + Destination: &rcPrepTimeoutMs, + }, + &cli.Uint64Flag{ + Name: "rdma-rc-exec-timeout-ms", + Usage: "milliseconds a hipobj-rc-v2 READY transfer may run (default 30000)", + EnvVars: []string{"VGW_RDMA_RC_EXEC_TIMEOUT_MS"}, + Value: 30000, + Destination: &rcExecTimeoutMs, + }, &cli.UintFlag{ Name: "rdma-port", Usage: "port for RDMA listener", @@ -1006,6 +1081,25 @@ func runGateway(ctx context.Context, be backend.Backend) error { return errors.New(msg) } } + if v2On { + // RC-only settings are irrelevant when the hipobj-rc-v2 + // data plane is not running; stale environment values + // must not block v1-only or plain-S3 startup. + v2s := rdmamode.V2Settings{ + MaxSessions: rcMaxSessions, + MaxUserSessions: rcMaxUserSessions, + MaxStagingBytes: rcMaxStagingBytes, + MaxUserStagingBytes: rcMaxUserStagingBytes, + MaxQPs: rcMaxQPs, + MaxUserQPs: rcMaxUserQPs, + MaxReadySlots: rcMaxReadySlots, + TPrepMs: rcPrepTimeoutMs, + TExecMs: rcExecTimeoutMs, + } + if msg := rdmamode.V2ValidationError(v2s); msg != "" { + return errors.New(msg) + } + } var s3Opts []s3api.Option if v1On { @@ -1163,14 +1257,16 @@ func runGateway(ctx context.Context, be backend.Backend) error { rcSvc, err := rcserver.Init(rcserver.DeviceOpts{ GidHint: rcGidHint, Port: 1, - MaxSessions: 1024, - MaxUserSessions: 64, - MaxStagingBytes: 4 << 30, - MaxUserStagingBytes: 1 << 30, - MaxQPs: 1024, - MaxUserQPs: 16, - TPrepMs: 100000, - TExecMs: 30000, + MaxSessions: uint32(rcMaxSessions), + MaxUserSessions: uint32(rcMaxUserSessions), + MaxStagingBytes: rcMaxStagingBytes, + MaxUserStagingBytes: rcMaxUserStagingBytes, + MaxQPs: uint32(rcMaxQPs), + MaxUserQPs: uint32(rcMaxUserQPs), + MaxReadySlots: uint32(rcMaxReadySlots), + TPrepMs: rcPrepTimeoutMs, + TExecMs: rcExecTimeoutMs, + Debug: debug, }) if err != nil { return err @@ -1203,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/cuwrapper/rc/rc_ibv_host.cpp b/cuwrapper/rc/rc_ibv_host.cpp index 3ebcdadd..0d0c9cd9 100644 --- a/cuwrapper/rc/rc_ibv_host.cpp +++ b/cuwrapper/rc/rc_ibv_host.cpp @@ -75,7 +75,33 @@ bool IBVWrapper::ensureLoaded() { struct ibv_send_wr *, struct ibv_send_wr **)>(load("ibv_post_send")); loaded = funcs_.get_device_list != nullptr && funcs_.open_device != nullptr && funcs_.alloc_pd != nullptr && funcs_.create_qp != nullptr && - funcs_.modify_qp != nullptr && funcs_.poll_cq != nullptr; + funcs_.modify_qp != nullptr; + if (loaded) { + /* poll_cq/post_send/post_recv are static inline wrappers in + * modern verbs.h (they dispatch through cq->context->ops), so + * dlsym cannot find them on rdma-core 61+. Resolve them from + * the ops table of the first successfully opened context + * instead; every context from the same device shares these + * providers. */ + int n = 0; + struct ibv_device **devs = funcs_.get_device_list(&n); + struct ibv_context *probe = nullptr; + if (devs && n > 0) probe = funcs_.open_device(devs[0]); + if (devs) funcs_.free_device_list(devs); + if (!probe) { + fprintf(stderr, "rc: no RDMA device to resolve verbs ops\n"); + loaded = false; + } else { + funcs_.poll_cq = probe->ops.poll_cq; + funcs_.post_recv = probe->ops.post_recv; + funcs_.post_send = probe->ops.post_send; + funcs_.close_device(probe); + if (!funcs_.poll_cq || !funcs_.post_recv || !funcs_.post_send) { + fprintf(stderr, "rc: provider ops table incomplete\n"); + loaded = false; + } + } + } if (loaded) { /* Mirror into the member seam for direct ibv.x() calls. */ get_device_list = funcs_.get_device_list; diff --git a/cuwrapper/rc/rc_server_abi.cpp b/cuwrapper/rc/rc_server_abi.cpp index dfcfa2b1..7d7c5e6a 100644 --- a/cuwrapper/rc/rc_server_abi.cpp +++ b/cuwrapper/rc/rc_server_abi.cpp @@ -15,6 +15,8 @@ #include #include #include +#include +#include #include "rc_ibv_host.h" #include "v2_data_phase.h" @@ -41,6 +43,9 @@ struct RcSession { V2Session core; uint64_t epoch = 0; std::atomic next_nonce{1}; + /* Monotonic creation timestamp for session-age observability; + * deadlines cannot serve that role because READY moves them. */ + uint64_t created_ms = 0; /* staged metadata from finish_staging (GET) or finish_put. */ std::string etag; std::string version_id; @@ -89,6 +94,13 @@ struct rc_server { hipObj::DeviceHandle *device = nullptr; SessionTable table; rc_device_opts opts{}; + /* Diagnostic sink: null keeps stderr-only error reporting. + * Reads/writes are plain loads/stores; the sink is installed + * once at init time (before the reaper starts) and only + * cleared by destroy after the reaper joined, so no thread + * races an in-flight sink pointer swap. */ + rc_log_fn log_fn = nullptr; + void *log_ctx = nullptr; std::atomic epoch_counter{1}; /* resource accounting (global buckets; per-principal map). */ std::mutex acct_mtx; @@ -117,6 +129,22 @@ struct rc_server { namespace { +/* Emits a diagnostic line to the installed sink (level 0 keeps + * the stderr error stream intact by also printing there, so + * existing deployments do not lose the only log they had). + * Callers must not hold map_mtx/acct_mtx when calling. */ +void rcLog(const rc_server *srv, int level, const char *file, int line, + const char *fmt, ...) { + char buf[256]; + va_list ap; + va_start(ap, fmt); + vsnprintf(buf, sizeof(buf), fmt, ap); + va_end(ap); + if (level <= 0) fprintf(stderr, "%s\n", buf); + rc_log_fn fn = srv->log_fn; + if (fn) fn(srv->log_ctx, level, buf, file, line); +} + RcSession *findSession(rc_server *srv, const std::string &id) { auto it = srv->sessions_map.find(id); return it == srv->sessions_map.end() ? nullptr : it->second.get(); @@ -187,6 +215,16 @@ void reapSession(rc_server *srv, RcSession *s) { s->core.qp = conn.qp; /* null on success, survivor on failure */ s->core.cq = conn.cq; bool destroyed = q_ok && c_ok; + /* Terminal record for every session teardown path (expiry, + * CANCEL, and destroy); reap_pass may have missed the final + * state, so the last outcome observed at READY time travels + * with the log line. */ + rcLog(srv, 2, __FILE__, __LINE__, + "rc: session reaped id=%s op=%s target=%.96s staged=%llu " + "qp_destroyed=%d", + s->core.id.c_str(), s->core.op.c_str(), + s->core.target.c_str(), + (unsigned long long)s->staging_len, (int)destroyed); if (destroyed) { /* Same policy as releaseStaging: a failed dereg leaves the * MR registered against the shared PD, so the buffer stays @@ -257,9 +295,18 @@ std::string encodeReplyToken(hipObj::DeviceHandle *dh, uint32_t qpn) { extern "C" { +void rc_server_set_log_sink(rc_server *srv, rc_log_fn fn, void *ctx) { + if (!srv) return; + srv->log_fn = fn; + srv->log_ctx = ctx; +} + int rc_server_init(const rc_device_opts *opts, rc_server **out) { if (!opts || !out) return RC_E_ARG; - if (!hipObj::ibv.ensureLoaded()) return RC_E_INTERNAL; + if (!hipObj::ibv.ensureLoaded()) { + fprintf(stderr, "rc: cannot load libibverbs (dlopen/dlsym failed)\n"); + return RC_E_INTERNAL; + } std::unique_ptr srv(new rc_server()); srv->opts = *opts; /* ibv port numbers are 1-based; treat an unset (0) port as 1 so @@ -275,7 +322,10 @@ int rc_server_init(const rc_device_opts *opts, rc_server **out) { int n = 0; struct ibv_device **devs = hipObj::ibv.get_device_list(&n); - if (!devs || n == 0) return RC_E_INTERNAL; + if (!devs || n == 0) { + fprintf(stderr, "rc: no RDMA devices found (ibv_get_device_list)\n"); + return RC_E_INTERNAL; + } struct ibv_device *chosen = devs[0]; /* GID hint: pick the first device/port whose GID starts with it. * Query with srv->opts.port, which the normalization above has @@ -309,12 +359,15 @@ int rc_server_init(const rc_device_opts *opts, rc_server **out) { } if (!ctx) { hipObj::ibv.free_device_list(devs); + fprintf(stderr, "rc: no verbs device matches gid_hint %.32s\n", + opts->gid_hint ? opts->gid_hint : ""); return RC_E_INTERNAL; } struct ibv_pd *pd = hipObj::ibv.alloc_pd(ctx); hipObj::ibv.free_device_list(devs); if (!pd) { hipObj::ibv.close_device(ctx); + fprintf(stderr, "rc: alloc_pd failed\n"); return RC_E_INTERNAL; } srv->device = new hipObj::DeviceHandle(); @@ -563,6 +616,11 @@ int rc_prepare(rc_server *srv, const rc_prepare_req *req, srv->opts.t_prep_ms ? hipObj::v2::clockSource().nowMs() + srv->opts.t_prep_ms : 0; + rs.created_ms = hipObj::v2::clockSource().nowMs(); + /* The session record carries its own id copy: reap logging and the + * terminal teardown record read core.id, while the map key is the + * only other place the id lives. */ + rs.core.id = id; { std::lock_guard g(srv->map_mtx); rs.staging_buf = reinterpret_cast(buf); @@ -703,6 +761,44 @@ int rc_session_info(rc_server *srv, rc_str_in session_id, return RC_OK; } +int rc_server_sessions_snapshot(rc_server *srv, rc_snapshot_cb cb, + void *ctx) { + if (!srv || !cb) return RC_E_ARG; + if (srv->closing.load()) return RC_E_INTERNAL; + + /* Fixed records so the vector can move without invalidating the + * string pointers inside; the copies own everything the callback + * reads, so delivery happens outside the map lock and cannot race + * the reaper moving or erasing entries. */ + std::vector recs; + uint64_t now = hipObj::v2::clockSource().nowMs(); + { + std::lock_guard g(srv->map_mtx); + recs.reserve(srv->sessions_map.size()); + for (const auto &kv : srv->sessions_map) { + const RcSession &s = *kv.second; + rc_session_snapshot r{}; + if (s.core.id.size() != 32) continue; /* live ids are 32 hex */ + if (s.core.op.size() >= sizeof(r.op)) continue; + if (s.core.target.size() > sizeof(r.target) - 1) continue; + memcpy(r.session_id, s.core.id.data(), 32); + r.session_id[32] = 0; + memcpy(r.op, s.core.op.data(), s.core.op.size()); + r.op[s.core.op.size()] = 0; + memcpy(r.target, s.core.target.data(), s.core.target.size()); + r.target[s.core.target.size()] = 0; + r.target_len = (uint32_t)s.core.target.size(); + r.state = (uint8_t)s.core.state; + if (s.reap_pending) r.state |= RC_SNAPSHOT_REAP_PENDING; + r.age_ms = s.created_ms ? now - s.created_ms : 0; + r.staging_bytes = s.staging_len; + recs.push_back(r); + } + } + for (const auto &r : recs) cb(&r, ctx); + return RC_OK; +} + int rc_ready_transfer(rc_server *srv, const rc_ready_req *req, rc_ready_resp *resp) { if (!srv || !req || !resp) return RC_E_ARG; @@ -798,6 +894,19 @@ int rc_ready_transfer(rc_server *srv, const rc_ready_req *req, RcSession *after = findSession(srv, id); if (!after) return RC_E_NO_SESSION; + /* Keep the wire-level reason (poll status vs post failure vs + * timeout) alongside the outcome the response carries, so a + * VerifyFail is diagnosable without re-running the transfer. + * Logged with the lock dropped: the sink must not block under + * map_mtx. */ + int rlog = (r == hipObj::v2::DataPhaseResult::Ok) + ? 2 + : (r == hipObj::v2::DataPhaseResult::Busy ? 2 : 0); + rcLog(srv, rlog, __FILE__, __LINE__, + "rc: ready data phase session=%s op=%s outcome=%d bytes=%llu", + id.c_str(), after->core.op.c_str(), (int)r, + (unsigned long long)stats.bytes); + switch (r) { case hipObj::v2::DataPhaseResult::Ok: after->last_outcome = RC_READY_OK; diff --git a/cuwrapper/rc/rc_server_abi.h b/cuwrapper/rc/rc_server_abi.h index 55c7d191..f852c19b 100644 --- a/cuwrapper/rc/rc_server_abi.h +++ b/cuwrapper/rc/rc_server_abi.h @@ -39,8 +39,19 @@ enum { typedef struct { const char *ptr; uint32_t len; } rc_str_in; +/* Diagnostic log sink. Called from RC server threads (request + * goroutines via cgo and the expiry reaper) without any server + * lock held. `msg` and `file` are only valid for the duration of + * the call; the sink must copy them if it needs them longer. + * `level` is 0 (error) or 2 (debug). Passing a null sink (or + * installing it after init) keeps stderr-only error reporting. */ +typedef void (*rc_log_fn)(void *ctx, int level, const char *msg, + const char *file, int line); + typedef struct rc_server rc_server; +void rc_server_set_log_sink(rc_server *srv, rc_log_fn fn, void *ctx); + /* Device selection: matching GID prefix when gid_hint is set, * otherwise the first verbs device. */ typedef struct { @@ -140,6 +151,30 @@ typedef struct { uint8_t op; } rc_session_info_resp; +/* Session observability snapshot. The record is a point-in-time copy; + * the fields are valid only inside the callback invocation. + * state combines the session state machine value with the reap-pending + * marker (RC_SNAPSHOT_REAP_PENDING) because CANCEL and expiry only set + * that marker without moving the state. */ +enum { RC_SNAPSHOT_REAP_PENDING = 0x80 }; + +typedef struct { + char session_id[33]; /* 32 hex + NUL */ + char op[4]; /* "GET" or "PUT" */ + char target[2048]; + uint32_t target_len; + uint8_t state; /* SessState value | RC_SNAPSHOT_REAP_PENDING */ + uint64_t age_ms; /* now - created_ms; 0 when the clock is absent */ + uint64_t staging_bytes; +} rc_session_snapshot; + +/* Copies every live session under the map lock into fixed records and + * invokes cb(rec, ctx) once per record outside the lock, in map order. + * A session whose op or target does not fit is skipped (record too + * small), not truncated. Returns RC_E_ARG for null arguments. */ +typedef void (*rc_snapshot_cb)(const rc_session_snapshot *rec, void *ctx); +int rc_server_sessions_snapshot(rc_server *srv, rc_snapshot_cb cb, void *ctx); + /* Lifecycle. destroy waits for active calls and the reaper. */ int rc_server_init(const rc_device_opts *opts, rc_server **out); void rc_server_destroy(rc_server *srv); 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/internal/rdmamode/rdmamode.go b/internal/rdmamode/rdmamode.go index a5b40aca..92cad9fa 100644 --- a/internal/rdmamode/rdmamode.go +++ b/internal/rdmamode/rdmamode.go @@ -21,6 +21,7 @@ import ( "math" "strings" "sync" + "time" "github.com/versity/versitygw/backend" ) @@ -68,6 +69,57 @@ func V1ValidationError(s V1Settings) string { } } +// V2Settings carries the hipobj-rc-v2 tunings that are validated +// only while the RC data plane runs. Counts arrive as uint64 from +// the CLI and narrow to uint32 at the DeviceOpts boundary, so the +// range is checked before the narrowing cast. Timeouts feed +// deadline arithmetic (nowMs + timeout), so an upper bound keeps +// the sum from wrapping. +type V2Settings struct { + MaxSessions uint64 + MaxUserSessions uint64 + MaxStagingBytes uint64 + MaxUserStagingBytes uint64 + MaxQPs uint64 + MaxUserQPs uint64 + MaxReadySlots uint64 + TPrepMs uint64 + TExecMs uint64 +} + +// v2TimeoutCeiling bounds the RC timeouts well below the uint64 +// wrap point of nowMs + timeout arithmetic in the C core. +const v2TimeoutCeiling = uint64(24 * time.Hour / time.Millisecond) + +// V2ValidationError describes the first invalid v2 setting, or +// the empty string when every setting is valid. Stale v2 values +// in the environment must not block v1-only or plain-S3 startup, +// so the gateway consults this only when v2 is on. +func V2ValidationError(s V2Settings) string { + switch { + case s.MaxSessions < 1 || s.MaxSessions > math.MaxUint32: + return fmt.Sprintf("rdma-rc-max-sessions %d is out of range (1-%d)", s.MaxSessions, uint64(math.MaxUint32)) + case s.MaxUserSessions < 1 || s.MaxUserSessions > math.MaxUint32: + return fmt.Sprintf("rdma-rc-max-user-sessions %d is out of range (1-%d)", s.MaxUserSessions, uint64(math.MaxUint32)) + case s.MaxStagingBytes < 1: + return fmt.Sprintf("rdma-rc-max-staging-bytes %d must be positive", s.MaxStagingBytes) + case s.MaxUserStagingBytes < 1: + return fmt.Sprintf("rdma-rc-max-user-staging-bytes %d must be positive", s.MaxUserStagingBytes) + case s.MaxQPs < 1 || s.MaxQPs > math.MaxUint32: + return fmt.Sprintf("rdma-rc-max-qps %d is out of range (1-%d)", s.MaxQPs, uint64(math.MaxUint32)) + case s.MaxUserQPs < 1 || s.MaxUserQPs > math.MaxUint32: + return fmt.Sprintf("rdma-rc-max-user-qps %d is out of range (1-%d)", s.MaxUserQPs, uint64(math.MaxUint32)) + case s.MaxReadySlots < 1 || s.MaxReadySlots > math.MaxUint32: + return fmt.Sprintf("rdma-rc-max-ready-slots %d is out of range (1-%d)", s.MaxReadySlots, uint64(math.MaxUint32)) + case s.TPrepMs < 1 || s.TPrepMs > v2TimeoutCeiling: + return fmt.Sprintf("rdma-rc-prep-timeout-ms %d is out of range (1-%d)", s.TPrepMs, v2TimeoutCeiling) + case s.TExecMs < 1 || s.TExecMs > v2TimeoutCeiling: + return fmt.Sprintf("rdma-rc-exec-timeout-ms %d is out of range (1-%d)", s.TExecMs, v2TimeoutCeiling) + default: + return "" + } +} + // Closer is the close operation of the RC session service. It is // idempotent. type Closer interface{ Close() } diff --git a/internal/rdmamode/rdmamode_test.go b/internal/rdmamode/rdmamode_test.go index abeaa515..3da8d035 100644 --- a/internal/rdmamode/rdmamode_test.go +++ b/internal/rdmamode/rdmamode_test.go @@ -256,3 +256,100 @@ func TestRCWrapperChainsOntoOnceBackend(t *testing.T) { t.Fatalf("base backend shutdown %d times, want 1", shutdowns) } } + +func v2Defaults() V2Settings { + return V2Settings{ + MaxSessions: 1024, + MaxUserSessions: 64, + MaxStagingBytes: 4 << 30, + MaxUserStagingBytes: 1 << 30, + MaxQPs: 1024, + MaxUserQPs: 16, + MaxReadySlots: 64, + TPrepMs: 100000, + TExecMs: 30000, + } +} + +func TestV2ValidationDefaultsPass(t *testing.T) { + // The defaults mirror the values the gateway passed before + // the flags existed, so a deployment that sets nothing must + // keep starting. + if msg := V2ValidationError(v2Defaults()); msg != "" { + t.Fatalf("defaults rejected: %q", msg) + } +} + +func TestV2ValidationRejectsZero(t *testing.T) { + // Every knob must be positive: counts narrow to uint32 and + // zero would disable a limit or a deadline in the C core. + for name, mutate := range map[string]func(*V2Settings){ + "max-sessions": func(s *V2Settings) { s.MaxSessions = 0 }, + "max-user-sessions": func(s *V2Settings) { s.MaxUserSessions = 0 }, + "max-staging-bytes": func(s *V2Settings) { s.MaxStagingBytes = 0 }, + "max-user-staging-bytes": func(s *V2Settings) { s.MaxUserStagingBytes = 0 }, + "max-qps": func(s *V2Settings) { s.MaxQPs = 0 }, + "max-user-qps": func(s *V2Settings) { s.MaxUserQPs = 0 }, + "max-ready-slots": func(s *V2Settings) { s.MaxReadySlots = 0 }, + "prep-timeout": func(s *V2Settings) { s.TPrepMs = 0 }, + "exec-timeout": func(s *V2Settings) { s.TExecMs = 0 }, + } { + s := v2Defaults() + mutate(&s) + if msg := V2ValidationError(s); msg == "" { + t.Fatalf("%s: zero accepted", name) + } + } +} + +func TestV2ValidationCountBoundaries(t *testing.T) { + // Counts arrive as uint64 and narrow to uint32 at the + // DeviceOpts boundary: MaxUint32 passes, the first value + // beyond it fails. + fields := []struct { + name string + field *uint64 + }{ + {"max-sessions", new(uint64)}, + {"max-user-sessions", new(uint64)}, + {"max-qps", new(uint64)}, + {"max-user-qps", new(uint64)}, + {"max-ready-slots", new(uint64)}, + } + for _, f := range fields { + base := v2Defaults() + f.field = &base.MaxSessions + switch f.name { + case "max-user-sessions": + f.field = &base.MaxUserSessions + case "max-qps": + f.field = &base.MaxQPs + case "max-user-qps": + f.field = &base.MaxUserQPs + case "max-ready-slots": + f.field = &base.MaxReadySlots + } + *f.field = math.MaxUint32 + if msg := V2ValidationError(base); msg != "" { + t.Fatalf("%s at MaxUint32 rejected: %q", f.name, msg) + } + *f.field = math.MaxUint32 + 1 + if msg := V2ValidationError(base); msg == "" { + t.Fatalf("%s beyond MaxUint32 accepted", f.name) + } + } +} + +func TestV2ValidationTimeoutCeiling(t *testing.T) { + // Timeouts feed nowMs + timeout deadline arithmetic in the + // C core, so values past the ceiling are refused. + base := v2Defaults() + base.TPrepMs = v2TimeoutCeiling + if msg := V2ValidationError(base); msg != "" { + t.Fatalf("ceiling rejected: %q", msg) + } + base.TPrepMs = v2TimeoutCeiling + 1 + if msg := V2ValidationError(base); msg == "" { + t.Fatal("ceiling+1 accepted") + } +} 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 new file mode 100644 index 00000000..7a0c3c52 --- /dev/null +++ b/rdma/rcserver/rcserver_cgo_linux_test.go @@ -0,0 +1,73 @@ +// 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 + +import ( + "errors" + "strings" + "testing" + "unsafe" +) + +// heapSubstring returns a substring whose backing array is heap-allocated +// and whose data pointer is an interior pointer, mirroring how header +// strings reach the binding from the gateway routes. +func heapSubstring(value string) string { + const prefix = "prefix:" + backing := strings.Clone(prefix + value + ":suffix") + return backing[len(prefix) : len(prefix)+len(value)] +} + +// TestPrepareHeapStringsReachCValidation calls the real Prepare wrapper +// with heap-backed strings. The request uses an invalid opcode so the C +// entrypoint returns RC_E_ARG from its argument validation before the +// server handle is dereferenced; the dummy handle below is never touched. +// Before the pinnedStrIn fix this call panics at the cgo pointer check; +// after it, the C argument validation runs and the error surfaces. +func TestPrepareHeapStringsReachCValidation(t *testing.T) { + const literalTarget = "/bucket1/obj1" + heapTarget := heapSubstring(literalTarget) + heapToken := heapSubstring(strings.Repeat("0", 88)) + for _, tc := range []struct { + name, target, token string + }{ + {"literal_control", literalTarget, ""}, + {"heap_target", heapTarget, ""}, + {"heap_token", literalTarget, heapToken}, + {"both_heap", heapTarget, heapToken}, + {"empty_control", "", ""}, + } { + t.Run(tc.name, func(t *testing.T) { + var svc RCSvc + // Test-only opaque sentinel, not an initialized rc_server. + // Op=255 returns in C argument validation before the + // server pointer is dereferenced. + dummy := new(uint64) + *(*unsafe.Pointer)(unsafe.Pointer(&svc.srv)) = unsafe.Pointer(dummy) + resp, err := svc.Prepare(PrepareRequest{ + Op: 255, + Size: 1, + Target: tc.target, + ClientToken: tc.token, + }) + if resp != nil || !errors.Is(err, ErrArg) { + t.Fatalf("Prepare = (%v, %v), want (nil, ErrArg)", resp, err) + } + }) + } +} diff --git a/rdma/rcserver/rcserver_linux.go b/rdma/rcserver/rcserver_linux.go index b2f3b9a5..2713c1c7 100644 --- a/rdma/rcserver/rcserver_linux.go +++ b/rdma/rcserver/rcserver_linux.go @@ -27,6 +27,13 @@ package rcserver #cgo LDFLAGS: -L${SRCDIR}/.. -l:librcserver.a -lstdc++ -ldl -lpthread #include "rc_server_abi.h" #include + +// Shim: a C function pointer that forwards into the exported Go +// sink. Go closures cannot be stored as C callbacks, so the +// 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" @@ -34,6 +41,7 @@ import ( "context" "errors" "fmt" + "log" "runtime" "sync" "sync/atomic" @@ -76,6 +84,10 @@ type DeviceOpts struct { // Concurrency slots (0 picks the C-side default 64/32). MaxReadySlots uint32 MaxStageSlots uint32 + // Debug enables level-2 (debug) RC diagnostics through the + // log sink; false keeps error-only reporting. Mirrors the + // gateway --debug flag. + Debug bool } // PrincipalID is the SHA-256 digest identifying the requester. @@ -174,6 +186,52 @@ func (s *RCSvc) Context() context.Context { return s.ctx } +// rcLogSink receives every diagnostic line the C server emits. +// It is stateless and process-global on purpose: the sink must be +// valid from init through destroy, independent of any single +// RCSvc instance. Lines are copied out immediately (the C msg +// buffer is only valid for the duration of the callback) and +// written to the standard error logger. rcLogMu serializes the +// copy because log.Logger is safe for concurrent use but keeps +// a single writer cheap. +var ( + rcLogMu sync.Mutex + rcLogLevel atomic.Int32 + rcLog = log.New(log.Writer(), "vgwrdma rc: ", 0) +) + +//export rcgo_log_sink +func rcgo_log_sink(_ unsafe.Pointer, level C.int, msg *C.char, + file *C.char, line C.int) { + // Level gating mirrors the C contract: 0 (error) always + // arrives here; 2 (debug) only when the service opted in. + if int32(level) > rcLogLevel.Load() { + return + } + m := C.GoString(msg) + f := "" + if file != nil { + f = C.GoString(file) + } + rcLogMu.Lock() + rcLog.Printf("%s (%s:%d) level=%d", m, f, int(line), int(level)) + rcLogMu.Unlock() +} + +// installLogSink wires the C server's diagnostics into the Go +// sink. debug selects level 2 (debug) versus level 0 (errors +// only); the C side keeps stderr error reporting when the sink +// is absent, so a nil install is a no-op by design. +func installLogSink(srv *C.rc_server, debug bool) { + if debug { + rcLogLevel.Store(2) + } else { + rcLogLevel.Store(0) + } + C.rc_server_set_log_sink(srv, + (*[0]byte)(C.rcgo_log_sink), nil) +} + // Init opens the verbs device and returns a service. func Init(opts DeviceOpts) (*RCSvc, error) { copts := C.rc_device_opts{ @@ -201,6 +259,7 @@ func Init(opts DeviceOpts) (*RCSvc, error) { if rc := C.rc_server_init(&copts, &srv); rc != C.RC_OK { return nil, fmt.Errorf("rcserver: init failed: status %d", int(rc)) } + installLogSink(srv, opts.Debug) ctx, cancel := context.WithCancel(context.Background()) return &RCSvc{srv: srv, ctx: ctx, cancel: cancel}, nil } @@ -313,18 +372,33 @@ func statusError(rc C.int) error { } } +// strIn borrows string bytes for a synchronous rc_str_in value argument. +// For a string view embedded in a request struct passed by pointer, use +// pinnedStrIn instead: the cgo pointer check rejects request structs that +// contain unpinned Go string data. C must not retain or modify the borrowed +// bytes. func strIn(s string) C.rc_str_in { if s == "" { return C.rc_str_in{ptr: nil, len: 0} } - // The pointer must stay alive across the cgo call; every - // wrapper keeps the string referenced until after the call. return C.rc_str_in{ ptr: (*C.char)(unsafe.Pointer(unsafe.StringData(s))), len: C.uint32_t(len(s)), } } +// pinnedStrIn builds a strIn view and pins the string bytes for the +// duration of the enclosing cgo call. The caller owns the Pinner and +// must defer Unpin before the first pinnedStrIn call. Pinning an +// interior pointer pins the whole backing allocation. +func pinnedStrIn(s string, pins *runtime.Pinner) C.rc_str_in { + in := strIn(s) + if in.ptr != nil { + pins.Pin(unsafe.Pointer(in.ptr)) + } + return in +} + // cStr reads a fixed C char array of length n into a Go string. func cStr(p *C.char, n C.uint32_t) string { if n == 0 { @@ -338,14 +412,13 @@ func (s *RCSvc) Prepare(req PrepareRequest) (*PrepareResponse, error) { if s.srv == nil { return nil, errors.New("rcserver: service closed") } + var pins runtime.Pinner + defer pins.Unpin() var principal C.rc_principal_id copy((*[32]byte)(unsafe.Pointer(&principal.id[0]))[:], req.Principal[:]) - var token C.rc_str_in - if req.ClientToken != "" { - token = strIn(req.ClientToken) - } - target := strIn(req.Target) + token := pinnedStrIn(req.ClientToken, &pins) + target := pinnedStrIn(req.Target, &pins) creq := C.rc_prepare_req{ principal: principal, op: C.uint8_t(req.Op), @@ -357,8 +430,6 @@ func (s *RCSvc) Prepare(req PrepareRequest) (*PrepareResponse, error) { client_token: token, } var cresp C.rc_prepare_resp - runtime.KeepAlive(req.ClientToken) - runtime.KeepAlive(req.Target) rc := C.rc_prepare(s.srv, &creq, &cresp) if rc != C.RC_OK { return nil, statusError(rc) @@ -447,14 +518,88 @@ 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 { return nil, errors.New("rcserver: service closed") } + var pins runtime.Pinner + defer pins.Unpin() var principal C.rc_principal_id copy((*[32]byte)(unsafe.Pointer(&principal.id[0]))[:], req.Principal[:]) - in := strIn(req.SessionID) + in := pinnedStrIn(req.SessionID, &pins) creq := C.rc_ready_req{ principal: principal, session_id: in, @@ -465,7 +610,6 @@ func (s *RCSvc) ReadyTransfer(req ReadyRequest) (*ReadyResponse, error) { } var cresp C.rc_ready_resp rc := C.rc_ready_transfer(s.srv, &creq, &cresp) - runtime.KeepAlive(req.SessionID) if rc != C.RC_OK { return nil, statusError(rc) } diff --git a/rdma/rcserver/rcserver_stub.go b/rdma/rcserver/rcserver_stub.go index 6bb051e8..99e6b220 100644 --- a/rdma/rcserver/rcserver_stub.go +++ b/rdma/rcserver/rcserver_stub.go @@ -37,6 +37,7 @@ type DeviceOpts struct { TExecMs uint64 MaxReadySlots uint32 MaxStageSlots uint32 + Debug bool } // PrincipalID is the SHA-256 digest identifying the requester. @@ -110,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{} @@ -151,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"]).