CoordinationService.java

Go to the documentation of this file.

/*
 * Autopsy Forensic Browser
 *
 * Copyright 2011-2018 Basis Technology Corp.
 * Contact: carrier <at> sleuthkit <dot> org
 *
 * 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.
 */
package org.sleuthkit.autopsy.coordinationservice;
 
import java.io.IOException;
import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import javax.annotation.concurrent.GuardedBy;
import javax.annotation.concurrent.ThreadSafe;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.locks.InterProcessMutex;
import org.apache.curator.framework.recipes.locks.InterProcessReadWriteLock;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.curator.utils.ZKPaths;
import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.NoNodeException;
import org.apache.zookeeper.ZooDefs;
import org.openide.util.Lookup;
import org.sleuthkit.autopsy.coordinationservice.utils.CoordinationServiceUtils;
import org.sleuthkit.autopsy.core.UserPreferences;

@ThreadSafe
public final class CoordinationService {
 
    private static final int SESSION_TIMEOUT_MILLISECONDS = 300000;
    private static final int CONNECTION_TIMEOUT_MILLISECONDS = 300000;
    private static final int PORT_OFFSET = 1000; // When run in Solr, ZooKeeper defaults to Solr port + 1000
    private static final String DEFAULT_NAMESPACE_ROOT = "autopsy";
    @GuardedBy("CoordinationService.class")
    private static CoordinationService instance;
    private final CuratorFramework curator;
    @GuardedBy("categoryNodeToPath")
    private final Map<String, String> categoryNodeToPath;

    public synchronized static CoordinationService getInstance() throws CoordinationServiceException {
        if (null == instance) {
            String rootNode;
            Collection<? extends CoordinationServiceNamespace> providers = Lookup.getDefault().lookupAll(CoordinationServiceNamespace.class);
            Iterator<? extends CoordinationServiceNamespace> it = providers.iterator();
            if (it.hasNext()) {
                rootNode = it.next().getNamespaceRoot();
            } else {
                rootNode = DEFAULT_NAMESPACE_ROOT;
            }
            try {
                instance = new CoordinationService(rootNode);
            } catch (IOException | KeeperException | CoordinationServiceException ex) {
                throw new CoordinationServiceException("Failed to create coordination service", ex);
            } catch (InterruptedException ex) {
                /*
                 * The interrupted exception should be propagated to support
                 * task cancellation. To avoid a public API change here, restore
                 * the interrupted flag and then throw the InterruptedException
                 * in its wrapper.
                 */
                Thread.currentThread().interrupt();
                throw new CoordinationServiceException("Failed to create coordination service", ex);
            }
        }
        return instance;
    }

    private CoordinationService(String rootNodeName) throws InterruptedException, IOException, KeeperException, CoordinationServiceException {
 
        // read ZK connection info
        String hostName = UserPreferences.getZkServerHost();
        String port = UserPreferences.getZkServerPort();
        if (hostName.isEmpty() || port.isEmpty()) {
            // use defaults for embedded ZK that runs on Solr server
            hostName = UserPreferences.getIndexingServerHost();
            int portInt = Integer.valueOf(UserPreferences.getIndexingServerPort()) + PORT_OFFSET;
            port = Integer.toString(portInt);
        }
        if (false == CoordinationServiceUtils.isZooKeeperAccessible(hostName, port)) {
            throw new CoordinationServiceException("Unable to access ZooKeeper");
        }
        
        // We are using ZK for all coordination/locking, so ZK connection info cannot be changed.
        // A reboot is required in order to use a different ZK server for coordination services.
        /*
         * Connect to ZooKeeper via Curator.
         */
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        String connectString = hostName + ":" + port;
        curator = CuratorFrameworkFactory.newClient(connectString, SESSION_TIMEOUT_MILLISECONDS, CONNECTION_TIMEOUT_MILLISECONDS, retryPolicy);
        curator.start();
 
        /*
         * Create the top-level root and category nodes.
         */
        String rootNode = rootNodeName;
 
        if (!rootNode.startsWith("/")) {
            rootNode = "/" + rootNode;
        }
        categoryNodeToPath = new ConcurrentHashMap<>();
        for (CategoryNode node : CategoryNode.values()) {
            String nodePath = rootNode + "/" + node.getDisplayName();
            try {
                curator.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT).withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE).forPath(nodePath);
            } catch (KeeperException ex) {
                if (ex.code() != KeeperException.Code.NODEEXISTS) {
                    throw ex;
                }
            } catch (Exception ex) {
                throw new CoordinationServiceException("Curator experienced an error", ex);
            }
            categoryNodeToPath.put(node.getDisplayName(), nodePath);
        }
    }
    
    private String upsertNodePath(CategoryNode category, String nodePath) throws CoordinationServiceException {
        String fullNodePath = getFullyQualifiedNodePath(category, nodePath);
 
        // ZKPaths.mkdirs throws an exception with trailing slash.  
        // Remove trailing slash if slash is present to prevent this issue.
        while(fullNodePath.endsWith("/")) {
            fullNodePath = fullNodePath.substring(0, fullNodePath.length() - 1);
        }
            
        try {
            // ensure leading path is present
            ZKPaths.mkdirs(curator.getZookeeperClient().getZooKeeper(), fullNodePath);
            return fullNodePath;
        } catch (Exception ex) {
            throw new CoordinationServiceException("An error occurred while creating node path at: " + fullNodePath, ex);
        }   
    }

    public Lock tryGetExclusiveLock(CategoryNode category, String nodePath, int timeOut, TimeUnit timeUnit) throws CoordinationServiceException, InterruptedException {
        String fullNodePath = "";
        try {
            // ensure node is present
            fullNodePath = upsertNodePath(category, nodePath);
            InterProcessReadWriteLock lock = new InterProcessReadWriteLock(curator, fullNodePath);
            if (lock.writeLock().acquire(timeOut, timeUnit)) {
                return new Lock(nodePath, lock.writeLock());
            } else {
                return null;
            }
        } catch (Exception ex) {
            if (ex instanceof InterruptedException) {
                throw (InterruptedException) ex;
            } else {
                throw new CoordinationServiceException(String.format("Failed to get exclusive lock for %s", fullNodePath), ex);
            }
        }
    }

    public Lock tryGetExclusiveLock(CategoryNode category, String nodePath) throws CoordinationServiceException {
        String fullNodePath = "";
        try {
            // ensure node is present
            fullNodePath = upsertNodePath(category, nodePath);
            InterProcessReadWriteLock lock = new InterProcessReadWriteLock(curator, fullNodePath);
            if (!lock.writeLock().acquire(0, TimeUnit.SECONDS)) {
                return null;
            }
            return new Lock(nodePath, lock.writeLock());
        } catch (Exception ex) {
            throw new CoordinationServiceException(String.format("Failed to get exclusive lock for %s", fullNodePath), ex);
        }
    }

    public Lock tryGetSharedLock(CategoryNode category, String nodePath, int timeOut, TimeUnit timeUnit) throws CoordinationServiceException, InterruptedException {
        String fullNodePath = "";
        try {
            // ensure node is present
            fullNodePath = upsertNodePath(category, nodePath);
            InterProcessReadWriteLock lock = new InterProcessReadWriteLock(curator, fullNodePath);
            if (lock.readLock().acquire(timeOut, timeUnit)) {
                return new Lock(nodePath, lock.readLock());
            } else {
                return null;
            }
        } catch (Exception ex) {
            if (ex instanceof InterruptedException) {
                throw (InterruptedException) ex;
            } else {
                throw new CoordinationServiceException(String.format("Failed to get shared lock for %s", fullNodePath), ex);
            }
        }
    }

    public Lock tryGetSharedLock(CategoryNode category, String nodePath) throws CoordinationServiceException {
        String fullNodePath = "";
        try {
            // ensure node is present
            fullNodePath = upsertNodePath(category, nodePath);
            InterProcessReadWriteLock lock = new InterProcessReadWriteLock(curator, fullNodePath);
            if (!lock.readLock().acquire(0, TimeUnit.SECONDS)) {
                return null;
            }
            return new Lock(nodePath, lock.readLock());
        } catch (Exception ex) {
            throw new CoordinationServiceException(String.format("Failed to get shared lock for %s", fullNodePath), ex);
        }
    }

    public byte[] getNodeData(CategoryNode category, String nodePath) throws CoordinationServiceException, InterruptedException {
        String fullNodePath = "";
        try {
            // ensure node is present
            fullNodePath = upsertNodePath(category, nodePath);
            
            // return node data path
            return curator.getData().forPath(fullNodePath);
        } catch (NoNodeException ex) {
            return null;
        } catch (Exception ex) {
            if (ex instanceof InterruptedException) {
                throw (InterruptedException) ex;
            } else {
                throw new CoordinationServiceException(String.format("Failed to get data for %s", fullNodePath), ex);
            }
        }
    }

    public void setNodeData(CategoryNode category, String nodePath, byte[] data) throws CoordinationServiceException, InterruptedException {
        String fullNodePath = getFullyQualifiedNodePath(category, nodePath);
        try {
            curator.setData().forPath(fullNodePath, data);
        } catch (Exception ex) {
            if (ex instanceof InterruptedException) {
                throw (InterruptedException) ex;
            } else {
                throw new CoordinationServiceException(String.format("Failed to set data for %s", fullNodePath), ex);
            }
        }
    }

    public void deleteNode(CategoryNode category, String nodePath) throws CoordinationServiceException, InterruptedException {
        String fullNodePath = getFullyQualifiedNodePath(category, nodePath);
        try {
            curator.delete().forPath(fullNodePath);
        } catch (Exception ex) {
            if (ex instanceof InterruptedException) {
                throw (InterruptedException) ex;
            } else {
                throw new CoordinationServiceException(String.format("Failed to delete node %s", fullNodePath), ex);
            }
        }
    }

    public List<String> getNodeList(CategoryNode category) throws CoordinationServiceException, InterruptedException {
        try {
            List<String> list = curator.getChildren().forPath(categoryNodeToPath.get(category.getDisplayName()));
            return list;
        } catch (Exception ex) {
            if (ex instanceof InterruptedException) {
                throw (InterruptedException) ex;
            } else {
                throw new CoordinationServiceException(String.format("Failed to get node list for %s", category.getDisplayName()), ex);
            }
        }
    }

    private String getFullyQualifiedNodePath(CategoryNode category, String nodePath) {
        // nodePath on Unix systems starts with a "/" and ZooKeeper doesn't like two slashes in a row
        if (nodePath.startsWith("/")) {
            return categoryNodeToPath.get(category.getDisplayName()) + nodePath.toUpperCase();
        } else {
            return categoryNodeToPath.get(category.getDisplayName()) + "/" + nodePath.toUpperCase();
        }
    }

    public final static class CoordinationServiceException extends Exception {
 
        private static final long serialVersionUID = 1L;
 
        private CoordinationServiceException(String message) {
            super(message);
        }
 
        private CoordinationServiceException(String message, Throwable cause) {
            super(message, cause);
        }
    }

    public static class Lock implements AutoCloseable {

        private final InterProcessMutex interProcessLock;
        private final String nodePath;
 
        private Lock(String nodePath, InterProcessMutex lock) {
            this.nodePath = nodePath;
            this.interProcessLock = lock;
        }
 
        public String getNodePath() {
            return nodePath;
        }
 
        public void release() throws CoordinationServiceException {
            try {
                this.interProcessLock.release();
            } catch (Exception ex) {
                throw new CoordinationServiceException(String.format("Failed to release the lock on %s", nodePath), ex);
            }
        }
 
        @Override
        public void close() throws CoordinationServiceException {
            release();
        }
    }

    public enum CategoryNode {
 
        CASES("cases"),
        MANIFESTS("manifests"),
        CONFIG("config"),
        CENTRAL_REPO("centralRepository"),
        HEALTH_MONITOR("healthMonitor");
 
        private final String displayName;
 
        private CategoryNode(String displayName) {
            this.displayName = displayName;
        }
 
        public String getDisplayName() {
            return displayName;
        }
    }
}