1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.chukwa.datacollection.agent.rest;
19
20 import org.apache.hadoop.chukwa.datacollection.agent.ChukwaAgent;
21 import org.apache.hadoop.chukwa.datacollection.adaptor.AdaptorException;
22 import org.apache.hadoop.chukwa.datacollection.adaptor.Adaptor;
23 import org.apache.hadoop.chukwa.datacollection.OffsetStatsManager;
24 import org.apache.commons.logging.Log;
25 import org.apache.commons.logging.LogFactory;
26
27 import javax.ws.rs.Path;
28 import javax.ws.rs.GET;
29 import javax.ws.rs.Produces;
30 import javax.ws.rs.PathParam;
31 import javax.ws.rs.DELETE;
32 import javax.ws.rs.POST;
33 import javax.ws.rs.Consumes;
34 import javax.ws.rs.core.Response;
35 import javax.ws.rs.core.MediaType;
36 import javax.servlet.http.HttpServletResponse;
37
38 import java.text.DecimalFormat;
39 import java.util.ArrayList;
40 import java.util.List;
41 import java.util.Map;
42
43
44
45
46 @Path("/adaptor")
47 public class AdaptorController {
48
49 private static final DecimalFormat DECIMAL_FORMAT = new DecimalFormat();
50 private static final Log LOG = LogFactory.getLog(AdaptorController.class);
51
52 static {
53 DECIMAL_FORMAT.setMinimumFractionDigits(2);
54 DECIMAL_FORMAT.setMaximumFractionDigits(2);
55 DECIMAL_FORMAT.setGroupingUsed(false);
56 }
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71 @POST
72 @Consumes({MediaType.APPLICATION_JSON, MediaType.APPLICATION_XML})
73 @Produces({MediaType.APPLICATION_JSON, MediaType.APPLICATION_XML, MediaType.TEXT_PLAIN})
74 public Response addAdaptor(AdaptorConfig ac) {
75 ChukwaAgent agent = ChukwaAgent.getAgent();
76 if (ac.getAdaptorClass() == null ||
77 ac.getDataType() == null) {
78 return badRequestResponse("Bad adaptor config.");
79 }
80
81 StringBuilder addCommand = new StringBuilder("add ");
82 addCommand.append(ac.getAdaptorClass()).append(' ');
83 addCommand.append(ac.getDataType());
84 if (ac.getAdaptorParams() != null)
85 addCommand.append(' ').append(ac.getAdaptorParams());
86 addCommand.append(' ').append(ac.getOffset());
87
88
89 try {
90 String adaptorId = agent.processAddCommandE(addCommand.toString());
91 return doGetAdaptor(adaptorId);
92 } catch (AdaptorException e) {
93 LOG.warn("Could not add adaptor for data type: '" + ac.getDataType() +
94 "', error: " + e.getMessage());
95 return badRequestResponse("Could not add adaptor for data type: '" + ac.getDataType() +
96 "', error: " + e.getMessage());
97 }
98 }
99
100
101
102
103
104
105
106
107
108 @DELETE
109 @Path("/{adaptorId}")
110 @Produces({MediaType.TEXT_PLAIN})
111 public Response removeAdaptor(@PathParam("adaptorId") String adaptorId) {
112 ChukwaAgent agent = ChukwaAgent.getAgent();
113
114
115 if (adaptorId == null) {
116 return badRequestResponse("Missing adaptorId.");
117 }
118
119
120 if (agent.getAdaptor(adaptorId) == null) {
121 return badRequestResponse("Invalid adaptorId: " + adaptorId);
122 }
123
124
125 agent.stopAdaptor(adaptorId, true);
126 return Response.ok().build();
127 }
128
129
130
131
132
133
134
135
136
137 @GET
138 @Produces({MediaType.APPLICATION_JSON, MediaType.APPLICATION_XML})
139 public Response getAdaptors() {
140 return doGetAdaptors();
141 }
142
143
144
145
146
147
148
149
150
151
152 @GET
153 @Path("/{adaptorId}")
154 @Produces({MediaType.APPLICATION_JSON, MediaType.APPLICATION_XML})
155 public Response getAdaptor(@PathParam("adaptorId") String adaptorId) {
156 return doGetAdaptor(adaptorId);
157 }
158
159
160
161
162
163
164 private Response doGetAdaptor(String adaptorId) {
165 return Response.ok(buildAdaptor(adaptorId)).build();
166 }
167
168
169
170
171
172
173 private Response doGetAdaptors() {
174 return Response.ok(buildAdaptors()).build();
175 }
176
177
178
179
180 protected AdaptorInfo buildAdaptor(String adaptorId) {
181 ChukwaAgent agent = ChukwaAgent.getAgent();
182 Adaptor adaptor = agent.getAdaptor(adaptorId);
183 OffsetStatsManager<Adaptor> adaptorStats = agent.getAdaptorStatsManager();
184
185 AdaptorInfo info = new AdaptorInfo();
186 info.setId(adaptorId);
187 info.setDataType(adaptor.getType());
188 info.setAdaptorClass(adaptor.getClass().getName());
189 String[] status = adaptor.getCurrentStatus().split(" ",2);
190 info.setAdaptorParams(status[1]);
191 List<AdaptorAveragedRate> rates = new ArrayList<AdaptorAveragedRate>();
192 rates.add(new AdaptorAveragedRate(60, adaptorStats.calcAverageRate(adaptor, 60)));
193 rates.add(new AdaptorAveragedRate(300, adaptorStats.calcAverageRate(adaptor, 300)));
194 rates.add(new AdaptorAveragedRate(600, adaptorStats.calcAverageRate(adaptor, 600)));
195 info.setAdaptorRates(rates);
196 return info;
197 }
198
199
200
201
202 protected AdaptorList buildAdaptors() {
203 ChukwaAgent agent = ChukwaAgent.getAgent();
204 Map<String, String> adaptorMap = agent.getAdaptorList();
205 AdaptorList list = new AdaptorList();
206 for(String name : adaptorMap.keySet()) {
207 Adaptor adaptor = agent.getAdaptor(name);
208 OffsetStatsManager<Adaptor> adaptorStats = agent.getAdaptorStatsManager();
209
210 AdaptorInfo info = new AdaptorInfo();
211 info.setId(name);
212 info.setDataType(adaptor.getType());
213 info.setAdaptorClass(adaptor.getClass().getName());
214 String[] status = adaptor.getCurrentStatus().split(" ",2);
215 info.setAdaptorParams(status[1]);
216 List<AdaptorAveragedRate> rates = new ArrayList<AdaptorAveragedRate>();
217 rates.add(new AdaptorAveragedRate(60, adaptorStats.calcAverageRate(adaptor, 60)));
218 rates.add(new AdaptorAveragedRate(300, adaptorStats.calcAverageRate(adaptor, 300)));
219 rates.add(new AdaptorAveragedRate(600, adaptorStats.calcAverageRate(adaptor, 600)));
220 info.setAdaptorRates(rates);
221 list.add(info);
222 }
223 return list;
224 }
225
226
227
228
229 private static Response badRequestResponse(String content) {
230 return Response.status(HttpServletResponse.SC_BAD_REQUEST)
231 .entity(content).build();
232 }
233
234 }