Source code

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
019package org.apache.oozie.command.bundle;
020
021import java.io.IOException;
022import java.io.StringReader;
023import java.util.Date;
024import java.util.HashMap;
025import java.util.List;
026import java.util.Map;
027import java.util.Map.Entry;
028
029import org.apache.hadoop.conf.Configuration;
030import org.apache.oozie.BundleActionBean;
031import org.apache.oozie.BundleJobBean;
032import org.apache.oozie.ErrorCode;
033import org.apache.oozie.XException;
034import org.apache.oozie.action.hadoop.OozieJobInfo;
035import org.apache.oozie.client.Job;
036import org.apache.oozie.client.OozieClient;
037import org.apache.oozie.client.rest.JsonBean;
038import org.apache.oozie.command.CommandException;
039import org.apache.oozie.command.PreconditionException;
040import org.apache.oozie.command.StartTransitionXCommand;
041import org.apache.oozie.command.coord.CoordSubmitXCommand;
042import org.apache.oozie.executor.jpa.BatchQueryExecutor;
043import org.apache.oozie.executor.jpa.BundleJobQueryExecutor;
044import org.apache.oozie.executor.jpa.BundleJobQueryExecutor.BundleJobQuery;
045import org.apache.oozie.executor.jpa.JPAExecutorException;
046import org.apache.oozie.executor.jpa.BatchQueryExecutor.UpdateEntry;
047import org.apache.oozie.util.ELUtils;
048import org.apache.oozie.util.JobUtils;
049import org.apache.oozie.util.LogUtils;
050import org.apache.oozie.util.ParamChecker;
051import org.apache.oozie.util.XConfiguration;
052import org.apache.oozie.util.XmlUtils;
053import org.jdom.Attribute;
054import org.jdom.Element;
055import org.jdom.JDOMException;
056
057/**
058 * The command to start Bundle job
059 */
060public class BundleStartXCommand extends StartTransitionXCommand {
061    private final String jobId;
062    private BundleJobBean bundleJob;
063
064    /**
065     * The constructor for class {@link BundleStartXCommand}
066     *
067     * @param jobId the bundle job id
068     */
069    public BundleStartXCommand(String jobId) {
070        super("bundle_start", "bundle_start", 1);
071        this.jobId = ParamChecker.notEmpty(jobId, "jobId");
072    }
073
074    /**
075     * The constructor for class {@link BundleStartXCommand}
076     *
077     * @param jobId the bundle job id
078     * @param dryrun true if dryrun is enable
079     */
080    public BundleStartXCommand(String jobId, boolean dryrun) {
081        super("bundle_start", "bundle_start", 1, dryrun);
082        this.jobId = ParamChecker.notEmpty(jobId, "jobId");
083    }
084
085    /* (non-Javadoc)
086     * @see org.apache.oozie.command.XCommand#getEntityKey()
087     */
088    @Override
089    public String getEntityKey() {
090        return jobId;
091    }
092
093    @Override
094    public String getKey() {
095        return getName() + "_" + jobId;
096    }
097
098    /* (non-Javadoc)
099     * @see org.apache.oozie.command.XCommand#isLockRequired()
100     */
101    @Override
102    protected boolean isLockRequired() {
103        return true;
104    }
105
106    @Override
107    protected void verifyPrecondition() throws CommandException, PreconditionException {
108        if (bundleJob.getStatus() != Job.Status.PREP) {
109            String msg = "Bundle " + bundleJob.getId() + " is not in PREP status. It is in : " + bundleJob.getStatus();
110            LOG.info(msg);
111            throw new PreconditionException(ErrorCode.E1100, msg);
112        }
113    }
114
115    @Override
116    public void loadState() throws CommandException {
117        try {
118            this.bundleJob = BundleJobQueryExecutor.getInstance().get(BundleJobQuery.GET_BUNDLE_JOB, jobId);
119            LogUtils.setLogInfo(bundleJob);
120            super.setJob(bundleJob);
121        }
122        catch (XException ex) {
123            throw new CommandException(ex);
124        }
125    }
126
127    /* (non-Javadoc)
128     * @see org.apache.oozie.command.StartTransitionXCommand#StartChildren()
129     */
130    @Override
131    public void StartChildren() throws CommandException {
132        LOG.debug("Started coord jobs for the bundle=[{0}]", jobId);
133        insertBundleActions();
134        startCoordJobs();
135        LOG.debug("Ended coord jobs for the bundle=[{0}]", jobId);
136    }
137
138    /* (non-Javadoc)
139     * @see org.apache.oozie.command.TransitionXCommand#notifyParent()
140     */
141    @Override
142    public void notifyParent() {
143    }
144
145    /* (non-Javadoc)
146     * @see org.apache.oozie.command.StartTransitionXCommand#performWrites()
147     */
148    @Override
149    public void performWrites() throws CommandException {
150        try {
151            BatchQueryExecutor.getInstance().executeBatchInsertUpdateDelete(insertList, updateList, null);
152        }
153        catch (JPAExecutorException e) {
154            throw new CommandException(e);
155        }
156    }
157
158    /**
159     * Insert bundle actions
160     *
161     * @throws CommandException thrown if failed to create bundle actions
162     */
163    @SuppressWarnings("unchecked")
164    private void insertBundleActions() throws CommandException {
165        if (bundleJob != null) {
166            Map<String, Boolean> map = new HashMap<String, Boolean>();
167            try {
168                Element bAppXml = XmlUtils.parseXml(bundleJob.getJobXml());
169                List<Element> coordElems = bAppXml.getChildren("coordinator", bAppXml.getNamespace());
170                for (Element elem : coordElems) {
171                    Attribute name = elem.getAttribute("name");
172                    Attribute critical = elem.getAttribute("critical");
173                    if (name != null) {
174                        if (map.containsKey(name.getValue())) {
175                            throw new CommandException(ErrorCode.E1304, name);
176                        }
177                        boolean isCritical = false;
178                        if (critical != null && Boolean.parseBoolean(critical.getValue())) {
179                            isCritical = true;
180                        }
181                        map.put(name.getValue(), isCritical);
182                    }
183                    else {
184                        throw new CommandException(ErrorCode.E1305);
185                    }
186                }
187            }
188            catch (JDOMException jex) {
189                throw new CommandException(ErrorCode.E1301, jex.getMessage(), jex);
190            }
191
192            // if there is no coordinator for this bundle, failed it.
193            if (map.isEmpty()) {
194                bundleJob.setStatus(Job.Status.FAILED);
195                bundleJob.resetPending();
196                try {
197                    BundleJobQueryExecutor.getInstance().executeUpdate(BundleJobQuery.UPDATE_BUNDLE_JOB_STATUS_PENDING, bundleJob);
198                }
199                catch (JPAExecutorException jex) {
200                    throw new CommandException(jex);
201                }
202
203                LOG.debug("No coord jobs for the bundle=[{0}], failed it!!", jobId);
204                throw new CommandException(ErrorCode.E1318, jobId);
205            }
206
207            for (Entry<String, Boolean> coordName : map.entrySet()) {
208                BundleActionBean action = createBundleAction(jobId, coordName.getKey(), coordName.getValue());
209                insertList.add(action);
210            }
211        }
212        else {
213            throw new CommandException(ErrorCode.E0604, jobId);
214        }
215    }
216
217    private BundleActionBean createBundleAction(String jobId, String coordName, boolean isCritical) {
218        BundleActionBean action = new BundleActionBean();
219        action.setBundleActionId(jobId + "_" + coordName);
220        action.setBundleId(jobId);
221        action.setCoordName(coordName);
222        action.setStatus(Job.Status.PREP);
223        action.setLastModifiedTime(new Date());
224        if (isCritical) {
225            action.setCritical();
226        }
227        else {
228            action.resetCritical();
229        }
230        return action;
231    }
232
233    /**
234     * Start Coord Jobs
235     *
236     * @throws CommandException thrown if failed to start coord jobs
237     */
238    @SuppressWarnings("unchecked")
239    private void startCoordJobs() throws CommandException {
240        if (bundleJob != null) {
241            try {
242                Element bAppXml = XmlUtils.parseXml(bundleJob.getJobXml());
243                List<Element> coordElems = bAppXml.getChildren("coordinator", bAppXml.getNamespace());
244                for (Element coordElem : coordElems) {
245                    Attribute name = coordElem.getAttribute("name");
246
247                    Configuration coordConf = mergeConfig(coordElem);
248                    coordConf.set(OozieClient.BUNDLE_ID, jobId);
249                    if (OozieJobInfo.isJobInfoEnabled()) {
250                        coordConf.set(OozieJobInfo.BUNDLE_NAME, bundleJob.getAppName());
251                    }
252                    String coordName=name.getValue();
253                    try {
254                        coordName = ELUtils.resolveAppName(coordName, coordConf);
255                    }
256                    catch (Exception e) {
257                        throw new CommandException(ErrorCode.E1321, e.getMessage(), e);
258
259                    }
260                    queue(new CoordSubmitXCommand(coordConf, bundleJob.getId(), name.getValue()));
261
262                }
263                updateBundleAction();
264            }
265            catch (JDOMException jex) {
266                throw new CommandException(ErrorCode.E1301, jex.getMessage(), jex);
267            }
268            catch (JPAExecutorException je) {
269                throw new CommandException(je);
270            }
271        }
272        else {
273            throw new CommandException(ErrorCode.E0604, jobId);
274        }
275    }
276
277    private void updateBundleAction() throws JPAExecutorException {
278        for(JsonBean bAction : insertList) {
279            BundleActionBean action = (BundleActionBean) bAction;
280            action.incrementAndGetPending();
281            action.setLastModifiedTime(new Date());
282        }
283    }
284
285    /**
286     * Merge Bundle job config and the configuration from the coord job to pass
287     * to Coord Engine
288     *
289     * @param coordElem the coordinator configuration
290     * @return Configuration merged configuration
291     * @throws CommandException thrown if failed to merge configuration
292     */
293    private Configuration mergeConfig(Element coordElem) throws CommandException {
294        String jobConf = bundleJob.getConf();
295        // Step 1: runConf = jobConf
296        Configuration runConf = null;
297        try {
298            runConf = new XConfiguration(new StringReader(jobConf));
299        }
300        catch (IOException e1) {
301            LOG.warn("Configuration parse error in:" + jobConf);
302            throw new CommandException(ErrorCode.E1306, e1.getMessage(), e1);
303        }
304        // Step 2: Merge local properties into runConf
305        // extract 'property' tags under 'configuration' block in the coordElem
306        // convert Element to XConfiguration
307        Element localConfigElement = coordElem.getChild("configuration", coordElem.getNamespace());
308
309        if (localConfigElement != null) {
310            String strConfig = XmlUtils.prettyPrint(localConfigElement).toString();
311            Configuration localConf;
312            try {
313                localConf = new XConfiguration(new StringReader(strConfig));
314            }
315            catch (IOException e1) {
316                LOG.warn("Configuration parse error in:" + strConfig);
317                throw new CommandException(ErrorCode.E1307, e1.getMessage(), e1);
318            }
319
320            // copy configuration properties in the coordElem to the runConf
321            XConfiguration.copy(localConf, runConf);
322        }
323
324        // Step 3: Extract value of 'app-path' in coordElem, save it as a
325        // new property called 'oozie.coord.application.path', and normalize.
326        String appPath = coordElem.getChild("app-path", coordElem.getNamespace()).getValue();
327        runConf.set(OozieClient.COORDINATOR_APP_PATH, appPath);
328        // Normalize coordinator appPath here;
329        try {
330            JobUtils.normalizeAppPath(runConf.get(OozieClient.USER_NAME), runConf.get(OozieClient.GROUP_NAME), runConf);
331        }
332        catch (IOException e) {
333            throw new CommandException(ErrorCode.E1001, runConf.get(OozieClient.COORDINATOR_APP_PATH));
334        }
335        return runConf;
336    }
337
338    /* (non-Javadoc)
339     * @see org.apache.oozie.command.TransitionXCommand#getJob()
340     */
341    @Override
342    public Job getJob() {
343        return bundleJob;
344    }
345
346    /* (non-Javadoc)
347     * @see org.apache.oozie.command.TransitionXCommand#updateJob()
348     */
349    @Override
350    public void updateJob() throws CommandException {
351        updateList.add(new UpdateEntry<BundleJobQuery>(BundleJobQuery.UPDATE_BUNDLE_JOB_STATUS_PENDING, bundleJob));
352    }
353}