HADOOP-5257. HDFS servers may start and stop external components through

a plugin interface. (Carlos Valiente via dhruba)



git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/trunk@761530 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -71,6 +71,9 @@ Trunk (unreleased changes)
     HADOOP-5528. Add a configurable hash partitioner operating on ranges of
     BinaryComparable keys. (Klaas Bosteels via shv)
 
+    HADOOP-5257. HDFS servers may start and stop external components through
+    a plugin interface. (Carlos Valiente via dhruba)
+
   IMPROVEMENTS
 
     HADOOP-4565. Added CombineFileInputFormat to use data locality information

src/core/org/apache/hadoop/conf/Configuration.java

@@ -59,6 +59,7 @@ import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableUtils;
+import org.apache.hadoop.util.ReflectionUtils;
 import org.apache.hadoop.util.StringUtils;
 import org.w3c.dom.DOMException;
 import org.w3c.dom.Document;
@@ -841,6 +842,31 @@ public class Configuration implements Iterable<Map.Entry<String,String>>,
     }
   }
 
+  /**
+   * Get the value of the <code>name</code> property as a <code>List</code>
+   * of objects implementing the interface specified by <code>xface</code>.
+   * 
+   * An exception is thrown if any of the classes does not exist, or if it does
+   * not implement the named interface.
+   * 
+   * @param name the property name.
+   * @param xface the interface implemented by the classes named by
+   *        <code>name</code>.
+   * @return a <code>List</code> of objects implementing <code>xface</code>.
+   */
+  @SuppressWarnings("unchecked")
+  public <U> List<U> getInstances(String name, Class<U> xface) {
+    List<U> ret = new ArrayList<U>();
+    Class<?>[] classes = getClasses(name);
+    for (Class<?> cl: classes) {
+      if (!xface.isAssignableFrom(cl)) {
+        throw new RuntimeException(cl + " does not implement " + xface);
+      }
+      ret.add((U)ReflectionUtils.newInstance(cl, this));
+    }
+    return ret;
+  }
+
   /** 
    * Set the value of the <code>name</code> property to the name of a 
    * <code>theClass</code> implementing the given interface <code>xface</code>.

src/core/org/apache/hadoop/util/ServicePlugin.java

@@ -0,0 +1,46 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.hadoop.util;
+
+import java.io.Closeable;
+
+/**
+ * Service plug-in interface.
+ * 
+ * Service plug-ins may be used to expose functionality of datanodes or
+ * namenodes using arbitrary RPC protocols. Plug-ins are instantiated by the
+ * service instance, and are notified of service life-cycle events using the
+ * methods defined by this class.
+ * 
+ * Service plug-ins are started after the service instance is started, and
+ * stopped before the service instance is stopped.
+ */
+public interface ServicePlugin extends Closeable {
+
+  /**
+   * This method is invoked when the service instance has been started.
+   *
+   * @param service The service instance invoking this method
+   */
+  void start(Object service);
+  
+  /**
+   * This method is invoked when the service instance is about to be shut down.
+   */
+  void stop();
+}

src/hdfs/hdfs-default.xml

@@ -378,4 +378,18 @@ creations/deletions), or "all".</description>
   </description>
 </property>
 
+<property>
+  <name>dfs.datanode.plugins</name>
+  <value></value>
+  <description>Comma-separated list of datanode plug-ins to be activated.
+  </description>
+</property>
+
+<property>
+  <name>dfs.namenode.plugins</name>
+  <value></value>
+  <description>Comma-separated list of namenode plug-ins to be activated.
+  </description>
+</property>
+
 </configuration>

src/hdfs/org/apache/hadoop/hdfs/server/datanode/DataNode.java

@@ -88,6 +88,7 @@ import org.apache.hadoop.security.authorize.PolicyProvider;
 import org.apache.hadoop.security.authorize.ServiceAuthorizationManager;
 import org.apache.hadoop.util.Daemon;
 import org.apache.hadoop.util.DiskChecker;
+import org.apache.hadoop.util.ServicePlugin;
 import org.apache.hadoop.util.ReflectionUtils;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.util.DiskChecker.DiskErrorException;
@@ -191,6 +192,9 @@ public class DataNode extends Configured
   public DataBlockScanner blockScanner = null;
   public Daemon blockScannerThread = null;
   
+  /** Activated plug-ins. */
+  private List<ServicePlugin> plugins;
+  
   private static final Random R = new Random();
   
   // For InterDataNodeProtocol
@@ -390,6 +394,16 @@ public class DataNode extends Configured
     dnRegistration.setIpcPort(ipcServer.getListenerAddress().getPort());
 
     LOG.info("dnRegistration = " + dnRegistration);
+    
+    plugins = conf.getInstances("dfs.datanode.plugins", ServicePlugin.class);
+    for (ServicePlugin p: plugins) {
+      try {
+        p.start(this);
+        LOG.info("Started plug-in " + p);
+      } catch (Throwable t) {
+        LOG.warn("ServicePlugin " + p + " could not be started", t);
+      }
+    }
   }
 
   /**
@@ -557,6 +571,17 @@ public class DataNode extends Configured
    * Otherwise, deadlock might occur.
    */
   public void shutdown() {
+    if (plugins != null) {
+      for (ServicePlugin p : plugins) {
+        try {
+          p.stop();
+          LOG.info("Stopped plug-in " + p);
+        } catch (Throwable t) {
+          LOG.warn("ServicePlugin " + p + " could not be stopped", t);
+        }
+      }
+    }
+    
     if (infoServer != null) {
       try {
         infoServer.stop();

src/hdfs/org/apache/hadoop/hdfs/server/namenode/NameNode.java

@@ -47,6 +47,7 @@ import org.apache.hadoop.hdfs.server.protocol.UpgradeCommand;
 import org.apache.hadoop.http.HttpServer;
 import org.apache.hadoop.ipc.*;
 import org.apache.hadoop.conf.*;
+import org.apache.hadoop.util.ServicePlugin;
 import org.apache.hadoop.util.ReflectionUtils;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.net.NetUtils;
@@ -64,6 +65,7 @@ import java.io.*;
 import java.net.*;
 import java.util.Collection;
 import java.util.Iterator;
+import java.util.List;
 
 /**********************************************************
  * NameNode serves as both directory namespace manager and
@@ -144,6 +146,8 @@ public class NameNode implements ClientProtocol, DatanodeProtocol,
   protected NamenodeRegistration nodeRegistration;
   /** Is service level authorization enabled? */
   private boolean serviceAuthEnabled = false;
+  /** Activated plug-ins. */
+  private List<ServicePlugin> plugins;
   
   /** Format a new filesystem.  Destroys any filesystem that may already
    * exist at this location.  **/
@@ -273,6 +277,15 @@ public class NameNode implements ClientProtocol, DatanodeProtocol,
     startHttpServer(conf);
     server.start();  //start RPC server
     startTrashEmptier(conf);
+    
+    plugins = conf.getInstances("dfs.namenode.plugins", ServicePlugin.class);
+    for (ServicePlugin p: plugins) {
+      try {
+        p.start(this);
+      } catch (Throwable t) {
+        LOG.warn("ServicePlugin " + p + " could not be started", t);
+      }
+    }
   }
 
   private void startTrashEmptier(Configuration conf) throws IOException {
@@ -382,6 +395,15 @@ public class NameNode implements ClientProtocol, DatanodeProtocol,
     if (stopRequested)
       return;
     stopRequested = true;
+    if (plugins != null) {
+      for (ServicePlugin p : plugins) {
+        try {
+          p.stop();
+        } catch (Throwable t) {
+          LOG.warn("ServicePlugin " + p + " could not be stopped", t);
+        }
+      }
+    }
     try {
       if (httpServer != null) httpServer.stop();
     } catch (Exception e) {

src/test/org/apache/hadoop/conf/TestGetInstances.java

@@ -0,0 +1,74 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you 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.apache.hadoop.conf;
+
+import java.util.List;
+
+import junit.framework.TestCase;
+
+public class TestGetInstances extends TestCase {
+  
+  interface SampleInterface {}
+  
+  interface ChildInterface extends SampleInterface {}
+  
+  static class SampleClass implements SampleInterface {
+    SampleClass() {}
+  }
+	
+  static class AnotherClass implements ChildInterface {
+    AnotherClass() {}
+  }
+  
+  /**
+   * Makes sure <code>Configuration.getInstances()</code> returns
+   * instances of the required type.
+   */
+  public void testGetInstances() throws Exception {
+    Configuration conf = new Configuration();
+    
+    List<SampleInterface> classes =
+      conf.getInstances("no.such.property", SampleInterface.class);
+    assertTrue(classes.isEmpty());
+
+    conf.set("empty.property", "");
+    classes = conf.getInstances("empty.property", SampleInterface.class);
+    assertTrue(classes.isEmpty());
+    
+    conf.setStrings("some.classes",
+        SampleClass.class.getName(), AnotherClass.class.getName());
+    classes = conf.getInstances("some.classes", SampleInterface.class);
+    assertEquals(2, classes.size());
+    
+    try {
+      conf.setStrings("some.classes",
+          SampleClass.class.getName(), AnotherClass.class.getName(),
+          String.class.getName());
+      conf.getInstances("some.classes", SampleInterface.class);
+      fail("java.lang.String does not implement SampleInterface");
+    } catch (RuntimeException e) {}
+    
+    try {
+      conf.setStrings("some.classes",
+          SampleClass.class.getName(), AnotherClass.class.getName(),
+          "no.such.Class");
+      conf.getInstances("some.classes", SampleInterface.class);
+      fail("no.such.Class does not exist");
+    } catch (RuntimeException e) {}
+  }
+}