From f3888a31837038cb98de1b6f6a3a01c27672635a Mon Sep 17 00:00:00 2001 From: Dmitry Litvintsev Date: Thu, 3 Sep 2026 13:01:29 -0500 Subject: [PATCH] xrootd: implement prepare function Motivation: ----------- dCache XRootD never implemented the prepare XRootD protocol call. In the beginning it returned an OK w/o doing anything and recently was implemented to fail with "unsupported". Meanwhile some customers (like the ALICE experiment) have to rely on this function to actually stage files. Modification: ------------ Implemented the prepare function. Regular CUSTODIAL files are staged and pinned via a new pin manager stub, using a configurable pin lifetime (xrootd.pin.lifetime[.unit], default 12 hours). REPLICA and OUTPUT files, as well as non-regular files (directories, links), are ignored for efficiency. Evict/cancel requests release the pin via the same stub. Conflicting option combinations (e.g. stage together with evict/cancel) are rejected with an error. New configuration properties were added to configure the pin manager cell stub (xrootd.service.pinmanager[.timeout[.unit]]) and the pin lifetime. The xrootd4j library dependency was bumped to 4.6.3 to pick up the required Prepare protocol message support. Result: ------ Implemented ability to stage and release files via xrootd prepare call. Target: trunk Request: 12.x Issue: https://github.com/dCache/dcache/issues/8121 Require-book: no Require-notes: yes Patch: https://rb.dcache.org/r/14770/ Acked-by: Tigran --- .../org/dcache/xrootd/door/XrootdDoor.java | 90 +++++++ .../xrootd/door/XrootdRedirectHandler.java | 119 ++++++--- .../org/dcache/xrootd/door/xrootd.xml | 10 + .../xrootd/door/XrootdDoorPinUnpinTest.java | 231 ++++++++++++++++++ pom.xml | 3 +- skel/share/defaults/xrootd.properties | 14 ++ skel/share/services/xrootd.batch | 6 +- 7 files changed, 435 insertions(+), 38 deletions(-) create mode 100644 modules/dcache-xrootd/src/test/java/org/dcache/xrootd/door/XrootdDoorPinUnpinTest.java diff --git a/modules/dcache-xrootd/src/main/java/org/dcache/xrootd/door/XrootdDoor.java b/modules/dcache-xrootd/src/main/java/org/dcache/xrootd/door/XrootdDoor.java index f39d12865d5..a4c713b1bf6 100644 --- a/modules/dcache-xrootd/src/main/java/org/dcache/xrootd/door/XrootdDoor.java +++ b/modules/dcache-xrootd/src/main/java/org/dcache/xrootd/door/XrootdDoor.java @@ -18,6 +18,7 @@ package org.dcache.xrootd.door; import static diskCacheV111.util.MissingResourceCacheException.checkResourceNotMissing; +import static diskCacheV111.util.RetentionPolicy.CUSTODIAL; import static java.util.Objects.requireNonNull; import static org.dcache.namespace.FileAttribute.CHECKSUM; import static org.dcache.namespace.FileAttribute.MODIFICATION_TIME; @@ -47,6 +48,7 @@ import diskCacheV111.util.PermissionDeniedCacheException; import diskCacheV111.util.PnfsHandler; import diskCacheV111.util.PnfsId; +import diskCacheV111.util.ServiceUnavailableException; import diskCacheV111.vehicles.DoorRequestInfoMessage; import diskCacheV111.vehicles.DoorTransferFinishedMessage; import diskCacheV111.vehicles.IoDoorEntry; @@ -54,6 +56,7 @@ import diskCacheV111.vehicles.PnfsCreateUploadPath; import diskCacheV111.vehicles.PoolIoFileMessage; import diskCacheV111.vehicles.PoolMoverKillMessage; +import diskCacheV111.vehicles.ProtocolInfo; import dmg.cells.nucleus.AbstractCellComponent; import dmg.cells.nucleus.CellCommandListener; import dmg.cells.nucleus.CellInfoProvider; @@ -110,6 +113,8 @@ import org.dcache.namespace.FileType; import org.dcache.namespace.PermissionHandler; import org.dcache.namespace.PosixPermissionHandler; +import org.dcache.pinmanager.PinManagerPinMessage; +import org.dcache.pinmanager.PinManagerUnpinMessage; import org.dcache.poolmanager.PoolManagerStub; import org.dcache.poolmanager.PoolMonitor; import org.dcache.util.Checksum; @@ -172,12 +177,18 @@ public class XrootdDoor private static final TransferRetryPolicy RETRY_POLICY = tryOnce().doNotTimeout(); + public static final Set REQUIRED_ATTRIBUTES = Collections + .unmodifiableSet(EnumSet.of(FileAttribute.PNFSID, FileAttribute.TYPE, + FileAttribute.OWNER_GROUP, FileAttribute.OWNER, FileAttribute.ACCESS_LATENCY, + FileAttribute.RETENTION_POLICY)); + private List _readPaths = Collections.singletonList(FsPath.ROOT); private List _writePaths = Collections.singletonList(FsPath.ROOT); private CellStub _pnfsStub; private CellStub _poolStub; private PoolManagerStub _poolManagerStub; + private CellStub pinManagerStub; private CellStub _billingStub; private PoolMonitor _poolMonitor; @@ -189,6 +200,9 @@ public class XrootdDoor private int _moverTimeout = 180000; private TimeUnit _moverTimeoutUnit = TimeUnit.MILLISECONDS; + private int pinLifetime = 12; + private TimeUnit pinLifetimeUnit = TimeUnit.HOURS; + private PnfsHandler _pnfs; private String _ioQueue; @@ -266,6 +280,11 @@ public void setProxyResponseTimeoutInSeconds(int proxyResponseTimeoutInSeconds) this.proxyResponseTimeoutInSeconds = proxyResponseTimeoutInSeconds; } + @Required + public void setPinManagerStub(CellStub pinManagerStub) { + this.pinManagerStub = pinManagerStub; + } + @Required public void setPnfsStub(CellStub pnfsStub) { _pnfsStub = pnfsStub; @@ -408,6 +427,30 @@ public void setMoverTimeoutUnit(TimeUnit unit) { _moverTimeoutUnit = requireNonNull(unit); } + /** + * Returns the pin lifetime on prepare call. + */ + public int getPinLifetime() { + return pinLifetime; + } + + /** + * Pin lifetime on prepare call. + * + * @param lifetime The pin lifetime in hours. + */ + @Required + public void setPinLifetime(int lifetime) { + if (lifetime < 0) { + throw new IllegalArgumentException("Pin lifetime must be positive or 0"); + } + pinLifetime = lifetime; + } + + public void setPinLifetimeUnit(TimeUnit unit) { + pinLifetimeUnit = requireNonNull(unit); + } + @Required public void setTriedHostsEnabled(boolean triedHostsEnabled) { this.triedHostsEnabled = triedHostsEnabled; @@ -1208,6 +1251,45 @@ public int[] getMultipleFileStatuses(FsPath[] allPaths, Subject subject, return flags; } + public void pin(FsPath[] paths, InetSocketAddress client, Subject subject, + Restriction restriction) throws CacheException { + PnfsHandler pnfsHandler = new PnfsHandler(_pnfs, subject, restriction); + for (FsPath path : paths) { + PnfsId pnfsId = pnfsHandler.getPnfsIdByPath(path.toString()); + FileAttributes attr = pnfsHandler.getFileAttributes(path, REQUIRED_ATTRIBUTES); + if (attr.getRetentionPolicy() != CUSTODIAL || attr.getFileType() != FileType.REGULAR) { + continue; + } + ProtocolInfo protocolInfo = new XrootdProtocolInfo(XROOTD_PROTOCOL_STRING, + XrootdProtocol.PROTOCOL_VERSION_MAJOR, XrootdProtocol.PROTOCOL_VERSION_MINOR, + client, new CellPath(getCellName(), getCellDomainName()), pnfsId, 0, null, null); + long lifetime = pinLifetimeUnit.toMillis(pinLifetime); + try { + PinManagerPinMessage message = new PinManagerPinMessage(attr, protocolInfo, + restriction, getRequestId(subject), lifetime); + message.setReplyWhenStarted(true); + pinManagerStub.sendAndWait(message); + } catch (NoRouteToCellException | InterruptedException e) { + throw new ServiceUnavailableException(e.getMessage()); + } + } + } + + public void unpin(FsPath[] paths, Subject subject, Restriction restriction) + throws CacheException { + PnfsHandler pnfsHandler = new PnfsHandler(_pnfs, subject, restriction); + for (FsPath path : paths) { + PnfsId pnfsId = pnfsHandler.getPnfsIdByPath(path.toString()); + try { + PinManagerUnpinMessage message = new PinManagerUnpinMessage(pnfsId); + message.setRequestId(getRequestId(subject)); + pinManagerStub.sendAndWait(message); + } catch (NoRouteToCellException | InterruptedException e) { + throw new ServiceUnavailableException(e.getMessage()); + } + } + } + public int nextTpcPlaceholder() { synchronized (_tpcFdIndex) { Integer next = _tpcPlaceholder.getAndIncrement(); @@ -1343,4 +1425,12 @@ public String call() throws Exception { return String.format("Mover %s not found on pool %s.", id, pool); } } + + private String getRequestId(Subject subject) throws PermissionDeniedCacheException { + if (Subjects.isNobody(subject)) { + throw new PermissionDeniedCacheException("cannot get request id for user."); + } + + return String.valueOf(Subjects.getUid(subject)); + } } diff --git a/modules/dcache-xrootd/src/main/java/org/dcache/xrootd/door/XrootdRedirectHandler.java b/modules/dcache-xrootd/src/main/java/org/dcache/xrootd/door/XrootdRedirectHandler.java index fded6a3b81b..834d325a9df 100644 --- a/modules/dcache-xrootd/src/main/java/org/dcache/xrootd/door/XrootdRedirectHandler.java +++ b/modules/dcache-xrootd/src/main/java/org/dcache/xrootd/door/XrootdRedirectHandler.java @@ -39,6 +39,7 @@ import static org.dcache.xrootd.protocol.XrootdProtocol.kXR_gx; import static org.dcache.xrootd.protocol.XrootdProtocol.kXR_mkpath; import static org.dcache.xrootd.protocol.XrootdProtocol.kXR_new; +import static org.dcache.xrootd.protocol.XrootdProtocol.kXR_ok; import static org.dcache.xrootd.protocol.XrootdProtocol.kXR_open_apnd; import static org.dcache.xrootd.protocol.XrootdProtocol.kXR_open_read; import static org.dcache.xrootd.protocol.XrootdProtocol.kXR_open_updt; @@ -53,6 +54,7 @@ import static org.dcache.xrootd.protocol.XrootdProtocol.kXR_ur; import static org.dcache.xrootd.protocol.XrootdProtocol.kXR_uw; import static org.dcache.xrootd.protocol.XrootdProtocol.kXR_ux; +import static org.dcache.xrootd.protocol.XrootdProtocol.kXR_Unsupported; import static org.dcache.xrootd.util.TriedRc.ENOENT; import static org.dcache.xrootd.util.TriedRc.IOERR; @@ -69,6 +71,7 @@ import diskCacheV111.util.QuotaExceededCacheException; import diskCacheV111.util.TimeoutCacheException; import dmg.cells.nucleus.CellPath; +import dmg.cells.nucleus.NoRouteToCellException; import io.netty.channel.ChannelHandlerContext; import java.io.IOException; import java.net.InetSocketAddress; @@ -114,6 +117,7 @@ import org.dcache.xrootd.protocol.messages.OpenRequest; import org.dcache.xrootd.protocol.messages.OpenResponse; import org.dcache.xrootd.protocol.messages.PrepareRequest; +import org.dcache.xrootd.protocol.messages.PrepareResponse; import org.dcache.xrootd.protocol.messages.QueryRequest; import org.dcache.xrootd.protocol.messages.QueryResponse; import org.dcache.xrootd.protocol.messages.RedirectResponse; @@ -141,7 +145,7 @@ */ public class XrootdRedirectHandler extends ConcurrentXrootdRequestHandler { - private static final Logger _log = + private static final Logger LOGGER = LoggerFactory.getLogger(XrootdRedirectHandler.class); private static final String EFFECTIVE_ROOT_NAME = "org.dcache.effectiveRoot"; @@ -157,7 +161,7 @@ private static Map safelyExtractOpaque(String opaqueString) { opaque = new HashMap<>(); } } catch (ParseException e) { - _log.warn("Ignoring malformed open opaque {}: {}", opaqueString, + LOGGER.warn("Ignoring malformed open opaque {}: {}", opaqueString, e.getMessage()); opaque = new HashMap<>(); } @@ -279,23 +283,23 @@ public void userEventTriggered(ChannelHandlerContext ctx, Object event) throws E @Override public void exceptionCaught(ChannelHandlerContext ctx, Throwable t) { if (t instanceof ClosedChannelException) { - _log.info("Connection unexpectedly closed on {}, cause {}.", ctx.channel(), + LOGGER.info("Connection unexpectedly closed on {}, cause {}.", ctx.channel(), Throwables.getRootCause(t).toString()); } else if (t instanceof RuntimeException || t instanceof Error) { Thread me = Thread.currentThread(); me.getUncaughtExceptionHandler().uncaughtException(me, t); } else if (!isHealthCheck() || !(t instanceof IOException)) { - _log.warn("exception caught on {}: {}, cause {}.", ctx.channel(), t.getMessage(), + LOGGER.warn("exception caught on {}: {}, cause {}.", ctx.channel(), t.getMessage(), Throwables.getRootCause(t).toString()); } else { - _log.info("IO exception caught during health check on {}: {}, cause {}.", ctx.channel(), + LOGGER.info("IO exception caught during health check on {}: {}, cause {}.", ctx.channel(), t.getMessage(), Throwables.getRootCause(t).toString()); } } @Override public void channelInactive(ChannelHandlerContext ctx) throws Exception { - _log.info("channel inactive event received on {}.", ctx.channel()); + LOGGER.info("channel inactive event received on {}.", ctx.channel()); /** * If the doOnOpen call has not yet returned, interrupt its thread. @@ -376,8 +380,8 @@ protected XrootdResponse doOnOpen(ChannelHandlerContext ctx, OpenRe FilePerm neededPerm = req.getRequiredPermission(); - _log.info("Opening {} for {}", req.getPath(), neededPerm.xmlText()); - if (_log.isDebugEnabled()) { + LOGGER.info("Opening {} for {}", req.getPath(), neededPerm.xmlText()); + if (LOGGER.isDebugEnabled()) { logDebugOnOpen(req); } @@ -390,11 +394,11 @@ protected XrootdResponse doOnOpen(ChannelHandlerContext ctx, OpenRe size = Long.valueOf(value); } } catch (NumberFormatException exception) { - _log.warn("Ignoring malformed oss.asize: {}", + LOGGER.warn("Ignoring malformed oss.asize: {}", exception.getMessage()); } - _log.info("OPAQUE : {}", opaque); + LOGGER.info("OPAQUE : {}", opaque); Set triedHosts = extractTriedHosts(opaque); UUID uuid = UUID.randomUUID(); @@ -556,11 +560,11 @@ private InetSocketAddress getRedirect(XrootdTransfer transfer) throws IOExceptio */ String host = redirectAddress.getHostName(); if (InetAddresses.isInetAddress(host)) { - _log.warn("Unable to resolve IP address {} " + LOGGER.warn("Unable to resolve IP address {} " + "to a canonical name", host); } - _log.info("Redirecting to {}, {}", host, redirectAddress.getPort()); + LOGGER.info("Redirecting to {}, {}", host, redirectAddress.getPort()); return redirectAddress; } @@ -601,7 +605,7 @@ private InetSocketAddress getRedirect(XrootdTransfer transfer) throws IOExceptio restriction, remoteHost); int fd = _door.nextTpcPlaceholder(); - _log.debug("placement response to {} sent to {} with fhandle {}.", + LOGGER.debug("placement response to {} sent to {} with fhandle {}.", req, remoteHost, fd); return new OpenResponse(req, fd, null, null, @@ -610,7 +614,7 @@ private InetSocketAddress getRedirect(XrootdTransfer transfer) throws IOExceptio String tpcKey = opaque.get("tpc.key"); if (tpcKey == null) { - _log.debug("{} -- not a third-party request.", req); + LOGGER.debug("{} -- not a third-party request.", req); return null; // proceed as usual with mover + redirect } @@ -620,7 +624,7 @@ private InetSocketAddress getRedirect(XrootdTransfer transfer) throws IOExceptio * to the TPC client. */ if (req.getSession().getDelegatedCredential() != null) { - _log.debug("{} -- third-party request with delegation.", req); + LOGGER.debug("{} -- third-party request with delegation.", req); return null; // proceed as usual with mover + redirect } @@ -651,14 +655,14 @@ private InetSocketAddress getRedirect(XrootdTransfer transfer) throws IOExceptio * remove the key and return immediately. */ _door.removeTpcPlaceholder(tpcKey); - _log.debug("{} -- request contains authorization token.", req); + LOGGER.debug("{} -- request contains authorization token.", req); return null; // proceed as usual with mover + redirect } info.addInfoFromOpaque(slfn, opaque); /** updates the status **/ switch (info.verify(remoteHost, slfn, opaque.get("tpc.org"))) { case READY: - _log.debug("Open request {} from destination server, info {}: " + LOGGER.debug("Open request {} from destination server, info {}: " + "OK to proceed.", req, info); /* @@ -669,7 +673,7 @@ private InetSocketAddress getRedirect(XrootdTransfer transfer) throws IOExceptio */ return null; case PENDING: - _log.debug("Open request {} from destination server, info {}: " + LOGGER.debug("Open request {} from destination server, info {}: " + "PENDING client open; sending WAIT-RETRY.", req, info); /* @@ -690,7 +694,7 @@ private InetSocketAddress getRedirect(XrootdTransfer transfer) throws IOExceptio * read permissions on this file. */ String error = "invalid open request (file permissions)."; - _log.warn("Open request {} from destination server, info {}: " + LOGGER.warn("Open request {} from destination server, info {}: " + "ERROR: {}.", req, info, error); _door.removeTpcPlaceholder(info.getFd()); @@ -700,7 +704,7 @@ private InetSocketAddress getRedirect(XrootdTransfer transfer) throws IOExceptio case CANCELLED: error = info.isExpired() ? "ttl expired" : "dst, path or org" + " did not match"; - _log.warn("Open request {} from destination server, info {}: " + LOGGER.warn("Open request {} from destination server, info {}: " + "CANCELLED: {}.", req, info, error); _door.removeTpcPlaceholder(info.getFd()); @@ -714,7 +718,7 @@ private InetSocketAddress getRedirect(XrootdTransfer transfer) throws IOExceptio * The request originated from the client, indicating that this door is the source. */ if (opaque.containsKey("tpc.dst")) { - _log.debug("Open request {} from client to door as source, " + LOGGER.debug("Open request {} from client to door as source, " + "info {}: OK.", req, info); FileStatus status = _door.getFileStatus(fsPath, subject, restriction, remoteHost); int flags = status.getFlags(); @@ -741,7 +745,7 @@ private InetSocketAddress getRedirect(XrootdTransfer transfer) throws IOExceptio * allow the write mover to be started on the selected pool. */ if (opaque.containsKey("tpc.src")) { - _log.debug("Open request {} from client to door as destination: OK;" + LOGGER.debug("Open request {} from client to door as destination: OK;" + "removing info {}.", req, info); _door.removeTpcPlaceholder(info.getFd()); /* @@ -772,13 +776,13 @@ private Set extractTriedHosts(Map opaque) { String triedrc = Strings.emptyToNull(opaque.get("triedrc")); if (!_door.isTriedHostsEnabled()) { - _log.debug("tried hosts option not enabled, ignoring 'tried={},triedrc={}'.", + LOGGER.debug("tried hosts option not enabled, ignoring 'tried={},triedrc={}'.", tried, triedrc); return Collections.EMPTY_SET; } if (tried == null || triedrc == null) { - _log.debug("tried {}, triedrc {}, ignoring.", tried, triedrc); + LOGGER.debug("tried {}, triedrc {}, ignoring.", tried, triedrc); return Collections.EMPTY_SET; } @@ -802,12 +806,12 @@ private Set extractTriedHosts(Map opaque) { if (value.equals(ENOENT.name()) || value.equals(IOERR.name())) { String host = hostNames.get(i); triedHosts.add(host); - _log.debug("tried {}, triedrc {}, {}.", + LOGGER.debug("tried {}, triedrc {}, {}.", host, value, TriedRc.valueOf(value).description()); } } - _log.debug("tried hosts : {}", triedHosts); + LOGGER.debug("tried hosts : {}", triedHosts); return triedHosts; } @@ -834,7 +838,7 @@ private String appSpecificQueue(OpenRequest req) { Map attr = OpaqueStringParser.getOpaqueMap(token); ioqueue = _appIoQueues.get(attr.get("xrd.appname")); } catch (ParseException e) { - _log.debug("Ignoring malformed login token {}: {}", token, e.getMessage()); + LOGGER.debug("Ignoring malformed login token {}: {}", token, e.getMessage()); } return ioqueue; @@ -848,7 +852,7 @@ private String appSpecificQueue(OpenRequest req) { protected XrootdResponse doOnClose(ChannelHandlerContext ctx, CloseRequest msg) throws XrootdException { int fd = msg.getFileHandle(); - _log.debug("doOnClose: removing tpc info for {}.", fd); + LOGGER.debug("doOnClose: removing tpc info for {}.", fd); if (_door.removeTpcPlaceholder(fd)) { return withOk(msg); } else { @@ -921,7 +925,7 @@ protected XrootdResponse doOnRm(ChannelHandlerContext ctx, RmRequest throw new XrootdException(kXR_ArgMissing, "no path specified"); } - _log.info("Trying to delete {}", req.getPath()); + LOGGER.info("Trying to delete {}", req.getPath()); try { LoginSessionInfo loginSessionInfo = sessionInfo(); @@ -949,7 +953,7 @@ protected XrootdResponse doOnRmDir(ChannelHandlerContext ctx, RmDi throw new XrootdException(kXR_ArgMissing, "no path specified"); } - _log.info("Trying to delete directory {}", req.getPath()); + LOGGER.info("Trying to delete directory {}", req.getPath()); try { LoginSessionInfo loginSessionInfo = sessionInfo(); @@ -976,7 +980,7 @@ protected XrootdResponse doOnMkDir(ChannelHandlerContext ctx, MkDi throw new XrootdException(kXR_ArgMissing, "no path specified"); } - _log.info("Trying to create directory {}", req.getPath()); + LOGGER.info("Trying to create directory {}", req.getPath()); try { LoginSessionInfo loginSessionInfo = sessionInfo(); @@ -1011,7 +1015,7 @@ protected XrootdResponse doOnMv(ChannelHandlerContext ctx, MvRequest throw new XrootdException(kXR_ArgMissing, "no target path specified"); } - _log.info("Trying to rename {} to {}", req.getSourcePath(), req.getTargetPath()); + LOGGER.info("Trying to rename {} to {}", req.getSourcePath(), req.getTargetPath()); try { LoginSessionInfo loginSessionInfo = sessionInfo(); @@ -1105,7 +1109,7 @@ protected XrootdResponse doOnDirList(ChannelHandlerContext ctx, throw new XrootdException(kXR_ArgMissing, "no source path specified"); } - _log.info("Listing directory {}", listPath); + LOGGER.info("Listing directory {}", listPath); FsPath fullListPath = createFullPath(listPath, safelyExtractOpaque(request.getOpaque())); if (!_door.isReadAllowed(fullListPath)) { @@ -1129,6 +1133,49 @@ protected XrootdResponse doOnDirList(ChannelHandlerContext ctx, } } + @Override + protected XrootdResponse doOnPrepare(ChannelHandlerContext ctx, + PrepareRequest request) + throws XrootdException { + + if (request.getPaths().length == 0) { + throw new XrootdException(kXR_ArgMissing, "no paths specified"); + } + + try { + FsPath[] paths = new FsPath[request.getPaths().length]; + for (int i = 0; i < paths.length; i++) { + paths[i] = createFullPath(request.getPaths()[i], + safelyExtractOpaque(request.getOpaques()[i])); + } + LoginSessionInfo loginSessionInfo = sessionInfo(); + Subject subject = loginSessionInfo.getSubject(); + Restriction restriction = loginSessionInfo.getRestriction(); + InetSocketAddress clientAddress = getSourceAddress(); + + if (request.getOptions() == 0 + || (request.isStage() && !(request.isEvict() || request.isCancel()))) { + _door.pin(paths, clientAddress, subject, restriction); + return new PrepareResponse(request, kXR_ok, new byte[paths.length]); + } else if (request.isStage() && (request.isEvict() || request.isCancel())) { + throw new XrootdException(kXR_InvalidRequest, + "Invalid parameters / conflicting options"); + } else if (request.isCancel() || request.isEvict()) { + _door.unpin(paths, subject, restriction); + return new PrepareResponse(request, kXR_ok, new byte[paths.length]); + } else { + throw new XrootdException(kXR_Unsupported, "Unsupported option(s)"); + } + } catch (PermissionDeniedCacheException e) { + throw xrootdException(e); + } catch (TimeoutCacheException e) { + throw xrootdException(e.getRc(), "Internal timeout"); + } catch (CacheException e) { + throw xrootdException(e.getRc(), + String.format("Failed to prepare files (%s [%d])", e.getMessage(), e.getRc())); + } + } + private void logDebugOnOpen(OpenRequest req) { int options = req.getOptions(); String openFlags = @@ -1171,7 +1218,7 @@ private void logDebugOnOpen(OpenRequest req) { openFlags += " kXR_posc"; } - _log.debug("open flags: {}", openFlags); + LOGGER.debug("open flags: {}", openFlags); int mode = req.getUMask(); String s = ""; @@ -1228,7 +1275,7 @@ private void logDebugOnOpen(OpenRequest req) { s += "-"; } - _log.debug("mode to apply to open path: {}", s); + LOGGER.debug("mode to apply to open path: {}", s); } /** @@ -1442,7 +1489,7 @@ private synchronized void unsetOnOpenThread() { private synchronized void interruptOnOpenThread() { if (onOpenThread != null) { - _log.info("{} called interruptOnOpenThread; interrupting {}.", Thread.currentThread(), + LOGGER.info("{} called interruptOnOpenThread; interrupting {}.", Thread.currentThread(), onOpenThread); onOpenThread.interrupt(); } diff --git a/modules/dcache-xrootd/src/main/resources/org/dcache/xrootd/door/xrootd.xml b/modules/dcache-xrootd/src/main/resources/org/dcache/xrootd/door/xrootd.xml index 5598f9e4c08..a91b1da2356 100644 --- a/modules/dcache-xrootd/src/main/resources/org/dcache/xrootd/door/xrootd.xml +++ b/modules/dcache-xrootd/src/main/resources/org/dcache/xrootd/door/xrootd.xml @@ -27,6 +27,13 @@ + + Pin manager cell stub + + + + + PNFS manager communication stub @@ -175,6 +182,7 @@ Gateway between xroot protocol handler and dCache + @@ -191,6 +199,8 @@ + + diff --git a/modules/dcache-xrootd/src/test/java/org/dcache/xrootd/door/XrootdDoorPinUnpinTest.java b/modules/dcache-xrootd/src/test/java/org/dcache/xrootd/door/XrootdDoorPinUnpinTest.java new file mode 100644 index 00000000000..c0e6ccaccdd --- /dev/null +++ b/modules/dcache-xrootd/src/test/java/org/dcache/xrootd/door/XrootdDoorPinUnpinTest.java @@ -0,0 +1,231 @@ +/* +COPYRIGHT STATUS: +Dec 1st 2001, Fermi National Accelerator Laboratory (FNAL) documents and +software are sponsored by the U.S. Department of Energy under Contract No. +DE-AC02-76CH03000. Therefore, the U.S. Government retains a world-wide +non-exclusive, royalty-free license to publish or reproduce these documents +and software for U.S. Government purposes. All documents and software +available from this server are protected under the U.S. and Foreign +Copyright Laws, and FNAL reserves all rights. + +Distribution of the software available from this server is free of +charge subject to the user following the terms of the Fermitools +Software Legal Information. + +Redistribution and/or modification of the software shall be accompanied +by the Fermitools Software Legal Information (including the copyright +notice). + +The user is asked to feed back problems, benefits, and/or suggestions +about the software to the Fermilab Software Providers. + +Neither the name of Fermilab, the URA, nor the names of the contributors +may be used to endorse or promote products derived from this software +without specific prior written permission. + +DISCLAIMER OF LIABILITY (BSD): + +THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS +"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT +LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS +FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL FERMILAB, +OR THE URA, OR THE U.S. DEPARTMENT of ENERGY, OR CONTRIBUTORS BE LIABLE +FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR +CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT +OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR +BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF +LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING +NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS +SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + +Liabilities of the Government: + +This software is provided by URA, independent from its Prime Contract +with the U.S. Department of Energy. URA is acting independently from +the Government and in its own private capacity and is not acting on +behalf of the U.S. Government, nor as its contractor nor its agent. +Correspondingly, it is understood and agreed that the U.S. Government +has no connection to this software and in no manner whatsoever shall +be liable for nor assume any responsibility or obligation for any claim, +cost, or damages arising out of or resulting from the use of the software +available from this server. + +Export Control: + +All documents and software available from this server are subject to U.S. +export control laws. Anyone downloading information from this server is +obligated to secure any necessary Government licenses before exporting +documents or software obtained from this server. +*/ +package org.dcache.xrootd.door; + +import static org.hamcrest.MatcherAssert.assertThat; +import static org.hamcrest.Matchers.is; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import com.google.common.util.concurrent.Futures; +import diskCacheV111.util.AccessLatency; +import diskCacheV111.util.FsPath; +import diskCacheV111.util.PermissionDeniedCacheException; +import diskCacheV111.util.PnfsHandler; +import diskCacheV111.util.PnfsId; +import diskCacheV111.util.RetentionPolicy; +import diskCacheV111.util.ServiceUnavailableException; +import dmg.cells.nucleus.CellAddressCore; +import dmg.cells.nucleus.CellMessage; +import dmg.cells.nucleus.CellPath; +import dmg.cells.nucleus.NoRouteToCellException; +import java.net.InetSocketAddress; +import java.util.concurrent.TimeUnit; +import javax.security.auth.Subject; +import org.dcache.auth.Subjects; +import org.dcache.auth.attributes.Restrictions; +import org.dcache.cells.CellStub; +import org.dcache.namespace.FileAttribute; +import org.dcache.namespace.FileType; +import org.dcache.pinmanager.PinManagerPinMessage; +import org.dcache.pinmanager.PinManagerUnpinMessage; +import org.dcache.vehicles.FileAttributes; +import org.dcache.vehicles.PnfsGetFileAttributes; +import org.junit.Before; +import org.junit.Test; +import org.mockito.ArgumentCaptor; + +/** + * Unit tests for {@link XrootdDoor#pin} and {@link XrootdDoor#unpin}, the methods backing the + * xrootd "prepare" (stage/pin, evict/cancel-unpin) support. + */ +public class XrootdDoorPinUnpinTest { + + private static final PnfsId PNFS_ID = new PnfsId("000000000000000000000000000000000001"); + private static final FsPath PATH = FsPath.create("/data/file"); + private static final InetSocketAddress CLIENT = new InetSocketAddress("127.0.0.1", 1094); + + private CellStub pnfsCellStub; + private CellStub pinManagerStub; + private XrootdDoor door; + private Subject subject; + + @Before + public void setUp() { + pnfsCellStub = mock(CellStub.class); + pinManagerStub = mock(CellStub.class); + + door = new XrootdDoor(); + door.setCellAddress(new CellAddressCore("xrootd", "local")); + door.setPnfsHandler(new PnfsHandler(pnfsCellStub)); + door.setPinManagerStub(pinManagerStub); + + subject = Subjects.of(1000, 1000, new int[]{1000}); + } + + /** + * Stubs the mocked PNFS CellStub so that any {@link PnfsGetFileAttributes} request is + * answered with a PNFSID plus, if RETENTION_POLICY was requested, the given retention policy + * (mimicking what {@link XrootdDoor#pin} needs to make its CUSTODIAL check). + */ + private void stubFileAttributes(RetentionPolicy retentionPolicy) { + when(pnfsCellStub.send(any(PnfsGetFileAttributes.class), anyLong())) + .thenAnswer(invocation -> { + PnfsGetFileAttributes request = invocation.getArgument(0); + FileAttributes attributes = new FileAttributes(); + attributes.setPnfsId(PNFS_ID); + if (request.getRequestedAttributes().contains(FileAttribute.RETENTION_POLICY)) { + attributes.setRetentionPolicy(retentionPolicy); + attributes.setAccessLatency(AccessLatency.NEARLINE); + attributes.setOwner(0); + attributes.setGroup(0); + attributes.setFileType(FileType.REGULAR); + } + request.setFileAttributes(attributes); + return Futures.immediateFuture(request); + }); + } + + @Test + public void pinSendsPinMessageForCustodialFile() throws Exception { + stubFileAttributes(RetentionPolicy.CUSTODIAL); + + door.pin(new FsPath[]{PATH}, CLIENT, subject, Restrictions.none()); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(PinManagerPinMessage.class); + verify(pinManagerStub).sendAndWait(captor.capture()); + + PinManagerPinMessage message = captor.getValue(); + assertThat(message.getPnfsId(), is(PNFS_ID)); + assertThat(message.getRequestId(), is("1000")); + assertThat(message.getLifetime(), is(TimeUnit.HOURS.toMillis(12))); + assertThat(message.isReplyWhenStarted(), is(true)); + } + + @Test + public void pinSkipsNonCustodialFile() throws Exception { + stubFileAttributes(RetentionPolicy.REPLICA); + + door.pin(new FsPath[]{PATH}, CLIENT, subject, Restrictions.none()); + + verify(pinManagerStub, never()).sendAndWait(any(PinManagerPinMessage.class)); + } + + @Test(expected = ServiceUnavailableException.class) + public void pinWrapsNoRouteToCellExceptionAsServiceUnavailable() throws Exception { + stubFileAttributes(RetentionPolicy.CUSTODIAL); + when(pinManagerStub.sendAndWait(any(PinManagerPinMessage.class))) + .thenThrow(noRouteToCellException()); + + door.pin(new FsPath[]{PATH}, CLIENT, subject, Restrictions.none()); + } + + @Test(expected = PermissionDeniedCacheException.class) + public void pinRejectsAnonymousSubject() throws Exception { + stubFileAttributes(RetentionPolicy.CUSTODIAL); + + door.pin(new FsPath[]{PATH}, CLIENT, Subjects.NOBODY, Restrictions.none()); + } + + @Test + public void unpinSendsUnpinMessageWithRequestId() throws Exception { + stubFileAttributes(RetentionPolicy.CUSTODIAL); + + door.unpin(new FsPath[]{PATH}, subject, Restrictions.none()); + + ArgumentCaptor captor = + ArgumentCaptor.forClass(PinManagerUnpinMessage.class); + verify(pinManagerStub).sendAndWait(captor.capture()); + + PinManagerUnpinMessage message = captor.getValue(); + assertThat(message.getPnfsId(), is(PNFS_ID)); + assertThat(message.getRequestId(), is("1000")); + } + + @Test(expected = ServiceUnavailableException.class) + public void unpinWrapsNoRouteToCellExceptionAsServiceUnavailable() throws Exception { + stubFileAttributes(RetentionPolicy.CUSTODIAL); + when(pinManagerStub.sendAndWait(any(PinManagerUnpinMessage.class))) + .thenThrow(noRouteToCellException()); + + door.unpin(new FsPath[]{PATH}, subject, Restrictions.none()); + } + + @Test(expected = PermissionDeniedCacheException.class) + public void unpinRejectsAnonymousSubject() throws Exception { + stubFileAttributes(RetentionPolicy.CUSTODIAL); + + door.unpin(new FsPath[]{PATH}, Subjects.NOBODY, Restrictions.none()); + } + + /** + * Builds a {@link NoRouteToCellException}, which requires a real {@link CellMessage} + * envelope rather than a mock. + */ + private static NoRouteToCellException noRouteToCellException() { + CellMessage envelope = new CellMessage(new CellPath("pinmanager"), "noop"); + return new NoRouteToCellException(envelope, "no route"); + } +} diff --git a/pom.xml b/pom.xml index 14c86ed121f..2bb8f030b4a 100644 --- a/pom.xml +++ b/pom.xml @@ -1,3 +1,4 @@ + 4.0.0 org.dcache @@ -68,7 +69,7 @@ 7.6.0 2.12.0 9.4.57.v20241219 - 4.6.2 + 4.6.3 2.47 3.0.0 4.2.15.Final diff --git a/skel/share/defaults/xrootd.properties b/skel/share/defaults/xrootd.properties index a4c94f65131..0d97a265283 100644 --- a/skel/share/defaults/xrootd.properties +++ b/skel/share/defaults/xrootd.properties @@ -70,6 +70,13 @@ xrootd.limits.login-cache.size=500 # (one-of?true|false)xrootd.enable.proxy-protocol = false +# Cell address of pinmanager service +xrootd.service.pinmanager=${dcache.service.pinmanager} + +# Timeout for pinmanager requests +xrootd.service.pinmanager.timeout=300000 +(one-of?MILLISECONDS|SECONDS|MINUTES|HOURS|DAYS)xrootd.service.pinmanager.timeout.unit=MILLISECONDS + # Cell address of poolmanager service xrootd.service.poolmanager=${xrootd.service.poolmanager-space-${xrootd.enable.space-reservation}} (immutable)xrootd.service.poolmanager-space-false=${dcache.service.poolmanager} @@ -263,6 +270,13 @@ xrootd.mover.queue = xrootd.mover.timeout = 180000 (one-of?MILLISECONDS|SECONDS|MINUTES|HOURS|DAYS)xrootd.mover.timeout.unit=MILLISECONDS +# ---- Pin lifetime on prepare call +# +# Lifetime of pins created by xrootd prepare call +# +xrootd.pin.lifetime = 12 +(one-of?MILLISECONDS|SECONDS|MINUTES|HOURS|DAYS)xrootd.pin.lifetime.unit = HOURS + # ----- Custom kXR_Qconfig responses # # xroot clients may query the server configuration using a diff --git a/skel/share/services/xrootd.batch b/skel/share/services/xrootd.batch index eb5ca7db510..fd08f0d1944 100644 --- a/skel/share/services/xrootd.batch +++ b/skel/share/services/xrootd.batch @@ -24,6 +24,9 @@ check -strong xrootd.limits.login-cache.size check -strong xrootd.root check -strong xrootd.service.pool.timeout check -strong xrootd.service.pool.timeout.unit +check -strong xrootd.service.pinmanager +check -strong xrootd.service.pinmanager.timeout +check -strong xrootd.service.pinmanager.timeout.unit check -strong xrootd.service.poolmanager check -strong xrootd.service.poolmanager.timeout check -strong xrootd.service.poolmanager.timeout.unit @@ -56,6 +59,8 @@ check -strong xrootd.security.tls.ca.ocsp-mode check -strong xrootd.mover.timeout check -strong xrootd.mover.timeout.unit +check -strong xrootd.pin.lifetime +check -strong xrootd.pin.lifetime.unit check xrootd.mover.queue check -strong xrootd.plugins check -strong xrootd.authz.user @@ -69,4 +74,3 @@ create org.dcache.cells.UniversalSpringCell ${xrootd.cell.name} \ -profiles=kafka-${xrootd.enable.kafka}\ -subscribe=${xrootd.cell.subscribe} \ -cellClass=XrootdDoor" -