client/src/main/java/org/apache/falcon/ExtensionHandler.java - falcon - Git at Google

 /**
  * 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.falcon;

 import org.apache.commons.codec.CharEncoding;
 import org.apache.falcon.client.FalconCLIException;
 import org.apache.falcon.client.FalconExtensionConstants;
 import org.apache.falcon.entity.v0.Entity;
 import org.apache.falcon.entity.v0.EntityType;
 import org.apache.falcon.extensions.ExtensionBuilder;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.codehaus.jettison.json.JSONException;
 import org.codehaus.jettison.json.JSONObject;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

 import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.OutputStream;
 import java.net.MalformedURLException;
 import java.net.URL;
 import java.net.URLEncoder;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.ServiceLoader;

 /**
  * Handler class that is responsible for preparing Extension entities.
  */
 public final class ExtensionHandler {

     public static final Logger LOG = LoggerFactory.getLogger(ExtensionHandler.class);
     private static final String UTF_8 = CharEncoding.UTF_8;
     private static final String TMP_BASE_DIR = String.format("file://%s", System.getProperty("java.io.tmpdir"));
     private static final String LOCATION = "location";
     private static final String TYPE = "type";
     private static final String NAME = "extensionName";
     private static final String EXTENSION_BUILDER_INTERFACE_SERVICE_FILE =
             "META-INF/services/org.apache.falcon.extensions.ExtensionBuilder";

     private ExtensionHandler(){}

     private List<Entity> getEntities(ClassLoader extensionClassloader, String extensionName, String jobName,
                                      InputStream configStream) throws IOException, FalconException {
         Thread.currentThread().setContextClassLoader(extensionClassloader);

         ServiceLoader<ExtensionBuilder> extensionBuilders = ServiceLoader.load(ExtensionBuilder.class);

         List<Class<? extends ExtensionBuilder>> result = new ArrayList<>();

         for (ExtensionBuilder extensionBuilder : extensionBuilders) {
             result.add(extensionBuilder.getClass());
         }

         if (result.isEmpty()) {
             throw new FalconException("Extension Implementation not found in the package of : " + extensionName);
         } else if (result.size() > 1) {
             throw new FalconException("Found more than one extension Implementation in the package of : "
                     + extensionName);
         }

         ExtensionBuilder extensionBuilder = null;
         try {
             @SuppressWarnings("unchecked")
             Class<ExtensionBuilder> clazz = (Class<ExtensionBuilder>) extensionClassloader
                     .loadClass(result.get(0).getCanonicalName());
             extensionBuilder = clazz.newInstance();
         } catch (ClassNotFoundException | InstantiationException | IllegalAccessException e) {
             throw new FalconCLIException("Failed to instantiate extension implementation " + extensionName, e);
         }

         extensionBuilder.validateExtensionConfig(extensionName, configStream);

         return extensionBuilder.getEntities(jobName, configStream);
     }

     public static List<Entity> loadAndPrepare(String extensionName, String jobName, InputStream configStream,
                                               String extensionBuildLocation) throws IOException, FalconException {
         Configuration conf = new Configuration();
         FileSystem fs = FileSystem.get(conf);
         String stagePath = createStagePath(extensionName, jobName);
         List<URL> urls = ExtensionHandler.copyExtensionPackage(extensionBuildLocation, fs, stagePath);

         List<Entity> entities = prepare(extensionName, jobName, configStream, urls);
         ExtensionHandler.stageEntities(entities, stagePath);
         return entities;
     }

     public static List<Entity> prepare(String extensionName, String jobName, InputStream configStream, List<URL> urls)
         throws IOException, FalconException {
         ClassLoader extensionClassLoader = ExtensionClassLoader.load(urls);
         if (extensionClassLoader.getResourceAsStream(EXTENSION_BUILDER_INTERFACE_SERVICE_FILE) == null) {
             throw new FalconCLIException("The extension build time jars do not contain "
                     + EXTENSION_BUILDER_INTERFACE_SERVICE_FILE);
         }
         ExtensionHandler extensionHandler = new ExtensionHandler();

         return extensionHandler.getEntities(extensionClassLoader, extensionName, jobName, configStream);
     }

     // This method is only for debugging, the staged entities can be found in /tmp path.
     private static void stageEntities(List<Entity> entities, String stagePath) {
         File entityFile;
         EntityType type;
         for (Entity entity : entities) {
             type = entity.getEntityType();
             OutputStream out;
             try {
                 entityFile = new File(new Path(stagePath + File.separator + entity.getEntityType().toString() + "_"
                     + URLEncoder.encode(entity.getName(), UTF_8)).toUri().toURL().getPath());
                 if (!entityFile.createNewFile()) {
                     LOG.debug("Not able to stage the entities in the tmp path");
                     return;
                 }
                 out = new FileOutputStream(entityFile);
                 type.getMarshaller().marshal(entity, out);
                 LOG.debug("Staged configuration {}/{}", type, entity.getName());
                 out.close();
             } catch (Exception e) {
                 LOG.error("Unable to serialize the entity object {}/{}", type, entity.getName(), e);
             }
         }
     }

     private static String createStagePath(String extensionName, String jobName) {
         String stagePath = TMP_BASE_DIR + File.separator + extensionName + File.separator + jobName
                 + File.separator + System.currentTimeMillis()/1000;
         File tmpPath = new File(stagePath);
         if (tmpPath.mkdir()) {
             throw new FalconCLIException("Failed to create stage directory" + tmpPath.toString());
         }
         return stagePath;
     }

     private static List<URL> copyExtensionPackage(String extensionBuildUrl, FileSystem fs, String stagePath)
         throws IOException {

         Path libsPath = new Path(extensionBuildUrl, FalconExtensionConstants.LIBS);
         Path buildLibsPath = new Path(libsPath, FalconExtensionConstants.BUILD);
         Path localStagePath = new Path(stagePath);
         Path localBuildLibsPath = new Path(localStagePath, FalconExtensionConstants.LIBS);
         LOG.info("Copying build time libs from {} to {}", buildLibsPath, localBuildLibsPath);
         fs.copyToLocalFile(buildLibsPath, localBuildLibsPath);

         Path resourcesPath = new Path(extensionBuildUrl, FalconExtensionConstants.RESOURCES);
         Path buildResourcesPath = new Path(resourcesPath, FalconExtensionConstants.BUILD);
         Path localBuildResourcesPath = new Path(localStagePath, FalconExtensionConstants.RESOURCES);
         LOG.info("Copying build time resources from {} to {}", buildLibsPath, localBuildResourcesPath);
         fs.copyToLocalFile(buildResourcesPath, localBuildResourcesPath);

         List<URL> urls = new ArrayList<>();
         urls.addAll(getFilesInPath(localBuildLibsPath.toUri().toURL()));
         urls.add(localBuildResourcesPath.toUri().toURL());
         return urls;
     }

     static List<URL> getFilesInPath(URL fileURL) throws MalformedURLException {
         List<URL> urls = new ArrayList<>();

         File file = new File(fileURL.getPath());
         if (file.isDirectory()) {
             File[] files = file.listFiles();

             if (files != null) {
                 for (File innerFile : files) {
                     if (innerFile.isFile()) {
                         urls.add(innerFile.toURI().toURL());
                     }
                 }
             }

             if (!fileURL.toString().endsWith("/")) {
                 fileURL = new URL(fileURL.toString() + "/");
             }
         }

         urls.add(fileURL);
         return urls;
     }


     public static String getExtensionLocation(String extensionName, JSONObject extensionDetailJson) {
         String extensionBuildPath;
         try {
             extensionBuildPath = extensionDetailJson.get(LOCATION).toString();
         } catch (JSONException e) {
             throw new FalconCLIException("Failed to get extension location for the given extension:" + extensionName,
                     e);
         }
         return extensionBuildPath;
     }

     public static  String getExtensionType(String extensionName, JSONObject extensionDetailJson) {
         String extensionType;
         try {
             extensionType = extensionDetailJson.get(TYPE).toString();
         } catch (JSONException e) {
             throw new FalconCLIException("Failed to get extension type for the given extension:" + extensionName, e);
         }
         return extensionType;
     }

     public static  String getExtensionName(String jobName, JSONObject extensionJobDetailJson) {
         String extensionType;
         try {
             extensionType = extensionJobDetailJson.get(NAME).toString();
         } catch (JSONException e) {
             throw new FalconCLIException("Failed to get extension name for the given extension job:" + jobName, e);
         }
         return extensionType;
     }
 }
	/**
	* 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.falcon;

	import org.apache.commons.codec.CharEncoding;
	import org.apache.falcon.client.FalconCLIException;
	import org.apache.falcon.client.FalconExtensionConstants;
	import org.apache.falcon.entity.v0.Entity;
	import org.apache.falcon.entity.v0.EntityType;
	import org.apache.falcon.extensions.ExtensionBuilder;
	import org.apache.hadoop.conf.Configuration;
	import org.apache.hadoop.fs.FileSystem;
	import org.apache.hadoop.fs.Path;
	import org.codehaus.jettison.json.JSONException;
	import org.codehaus.jettison.json.JSONObject;
	import org.slf4j.Logger;
	import org.slf4j.LoggerFactory;

	import java.io.File;
	import java.io.FileOutputStream;
	import java.io.IOException;
	import java.io.InputStream;
	import java.io.OutputStream;
	import java.net.MalformedURLException;
	import java.net.URL;
	import java.net.URLEncoder;
	import java.util.ArrayList;
	import java.util.List;
	import java.util.ServiceLoader;

	/**
	* Handler class that is responsible for preparing Extension entities.
	*/
	public final class ExtensionHandler {

	public static final Logger LOG = LoggerFactory.getLogger(ExtensionHandler.class);
	private static final String UTF_8 = CharEncoding.UTF_8;
	private static final String TMP_BASE_DIR = String.format("file://%s", System.getProperty("java.io.tmpdir"));
	private static final String LOCATION = "location";
	private static final String TYPE = "type";
	private static final String NAME = "extensionName";
	private static final String EXTENSION_BUILDER_INTERFACE_SERVICE_FILE =
	"META-INF/services/org.apache.falcon.extensions.ExtensionBuilder";

	private ExtensionHandler(){}

	private List<Entity> getEntities(ClassLoader extensionClassloader, String extensionName, String jobName,
	InputStream configStream) throws IOException, FalconException {
	Thread.currentThread().setContextClassLoader(extensionClassloader);

	ServiceLoader<ExtensionBuilder> extensionBuilders = ServiceLoader.load(ExtensionBuilder.class);

	List<Class<? extends ExtensionBuilder>> result = new ArrayList<>();

	for (ExtensionBuilder extensionBuilder : extensionBuilders) {
	result.add(extensionBuilder.getClass());
	}

	if (result.isEmpty()) {
	throw new FalconException("Extension Implementation not found in the package of : " + extensionName);
	} else if (result.size() > 1) {
	throw new FalconException("Found more than one extension Implementation in the package of : "
	+ extensionName);
	}

	ExtensionBuilder extensionBuilder = null;
	try {
	@SuppressWarnings("unchecked")
	Class<ExtensionBuilder> clazz = (Class<ExtensionBuilder>) extensionClassloader
	.loadClass(result.get(0).getCanonicalName());
	extensionBuilder = clazz.newInstance();
	} catch (ClassNotFoundException \| InstantiationException \| IllegalAccessException e) {
	throw new FalconCLIException("Failed to instantiate extension implementation " + extensionName, e);
	}

	extensionBuilder.validateExtensionConfig(extensionName, configStream);

	return extensionBuilder.getEntities(jobName, configStream);
	}

	public static List<Entity> loadAndPrepare(String extensionName, String jobName, InputStream configStream,
	String extensionBuildLocation) throws IOException, FalconException {
	Configuration conf = new Configuration();
	FileSystem fs = FileSystem.get(conf);
	String stagePath = createStagePath(extensionName, jobName);
	List<URL> urls = ExtensionHandler.copyExtensionPackage(extensionBuildLocation, fs, stagePath);

	List<Entity> entities = prepare(extensionName, jobName, configStream, urls);
	ExtensionHandler.stageEntities(entities, stagePath);
	return entities;
	}

	public static List<Entity> prepare(String extensionName, String jobName, InputStream configStream, List<URL> urls)
	throws IOException, FalconException {
	ClassLoader extensionClassLoader = ExtensionClassLoader.load(urls);
	if (extensionClassLoader.getResourceAsStream(EXTENSION_BUILDER_INTERFACE_SERVICE_FILE) == null) {
	throw new FalconCLIException("The extension build time jars do not contain "
	+ EXTENSION_BUILDER_INTERFACE_SERVICE_FILE);
	}
	ExtensionHandler extensionHandler = new ExtensionHandler();

	return extensionHandler.getEntities(extensionClassLoader, extensionName, jobName, configStream);
	}

	// This method is only for debugging, the staged entities can be found in /tmp path.
	private static void stageEntities(List<Entity> entities, String stagePath) {
	File entityFile;
	EntityType type;
	for (Entity entity : entities) {
	type = entity.getEntityType();
	OutputStream out;
	try {
	entityFile = new File(new Path(stagePath + File.separator + entity.getEntityType().toString() + "_"
	+ URLEncoder.encode(entity.getName(), UTF_8)).toUri().toURL().getPath());
	if (!entityFile.createNewFile()) {
	LOG.debug("Not able to stage the entities in the tmp path");
	return;
	}
	out = new FileOutputStream(entityFile);
	type.getMarshaller().marshal(entity, out);
	LOG.debug("Staged configuration {}/{}", type, entity.getName());
	out.close();
	} catch (Exception e) {
	LOG.error("Unable to serialize the entity object {}/{}", type, entity.getName(), e);
	}
	}
	}

	private static String createStagePath(String extensionName, String jobName) {
	String stagePath = TMP_BASE_DIR + File.separator + extensionName + File.separator + jobName
	+ File.separator + System.currentTimeMillis()/1000;
	File tmpPath = new File(stagePath);
	if (tmpPath.mkdir()) {
	throw new FalconCLIException("Failed to create stage directory" + tmpPath.toString());
	}
	return stagePath;
	}

	private static List<URL> copyExtensionPackage(String extensionBuildUrl, FileSystem fs, String stagePath)
	throws IOException {

	Path libsPath = new Path(extensionBuildUrl, FalconExtensionConstants.LIBS);
	Path buildLibsPath = new Path(libsPath, FalconExtensionConstants.BUILD);
	Path localStagePath = new Path(stagePath);
	Path localBuildLibsPath = new Path(localStagePath, FalconExtensionConstants.LIBS);
	LOG.info("Copying build time libs from {} to {}", buildLibsPath, localBuildLibsPath);
	fs.copyToLocalFile(buildLibsPath, localBuildLibsPath);

	Path resourcesPath = new Path(extensionBuildUrl, FalconExtensionConstants.RESOURCES);
	Path buildResourcesPath = new Path(resourcesPath, FalconExtensionConstants.BUILD);
	Path localBuildResourcesPath = new Path(localStagePath, FalconExtensionConstants.RESOURCES);
	LOG.info("Copying build time resources from {} to {}", buildLibsPath, localBuildResourcesPath);
	fs.copyToLocalFile(buildResourcesPath, localBuildResourcesPath);

	List<URL> urls = new ArrayList<>();
	urls.addAll(getFilesInPath(localBuildLibsPath.toUri().toURL()));
	urls.add(localBuildResourcesPath.toUri().toURL());
	return urls;
	}

	static List<URL> getFilesInPath(URL fileURL) throws MalformedURLException {
	List<URL> urls = new ArrayList<>();

	File file = new File(fileURL.getPath());
	if (file.isDirectory()) {
	File[] files = file.listFiles();

	if (files != null) {
	for (File innerFile : files) {
	if (innerFile.isFile()) {
	urls.add(innerFile.toURI().toURL());
	}
	}
	}

	if (!fileURL.toString().endsWith("/")) {
	fileURL = new URL(fileURL.toString() + "/");
	}
	}

	urls.add(fileURL);
	return urls;
	}


	public static String getExtensionLocation(String extensionName, JSONObject extensionDetailJson) {
	String extensionBuildPath;
	try {
	extensionBuildPath = extensionDetailJson.get(LOCATION).toString();
	} catch (JSONException e) {
	throw new FalconCLIException("Failed to get extension location for the given extension:" + extensionName,
	e);
	}
	return extensionBuildPath;
	}

	public static String getExtensionType(String extensionName, JSONObject extensionDetailJson) {
	String extensionType;
	try {
	extensionType = extensionDetailJson.get(TYPE).toString();
	} catch (JSONException e) {
	throw new FalconCLIException("Failed to get extension type for the given extension:" + extensionName, e);
	}
	return extensionType;
	}

	public static String getExtensionName(String jobName, JSONObject extensionJobDetailJson) {
	String extensionType;
	try {
	extensionType = extensionJobDetailJson.get(NAME).toString();
	} catch (JSONException e) {
	throw new FalconCLIException("Failed to get extension name for the given extension job:" + jobName, e);
	}
	return extensionType;
	}
	}