xrootd: implement prepare function

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,13 +48,15 @@
 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;
 import diskCacheV111.vehicles.IoDoorInfo;
 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<FileAttribute> REQUIRED_ATTRIBUTES = Collections
+          .unmodifiableSet(EnumSet.of(FileAttribute.PNFSID, FileAttribute.TYPE,
+                FileAttribute.OWNER_GROUP, FileAttribute.OWNER, FileAttribute.ACCESS_LATENCY,
+                FileAttribute.RETENTION_POLICY));
+
     private List<FsPath> _readPaths = Collections.singletonList(FsPath.ROOT);
     private List<FsPath> _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));
+    }
 }