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.servlet;
020
021import java.io.IOException;
022import java.util.Arrays;
023
024import javax.servlet.ServletException;
025import javax.servlet.http.HttpServletRequest;
026import javax.servlet.http.HttpServletResponse;
027
028import org.apache.hadoop.conf.Configuration;
029import org.apache.oozie.BaseEngineException;
030import org.apache.oozie.ErrorCode;
031import org.apache.oozie.client.OozieClient;
032import org.apache.oozie.client.XOozieClient;
033import org.apache.oozie.client.rest.JsonBean;
034import org.apache.oozie.client.rest.JsonTags;
035import org.apache.oozie.client.rest.RestConstants;
036import org.apache.oozie.service.AuthorizationException;
037import org.apache.oozie.service.AuthorizationService;
038import org.apache.oozie.service.Services;
039import org.apache.oozie.service.XLogService;
040import org.apache.oozie.util.ConfigUtils;
041import org.apache.oozie.util.JobUtils;
042import org.apache.oozie.util.XConfiguration;
043import org.apache.oozie.util.XLog;
044import org.json.simple.JSONObject;
045
046public abstract class BaseJobServlet extends JsonRestServlet {
047
048    private static final ResourceInfo RESOURCES_INFO[] = new ResourceInfo[1];
049
050    static {
051        RESOURCES_INFO[0] = new ResourceInfo("*", Arrays.asList("PUT", "GET"), Arrays.asList(new ParameterInfo(
052                RestConstants.ACTION_PARAM, String.class, true, Arrays.asList("PUT")), new ParameterInfo(
053                RestConstants.JOB_SHOW_PARAM, String.class, false, Arrays.asList("GET")), new ParameterInfo(
054                        RestConstants.ORDER_PARAM, String.class, false, Arrays.asList("GET"))));
055    }
056
057    public BaseJobServlet(String instrumentationName) {
058        super(instrumentationName, RESOURCES_INFO);
059    }
060
061    /**
062     * Perform various job related actions - start, suspend, resume, kill, etc.
063     */
064    @Override
065    protected void doPut(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
066        String jobId = getResourceName(request);
067        request.setAttribute(AUDIT_PARAM, jobId);
068        request.setAttribute(AUDIT_OPERATION, request.getParameter(RestConstants.ACTION_PARAM));
069        try {
070            AuthorizationService auth = Services.get().get(AuthorizationService.class);
071            auth.authorizeForJob(getUser(request), jobId, true);
072        }
073        catch (AuthorizationException ex) {
074            throw new XServletException(HttpServletResponse.SC_UNAUTHORIZED, ex);
075        }
076
077        String action = request.getParameter(RestConstants.ACTION_PARAM);
078        if (action.equals(RestConstants.JOB_ACTION_START)) {
079            stopCron();
080            startJob(request, response);
081            startCron();
082            response.setStatus(HttpServletResponse.SC_OK);
083        }
084        else if (action.equals(RestConstants.JOB_ACTION_RESUME)) {
085            stopCron();
086            resumeJob(request, response);
087            startCron();
088            response.setStatus(HttpServletResponse.SC_OK);
089        }
090        else if (action.equals(RestConstants.JOB_ACTION_SUSPEND)) {
091            stopCron();
092            suspendJob(request, response);
093            startCron();
094            response.setStatus(HttpServletResponse.SC_OK);
095        }
096        else if (action.equals(RestConstants.JOB_ACTION_KILL)) {
097            stopCron();
098            JSONObject json =  killJob(request, response);
099            startCron();
100            if (json != null) {
101                sendJsonResponse(response, HttpServletResponse.SC_OK, json);
102            }
103            else {
104                response.setStatus(HttpServletResponse.SC_OK);
105            }
106        }
107        else if (action.equals(RestConstants.JOB_ACTION_CHANGE)) {
108            stopCron();
109            changeJob(request, response);
110            startCron();
111            response.setStatus(HttpServletResponse.SC_OK);
112        }
113        else if (action.equals(RestConstants.JOB_ACTION_IGNORE)) {
114            stopCron();
115            JSONObject json = ignoreJob(request, response);
116            startCron();
117            if (json != null) {
118                sendJsonResponse(response, HttpServletResponse.SC_OK, json);
119            }
120            else {
121            response.setStatus(HttpServletResponse.SC_OK);
122            }
123        }
124        else if (action.equals(RestConstants.JOB_ACTION_RERUN)) {
125            validateContentType(request, RestConstants.XML_CONTENT_TYPE);
126            Configuration conf = new XConfiguration(request.getInputStream());
127            stopCron();
128            String requestUser = getUser(request);
129            if (!requestUser.equals(UNDEF)) {
130                conf.set(OozieClient.USER_NAME, requestUser);
131            }
132            if (conf.get(OozieClient.APP_PATH) != null) {
133                BaseJobServlet.checkAuthorizationForApp(conf);
134                JobUtils.normalizeAppPath(conf.get(OozieClient.USER_NAME), conf.get(OozieClient.GROUP_NAME), conf);
135            }
136            reRunJob(request, response, conf);
137            startCron();
138            response.setStatus(HttpServletResponse.SC_OK);
139        }
140        else if (action.equals(RestConstants.JOB_COORD_ACTION_RERUN)) {
141            validateContentType(request, RestConstants.XML_CONTENT_TYPE);
142            stopCron();
143            JSONObject json = reRunJob(request, response, null);
144            startCron();
145            if (json != null) {
146                sendJsonResponse(response, HttpServletResponse.SC_OK, json);
147            }
148            else {
149                response.setStatus(HttpServletResponse.SC_OK);
150            }
151        }
152        else if (action.equals(RestConstants.JOB_BUNDLE_ACTION_RERUN)) {
153            validateContentType(request, RestConstants.XML_CONTENT_TYPE);
154            stopCron();
155            JSONObject json = reRunJob(request, response, null);
156            startCron();
157            if (json != null) {
158                sendJsonResponse(response, HttpServletResponse.SC_OK, json);
159            }
160            else {
161                response.setStatus(HttpServletResponse.SC_OK);
162            }
163        }
164        else if (action.equals(RestConstants.JOB_COORD_UPDATE)) {
165            validateContentType(request, RestConstants.XML_CONTENT_TYPE);
166            Configuration conf = new XConfiguration(request.getInputStream());
167            stopCron();
168            String requestUser = getUser(request);
169            if (!requestUser.equals(UNDEF)) {
170                conf.set(OozieClient.USER_NAME, requestUser);
171            }
172            if (conf.get(OozieClient.COORDINATOR_APP_PATH) != null) {
173                //If coord is submitted from bundle, user may want to update individual coord job with bundle properties
174                //If COORDINATOR_APP_PATH is set, we should check only COORDINATOR_APP_PATH path permission
175                String bundlePath = conf.get(OozieClient.BUNDLE_APP_PATH);
176                if (bundlePath != null) {
177                    conf.unset(OozieClient.BUNDLE_APP_PATH);
178                }
179                BaseJobServlet.checkAuthorizationForApp(conf);
180                JobUtils.normalizeAppPath(conf.get(OozieClient.USER_NAME), conf.get(OozieClient.GROUP_NAME), conf);
181                if (bundlePath != null) {
182                    conf.set(OozieClient.BUNDLE_APP_PATH, bundlePath);
183                }
184            }
185            JSONObject json = updateJob(request, response, conf);
186            startCron();
187            sendJsonResponse(response, HttpServletResponse.SC_OK, json);
188        }
189        else {
190            throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ErrorCode.E0303,
191                    RestConstants.ACTION_PARAM, action);
192        }
193    }
194
195    abstract JSONObject ignoreJob(HttpServletRequest request, HttpServletResponse response) throws XServletException,
196            IOException;
197
198    /**
199     * Validate the configuration user/group. <p/>
200     *
201     * @param conf configuration.
202     * @throws XServletException thrown if the configuration does not have a property {@link
203     * org.apache.oozie.client.OozieClient#USER_NAME}.
204     */
205    static void checkAuthorizationForApp(Configuration conf) throws XServletException {
206        String user = conf.get(OozieClient.USER_NAME);
207        String acl = ConfigUtils.getWithDeprecatedCheck(conf, OozieClient.GROUP_NAME, OozieClient.JOB_ACL, null);
208        try {
209            if (user == null) {
210                throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ErrorCode.E0401, OozieClient.USER_NAME);
211            }
212            AuthorizationService auth = Services.get().get(AuthorizationService.class);
213
214            if (acl != null){
215                 conf.set(OozieClient.GROUP_NAME, acl);
216            }
217            else if (acl == null && auth.useDefaultGroupAsAcl()) {
218                acl = auth.getDefaultGroup(user);
219                conf.set(OozieClient.GROUP_NAME, acl);
220            }
221            XLog.Info.get().setParameter(XLogService.GROUP, acl);
222            String wfPath = conf.get(OozieClient.APP_PATH);
223            String coordPath = conf.get(OozieClient.COORDINATOR_APP_PATH);
224            String bundlePath = conf.get(OozieClient.BUNDLE_APP_PATH);
225
226            if (wfPath == null && coordPath == null && bundlePath == null) {
227                String[] libPaths = conf.getStrings(XOozieClient.LIBPATH);
228                if (libPaths != null && libPaths.length > 0 && libPaths[0].trim().length() > 0) {
229                    conf.set(OozieClient.APP_PATH, libPaths[0].trim());
230                    wfPath = libPaths[0].trim();
231                }
232                else {
233                    throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ErrorCode.E0405);
234                }
235            }
236            ServletUtilities.ValidateAppPath(wfPath, coordPath, bundlePath);
237
238            if (wfPath != null) {
239                auth.authorizeForApp(user, acl, wfPath, "workflow.xml", conf);
240            }
241            else if (coordPath != null){
242                auth.authorizeForApp(user, acl, coordPath, "coordinator.xml", conf);
243            }
244            else if (bundlePath != null){
245                auth.authorizeForApp(user, acl, bundlePath, "bundle.xml", conf);
246            }
247        }
248        catch (AuthorizationException ex) {
249            XLog.getLog(BaseJobServlet.class).info("AuthorizationException ", ex);
250            throw new XServletException(HttpServletResponse.SC_UNAUTHORIZED, ex);
251        }
252    }
253
254    /**
255     * Return information about jobs.
256     */
257    @Override
258    @SuppressWarnings("unchecked")
259    public void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
260        String jobId = getResourceName(request);
261        String show = request.getParameter(RestConstants.JOB_SHOW_PARAM);
262        String timeZoneId = request.getParameter(RestConstants.TIME_ZONE_PARAM) == null
263                ? "GMT" : request.getParameter(RestConstants.TIME_ZONE_PARAM);
264
265        try {
266            AuthorizationService auth = Services.get().get(AuthorizationService.class);
267            auth.authorizeForJob(getUser(request), jobId, false);
268        }
269        catch (AuthorizationException ex) {
270            throw new XServletException(HttpServletResponse.SC_UNAUTHORIZED, ex);
271        }
272
273        if (show == null || show.equals(RestConstants.JOB_SHOW_INFO)) {
274            stopCron();
275            JsonBean job = null;
276            try {
277                job = getJob(request, response);
278            }
279            catch (BaseEngineException e) {
280                // TODO Auto-generated catch block
281                // e.printStackTrace();
282
283                throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, e);
284            }
285            startCron();
286            sendJsonResponse(response, HttpServletResponse.SC_OK, job, timeZoneId);
287        }
288        else if (show.equals(RestConstants.ALL_WORKFLOWS_FOR_COORD_ACTION)) {
289            stopCron();
290            JSONObject json = getJobsByParentId(request, response);
291            startCron();
292            sendJsonResponse(response, HttpServletResponse.SC_OK, json);
293        }
294        else if (show.equals(RestConstants.JOB_SHOW_JMS_TOPIC)) {
295            stopCron();
296            String jmsTopicName = getJMSTopicName(request, response);
297            JSONObject json = new JSONObject();
298            json.put(JsonTags.JMS_TOPIC_NAME, jmsTopicName);
299            startCron();
300            sendJsonResponse(response, HttpServletResponse.SC_OK, json);
301        }
302
303        else if (show.equals(RestConstants.JOB_SHOW_LOG)) {
304            response.setContentType(TEXT_UTF8);
305            streamJobLog(request, response);
306        }
307        else if (show.equals(RestConstants.JOB_SHOW_DEFINITION)) {
308            stopCron();
309            response.setContentType(XML_UTF8);
310            String wfDefinition = getJobDefinition(request, response);
311            startCron();
312            response.setStatus(HttpServletResponse.SC_OK);
313            response.getWriter().write(wfDefinition);
314        }
315        else if (show.equals(RestConstants.JOB_SHOW_GRAPH)) {
316            stopCron();
317            streamJobGraph(request, response);
318            startCron(); // -- should happen before you stream anything in response?
319        } else if (show.equals(RestConstants.JOB_SHOW_STATUS)) {
320            stopCron();
321            String status = getJobStatus(request, response);
322            JSONObject json = new JSONObject();
323            json.put(JsonTags.STATUS, status);
324            startCron();
325            sendJsonResponse(response, HttpServletResponse.SC_OK, json);
326        }
327        else {
328            throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ErrorCode.E0303,
329                    RestConstants.JOB_SHOW_PARAM, show);
330        }
331    }
332
333    /**
334     * abstract method to start a job, either workflow or coordinator
335     *
336     * @param request
337     * @param response
338     * @throws XServletException
339     * @throws IOException TODO
340     */
341    abstract void startJob(HttpServletRequest request, HttpServletResponse response) throws XServletException,
342            IOException;
343
344    /**
345     * abstract method to resume a job, either workflow or coordinator
346     *
347     * @param request
348     * @param response
349     * @throws XServletException
350     * @throws IOException TODO
351     */
352    abstract void resumeJob(HttpServletRequest request, HttpServletResponse response) throws XServletException,
353            IOException;
354
355    /**
356     * abstract method to suspend a job, either workflow or coordinator
357     *
358     * @param request
359     * @param response
360     * @throws XServletException
361     * @throws IOException TODO
362     */
363    abstract void suspendJob(HttpServletRequest request, HttpServletResponse response) throws XServletException,
364            IOException;
365
366    /**
367     * abstract method to kill a job, either workflow or coordinator
368     *
369     * @param request
370     * @param response
371     * @return
372     * @throws XServletException
373     * @throws IOException TODO
374     */
375    abstract JSONObject killJob(HttpServletRequest request, HttpServletResponse response) throws XServletException,
376            IOException;
377
378    /**
379     * abstract method to change a coordinator job
380     *
381     * @param request
382     * @param response
383     * @throws XServletException
384     * @throws IOException TODO
385     */
386    abstract void changeJob(HttpServletRequest request, HttpServletResponse response) throws XServletException,
387            IOException;
388
389    /**
390     * abstract method to re-run a job, either workflow or coordinator
391     *
392     * @param request
393     * @param response
394     * @param conf
395     * @throws XServletException
396     * @throws IOException TODO
397     */
398    abstract JSONObject reRunJob(HttpServletRequest request, HttpServletResponse response, Configuration conf)
399            throws XServletException, IOException;
400
401    /**
402     * abstract method to get a job, either workflow or coordinator, in JsonBean representation
403     *
404     * @param request
405     * @param response
406     * @return JsonBean representation of a job, either workflow or coordinator
407     * @throws XServletException
408     * @throws IOException TODO
409     * @throws BaseEngineException
410     */
411    abstract JsonBean getJob(HttpServletRequest request, HttpServletResponse response) throws XServletException,
412            IOException, BaseEngineException;
413
414    /**
415     * abstract method to get definition of a job, either workflow or coordinator
416     *
417     * @param request
418     * @param response
419     * @return job, either workflow or coordinator, definition in string format
420     * @throws XServletException
421     * @throws IOException TODO
422     */
423    abstract String getJobDefinition(HttpServletRequest request, HttpServletResponse response)
424            throws XServletException, IOException;
425
426    /**
427     * abstract method to get and stream log information of job, either workflow or coordinator
428     *
429     * @param request
430     * @param response
431     * @throws XServletException
432     * @throws IOException
433     */
434    abstract void streamJobLog(HttpServletRequest request, HttpServletResponse response) throws XServletException,
435            IOException;
436
437    /**
438     * abstract method to create and stream image for runtime DAG -- workflow only
439     *
440     * @param request
441     * @param response
442     * @throws XServletException
443     * @throws IOException
444     */
445    abstract void streamJobGraph(HttpServletRequest request, HttpServletResponse response)
446            throws XServletException, IOException;
447
448    /**
449     * abstract method to get JMS topic name for a job
450     * @param request
451     * @param response
452     * @throws XServletException
453     * @throws IOException
454     */
455    abstract String getJMSTopicName(HttpServletRequest request, HttpServletResponse response)
456            throws XServletException, IOException;
457
458    /**
459     * abstract method to get workflow job ids from the parent id
460     * i.e. coordinator action
461     * @param request
462     * @param response
463     * @return comma-separated list of workflow job ids
464     * @throws XServletException
465     * @throws IOException
466     */
467    abstract JSONObject getJobsByParentId(HttpServletRequest request, HttpServletResponse response)
468            throws XServletException, IOException;
469
470    /**
471     * Abstract method to Update coord job.
472     *
473     * @param request the request
474     * @param response the response
475     * @param Configuration conf
476     * @return the JSON object
477     * @throws XServletException the x servlet exception
478     * @throws IOException Signals that an I/O exception has occurred.
479     */
480    abstract JSONObject updateJob(HttpServletRequest request, HttpServletResponse response, Configuration conf)
481            throws XServletException, IOException;
482
483    /**
484     * Abstract method to get status for a job
485     *
486     * @param request the request
487     * @param response the response
488     * @return the JSON object
489     * @throws XServletException the x servlet exception
490     * @throws IOException Signals that an I/O exception has occurred.
491     */
492    abstract String getJobStatus(HttpServletRequest request, HttpServletResponse response)
493            throws XServletException, IOException;
494}
495