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