This project has retired. For details please refer to its
Attic page.
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.command.coord;
019
020 import org.apache.hadoop.conf.Configuration;
021
022 import org.apache.oozie.client.CoordinatorAction;
023 import org.apache.oozie.client.OozieClient;
024 import org.apache.oozie.CoordinatorActionBean;
025 import org.apache.oozie.DagEngineException;
026 import org.apache.oozie.DagEngine;
027 import org.apache.oozie.ErrorCode;
028 import org.apache.oozie.WorkflowJobBean;
029 import org.apache.oozie.command.CommandException;
030 import org.apache.oozie.command.PreconditionException;
031 import org.apache.oozie.service.DagEngineService;
032 import org.apache.oozie.service.JPAService;
033 import org.apache.oozie.service.Services;
034 import org.apache.oozie.util.JobUtils;
035 import org.apache.oozie.util.LogUtils;
036 import org.apache.oozie.util.ParamChecker;
037 import org.apache.oozie.util.XLog;
038 import org.apache.oozie.util.XmlUtils;
039 import org.apache.oozie.util.XConfiguration;
040 import org.apache.oozie.util.db.SLADbOperations;
041 import org.apache.oozie.client.SLAEvent.SlaAppType;
042 import org.apache.oozie.client.SLAEvent.Status;
043 import org.apache.oozie.executor.jpa.JPAExecutorException;
044 import org.apache.oozie.executor.jpa.WorkflowJobGetJPAExecutor;
045 import org.apache.oozie.executor.jpa.WorkflowJobUpdateJPAExecutor;
046
047 import org.jdom.Element;
048 import org.jdom.JDOMException;
049
050 import java.io.IOException;
051 import java.io.StringReader;
052
053 public class CoordActionStartXCommand extends CoordinatorXCommand<Void> {
054
055 public static final String EL_ERROR = "EL_ERROR";
056 public static final String EL_EVAL_ERROR = "EL_EVAL_ERROR";
057 public static final String COULD_NOT_START = "COULD_NOT_START";
058 public static final String START_DATA_MISSING = "START_DATA_MISSING";
059 public static final String EXEC_DATA_MISSING = "EXEC_DATA_MISSING";
060
061 private final XLog log = getLog();
062 private String actionId = null;
063 private String user = null;
064 private String authToken = null;
065 private CoordinatorActionBean coordAction = null;
066 private JPAService jpaService = null;
067
068 public CoordActionStartXCommand(String id, String user, String token) {
069 //super("coord_action_start", "coord_action_start", 1, XLog.OPS);
070 super("coord_action_start", "coord_action_start", 1);
071 this.actionId = ParamChecker.notEmpty(id, "id");
072 this.user = ParamChecker.notEmpty(user, "user");
073 this.authToken = ParamChecker.notEmpty(token, "token");
074 }
075
076 /**
077 * Create config to pass to WF Engine 1. Get createdConf from coord_actions table 2. Get actionXml from
078 * coord_actions table. Extract all 'property' tags and merge createdConf (overwrite duplicate keys). 3. Extract
079 * 'app-path' from actionXML. Create a new property called 'oozie.wf.application.path' and merge with createdConf
080 * (overwrite duplicate keys) 4. Read contents of config-default.xml in workflow directory. 5. Merge createdConf
081 * with config-default.xml (overwrite duplicate keys). 6. Results is runConf which is saved in coord_actions table.
082 * Merge Action createdConf with actionXml to create new runConf with replaced variables
083 *
084 * @param action CoordinatorActionBean
085 * @return Configuration
086 * @throws CommandException
087 */
088 private Configuration mergeConfig(CoordinatorActionBean action) throws CommandException {
089 String createdConf = action.getCreatedConf();
090 String actionXml = action.getActionXml();
091 Element workflowProperties = null;
092 try {
093 workflowProperties = XmlUtils.parseXml(actionXml);
094 }
095 catch (JDOMException e1) {
096 log.warn("Configuration parse error in:" + actionXml);
097 throw new CommandException(ErrorCode.E1005, e1.getMessage(), e1);
098 }
099 // generate the 'runConf' for this action
100 // Step 1: runConf = createdConf
101 Configuration runConf = null;
102 try {
103 runConf = new XConfiguration(new StringReader(createdConf));
104 }
105 catch (IOException e1) {
106 log.warn("Configuration parse error in:" + createdConf);
107 throw new CommandException(ErrorCode.E1005, e1.getMessage(), e1);
108 }
109 // Step 2: Merge local properties into runConf
110 // extract 'property' tags under 'configuration' block in the
111 // coordinator.xml (saved in actionxml column)
112 // convert Element to XConfiguration
113 Element configElement = (Element) workflowProperties.getChild("action", workflowProperties.getNamespace())
114 .getChild("workflow", workflowProperties.getNamespace()).getChild("configuration",
115 workflowProperties.getNamespace());
116 if (configElement != null) {
117 String strConfig = XmlUtils.prettyPrint(configElement).toString();
118 Configuration localConf;
119 try {
120 localConf = new XConfiguration(new StringReader(strConfig));
121 }
122 catch (IOException e1) {
123 log.warn("Configuration parse error in:" + strConfig);
124 throw new CommandException(ErrorCode.E1005, e1.getMessage(), e1);
125 }
126
127 // copy configuration properties in coordinator.xml to the runConf
128 XConfiguration.copy(localConf, runConf);
129 }
130
131 // Step 3: Extract value of 'app-path' in actionxml, and save it as a
132 // new property called 'oozie.wf.application.path'
133 // WF Engine requires the path to the workflow.xml to be saved under
134 // this property name
135 String appPath = workflowProperties.getChild("action", workflowProperties.getNamespace()).getChild("workflow",
136 workflowProperties.getNamespace()).getChild("app-path", workflowProperties.getNamespace()).getValue();
137 runConf.set("oozie.wf.application.path", appPath);
138 return runConf;
139 }
140
141 @Override
142 protected Void execute() throws CommandException {
143 boolean makeFail = true;
144 String errCode = "";
145 String errMsg = "";
146 ParamChecker.notEmpty(user, "user");
147 ParamChecker.notEmpty(authToken, "authToken");
148
149 log.debug("actionid=" + actionId + ", status=" + coordAction.getStatus());
150 if (coordAction.getStatus() == CoordinatorAction.Status.SUBMITTED) {
151 // log.debug("getting.. job id: " + coordAction.getJobId());
152 // create merged runConf to pass to WF Engine
153 Configuration runConf = mergeConfig(coordAction);
154 coordAction.setRunConf(XmlUtils.prettyPrint(runConf).toString());
155 // log.debug("%%% merged runconf=" +
156 // XmlUtils.prettyPrint(runConf).toString());
157 DagEngine dagEngine = Services.get().get(DagEngineService.class).getDagEngine(user, authToken);
158 try {
159 boolean startJob = true;
160 Configuration conf = new XConfiguration(new StringReader(coordAction.getRunConf()));
161 SLADbOperations.writeStausEvent(coordAction.getSlaXml(), coordAction.getId(), Status.STARTED,
162 SlaAppType.COORDINATOR_ACTION, log);
163
164 // Normalize workflow appPath here;
165 JobUtils.normalizeAppPath(conf.get(OozieClient.USER_NAME), conf.get(OozieClient.GROUP_NAME), conf);
166 String wfId = dagEngine.submitJob(conf, startJob);
167 coordAction.setStatus(CoordinatorAction.Status.RUNNING);
168 coordAction.setExternalId(wfId);
169 coordAction.incrementAndGetPending();
170
171 //store.updateCoordinatorAction(coordAction);
172 JPAService jpaService = Services.get().get(JPAService.class);
173 if (jpaService != null) {
174 log.debug("Updating WF record for WFID :" + wfId + " with parent id: " + actionId);
175 WorkflowJobBean wfJob = jpaService.execute(new WorkflowJobGetJPAExecutor(wfId));
176 wfJob.setParentId(actionId);
177 jpaService.execute(new WorkflowJobUpdateJPAExecutor(wfJob));
178 jpaService.execute(new org.apache.oozie.executor.jpa.CoordActionUpdateJPAExecutor(coordAction));
179 }
180 else {
181 log.error(ErrorCode.E0610);
182 }
183
184 makeFail = false;
185 }
186 catch (DagEngineException dee) {
187 errMsg = dee.getMessage();
188 errCode = "E1005";
189 log.warn("can not create DagEngine for submitting jobs", dee);
190 }
191 catch (CommandException ce) {
192 errMsg = ce.getMessage();
193 errCode = ce.getErrorCode().toString();
194 log.warn("command exception occured ", ce);
195 }
196 catch (java.io.IOException ioe) {
197 errMsg = ioe.getMessage();
198 errCode = "E1005";
199 log.warn("Configuration parse error. read from DB :" + coordAction.getRunConf(), ioe);
200 }
201 catch (Exception ex) {
202 errMsg = ex.getMessage();
203 errCode = "E1005";
204 log.warn("can not create DagEngine for submitting jobs", ex);
205 }
206 finally {
207 if (makeFail == true) { // No DB exception occurs
208 log.warn("Failing the action " + coordAction.getId() + ". Because " + errCode + " : " + errMsg);
209 coordAction.setStatus(CoordinatorAction.Status.FAILED);
210 if (errMsg.length() > 254) { // Because table column size is 255
211 errMsg = errMsg.substring(0, 255);
212 }
213 coordAction.setErrorMessage(errMsg);
214 coordAction.setErrorCode(errCode);
215
216 JPAService jpaService = Services.get().get(JPAService.class);
217 if (jpaService != null) {
218 try {
219 jpaService.execute(new org.apache.oozie.executor.jpa.CoordActionUpdateJPAExecutor(coordAction));
220 }
221 catch (JPAExecutorException je) {
222 throw new CommandException(je);
223 }
224 }
225 else {
226 log.error(ErrorCode.E0610);
227 }
228 SLADbOperations.writeStausEvent(coordAction.getSlaXml(), coordAction.getId(), Status.FAILED,
229 SlaAppType.COORDINATOR_ACTION, log); //Update SLA events
230 queue(new CoordActionReadyXCommand(coordAction.getJobId()));
231 }
232 }
233 }
234 return null;
235 }
236
237 @Override
238 protected String getEntityKey() {
239 return coordAction.getJobId();
240 }
241
242 @Override
243 protected boolean isLockRequired() {
244 return true;
245 }
246
247 @Override
248 protected void loadState() throws CommandException {
249
250 }
251
252 @Override
253 protected void eagerLoadState() throws CommandException {
254 jpaService = Services.get().get(JPAService.class);
255 if (jpaService == null) {
256 throw new CommandException(ErrorCode.E0610);
257 }
258
259 try {
260 coordAction = jpaService.execute(new org.apache.oozie.executor.jpa.CoordActionGetJPAExecutor(actionId));
261 }
262 catch (JPAExecutorException je) {
263 throw new CommandException(je);
264 }
265 LogUtils.setLogInfo(coordAction, logInfo);
266 }
267
268 @Override
269 protected void verifyPrecondition() throws PreconditionException {
270 if (coordAction.getStatus() != CoordinatorAction.Status.SUBMITTED) {
271 throw new PreconditionException(ErrorCode.E1100);
272 }
273 }
274 }