This project has retired. For details please refer to its
Attic page.
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.servlet;
019
020 import org.apache.hadoop.conf.Configuration;
021 import org.apache.oozie.client.rest.RestConstants;
022 import org.apache.oozie.client.rest.JsonTags;
023 import org.apache.oozie.client.OozieClient;
024 import org.apache.oozie.util.XConfiguration;
025 import org.apache.oozie.service.Services;
026 import org.apache.oozie.service.DagEngineService;
027 import org.apache.oozie.service.WorkflowAppService;
028 import org.apache.oozie.DagEngine;
029 import org.apache.oozie.DagEngineException;
030 import org.apache.oozie.WorkflowsInfo;
031 import org.apache.oozie.WorkflowJobBean;
032 import org.apache.oozie.ErrorCode;
033 import org.json.simple.JSONObject;
034
035 import javax.servlet.ServletException;
036 import javax.servlet.http.HttpServletRequest;
037 import javax.servlet.http.HttpServletResponse;
038 import java.io.IOException;
039 import java.util.Arrays;
040 import java.util.List;
041
042 public class JobsServlet extends JsonRestServlet {
043 private static final String INSTRUMENTATION_NAME = "jobs";
044
045 private static final JsonRestServlet.ResourceInfo RESOURCES_INFO[] = new JsonRestServlet.ResourceInfo[1];
046
047 static {
048 RESOURCES_INFO[0] =
049 new JsonRestServlet.ResourceInfo("", Arrays.asList("POST", "GET"), Arrays.asList(
050 new JsonRestServlet.ParameterInfo(RestConstants.ACTION_PARAM, String.class, false, Arrays.asList("POST")),
051 new JsonRestServlet.ParameterInfo(RestConstants.JOBS_FILTER_PARAM, String.class, false, Arrays.asList("GET")),
052 new JsonRestServlet.ParameterInfo(RestConstants.JOBTYPE_PARAM, String.class, false, Arrays.asList("GET")),
053 new JsonRestServlet.ParameterInfo(RestConstants.OFFSET_PARAM, String.class, false, Arrays.asList("GET")),
054 new JsonRestServlet.ParameterInfo(RestConstants.LEN_PARAM, String.class, false, Arrays.asList("GET")),
055 new JsonRestServlet.ParameterInfo(RestConstants.JOBS_EXTERNAL_ID_PARAM, String.class, false, Arrays.asList("GET"))));
056 }
057
058 public JobsServlet() {
059 super(INSTRUMENTATION_NAME, RESOURCES_INFO);
060 }
061
062 /**
063 * Create a job.
064 */
065 @SuppressWarnings("unchecked")
066 protected void doPost(HttpServletRequest request, HttpServletResponse response)
067 throws ServletException, IOException {
068
069 validateContentType(request, RestConstants.XML_CONTENT_TYPE);
070
071 request.setAttribute(AUDIT_OPERATION, request.getParameter(RestConstants.ACTION_PARAM));
072
073 XConfiguration conf = new XConfiguration(request.getInputStream());
074
075 stopCron();
076
077 conf = conf.trim();
078 conf = conf.resolve();
079
080 validateJobConfiguration(conf);
081
082 JobServlet.checkAuthorizationForApp(getUser(request), conf);
083
084 String action = request.getParameter(RestConstants.ACTION_PARAM);
085 if (action != null && !action.equals(RestConstants.JOB_ACTION_START)) {
086 throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ErrorCode.E0303, RestConstants.ACTION_PARAM,
087 action);
088 }
089 try {
090 boolean startJob = (action != null);
091 String user = conf.get(OozieClient.USER_NAME);
092 DagEngine dagEngine = Services.get().get(DagEngineService.class).getDagEngine(user, getAuthToken(request));
093 String id = dagEngine.submitJob(conf, startJob);
094 JSONObject json = new JSONObject();
095 json.put(JsonTags.JOB_ID, id);
096 startCron();
097 sendJsonResponse(response, HttpServletResponse.SC_CREATED, json);
098 }
099 catch (DagEngineException ex) {
100 throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ex);
101 }
102 }
103
104 static void validateJobConfiguration(Configuration conf) throws XServletException {
105 if (conf.get(OozieClient.USER_NAME) == null) {
106 throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ErrorCode.E0401,
107 OozieClient.USER_NAME);
108 }
109
110 String localRealm = Services.get().getConf().get("local.realm");
111
112 //if the job properties don't define JT/NN Kerberos principals, add default value
113 if (conf.get(WorkflowAppService.HADOOP_JT_KERBEROS_NAME) == null) {
114 conf.set(WorkflowAppService.HADOOP_JT_KERBEROS_NAME, "mapred/_HOST@" + localRealm);
115 }
116 if (conf.get(WorkflowAppService.HADOOP_NN_KERBEROS_NAME) == null) {
117 conf.set(WorkflowAppService.HADOOP_NN_KERBEROS_NAME, "hdfs/_HOST@" + localRealm);
118 }
119 }
120
121 /**
122 * Return information about jobs.
123 */
124 @SuppressWarnings("unchecked")
125 public void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
126 try {
127 String externalId = request.getParameter(RestConstants.JOBS_EXTERNAL_ID_PARAM);
128 if (externalId != null) {
129 stopCron();
130 DagEngine dagEngine = Services.get().get(DagEngineService.class)
131 .getDagEngine(getUser(request), getAuthToken(request));
132 String jobId = dagEngine.getJobIdForExternalId(externalId);
133 JSONObject json = new JSONObject();
134 json.put(JsonTags.JOB_ID, jobId);
135 startCron();
136 sendJsonResponse(response, HttpServletResponse.SC_OK, json);
137 }
138 else {
139 String filter = request.getParameter(RestConstants.JOBS_FILTER_PARAM);
140 String startStr = request.getParameter(RestConstants.OFFSET_PARAM);
141 String lenStr = request.getParameter(RestConstants.LEN_PARAM);
142 int start = (startStr != null) ? Integer.parseInt(startStr) : 1;
143 start = (start < 1) ? 1 : start;
144 int len = (lenStr != null) ? Integer.parseInt(lenStr) : 50;
145 len = (len < 1) ? 50 : len;
146 stopCron();
147 DagEngine dagEngine = Services.get().get(DagEngineService.class)
148 .getDagEngine(getUser(request), getAuthToken(request));
149 WorkflowsInfo jobs = dagEngine.getJobs(filter, start, len);
150 List<WorkflowJobBean> jsonWorkflows = jobs.getWorkflows();
151 startCron();
152 JSONObject json = new JSONObject();
153 json.put(JsonTags.WORKFLOWS_JOBS, WorkflowJobBean.toJSONArray(jsonWorkflows));
154 json.put(JsonTags.WORKFLOWS_TOTAL, jobs.getTotal());
155 json.put(JsonTags.WORKFLOWS_OFFSET, jobs.getStart());
156 json.put(JsonTags.WORKFLOWS_LEN, jobs.getLen());
157 sendJsonResponse(response, HttpServletResponse.SC_OK, json);
158 }
159 }
160 catch (DagEngineException ex) {
161 throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ex);
162 }
163 }
164
165 }