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.service.ConfigurationService;
039import org.apache.oozie.service.DagEngineService;
040import org.apache.oozie.service.Services;
041import org.apache.oozie.util.ConfigUtils;
042import org.apache.oozie.util.JobUtils;
043import org.apache.oozie.util.PropertiesUtils;
044import org.apache.oozie.util.XConfiguration;
045import org.apache.oozie.util.XLog;
046import org.apache.oozie.util.XmlUtils;
047import org.jdom.Element;
048import org.jdom.Namespace;
049
050public class SubWorkflowActionExecutor extends ActionExecutor {
051    public static final String ACTION_TYPE = "sub-workflow";
052    public static final String LOCAL = "local";
053    public static final String PARENT_ID = "oozie.wf.parent.id";
054    public static final String SUPER_PARENT_ID = "oozie.wf.superparent.id";
055    public static final String SUBWORKFLOW_MAX_DEPTH = "oozie.action.subworkflow.max.depth";
056    public static final String SUBWORKFLOW_DEPTH = "oozie.action.subworkflow.depth";
057    public static final String SUBWORKFLOW_RERUN = "oozie.action.subworkflow.rerun";
058
059    private static final Set<String> DISALLOWED_USER_PROPERTIES = new HashSet<String>();
060    public XLog LOG = XLog.getLog(getClass());
061
062
063    static {
064        String[] badUserProps = {PropertiesUtils.DAYS, PropertiesUtils.HOURS, PropertiesUtils.MINUTES,
065                PropertiesUtils.KB, PropertiesUtils.MB, PropertiesUtils.GB, PropertiesUtils.TB, PropertiesUtils.PB,
066                PropertiesUtils.RECORDS, PropertiesUtils.MAP_IN, PropertiesUtils.MAP_OUT, PropertiesUtils.REDUCE_IN,
067                PropertiesUtils.REDUCE_OUT, PropertiesUtils.GROUPS};
068
069        PropertiesUtils.createPropertySet(badUserProps, DISALLOWED_USER_PROPERTIES);
070    }
071
072    protected SubWorkflowActionExecutor() {
073        super(ACTION_TYPE);
074    }
075
076    public void initActionType() {
077        super.initActionType();
078    }
079
080    protected OozieClient getWorkflowClient(Context context, String oozieUri) {
081        OozieClient oozieClient;
082        if (oozieUri.equals(LOCAL)) {
083            WorkflowJobBean workflow = (WorkflowJobBean) context.getWorkflow();
084            String user = workflow.getUser();
085            String group = workflow.getGroup();
086            DagEngine dagEngine = Services.get().get(DagEngineService.class).getDagEngine(user);
087            oozieClient = new LocalOozieClient(dagEngine);
088        }
089        else {
090            // TODO we need to add authToken to the WC for the remote case
091            oozieClient = new OozieClient(oozieUri);
092        }
093        return oozieClient;
094    }
095
096    protected void injectInline(Element eConf, Configuration subWorkflowConf) throws IOException,
097            ActionExecutorException {
098        if (eConf != null) {
099            String strConf = XmlUtils.prettyPrint(eConf).toString();
100            Configuration conf = new XConfiguration(new StringReader(strConf));
101            try {
102                PropertiesUtils.checkDisallowedProperties(conf, DISALLOWED_USER_PROPERTIES);
103                PropertiesUtils.checkDefaultDisallowedProperties(conf);
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
183                Configuration parentConf = new XConfiguration(new StringReader(context.getWorkflow().getConf()));
184
185                if (eConf.getChild(("propagate-configuration"), ns) != null) {
186                    XConfiguration.copy(parentConf, subWorkflowConf);
187                }
188
189                // Propagate coordinator and bundle info to subworkflow
190                if (OozieJobInfo.isJobInfoEnabled()) {
191                  if (parentConf.get(OozieJobInfo.COORD_ID) != null) {
192                    subWorkflowConf.set(OozieJobInfo.COORD_ID, parentConf.get(OozieJobInfo.COORD_ID));
193                    subWorkflowConf.set(OozieJobInfo.COORD_NAME, parentConf.get(OozieJobInfo.COORD_NAME));
194                    subWorkflowConf.set(OozieJobInfo.COORD_NOMINAL_TIME, parentConf.get(OozieJobInfo.COORD_NOMINAL_TIME));
195                  }
196                  if (parentConf.get(OozieJobInfo.BUNDLE_ID) != null) {
197                    subWorkflowConf.set(OozieJobInfo.BUNDLE_ID, parentConf.get(OozieJobInfo.BUNDLE_ID));
198                    subWorkflowConf.set(OozieJobInfo.BUNDLE_NAME, parentConf.get(OozieJobInfo.BUNDLE_NAME));
199                  }
200                }
201
202                // the proto has the necessary credentials
203                Configuration protoActionConf = context.getProtoActionConf();
204                XConfiguration.copy(protoActionConf, subWorkflowConf);
205                subWorkflowConf.set(OozieClient.APP_PATH, appPath);
206                String group = ConfigUtils.getWithDeprecatedCheck(parentConf, OozieClient.JOB_ACL, OozieClient.GROUP_NAME, null);
207                if(group != null) {
208                    subWorkflowConf.set(OozieClient.GROUP_NAME, group);
209                }
210
211                injectInline(eConf.getChild("configuration", ns), subWorkflowConf);
212                injectCallback(context, subWorkflowConf);
213                injectRecovery(extId, subWorkflowConf);
214                injectParent(context.getWorkflow().getId(), subWorkflowConf);
215                injectSuperParent(context.getWorkflow(), parentConf, subWorkflowConf);
216                verifyAndInjectSubworkflowDepth(parentConf, subWorkflowConf);
217
218                //TODO: this has to be refactored later to be done in a single place for REST calls and this
219                JobUtils.normalizeAppPath(context.getWorkflow().getUser(), context.getWorkflow().getGroup(),
220                                          subWorkflowConf);
221
222                subWorkflowConf.set(OOZIE_ACTION_YARN_TAG, getActionYarnTag(parentConf, context.getWorkflow(), action));
223
224                // if the rerun failed node option is provided during the time of rerun command, old subworkflow will
225                // rerun again.
226                if(action.getExternalId() != null && parentConf.getBoolean(OozieClient.RERUN_FAIL_NODES, false)) {
227                    subWorkflowConf.setBoolean(SUBWORKFLOW_RERUN, true);
228                    oozieClient.reRun(action.getExternalId(), subWorkflowConf.toProperties());
229                    subWorkflowId = action.getExternalId();
230                } else {
231                    subWorkflowId = oozieClient.run(subWorkflowConf.toProperties());
232                }
233            }
234            else {
235                subWorkflowId = runningJobId;
236            }
237            WorkflowJob workflow = oozieClient.getJobInfo(subWorkflowId);
238            String consoleUrl = workflow.getConsoleUrl();
239            context.setStartData(subWorkflowId, oozieUri, consoleUrl);
240            if (runningJobId != null) {
241                check(context, action);
242            }
243        }
244        catch (Exception ex) {
245            LOG.error(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}