001 /** 002 * Licensed to the Apache Software Foundation (ASF) under one 003 * or more contributor license agreements. See the NOTICE file 004 * distributed with this work for additional information 005 * regarding copyright ownership. The ASF licenses this file 006 * to you under the Apache License, Version 2.0 (the 007 * "License"); you may not use this file except in compliance 008 * with the License. You may obtain a copy of the License at 009 * 010 * http://www.apache.org/licenses/LICENSE-2.0 011 * 012 * Unless required by applicable law or agreed to in writing, software 013 * distributed under the License is distributed on an "AS IS" BASIS, 014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 015 * See the License for the specific language governing permissions and 016 * limitations under the License. 017 */ 018 package org.apache.oozie.service; 019 020 import org.apache.hadoop.conf.Configuration; 021 import org.apache.hadoop.fs.FileStatus; 022 import org.apache.hadoop.fs.FileSystem; 023 import org.apache.hadoop.fs.Path; 024 import org.apache.hadoop.fs.PathFilter; 025 import org.apache.hadoop.mapred.JobConf; 026 import org.apache.oozie.client.OozieClient; 027 import org.apache.oozie.workflow.WorkflowApp; 028 import org.apache.oozie.workflow.WorkflowException; 029 import org.apache.oozie.util.IOUtils; 030 import org.apache.oozie.util.XConfiguration; 031 import org.apache.oozie.util.XLog; 032 import org.apache.oozie.ErrorCode; 033 034 import java.io.IOException; 035 import java.io.InputStreamReader; 036 import java.io.Reader; 037 import java.io.StringWriter; 038 import java.net.URI; 039 import java.net.URISyntaxException; 040 import java.util.ArrayList; 041 import java.util.Arrays; 042 import java.util.Collection; 043 import java.util.LinkedHashSet; 044 import java.util.List; 045 import java.util.Map; 046 import java.util.Set; 047 048 /** 049 * Service that provides application workflow definition reading from the path and creation of the proto configuration. 050 */ 051 public abstract class WorkflowAppService implements Service { 052 053 public static final String CONF_PREFIX = Service.CONF_PREFIX + "WorkflowAppService."; 054 055 public static final String SYSTEM_LIB_PATH = CONF_PREFIX + "system.libpath"; 056 057 public static final String APP_LIB_PATH_LIST = "oozie.wf.application.lib"; 058 059 public static final String HADOOP_USER = "user.name"; 060 061 public static final String CONFG_MAX_WF_LENGTH = CONF_PREFIX + "WorkflowDefinitionMaxLength"; 062 063 public static final String OOZIE_SUBWORKFLOW_CLASSPATH_INHERITANCE = "oozie.subworkflow.classpath.inheritance"; 064 065 public static final String OOZIE_WF_SUBWORKFLOW_CLASSPATH_INHERITANCE = "oozie.wf.subworkflow.classpath.inheritance"; 066 067 private Path systemLibPath; 068 private long maxWFLength; 069 private boolean oozieSubWfCPInheritance; 070 071 /** 072 * Initialize the workflow application service. 073 * 074 * @param services services instance. 075 */ 076 public void init(Services services) { 077 Configuration conf = services.getConf(); 078 079 String path = conf.get(SYSTEM_LIB_PATH, " "); 080 if (path.trim().length() > 0) { 081 systemLibPath = new Path(path.trim()); 082 } 083 084 maxWFLength = conf.getInt(CONFG_MAX_WF_LENGTH, 100000); 085 086 oozieSubWfCPInheritance = conf.getBoolean(OOZIE_SUBWORKFLOW_CLASSPATH_INHERITANCE, false); 087 } 088 089 /** 090 * Destroy the workflow application service. 091 */ 092 public void destroy() { 093 } 094 095 /** 096 * Return the public interface for workflow application service. 097 * 098 * @return {@link WorkflowAppService}. 099 */ 100 public Class<? extends Service> getInterface() { 101 return WorkflowAppService.class; 102 } 103 104 /** 105 * Read workflow definition. 106 * 107 * 108 * @param appPath application path. 109 * @param user user name. 110 * @return workflow definition. 111 * @throws WorkflowException thrown if the definition could not be read. 112 */ 113 protected String readDefinition(String appPath, String user, Configuration conf) 114 throws WorkflowException { 115 try { 116 URI uri = new URI(appPath); 117 HadoopAccessorService has = Services.get().get(HadoopAccessorService.class); 118 JobConf jobConf = has.createJobConf(uri.getAuthority()); 119 FileSystem fs = has.createFileSystem(user, uri, jobConf); 120 121 // app path could be a directory 122 Path path = new Path(uri.getPath()); 123 if (!fs.isFile(path)) { 124 path = new Path(path, "workflow.xml"); 125 } 126 127 FileStatus fsStatus = fs.getFileStatus(path); 128 if (fsStatus.getLen() > this.maxWFLength) { 129 throw new WorkflowException(ErrorCode.E0736, fsStatus.getLen(), this.maxWFLength); 130 } 131 132 Reader reader = new InputStreamReader(fs.open(path)); 133 StringWriter writer = new StringWriter(); 134 IOUtils.copyCharStream(reader, writer); 135 return writer.toString(); 136 137 } 138 catch (WorkflowException wfe) { 139 throw wfe; 140 } 141 catch (IOException ex) { 142 throw new WorkflowException(ErrorCode.E0710, ex.getMessage(), ex); 143 } 144 catch (URISyntaxException ex) { 145 throw new WorkflowException(ErrorCode.E0711, appPath, ex.getMessage(), ex); 146 } 147 catch (HadoopAccessorException ex) { 148 throw new WorkflowException(ex); 149 } 150 catch (Exception ex) { 151 throw new WorkflowException(ErrorCode.E0710, ex.getMessage(), ex); 152 } 153 } 154 155 /** 156 * Create proto configuration. <p/> The proto configuration includes the user,group and the paths which need to be 157 * added to distributed cache. These paths include .jar,.so and the resource file paths. 158 * 159 * @param jobConf job configuration. 160 * @param isWorkflowJob indicates if the job is a workflow job or not. 161 * @return proto configuration. 162 * @throws WorkflowException thrown if the proto action configuration could not be created. 163 */ 164 public XConfiguration createProtoActionConf(Configuration jobConf, boolean isWorkflowJob) 165 throws WorkflowException { 166 try { 167 HadoopAccessorService has = Services.get().get(HadoopAccessorService.class); 168 URI uri = new URI(jobConf.get(OozieClient.APP_PATH)); 169 170 Configuration conf = has.createJobConf(uri.getAuthority()); 171 172 String user = jobConf.get(OozieClient.USER_NAME); 173 conf.set(OozieClient.USER_NAME, user); 174 175 FileSystem fs = has.createFileSystem(user, uri, conf); 176 177 Path appPath = new Path(uri); 178 XLog.getLog(getClass()).debug("jobConf.libPath = " + jobConf.get(OozieClient.LIBPATH)); 179 XLog.getLog(getClass()).debug("jobConf.appPath = " + appPath); 180 181 Collection<String> filePaths; 182 if (isWorkflowJob) { 183 // app path could be a directory 184 Path path = new Path(uri.getPath()); 185 if (!fs.isFile(path)) { 186 filePaths = getLibFiles(fs, new Path(appPath + "/lib")); 187 } else { 188 filePaths = getLibFiles(fs, new Path(appPath.getParent(), "lib")); 189 } 190 } 191 else { 192 filePaths = new LinkedHashSet<String>(); 193 } 194 195 String[] libPaths = jobConf.getStrings(OozieClient.LIBPATH); 196 if (libPaths != null && libPaths.length > 0) { 197 for (int i = 0; i < libPaths.length; i++) { 198 if (libPaths[i].trim().length() > 0) { 199 Path libPath = new Path(libPaths[i].trim()); 200 Collection<String> libFilePaths = getLibFiles(fs, libPath); 201 filePaths.addAll(libFilePaths); 202 } 203 } 204 } 205 206 // Check if a subworkflow should inherit the libs from the parent WF 207 // OOZIE_WF_SUBWORKFLOW_CLASSPATH_INHERITANCE has priority over OOZIE_SUBWORKFLOW_CLASSPATH_INHERITANCE from oozie-site 208 // If OOZIE_WF_SUBWORKFLOW_CLASSPATH_INHERITANCE isn't specified, we use OOZIE_SUBWORKFLOW_CLASSPATH_INHERITANCE 209 if (jobConf.getBoolean(OOZIE_WF_SUBWORKFLOW_CLASSPATH_INHERITANCE, oozieSubWfCPInheritance)) { 210 // Keep any libs from a parent workflow that might already be in APP_LIB_PATH_LIST and also remove duplicates 211 String[] parentFilePaths = jobConf.getStrings(APP_LIB_PATH_LIST); 212 if (parentFilePaths != null && parentFilePaths.length > 0) { 213 String[] filePathsNames = filePaths.toArray(new String[filePaths.size()]); 214 for (int i = 0; i < filePathsNames.length; i++) { 215 Path p = new Path(filePathsNames[i]); 216 filePathsNames[i] = p.getName(); 217 } 218 Arrays.sort(filePathsNames); 219 List<String> nonDuplicateParentFilePaths = new ArrayList<String>(); 220 for (String parentFilePath : parentFilePaths) { 221 Path p = new Path(parentFilePath); 222 if (Arrays.binarySearch(filePathsNames, p.getName()) < 0) { 223 nonDuplicateParentFilePaths.add(parentFilePath); 224 } 225 } 226 filePaths.addAll(nonDuplicateParentFilePaths); 227 } 228 } 229 230 conf.setStrings(APP_LIB_PATH_LIST, filePaths.toArray(new String[filePaths.size()])); 231 232 //Add all properties start with 'oozie.' 233 for (Map.Entry<String, String> entry : jobConf) { 234 if (entry.getKey().startsWith("oozie.")) { 235 String name = entry.getKey(); 236 String value = entry.getValue(); 237 // if property already exists, should not overwrite 238 if(conf.get(name) == null) { 239 conf.set(name, value); 240 } 241 } 242 } 243 XConfiguration retConf = new XConfiguration(); 244 XConfiguration.copy(conf, retConf); 245 return retConf; 246 } 247 catch (IOException ex) { 248 throw new WorkflowException(ErrorCode.E0712, jobConf.get(OozieClient.APP_PATH), ex.getMessage(), ex); 249 } 250 catch (URISyntaxException ex) { 251 throw new WorkflowException(ErrorCode.E0711, jobConf.get(OozieClient.APP_PATH), ex.getMessage(), ex); 252 } 253 catch (HadoopAccessorException ex) { 254 throw new WorkflowException(ex); 255 } 256 catch (Exception ex) { 257 throw new WorkflowException(ErrorCode.E0712, jobConf.get(OozieClient.APP_PATH), 258 ex.getMessage(), ex); 259 } 260 } 261 262 /** 263 * Parse workflow definition. 264 * 265 * @param jobConf job configuration. 266 * @return workflow application. 267 * @throws WorkflowException thrown if the workflow application could not be parsed. 268 */ 269 public abstract WorkflowApp parseDef(Configuration jobConf) throws WorkflowException; 270 271 /** 272 * Parse workflow definition. 273 * @param wfXml workflow. 274 * @param jobConf job configuration 275 * @return workflow application. 276 * @throws WorkflowException thrown if the workflow application could not be parsed. 277 */ 278 public abstract WorkflowApp parseDef(String wfXml, Configuration jobConf) throws WorkflowException; 279 280 /** 281 * Get all library paths. 282 * 283 * @param fs file system object. 284 * @param libPath hdfs library path. 285 * @return list of paths. 286 * @throws IOException thrown if the lib paths could not be obtained. 287 */ 288 private Collection<String> getLibFiles(FileSystem fs, Path libPath) throws IOException { 289 Set<String> libPaths = new LinkedHashSet<String>(); 290 if (fs.exists(libPath)) { 291 FileStatus[] files = fs.listStatus(libPath, new NoPathFilter()); 292 293 for (FileStatus file : files) { 294 libPaths.add(file.getPath().toUri().toString()); 295 } 296 } 297 else { 298 XLog.getLog(getClass()).warn("libpath [{0}] does not exist", libPath); 299 } 300 return libPaths; 301 } 302 303 /* 304 * Filter class doing no filtering. 305 * We dont need define this class, but seems fs.listStatus() is not working properly without this. 306 * So providing this dummy no filtering Filter class. 307 */ 308 private class NoPathFilter implements PathFilter { 309 @Override 310 public boolean accept(Path path) { 311 return true; 312 } 313 } 314 315 /** 316 * Returns Oozie system libpath. 317 * 318 * @return Oozie system libpath (sharelib) in HDFS if present, otherwise it returns <code>NULL</code>. 319 */ 320 public Path getSystemLibPath() { 321 return systemLibPath; 322 } 323 }