Merge pull request #2 from yeyuqiang/YARN-6507-decouple

decouple fpga framework from fpga plugin

Author: tangzhankun

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java

@@ -1303,6 +1303,13 @@ public static boolean isAclEnabled(Configuration conf) {
   public static final String NM_FPGA_RESOURCE_ENABLED =
       NM_FPGA_RESOURCE_PREFIX + "enabled";
 
+  /**
+   * Settings for fpga vendor plugin
+   */
+  @Private
+  public static final String NM_FPGA_PLUGIN_CLASS =
+          NM_PREFIX + "fpga.plugin.class";
+
   /**
    * FPGA as a resource is disabled by default.
    **/

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/java/org/apache/hadoop/yarn/util/resource/ResourceUtils.java

@@ -395,12 +395,9 @@ public static Map<String, ResourceInformation> getNodeResourceInformation(
   /**
    * Function to get the device allowed infomation. The value format should be comma separated majorNumber:minorNumber
    *
-   * <property>
-   *   <name>yarn.nodemanager.resource-types.MCP.allowed</name>
-   *   <value>244:0,245:1</value>
-   * </property>
+   * @param conf
    * @return a map of resource type and allowed value string
-   * */
+   */
   public static Map<String, String> getResourceTypeAllowedValue(Configuration conf) {
     Map<String, String> allowedDevices = new HashMap<>();
     for (Map.Entry<String, String> entry : conf) {

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/resources/Fpga/AbstractFpgaPlugin.java

@@ -21,7 +21,6 @@
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
-import org.apache.hadoop.conf.Configuration;
 
 import java.util.List;
 
@@ -35,7 +34,7 @@
 @InterfaceStability.Unstable
 public interface AbstractFpgaPlugin {
 
-  boolean initPlugin(String s, Configuration configuration);
+  boolean initPlugin();
 
   String getExistingIPID(int major, int minor);
 
@@ -46,7 +45,7 @@ public interface AbstractFpgaPlugin {
    * */
   String downloadIP(String id, String dstDir);
 
-  boolean configureIP(String ipPath, FpgaResourceAllocator.FpgaAllocation fpgaAllocations);
+  boolean configureIP(String ipPath, List<String> addresses);
 
-  boolean cleanupFpgas(FpgaResourceAllocator.FpgaAllocation fpgaAllocations);
+  boolean cleanupFpgas(List<String> address);
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/resources/Fpga/FpgaPluginChain.java

@@ -19,9 +19,6 @@
 
 package org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.resources.Fpga;
 
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.resources.ResourceHandlerException;
-
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
@@ -52,9 +49,9 @@ public List<AbstractFpgaPlugin> getPlugins() {
     return new ArrayList<>(plugins.values());
   }
 
-  public boolean initPlugin(String s, Configuration configuration) {
+  public boolean initPlugin() {
     for (AbstractFpgaPlugin plugin : plugins.values()) {
-      if (!plugin.initPlugin(s,configuration)) {
+      if (!plugin.initPlugin()) {
         return false;
       }
     }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/resources/Fpga/FpgaResourceHandlerImpl.java

@@ -27,16 +27,17 @@
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.yarn.api.records.ContainerId;
-import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.api.records.ResourceInformation;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.privileged.PrivilegedOperation;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.resources.CGroupsHandler;
-import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.resources.Fpga.plugins.IntelMCPFpgaPlugin;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.resources.ResourceHandlerException;
 import org.apache.hadoop.yarn.util.resource.ResourceUtils;
 
+import java.lang.reflect.Constructor;
+import java.lang.reflect.InvocationTargetException;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -55,11 +56,31 @@ public class FpgaResourceHandlerImpl implements FpgaResourceHandler {
 
   private CGroupsHandler cGroupsHandler;
 
-  public FpgaResourceHandlerImpl(CGroupsHandler cGroupsHandler) {
+  public FpgaResourceHandlerImpl(CGroupsHandler cGroupsHandler, Configuration conf) {
+
+    LOG.info("FPGA Plugin Chain init.");
+
     allocator = new FpgaResourceAllocator();
     //init all plugins based on configurations or hardcode
     pluginChain = new FpgaPluginChain();
-    pluginChain.addPlugin(new IntelMCPFpgaPlugin());
+
+    String[] fpgaPluginClassStrs = conf.getStrings(YarnConfiguration.NM_FPGA_PLUGIN_CLASS);
+    if(fpgaPluginClassStrs == null) {
+      LOG.info("No FPGA plugin can be loaded.");
+    } else {
+
+      for (String fpgaPluginClass : fpgaPluginClassStrs) {
+        LOG.info("FPGA Plugin Class " + fpgaPluginClass);
+        try {
+          Constructor<?> constructor = Class.forName(fpgaPluginClass).getConstructor();
+          AbstractFpgaPlugin fpgaPlugin = (AbstractFpgaPlugin) constructor.newInstance();
+          pluginChain.addPlugin(fpgaPlugin);
+          LOG.info(fpgaPluginClass + " loaded");
+        } catch (NoSuchMethodException | ClassNotFoundException | IllegalAccessException | InstantiationException | InvocationTargetException e) {
+          throw new YarnRuntimeException(e);
+        }
+      }
+    }
     this.cGroupsHandler = cGroupsHandler;
   }
 
@@ -89,7 +110,7 @@ public String getRequestedIPID(Container container) {
   public List<PrivilegedOperation> bootstrap(Configuration configuration) throws ResourceHandlerException {
     // get vendor plugin type, major and minor number from configuration
     // add FPGA devices to allocator
-    if (!pluginChain.initPlugin("", configuration)){
+    if (!pluginChain.initPlugin()){
       throw new ResourceHandlerException("Fpga plugin initialization failed", null);
     }
     //get major number and minor number from configuration node-resource.xml
@@ -161,7 +182,13 @@ public List<PrivilegedOperation> preStart(Container container) throws ResourceHa
           if (null == ipFilePath) {
             throw new ResourceHandlerException("Fpga plugin failed to download IP", null);
           }
-          if (!tempPlugin.configureIP(ipFilePath, allocation)) {
+          List<FpgaResourceAllocator.FpgaDevice> allowed = allocation.getAllowed();
+          List<String> addresses = new ArrayList<>();
+          for(int i = 0; i < allowed.size(); i++) {
+            addresses.add(allowed.get(i).getMajor() + ":" + allowed.get(i).getMinor());
+          }
+
+          if (!tempPlugin.configureIP(ipFilePath, addresses)) {
             throw new ResourceHandlerException("Fpga plugin failed to configure IP", null);
           }
           //update the allocator that we update an IP of a device

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/resources/Fpga/plugins/IntelMCPFpgaPlugin.java

@@ -95,7 +95,7 @@ private static FpgaResourceHandlerImpl getFpgaResourceHandler(
         YarnConfiguration.DEFAULT_NM_FPGA_RESOURCE_ENABLED);
     if (fpgaEnabled) {
       return new FpgaResourceHandlerImpl(
-          getInitializedCGroupsHandler(conf));
+          getInitializedCGroupsHandler(conf), conf);
     }
     return null;
   }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/resources/ResourceHandlerModule.java

@@ -1,3 +1,21 @@
+/**
+ * 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.yarn.server.nodemanager.containermanager.linux.resources;
 
 import org.apache.hadoop.conf.Configuration;
@@ -6,12 +24,10 @@
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.resources.Fpga.AbstractFpgaPlugin;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.resources.Fpga.FpgaResourceHandlerImpl;
-import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.resources.Fpga.plugins.IntelMCPFpgaPlugin;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Mockito;
-import org.omg.CORBA.Any;
 
 
 import java.util.HashMap;
@@ -21,32 +37,34 @@
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
 
+
 public class TestFpgaResourceHandler {
   private FpgaResourceHandlerImpl fpgaResourceHandler;
   private Configuration configuration;
+
+  /**
+   * it's better to define allowed devices in the node-resource.xml:
+   * <property>
+   *   <name>yarn.nodemanager.resource-types.MCP</name>
+   *   <value>2</value>
+   * </property>
+   * <property>
+   *   <name>yarn.nodemanager.resource-types.MCP.allowed</name>
+   *   <value>244:0,245:1</value>
+   * </property>
+   * <property>
+   *   <name>yarn.nodemanager.resource-types.DCP</name>
+   *   <value>2</value>
+   * </property>
+   * <property>
+   *   <name>yarn.nodemanager.resource-types.DCP.allowed</name>
+   *   <value>100:0,100:1</value>
+   * </property>
+   */
   @Before
   public void setup() {
-    /**
-     * it's better to define allowed devices in the node-resource.xml:
-     * <property>
-     *   <name>yarn.nodemanager.resource-types.MCP</name>
-     *   <value>2</value>
-     * </property>
-     * <property>
-     *   <name>yarn.nodemanager.resource-types.MCP.allowed</name>
-     *   <value>244:0,245:1</value>
-     * </property>
-     * <property>
-     *   <name>yarn.nodemanager.resource-types.DCP</name>
-     *   <value>2</value>
-     * </property>
-     * <property>
-     *   <name>yarn.nodemanager.resource-types.DCP.allowed</name>
-     *   <value>100:0,100:1</value>
-     * </property>
-     * */
-    fpgaResourceHandler = new FpgaResourceHandlerImpl(mock(CGroupsHandler.class));
     configuration = new YarnConfiguration();
+    fpgaResourceHandler = new FpgaResourceHandlerImpl(mock(CGroupsHandler.class), configuration);
     configuration.set(YarnConfiguration.NM_RESOURCES_PREFIX + "MCP.allowed", "244:0,245:1");
   }
 
@@ -111,7 +129,7 @@ public void testPreStartWithMultiplePlugins() throws ResourceHandlerException {
 
   private static AbstractFpgaPlugin mockPlugin(String type) {
     AbstractFpgaPlugin plugin = mock(AbstractFpgaPlugin.class);
-    when(plugin.initPlugin(Mockito.anyString(), Mockito.anyObject())).thenReturn(true);
+    when(plugin.initPlugin()).thenReturn(true);
     when(plugin.getFpgaType()).thenReturn(type);
     when(plugin.getExistingIPID(Mockito.anyInt(), Mockito.anyInt())).thenReturn("LZO");
     when(plugin.cleanupFpgas(Mockito.anyObject())).thenReturn(true);