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 }