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 java.io.IOException;
021
022 import javax.servlet.http.HttpServletRequest;
023 import javax.servlet.http.HttpServletResponse;
024
025 import org.apache.hadoop.conf.Configuration;
026 import org.apache.oozie.DagEngine;
027 import org.apache.oozie.DagEngineException;
028 import org.apache.oozie.client.rest.JsonBean;
029 import org.apache.oozie.service.DagEngineService;
030 import org.apache.oozie.service.Services;
031 import org.json.simple.JSONObject;
032 import org.apache.oozie.ErrorCode;
033
034 @SuppressWarnings("serial")
035 public class V0JobServlet extends BaseJobServlet {
036
037 private static final String INSTRUMENTATION_NAME = "v0job";
038
039 public V0JobServlet() {
040 super(INSTRUMENTATION_NAME);
041 }
042
043 /*
044 * v0 service method to start a job
045 */
046 @Override
047 protected void startJob(HttpServletRequest request, HttpServletResponse response) throws XServletException,
048 IOException {
049 DagEngine dagEngine = Services.get().get(DagEngineService.class).getDagEngine(getUser(request));
050
051 String jobId = getResourceName(request);
052 try {
053 dagEngine.start(jobId);
054 }
055 catch (DagEngineException ex) {
056 throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ex);
057 }
058 }
059
060 /*
061 * v0 service method to resume a job
062 */
063 @Override
064 protected void resumeJob(HttpServletRequest request, HttpServletResponse response) throws XServletException,
065 IOException {
066 DagEngine dagEngine = Services.get().get(DagEngineService.class).getDagEngine(getUser(request));
067
068 String jobId = getResourceName(request);
069 try {
070 dagEngine.resume(jobId);
071 }
072 catch (DagEngineException ex) {
073 throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ex);
074 }
075 }
076
077 /*
078 * v0 service method to suspend a job
079 */
080 @Override
081 protected void suspendJob(HttpServletRequest request, HttpServletResponse response) throws XServletException,
082 IOException {
083 DagEngine dagEngine = Services.get().get(DagEngineService.class).getDagEngine(getUser(request));
084
085 String jobId = getResourceName(request);
086 try {
087 dagEngine.suspend(jobId);
088 }
089 catch (DagEngineException ex) {
090 throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ex);
091 }
092 }
093
094 /*
095 * v0 service method to kill a job
096 */
097 @Override
098 protected void killJob(HttpServletRequest request, HttpServletResponse response) throws XServletException,
099 IOException {
100 DagEngine dagEngine = Services.get().get(DagEngineService.class).getDagEngine(getUser(request));
101
102 String jobId = getResourceName(request);
103 try {
104 dagEngine.kill(jobId);
105 }
106 catch (DagEngineException ex) {
107 throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ex);
108 }
109 }
110
111 /*
112 * v0 service method to change a job
113 */
114 protected void changeJob(HttpServletRequest request, HttpServletResponse response) throws XServletException,
115 IOException {
116 // This code should not be reached. But if it happens somehow, we throw
117 // bad request exception.
118 throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ErrorCode.E1014);
119 }
120
121 /*
122 * v0 service method to reRun a job
123 */
124 @Override
125 protected JSONObject reRunJob(HttpServletRequest request, HttpServletResponse response, Configuration conf)
126 throws XServletException, IOException {
127 DagEngine dagEngine = Services.get().get(DagEngineService.class).getDagEngine(getUser(request));
128
129 String jobId = getResourceName(request);
130 try {
131 dagEngine.reRun(jobId, conf);
132 }
133 catch (DagEngineException ex) {
134 throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ex);
135 }
136 return null;
137 }
138
139 /*
140 * v0 service method to get a job in JsonBean representation
141 */
142 @Override
143 protected JsonBean getJob(HttpServletRequest request, HttpServletResponse response) throws XServletException,
144 IOException {
145 DagEngine dagEngine = Services.get().get(DagEngineService.class).getDagEngine(getUser(request));
146
147 JsonBean jobBean = null;
148 String jobId = getResourceName(request);
149 try {
150 jobBean = (JsonBean) dagEngine.getJob(jobId);
151 }
152 catch (DagEngineException ex) {
153 throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ex);
154 }
155
156 return jobBean;
157 }
158
159 /*
160 * v0 service method to get a job definition in String format
161 */
162 @Override
163 protected String getJobDefinition(HttpServletRequest request, HttpServletResponse response)
164 throws XServletException, IOException {
165 DagEngine dagEngine = Services.get().get(DagEngineService.class).getDagEngine(getUser(request));
166
167 String wfDefinition = null;
168 String jobId = getResourceName(request);
169 try {
170 wfDefinition = dagEngine.getDefinition(jobId);
171 }
172 catch (DagEngineException ex) {
173 throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ex);
174 }
175 return wfDefinition;
176 }
177
178 /*
179 * v0 service method to stream a job log into response object
180 */
181 @Override
182 protected void streamJobLog(HttpServletRequest request, HttpServletResponse response) throws XServletException,
183 IOException {
184 DagEngine dagEngine = Services.get().get(DagEngineService.class).getDagEngine(getUser(request));
185
186 String jobId = getResourceName(request);
187 try {
188 dagEngine.streamLog(jobId, response.getWriter());
189 }
190 catch (DagEngineException ex) {
191 throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ex);
192 }
193 }
194
195 /*
196 * Not implemented in v0
197 */
198 @Override
199 protected void streamJobGraph(HttpServletRequest request, HttpServletResponse response)
200 throws XServletException, IOException {
201 // Should this error code be NOT_IMPLEMENTED?
202 throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ErrorCode.E0306);
203 }
204
205 @Override
206 protected String getJMSTopicName(HttpServletRequest request, HttpServletResponse response) throws XServletException,
207 IOException {
208 throw new XServletException(HttpServletResponse.SC_BAD_REQUEST, ErrorCode.E0306);
209 }
210 }