hadoop: return all replica hostnames

Updates CephFileSystem to return all replica locations,
and in addition attempts to use reverse DNS to convert
the OSD IPs into hostnames. Hadoop does not do well at
comparing the IP with hostnames, and locality is lost.

Signed-off-by: Noah Watkins <noahwatkins@gmail.com>

diff --git a/src/client/hadoop/ceph/CephFileSystem.java b/src/client/hadoop/ceph/CephFileSystem.java
index 5f2fa59bef4b28e88d6251c3d99f7dd5a5606d4a..95f22238b4d0aec03a03d0ac1df01707a0d6dbbe 100644 (file)
--- a/src/client/hadoop/ceph/CephFileSystem.java
+++ b/src/client/hadoop/ceph/CephFileSystem.java
@@ -25,8 +25,10 @@ import java.io.IOException;
 import java.io.FileNotFoundException;
 import java.io.OutputStream;
 import java.net.URI;
+import java.net.InetAddress;
 import java.util.EnumSet;
 import java.lang.Math;
+import java.util.ArrayList;
 
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
@@ -41,6 +43,7 @@ import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.util.Progressable;
 import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.net.DNS;
 
 
 /**
@@ -72,6 +75,10 @@ public class CephFileSystem extends FileSystem {
   private final Path root;
   private CephFS ceph = null;
 
+  private static String CEPH_NAMESERVER;
+  private static final String CEPH_NAMESERVER_KEY = "fs.ceph.nameserver";
+  private static final String CEPH_NAMESERVER_DEFAULT = "localhost";
+
   /**
    * Create a new CephFileSystem.
    */
@@ -115,6 +122,9 @@ public class CephFileSystem extends FileSystem {
     if (ceph == null) {
       ceph = new CephTalker(conf, LOG);
     }
+
+    CEPH_NAMESERVER = conf.get(CEPH_NAMESERVER_KEY, CEPH_NAMESERVER_DEFAULT);
+
     // build up the arguments for Ceph
     String arguments = "CephFSInterface";
 
@@ -580,6 +590,25 @@ public class CephFileSystem extends FileSystem {
     return result;
   }
 
+  /*
+   * Attempt to convert an IP into its hostname
+   */
+  private String[] ips2Hosts(String[] ips) {
+    ArrayList<String> hosts = new ArrayList<String>();
+    for (String ip : ips) {
+      try {
+        String host = DNS.reverseDns(InetAddress.getByName(ip), CEPH_NAMESERVER);
+        if (host.charAt(host.length()-1) == '.') {
+          host = host.substring(0, host.length()-1);
+        }
+        hosts.add(host); /* append */
+      } catch (Exception e) {
+        LOG.error("reverseDns ["+ip+"] failed: "+ e);
+      }
+    }
+    return hosts.toArray(new String[hosts.size()]);
+  }
+
   /**
    * Get a BlockLocation object for each block in a file.
    *
@@ -595,52 +624,27 @@ public class CephFileSystem extends FileSystem {
    */
   @Override
   public BlockLocation[] getFileBlockLocations(FileStatus file, long start, long len) throws IOException {
-    LOG.debug(
-        "getFileBlockLocations:enter with path " + file.getPath()
-        + ", start pos " + start + ", length " + len);
-    // sanitize and get the filehandle
     Path abs_path = makeAbsolute(file.getPath());
 
-    LOG.trace("getFileBlockLocations:call ceph_open_for_read from Java");
     int fh = ceph.ceph_open_for_read(getCephPath(abs_path));
-
-    LOG.trace(
-        "getFileBlockLocations:return from ceph_open_for_read to Java with fh "
-            + fh);
     if (fh < 0) {
-      LOG.error(
-          "getFileBlockLocations:got error " + fh
-          + ", exiting and returning null!");
+      LOG.error("getFileBlockLocations:got error " + fh + ", exiting and returning null!");
       return null;
     }
-    // get the block size
-    LOG.trace("getFileBlockLocations:call ceph_getblocksize from Java");
-    long blockSize = ceph.ceph_getblocksize(getCephPath(abs_path));
 
-    LOG.trace("getFileBlockLocations:return from ceph_getblocksize");
+    long blockSize = ceph.ceph_getblocksize(getCephPath(abs_path));
     BlockLocation[] locations = new BlockLocation[(int) Math.ceil(len / (float) blockSize)];
-    long offset;
 
     for (int i = 0; i < locations.length; ++i) {
-      offset = start + i * blockSize;
-      LOG.trace(
-          "getFileBlockLocations:call ceph_hosts from Java on fh " + fh
-          + " and offset " + offset);
-      String host[] = ceph.ceph_hosts(fh, offset);
-
-      LOG.trace(
-          "getFileBlockLocations:return from ceph_hosts to Java with host "
-              + host[0]);
-      String[] hostArray = new String[1];
-
-      hostArray[0] = host[0];
-      locations[i] = new BlockLocation(hostArray, hostArray,
-          start + i * blockSize - (start % blockSize), blockSize);
+      long offset = start + i * blockSize;
+      long blockStart = start + i * blockSize - (start % blockSize);
+      String ips[] = ceph.ceph_hosts(fh, offset);
+      String hosts[] = ips2Hosts(ips);
+      locations[i] = new BlockLocation(null, hosts, blockStart, blockSize);
+      LOG.debug("getFileBlockLocations: location[" + i + "]: " + locations[i]);
     }
-    LOG.trace("getFileBlockLocations:call ceph_close from Java on fh " + fh);
+
     ceph.ceph_close(fh);
-    LOG.debug(
-        "getFileBlockLocations:return with " + locations.length + " locations");
     return locations;
   }