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 }