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.action.oozie;
020
021import java.io.IOException;
022import java.io.StringReader;
023import java.util.HashSet;
024import java.util.Set;
025
026import org.apache.hadoop.conf.Configuration;
027import org.apache.oozie.DagEngine;
028import org.apache.oozie.LocalOozieClient;
029import org.apache.oozie.WorkflowJobBean;
030import org.apache.oozie.action.ActionExecutor;
031import org.apache.oozie.action.ActionExecutorException;
032import org.apache.oozie.action.hadoop.OozieJobInfo;
033import org.apache.oozie.client.OozieClient;
034import org.apache.oozie.client.OozieClientException;
035import org.apache.oozie.client.WorkflowAction;
036import org.apache.oozie.client.WorkflowJob;
037import org.apache.oozie.command.CommandException;
038import org.apache.oozie.command.wf.ActionStartXCommand;
039import org.apache.oozie.service.ConfigurationService;
040import org.apache.oozie.service.DagEngineService;
041import org.apache.oozie.service.Services;
042import org.apache.oozie.util.ConfigUtils;
043import org.apache.oozie.util.JobUtils;
044import org.apache.oozie.util.PropertiesUtils;
045import org.apache.oozie.util.XConfiguration;
046import org.apache.oozie.util.XLog;
047import org.apache.oozie.util.XmlUtils;
048import org.jdom.Element;
049import org.jdom.Namespace;
050
051public class SubWorkflowActionExecutor extends ActionExecutor {
052    public static final String ACTION_TYPE = "sub-workflow";
053    public static final String LOCAL = "local";
054    public static final String PARENT_ID = "oozie.wf.parent.id";
055    public static final String SUPER_PARENT_ID = "oozie.wf.superparent.id";
056    public static final String SUBWORKFLOW_MAX_DEPTH = "oozie.action.subworkflow.max.depth";
057    public static final String SUBWORKFLOW_DEPTH = "oozie.action.subworkflow.depth";
058    public static final String SUBWORKFLOW_RERUN = "oozie.action.subworkflow.rerun";
059
060    private static final Set<String> DISALLOWED_DEFAULT_PROPERTIES = new HashSet<String>();
061
062    static {
063        String[] badUserProps = {PropertiesUtils.DAYS, PropertiesUtils.HOURS, PropertiesUtils.MINUTES,
064                PropertiesUtils.KB, PropertiesUtils.MB, PropertiesUtils.GB, PropertiesUtils.TB, PropertiesUtils.PB,
065                PropertiesUtils.RECORDS, PropertiesUtils.MAP_IN, PropertiesUtils.MAP_OUT, PropertiesUtils.REDUCE_IN,
066                PropertiesUtils.REDUCE_OUT, PropertiesUtils.GROUPS};
067
068        String[] badDefaultProps = {PropertiesUtils.HADOOP_USER};
069        PropertiesUtils.createPropertySet(badUserProps, DISALLOWED_DEFAULT_PROPERTIES);
070        PropertiesUtils.createPropertySet(badDefaultProps, DISALLOWED_DEFAULT_PROPERTIES);
071    }
072
073    protected SubWorkflowActionExecutor() {
074        super(ACTION_TYPE);
075    }
076
077    public void initActionType() {
078        super.initActionType();
079    }
080
081    protected OozieClient getWorkflowClient(Context context, String oozieUri) {
082        OozieClient oozieClient;
083        if (oozieUri.equals(LOCAL)) {
084            WorkflowJobBean workflow = (WorkflowJobBean) context.getWorkflow();
085            String user = workflow.getUser();
086            String group = workflow.getGroup();
087            DagEngine dagEngine = Services.get().get(DagEngineService.class).getDagEngine(user);
088            oozieClient = new LocalOozieClient(dagEngine);
089        }
090        else {
091            // TODO we need to add authToken to the WC for the remote case
092            oozieClient = new OozieClient(oozieUri);
093        }
094        return oozieClient;
095    }
096
097    protected void injectInline(Element eConf, Configuration subWorkflowConf) throws IOException,
098            ActionExecutorException {
099        if (eConf != null) {
100            String strConf = XmlUtils.prettyPrint(eConf).toString();
101            Configuration conf = new XConfiguration(new StringReader(strConf));
102            try {
103                PropertiesUtils.checkDisallowedProperties(conf, DISALLOWED_DEFAULT_PROPERTIES);
104            }
105            catch (CommandException ex) {
106                throw convertException(ex);
107            }
108            XConfiguration.copy(conf, subWorkflowConf);
109        }
110    }
111
112    @SuppressWarnings("unchecked")
113    protected void injectCallback(Context context, Configuration conf) {
114        String callback = context.getCallbackUrl("$status");
115        if (conf.get(OozieClient.WORKFLOW_NOTIFICATION_URL) != null) {
116            XLog.getLog(getClass())
117                    .warn("Sub-Workflow configuration has a custom job end notification URI, overriding");
118        }
119        conf.set(OozieClient.WORKFLOW_NOTIFICATION_URL, callback);
120    }
121
122    protected void injectRecovery(String externalId, Configuration conf) {
123        conf.set(OozieClient.EXTERNAL_ID, externalId);
124    }
125
126    protected void injectParent(String parentId, Configuration conf) {
127        conf.set(PARENT_ID, parentId);
128    }
129
130    protected void injectSuperParent(WorkflowJob parentWorkflow, Configuration parentConf, Configuration conf) {
131        String superParentId = parentConf.get(SUPER_PARENT_ID);
132        if (superParentId == null) {
133            // This is a sub-workflow at depth 1
134            superParentId = parentWorkflow.getParentId();
135
136            // If the parent workflow is not submitted through a coordinator then the parentId will be the super parent id.
137            if (superParentId == null) {
138                superParentId = parentWorkflow.getId();
139            }
140            conf.set(SUPER_PARENT_ID, superParentId);
141        } else {
142            // Sub-workflow at depth 2 or more.
143            conf.set(SUPER_PARENT_ID, superParentId);
144        }
145    }
146
147    protected void verifyAndInjectSubworkflowDepth(Configuration parentConf, Configuration conf) throws ActionExecutorException {
148        int depth = parentConf.getInt(SUBWORKFLOW_DEPTH, 0);
149        int maxDepth = ConfigurationService.getInt(SUBWORKFLOW_MAX_DEPTH);
150        if (depth >= maxDepth) {
151            throw new ActionExecutorException(ActionExecutorException.ErrorType.ERROR, "SUBWF001",
152                    "Depth [{0}] cannot exceed maximum subworkflow depth [{1}]", (depth + 1), maxDepth);
153        }
154        conf.setInt(SUBWORKFLOW_DEPTH, depth + 1);
155    }
156
157    protected String checkIfRunning(OozieClient oozieClient, String extId) throws OozieClientException {
158        String jobId = oozieClient.getJobId(extId);
159        if (jobId.equals("")) {
160            return null;
161        }
162        return jobId;
163    }
164
165    public void start(Context context, WorkflowAction action) throws ActionExecutorException {
166        try {
167            Element eConf = XmlUtils.parseXml(action.getConf());
168            Namespace ns = eConf.getNamespace();
169            Element e = eConf.getChild("oozie", ns);
170            String oozieUri = (e == null) ? LOCAL : e.getTextTrim();
171            OozieClient oozieClient = getWorkflowClient(context, oozieUri);
172            String subWorkflowId = null;
173            String extId = context.getRecoveryId();
174            String runningJobId = null;
175            if (extId != null) {
176                runningJobId = checkIfRunning(oozieClient, extId);
177            }
178            if (runningJobId == null) {
179                String appPath = eConf.getChild("app-path", ns).getTextTrim();
180
181                XConfiguration subWorkflowConf = new XConfiguration();
182                Configuration parentConf = new XConfiguration(new StringReader(context.getWorkflow().getConf()));
183                if (eConf.getChild(("propagate-configuration"), ns) != null) {
184                    XConfiguration.copy(parentConf, subWorkflowConf);
185                }
186
187                // Propagate coordinator and bundle info to subworkflow
188                if (OozieJobInfo.isJobInfoEnabled()) {
189                  if (parentConf.get(OozieJobInfo.COORD_ID) != null) {
190                    subWorkflowConf.set(OozieJobInfo.COORD_ID, parentConf.get(OozieJobInfo.COORD_ID));
191                    subWorkflowConf.set(OozieJobInfo.COORD_NAME, parentConf.get(OozieJobInfo.COORD_NAME));
192                    subWorkflowConf.set(OozieJobInfo.COORD_NOMINAL_TIME, parentConf.get(OozieJobInfo.COORD_NOMINAL_TIME));
193                  }
194                  if (parentConf.get(OozieJobInfo.BUNDLE_ID) != null) {
195                    subWorkflowConf.set(OozieJobInfo.BUNDLE_ID, parentConf.get(OozieJobInfo.BUNDLE_ID));
196                    subWorkflowConf.set(OozieJobInfo.BUNDLE_NAME, parentConf.get(OozieJobInfo.BUNDLE_NAME));
197                  }
198                }
199
200                // the proto has the necessary credentials
201                Configuration protoActionConf = context.getProtoActionConf();
202                XConfiguration.copy(protoActionConf, subWorkflowConf);
203                subWorkflowConf.set(OozieClient.APP_PATH, appPath);
204                String group = ConfigUtils.getWithDeprecatedCheck(parentConf, OozieClient.JOB_ACL, OozieClient.GROUP_NAME, null);
205                if(group != null) {
206                    subWorkflowConf.set(OozieClient.GROUP_NAME, group);
207                }
208                injectInline(eConf.getChild("configuration", ns), subWorkflowConf);
209                injectCallback(context, subWorkflowConf);
210                injectRecovery(extId, subWorkflowConf);
211                injectParent(context.getWorkflow().getId(), subWorkflowConf);
212                injectSuperParent(context.getWorkflow(), parentConf, subWorkflowConf);
213                verifyAndInjectSubworkflowDepth(parentConf, subWorkflowConf);
214
215                //TODO: this has to be refactored later to be done in a single place for REST calls and this
216                JobUtils.normalizeAppPath(context.getWorkflow().getUser(), context.getWorkflow().getGroup(),
217                                          subWorkflowConf);
218
219                // pushing the tag to conf for using by Launcher.
220                if(context.getVar(ActionStartXCommand.OOZIE_ACTION_YARN_TAG) != null) {
221                    subWorkflowConf.set(ActionStartXCommand.OOZIE_ACTION_YARN_TAG,
222                            context.getVar(ActionStartXCommand.OOZIE_ACTION_YARN_TAG));
223                }
224
225                // if the rerun failed node option is provided during the time of rerun command, old subworkflow will
226                // rerun again.
227                if(action.getExternalId() != null && parentConf.getBoolean(OozieClient.RERUN_FAIL_NODES, false)) {
228                    subWorkflowConf.setBoolean(SUBWORKFLOW_RERUN, true);
229                    oozieClient.reRun(action.getExternalId(), subWorkflowConf.toProperties());
230                    subWorkflowId = action.getExternalId();
231                } else {
232                    subWorkflowId = oozieClient.run(subWorkflowConf.toProperties());
233                }
234            }
235            else {
236                subWorkflowId = runningJobId;
237            }
238            WorkflowJob workflow = oozieClient.getJobInfo(subWorkflowId);
239            String consoleUrl = workflow.getConsoleUrl();
240            context.setStartData(subWorkflowId, oozieUri, consoleUrl);
241            if (runningJobId != null) {
242                check(context, action);
243            }
244        }
245        catch (Exception ex) {
246            throw convertException(ex);
247        }
248    }
249
250    public void end(Context context, WorkflowAction action) throws ActionExecutorException {
251        try {
252            String externalStatus = action.getExternalStatus();
253            WorkflowAction.Status status = externalStatus.equals("SUCCEEDED") ? WorkflowAction.Status.OK
254                                           : WorkflowAction.Status.ERROR;
255            context.setEndData(status, getActionSignal(status));
256        }
257        catch (Exception ex) {
258            throw convertException(ex);
259        }
260    }
261
262    public void check(Context context, WorkflowAction action) throws ActionExecutorException {
263        try {
264            String subWorkflowId = action.getExternalId();
265            String oozieUri = action.getTrackerUri();
266            OozieClient oozieClient = getWorkflowClient(context, oozieUri);
267            WorkflowJob subWorkflow = oozieClient.getJobInfo(subWorkflowId);
268            WorkflowJob.Status status = subWorkflow.getStatus();
269            switch (status) {
270                case FAILED:
271                case KILLED:
272                case SUCCEEDED:
273                    context.setExecutionData(status.toString(), null);
274                    break;
275                default:
276                    context.setExternalStatus(status.toString());
277                    break;
278            }
279        }
280        catch (Exception ex) {
281            throw convertException(ex);
282        }
283    }
284
285    public void kill(Context context, WorkflowAction action) throws ActionExecutorException {
286        try {
287            String subWorkflowId = action.getExternalId();
288            String oozieUri = action.getTrackerUri();
289            if (subWorkflowId != null && oozieUri != null) {
290                OozieClient oozieClient = getWorkflowClient(context, oozieUri);
291                oozieClient.kill(subWorkflowId);
292            }
293            context.setEndData(WorkflowAction.Status.KILLED, getActionSignal(WorkflowAction.Status.KILLED));
294        }
295        catch (Exception ex) {
296            throw convertException(ex);
297        }
298    }
299
300    private static Set<String> FINAL_STATUS = new HashSet<String>();
301
302    static {
303        FINAL_STATUS.add("SUCCEEDED");
304        FINAL_STATUS.add("KILLED");
305        FINAL_STATUS.add("FAILED");
306    }
307
308    public boolean isCompleted(String externalStatus) {
309        return FINAL_STATUS.contains(externalStatus);
310    }
311
312    public boolean supportsConfigurationJobXML() {
313        return true;
314    }
315}