1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.regionserver;
20
21 import java.io.IOException;
22 import java.util.Comparator;
23 import java.util.List;
24
25 import org.apache.commons.lang.ClassUtils;
26 import org.apache.commons.logging.Log;
27 import org.apache.commons.logging.LogFactory;
28 import org.apache.hadoop.conf.Configuration;
29 import org.apache.hadoop.hbase.CellScanner;
30 import org.apache.hadoop.hbase.Coprocessor;
31 import org.apache.hadoop.hbase.CoprocessorEnvironment;
32 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
33 import org.apache.hadoop.hbase.MetaMutationAnnotation;
34 import org.apache.hadoop.hbase.classification.InterfaceAudience;
35 import org.apache.hadoop.hbase.classification.InterfaceStability;
36 import org.apache.hadoop.hbase.client.Mutation;
37 import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
38 import org.apache.hadoop.hbase.coprocessor.MetricsCoprocessor;
39 import org.apache.hadoop.hbase.coprocessor.ObserverContext;
40 import org.apache.hadoop.hbase.coprocessor.RegionServerCoprocessorEnvironment;
41 import org.apache.hadoop.hbase.coprocessor.RegionServerObserver;
42 import org.apache.hadoop.hbase.coprocessor.SingletonCoprocessorService;
43
44 import org.apache.hadoop.hbase.ipc.RpcServer;
45 import org.apache.hadoop.hbase.metrics.MetricRegistry;
46 import org.apache.hadoop.hbase.replication.ReplicationEndpoint;
47 import org.apache.hadoop.hbase.security.User;
48 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.WALEntry;
49
50 @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.COPROC)
51 @InterfaceStability.Evolving
52 public class RegionServerCoprocessorHost extends
53 CoprocessorHost<RegionServerCoprocessorHost.RegionServerEnvironment> {
54
55 private static final Log LOG = LogFactory.getLog(RegionServerCoprocessorHost.class);
56
57 private RegionServerServices rsServices;
58
59 public RegionServerCoprocessorHost(RegionServerServices rsServices,
60 Configuration conf) {
61 super(rsServices);
62 this.rsServices = rsServices;
63 this.conf = conf;
64
65
66
67 boolean coprocessorsEnabled = conf.getBoolean(COPROCESSORS_ENABLED_CONF_KEY,
68 DEFAULT_COPROCESSORS_ENABLED);
69 boolean tableCoprocessorsEnabled = conf.getBoolean(USER_COPROCESSORS_ENABLED_CONF_KEY,
70 DEFAULT_USER_COPROCESSORS_ENABLED);
71 LOG.info("System coprocessor loading is " + (coprocessorsEnabled ? "enabled" : "disabled"));
72 LOG.info("Table coprocessor loading is " +
73 ((coprocessorsEnabled && tableCoprocessorsEnabled) ? "enabled" : "disabled"));
74 loadSystemCoprocessors(conf, REGIONSERVER_COPROCESSOR_CONF_KEY);
75 }
76
77 @Override
78 public RegionServerEnvironment createEnvironment(Class<?> implClass,
79 Coprocessor instance, int priority, int sequence, Configuration conf) {
80 return new RegionServerEnvironment(implClass, instance, priority,
81 sequence, conf, this.rsServices);
82 }
83
84 public void preStop(String message) throws IOException {
85
86
87 execShutdown(coprocessors.isEmpty() ? null : new CoprocessorOperation() {
88 @Override
89 public void call(RegionServerObserver oserver,
90 ObserverContext<RegionServerCoprocessorEnvironment> ctx) throws IOException {
91 oserver.preStopRegionServer(ctx);
92 }
93 @Override
94 public void postEnvCall(RegionServerEnvironment env) {
95
96 shutdown(env);
97 }
98 });
99 }
100
101 public boolean preMerge(final HRegion regionA, final HRegion regionB) throws IOException {
102 return execOperation(coprocessors.isEmpty() ? null : new CoprocessorOperation() {
103 @Override
104 public void call(RegionServerObserver oserver,
105 ObserverContext<RegionServerCoprocessorEnvironment> ctx) throws IOException {
106 oserver.preMerge(ctx, regionA, regionB);
107 }
108 });
109 }
110
111 public void postMerge(final HRegion regionA, final HRegion regionB, final HRegion mergedRegion)
112 throws IOException {
113 execOperation(coprocessors.isEmpty() ? null : new CoprocessorOperation() {
114 @Override
115 public void call(RegionServerObserver oserver,
116 ObserverContext<RegionServerCoprocessorEnvironment> ctx) throws IOException {
117 oserver.postMerge(ctx, regionA, regionB, mergedRegion);
118 }
119 });
120 }
121
122 public boolean preMergeCommit(final HRegion regionA, final HRegion regionB,
123 final @MetaMutationAnnotation List<Mutation> metaEntries) throws IOException {
124 return execOperation(coprocessors.isEmpty() ? null : new CoprocessorOperation() {
125 @Override
126 public void call(RegionServerObserver oserver,
127 ObserverContext<RegionServerCoprocessorEnvironment> ctx) throws IOException {
128 oserver.preMergeCommit(ctx, regionA, regionB, metaEntries);
129 }
130 });
131 }
132
133 public void postMergeCommit(final HRegion regionA, final HRegion regionB,
134 final HRegion mergedRegion) throws IOException {
135 execOperation(coprocessors.isEmpty() ? null : new CoprocessorOperation() {
136 @Override
137 public void call(RegionServerObserver oserver,
138 ObserverContext<RegionServerCoprocessorEnvironment> ctx) throws IOException {
139 oserver.postMergeCommit(ctx, regionA, regionB, mergedRegion);
140 }
141 });
142 }
143
144 public void preRollBackMerge(final HRegion regionA, final HRegion regionB) throws IOException {
145 execOperation(coprocessors.isEmpty() ? null : new CoprocessorOperation() {
146 @Override
147 public void call(RegionServerObserver oserver,
148 ObserverContext<RegionServerCoprocessorEnvironment> ctx) throws IOException {
149 oserver.preRollBackMerge(ctx, regionA, regionB);
150 }
151 });
152 }
153
154 public void postRollBackMerge(final HRegion regionA, final HRegion regionB) throws IOException {
155 execOperation(coprocessors.isEmpty() ? null : new CoprocessorOperation() {
156 @Override
157 public void call(RegionServerObserver oserver,
158 ObserverContext<RegionServerCoprocessorEnvironment> ctx) throws IOException {
159 oserver.postRollBackMerge(ctx, regionA, regionB);
160 }
161 });
162 }
163
164 public void preRollWALWriterRequest() throws IOException {
165 execOperation(coprocessors.isEmpty() ? null : new CoprocessorOperation() {
166 @Override
167 public void call(RegionServerObserver oserver,
168 ObserverContext<RegionServerCoprocessorEnvironment> ctx) throws IOException {
169 oserver.preRollWALWriterRequest(ctx);
170 }
171 });
172 }
173
174 public void postRollWALWriterRequest() throws IOException {
175 execOperation(coprocessors.isEmpty() ? null : new CoprocessorOperation() {
176 @Override
177 public void call(RegionServerObserver oserver,
178 ObserverContext<RegionServerCoprocessorEnvironment> ctx) throws IOException {
179 oserver.postRollWALWriterRequest(ctx);
180 }
181 });
182 }
183
184 public void preReplicateLogEntries(final List<WALEntry> entries, final CellScanner cells)
185 throws IOException {
186 execOperation(coprocessors.isEmpty() ? null : new CoprocessorOperation() {
187 @Override
188 public void call(RegionServerObserver oserver,
189 ObserverContext<RegionServerCoprocessorEnvironment> ctx) throws IOException {
190 oserver.preReplicateLogEntries(ctx, entries, cells);
191 }
192 });
193 }
194
195 public void postReplicateLogEntries(final List<WALEntry> entries, final CellScanner cells)
196 throws IOException {
197 execOperation(coprocessors.isEmpty() ? null : new CoprocessorOperation() {
198 @Override
199 public void call(RegionServerObserver oserver,
200 ObserverContext<RegionServerCoprocessorEnvironment> ctx) throws IOException {
201 oserver.postReplicateLogEntries(ctx, entries, cells);
202 }
203 });
204 }
205
206 public ReplicationEndpoint postCreateReplicationEndPoint(final ReplicationEndpoint endpoint)
207 throws IOException {
208 return execOperationWithResult(endpoint, coprocessors.isEmpty() ? null
209 : new CoprocessOperationWithResult<ReplicationEndpoint>() {
210 @Override
211 public void call(RegionServerObserver oserver,
212 ObserverContext<RegionServerCoprocessorEnvironment> ctx) throws IOException {
213 setResult(oserver.postCreateReplicationEndPoint(ctx, getResult()));
214 }
215 });
216 }
217
218 private <T> T execOperationWithResult(final T defaultValue,
219 final CoprocessOperationWithResult<T> ctx) throws IOException {
220 if (ctx == null)
221 return defaultValue;
222 ctx.setResult(defaultValue);
223 execOperation(ctx);
224 return ctx.getResult();
225 }
226
227 private static abstract class CoprocessorOperation
228 extends ObserverContext<RegionServerCoprocessorEnvironment> {
229 public CoprocessorOperation() {
230 }
231
232 public abstract void call(RegionServerObserver oserver,
233 ObserverContext<RegionServerCoprocessorEnvironment> ctx) throws IOException;
234
235 public void postEnvCall(RegionServerEnvironment env) {
236 }
237 }
238
239 private static abstract class CoprocessOperationWithResult<T> extends CoprocessorOperation {
240 private T result = null;
241
242 public void setResult(final T result) {
243 this.result = result;
244 }
245
246 public T getResult() {
247 return this.result;
248 }
249 }
250
251 private boolean execOperation(final CoprocessorOperation ctx) throws IOException {
252 if (ctx == null) return false;
253 boolean bypass = false;
254 List<RegionServerEnvironment> envs = coprocessors.get();
255 for (int i = 0; i < envs.size(); i++) {
256 RegionServerEnvironment env = envs.get(i);
257 if (env.getInstance() instanceof RegionServerObserver) {
258 ctx.prepare(env);
259 Thread currentThread = Thread.currentThread();
260 ClassLoader cl = currentThread.getContextClassLoader();
261 try {
262 currentThread.setContextClassLoader(env.getClassLoader());
263 ctx.call((RegionServerObserver)env.getInstance(), ctx);
264 } catch (Throwable e) {
265 handleCoprocessorThrowable(env, e);
266 } finally {
267 currentThread.setContextClassLoader(cl);
268 }
269 bypass |= ctx.shouldBypass();
270 if (ctx.shouldComplete()) {
271 break;
272 }
273 }
274 ctx.postEnvCall(env);
275 }
276 return bypass;
277 }
278
279
280
281
282
283
284
285
286
287
288
289
290
291 private boolean execShutdown(final CoprocessorOperation ctx) throws IOException {
292 if (ctx == null) return false;
293 boolean bypass = false;
294 List<RegionServerEnvironment> envs = coprocessors.get();
295 int envsSize = envs.size();
296
297 for (int i = 0; i < envsSize; i++) {
298 RegionServerEnvironment env = envs.get(i);
299 if (env.getInstance() instanceof RegionServerObserver) {
300 ctx.prepare(env);
301 Thread currentThread = Thread.currentThread();
302 ClassLoader cl = currentThread.getContextClassLoader();
303 try {
304 currentThread.setContextClassLoader(env.getClassLoader());
305 ctx.call((RegionServerObserver) env.getInstance(), ctx);
306 } catch (Throwable e) {
307 handleCoprocessorThrowable(env, e);
308 } finally {
309 currentThread.setContextClassLoader(cl);
310 }
311 bypass |= ctx.shouldBypass();
312 if (ctx.shouldComplete()) {
313 break;
314 }
315 }
316 }
317
318
319 for (int i = 0; i < envsSize; i++) {
320 RegionServerEnvironment env = envs.get(i);
321 ctx.postEnvCall(env);
322 }
323 return bypass;
324 }
325
326
327
328
329
330 static class RegionServerEnvironment extends CoprocessorHost.Environment
331 implements RegionServerCoprocessorEnvironment {
332 private final RegionServerServices regionServerServices;
333 private final MetricRegistry metricRegistry;
334
335 @edu.umd.cs.findbugs.annotations.SuppressWarnings(value="BC_UNCONFIRMED_CAST",
336 justification="Intentional; FB has trouble detecting isAssignableFrom")
337 public RegionServerEnvironment(final Class<?> implClass,
338 final Coprocessor impl, final int priority, final int seq,
339 final Configuration conf, final RegionServerServices services) {
340 super(impl, priority, seq, conf);
341 this.regionServerServices = services;
342 for (Object itf : ClassUtils.getAllInterfaces(implClass)) {
343 Class<?> c = (Class<?>) itf;
344 if (SingletonCoprocessorService.class.isAssignableFrom(c)) {
345 this.regionServerServices.registerService(
346 ((SingletonCoprocessorService) impl).getService());
347 break;
348 }
349 }
350 this.metricRegistry =
351 MetricsCoprocessor.createRegistryForRSCoprocessor(implClass.getName());
352 }
353
354 @Override
355 public RegionServerServices getRegionServerServices() {
356 return regionServerServices;
357 }
358
359 @Override
360 public MetricRegistry getMetricRegistryForRegionServer() {
361 return metricRegistry;
362 }
363
364 @Override
365 protected void shutdown() {
366 super.shutdown();
367 MetricsCoprocessor.removeRegistry(metricRegistry);
368 }
369 }
370
371
372
373
374
375 static class EnvironmentPriorityComparator implements
376 Comparator<CoprocessorEnvironment> {
377 @Override
378 public int compare(final CoprocessorEnvironment env1,
379 final CoprocessorEnvironment env2) {
380 if (env1.getPriority() < env2.getPriority()) {
381 return -1;
382 } else if (env1.getPriority() > env2.getPriority()) {
383 return 1;
384 }
385 if (env1.getLoadSequence() < env2.getLoadSequence()) {
386 return -1;
387 } else if (env1.getLoadSequence() > env2.getLoadSequence()) {
388 return 1;
389 }
390 return 0;
391 }
392 }
393 }