1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 package org.apache.hadoop.hbase.coprocessor;
21
22 import java.io.IOException;
23 import java.util.ArrayList;
24 import java.util.Collections;
25 import java.util.Comparator;
26 import java.util.HashSet;
27 import java.util.List;
28 import java.util.Set;
29 import java.util.TreeSet;
30 import java.util.UUID;
31 import java.util.concurrent.ConcurrentSkipListSet;
32 import java.util.concurrent.ExecutorService;
33 import java.util.concurrent.atomic.AtomicInteger;
34
35 import org.apache.commons.logging.Log;
36 import org.apache.commons.logging.LogFactory;
37 import org.apache.hadoop.hbase.classification.InterfaceAudience;
38 import org.apache.hadoop.hbase.classification.InterfaceStability;
39 import org.apache.hadoop.conf.Configuration;
40 import org.apache.hadoop.fs.Path;
41 import org.apache.hadoop.hbase.Abortable;
42 import org.apache.hadoop.hbase.Coprocessor;
43 import org.apache.hadoop.hbase.CoprocessorEnvironment;
44 import org.apache.hadoop.hbase.DoNotRetryIOException;
45 import org.apache.hadoop.hbase.HBaseInterfaceAudience;
46 import org.apache.hadoop.hbase.HConstants;
47 import org.apache.hadoop.hbase.TableName;
48 import org.apache.hadoop.hbase.client.HTable;
49 import org.apache.hadoop.hbase.client.HTableInterface;
50 import org.apache.hadoop.hbase.client.HTableWrapper;
51 import org.apache.hadoop.hbase.util.Bytes;
52 import org.apache.hadoop.hbase.util.CoprocessorClassLoader;
53 import org.apache.hadoop.hbase.util.SortedList;
54 import org.apache.hadoop.hbase.util.VersionInfo;
55
56
57
58
59
60
61
62 @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.COPROC)
63 @InterfaceStability.Evolving
64 public abstract class CoprocessorHost<E extends CoprocessorEnvironment> {
65 public static final String REGION_COPROCESSOR_CONF_KEY =
66 "hbase.coprocessor.region.classes";
67 public static final String REGIONSERVER_COPROCESSOR_CONF_KEY =
68 "hbase.coprocessor.regionserver.classes";
69 public static final String USER_REGION_COPROCESSOR_CONF_KEY =
70 "hbase.coprocessor.user.region.classes";
71 public static final String MASTER_COPROCESSOR_CONF_KEY =
72 "hbase.coprocessor.master.classes";
73 public static final String WAL_COPROCESSOR_CONF_KEY =
74 "hbase.coprocessor.wal.classes";
75 public static final String ABORT_ON_ERROR_KEY = "hbase.coprocessor.abortonerror";
76 public static final boolean DEFAULT_ABORT_ON_ERROR = true;
77 public static final String COPROCESSORS_ENABLED_CONF_KEY = "hbase.coprocessor.enabled";
78 public static final boolean DEFAULT_COPROCESSORS_ENABLED = true;
79 public static final String USER_COPROCESSORS_ENABLED_CONF_KEY =
80 "hbase.coprocessor.user.enabled";
81 public static final boolean DEFAULT_USER_COPROCESSORS_ENABLED = true;
82
83 private static final Log LOG = LogFactory.getLog(CoprocessorHost.class);
84 protected Abortable abortable;
85
86 protected SortedList<E> coprocessors =
87 new SortedList<E>(new EnvironmentPriorityComparator());
88 protected Configuration conf;
89
90 protected String pathPrefix;
91 protected AtomicInteger loadSequence = new AtomicInteger();
92
93 public CoprocessorHost(Abortable abortable) {
94 this.abortable = abortable;
95 this.pathPrefix = UUID.randomUUID().toString();
96 }
97
98
99
100
101
102
103
104
105
106 private static Set<String> coprocessorNames =
107 Collections.synchronizedSet(new HashSet<String>());
108
109 public static Set<String> getLoadedCoprocessors() {
110 synchronized (coprocessorNames) {
111 return new HashSet(coprocessorNames);
112 }
113 }
114
115
116
117
118
119
120
121
122 public Set<String> getCoprocessors() {
123 Set<String> returnValue = new TreeSet<String>();
124 for (CoprocessorEnvironment e: coprocessors) {
125 returnValue.add(e.getInstance().getClass().getSimpleName());
126 }
127 return returnValue;
128 }
129
130
131
132
133
134 protected void loadSystemCoprocessors(Configuration conf, String confKey) {
135 boolean coprocessorsEnabled = conf.getBoolean(COPROCESSORS_ENABLED_CONF_KEY,
136 DEFAULT_COPROCESSORS_ENABLED);
137 if (!coprocessorsEnabled) {
138 return;
139 }
140
141 Class<?> implClass = null;
142
143
144 String[] defaultCPClasses = conf.getStrings(confKey);
145 if (defaultCPClasses == null || defaultCPClasses.length == 0)
146 return;
147
148 int priority = Coprocessor.PRIORITY_SYSTEM;
149 for (String className : defaultCPClasses) {
150 className = className.trim();
151 if (findCoprocessor(className) != null) {
152
153 LOG.warn("Attempted duplicate loading of " + className + "; skipped");
154 continue;
155 }
156 ClassLoader cl = this.getClass().getClassLoader();
157 Thread.currentThread().setContextClassLoader(cl);
158 try {
159 implClass = cl.loadClass(className);
160
161
162 this.coprocessors.add(loadInstance(implClass, priority, conf));
163 LOG.info("System coprocessor " + className + " was loaded " +
164 "successfully with priority (" + priority + ").");
165 ++priority;
166 } catch (Throwable t) {
167
168 abortServer(className, t);
169 }
170 }
171 }
172
173
174
175
176
177
178
179
180
181 public E load(Path path, String className, int priority,
182 Configuration conf) throws IOException {
183 String[] includedClassPrefixes = null;
184 if (conf.get(HConstants.CP_HTD_ATTR_INCLUSION_KEY) != null){
185 String prefixes = conf.get(HConstants.CP_HTD_ATTR_INCLUSION_KEY);
186 includedClassPrefixes = prefixes.split(";");
187 }
188 return load(path, className, priority, conf, includedClassPrefixes);
189 }
190
191
192
193
194
195
196
197
198
199
200 public E load(Path path, String className, int priority,
201 Configuration conf, String[] includedClassPrefixes) throws IOException {
202 Class<?> implClass = null;
203 LOG.debug("Loading coprocessor class " + className + " with path " +
204 path + " and priority " + priority);
205
206 ClassLoader cl = null;
207 if (path == null) {
208 try {
209 implClass = getClass().getClassLoader().loadClass(className);
210 } catch (ClassNotFoundException e) {
211 throw new IOException("No jar path specified for " + className);
212 }
213 } else {
214 cl = CoprocessorClassLoader.getClassLoader(
215 path, getClass().getClassLoader(), pathPrefix, conf);
216 try {
217 implClass = ((CoprocessorClassLoader)cl).loadClass(className, includedClassPrefixes);
218 } catch (ClassNotFoundException e) {
219 throw new IOException("Cannot load external coprocessor class " + className, e);
220 }
221 }
222
223
224 Thread currentThread = Thread.currentThread();
225 ClassLoader hostClassLoader = currentThread.getContextClassLoader();
226 try{
227
228 currentThread.setContextClassLoader(cl);
229 E cpInstance = loadInstance(implClass, priority, conf);
230 return cpInstance;
231 } finally {
232
233 currentThread.setContextClassLoader(hostClassLoader);
234 }
235 }
236
237
238
239
240
241
242
243 public void load(Class<?> implClass, int priority, Configuration conf)
244 throws IOException {
245 E env = loadInstance(implClass, priority, conf);
246 coprocessors.add(env);
247 }
248
249
250
251
252
253
254
255 public E loadInstance(Class<?> implClass, int priority, Configuration conf)
256 throws IOException {
257 if (!Coprocessor.class.isAssignableFrom(implClass)) {
258 throw new IOException("Configured class " + implClass.getName() + " must implement "
259 + Coprocessor.class.getName() + " interface ");
260 }
261
262
263 Coprocessor impl;
264 Object o = null;
265 try {
266 o = implClass.getDeclaredConstructor().newInstance();
267 impl = (Coprocessor)o;
268 } catch (Exception e) {
269 throw new IOException(e);
270 }
271
272 E env = createEnvironment(implClass, impl, priority, loadSequence.incrementAndGet(), conf);
273 if (env instanceof Environment) {
274 ((Environment)env).startup();
275 }
276
277
278 coprocessorNames.add(implClass.getName());
279 return env;
280 }
281
282
283
284
285 public abstract E createEnvironment(Class<?> implClass, Coprocessor instance,
286 int priority, int sequence, Configuration conf);
287
288 public void shutdown(CoprocessorEnvironment e) {
289 if (e instanceof Environment) {
290 if (LOG.isDebugEnabled()) {
291 LOG.debug("Stop coprocessor " + e.getInstance().getClass().getName());
292 }
293 ((Environment)e).shutdown();
294 } else {
295 LOG.warn("Shutdown called on unknown environment: "+
296 e.getClass().getName());
297 }
298 }
299
300
301
302
303
304
305 public Coprocessor findCoprocessor(String className) {
306 for (E env: coprocessors) {
307 if (env.getInstance().getClass().getName().equals(className) ||
308 env.getInstance().getClass().getSimpleName().equals(className)) {
309 return env.getInstance();
310 }
311 }
312 return null;
313 }
314
315
316
317
318
319
320 public <T extends Coprocessor> List<T> findCoprocessors(Class<T> cls) {
321 ArrayList<T> ret = new ArrayList<T>();
322
323 for (E env: coprocessors) {
324 Coprocessor cp = env.getInstance();
325
326 if(cp != null) {
327 if (cls.isAssignableFrom(cp.getClass())) {
328 ret.add((T)cp);
329 }
330 }
331 }
332 return ret;
333 }
334
335
336
337
338
339
340 public CoprocessorEnvironment findCoprocessorEnvironment(String className) {
341 for (E env: coprocessors) {
342 if (env.getInstance().getClass().getName().equals(className) ||
343 env.getInstance().getClass().getSimpleName().equals(className)) {
344 return env;
345 }
346 }
347 return null;
348 }
349
350
351
352
353
354
355 Set<ClassLoader> getExternalClassLoaders() {
356 Set<ClassLoader> externalClassLoaders = new HashSet<ClassLoader>();
357 final ClassLoader systemClassLoader = this.getClass().getClassLoader();
358 for (E env : coprocessors) {
359 ClassLoader cl = env.getInstance().getClass().getClassLoader();
360 if (cl != systemClassLoader ){
361
362 externalClassLoaders.add(cl);
363 }
364 }
365 return externalClassLoaders;
366 }
367
368
369
370
371
372 static class EnvironmentPriorityComparator
373 implements Comparator<CoprocessorEnvironment> {
374 @Override
375 public int compare(final CoprocessorEnvironment env1,
376 final CoprocessorEnvironment env2) {
377 if (env1.getPriority() < env2.getPriority()) {
378 return -1;
379 } else if (env1.getPriority() > env2.getPriority()) {
380 return 1;
381 }
382 if (env1.getLoadSequence() < env2.getLoadSequence()) {
383 return -1;
384 } else if (env1.getLoadSequence() > env2.getLoadSequence()) {
385 return 1;
386 }
387 return 0;
388 }
389 }
390
391
392
393
394 public static class Environment implements CoprocessorEnvironment {
395
396
397 public Coprocessor impl;
398
399 protected int priority = Coprocessor.PRIORITY_USER;
400
401 Coprocessor.State state = Coprocessor.State.UNINSTALLED;
402
403 protected final List<HTableInterface> openTables =
404 Collections.synchronizedList(new ArrayList<HTableInterface>());
405 private int seq;
406 private Configuration conf;
407 private ClassLoader classLoader;
408
409
410
411
412
413
414 public Environment(final Coprocessor impl, final int priority,
415 final int seq, final Configuration conf) {
416 this.impl = impl;
417 this.classLoader = impl.getClass().getClassLoader();
418 this.priority = priority;
419 this.state = Coprocessor.State.INSTALLED;
420 this.seq = seq;
421 this.conf = conf;
422 }
423
424
425 public void startup() throws IOException {
426 if (state == Coprocessor.State.INSTALLED ||
427 state == Coprocessor.State.STOPPED) {
428 state = Coprocessor.State.STARTING;
429 Thread currentThread = Thread.currentThread();
430 ClassLoader hostClassLoader = currentThread.getContextClassLoader();
431 try {
432 currentThread.setContextClassLoader(this.getClassLoader());
433 impl.start(this);
434 state = Coprocessor.State.ACTIVE;
435 } finally {
436 currentThread.setContextClassLoader(hostClassLoader);
437 }
438 } else {
439 LOG.warn("Not starting coprocessor "+impl.getClass().getName()+
440 " because not inactive (state="+state.toString()+")");
441 }
442 }
443
444
445 protected void shutdown() {
446 if (state == Coprocessor.State.ACTIVE) {
447 state = Coprocessor.State.STOPPING;
448 Thread currentThread = Thread.currentThread();
449 ClassLoader hostClassLoader = currentThread.getContextClassLoader();
450 try {
451 currentThread.setContextClassLoader(this.getClassLoader());
452 impl.stop(this);
453 state = Coprocessor.State.STOPPED;
454 } catch (IOException ioe) {
455 LOG.error("Error stopping coprocessor "+impl.getClass().getName(), ioe);
456 } finally {
457 currentThread.setContextClassLoader(hostClassLoader);
458 }
459 } else {
460 LOG.warn("Not stopping coprocessor "+impl.getClass().getName()+
461 " because not active (state="+state.toString()+")");
462 }
463 synchronized (openTables) {
464
465 for (HTableInterface table: openTables) {
466 try {
467 ((HTableWrapper)table).internalClose();
468 } catch (IOException e) {
469
470 LOG.warn("Failed to close " +
471 Bytes.toStringBinary(table.getTableName()), e);
472 }
473 }
474 }
475 }
476
477 @Override
478 public Coprocessor getInstance() {
479 return impl;
480 }
481
482 @Override
483 public ClassLoader getClassLoader() {
484 return classLoader;
485 }
486
487 @Override
488 public int getPriority() {
489 return priority;
490 }
491
492 @Override
493 public int getLoadSequence() {
494 return seq;
495 }
496
497
498 @Override
499 public int getVersion() {
500 return Coprocessor.VERSION;
501 }
502
503
504 @Override
505 public String getHBaseVersion() {
506 return VersionInfo.getVersion();
507 }
508
509 @Override
510 public Configuration getConfiguration() {
511 return conf;
512 }
513
514
515
516
517
518
519
520 @Override
521 public HTableInterface getTable(TableName tableName) throws IOException {
522 return this.getTable(tableName, HTable.getDefaultExecutor(getConfiguration()));
523 }
524
525
526
527
528
529
530
531 @Override
532 public HTableInterface getTable(TableName tableName, ExecutorService pool) throws IOException {
533 return HTableWrapper.createWrapper(openTables, tableName, this, pool);
534 }
535 }
536
537 protected void abortServer(final CoprocessorEnvironment environment, final Throwable e) {
538 abortServer(environment.getInstance().getClass().getName(), e);
539 }
540
541 protected void abortServer(final String coprocessorName, final Throwable e) {
542 String message = "The coprocessor " + coprocessorName + " threw " + e.toString();
543 LOG.error(message, e);
544 if (abortable != null) {
545 abortable.abort(message, e);
546 } else {
547 LOG.warn("No available Abortable, process was not aborted");
548 }
549 }
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566 protected void handleCoprocessorThrowable(final CoprocessorEnvironment env, final Throwable e)
567 throws IOException {
568 if (e instanceof IOException) {
569 throw (IOException)e;
570 }
571
572
573
574
575
576
577 if (env.getConfiguration().getBoolean(ABORT_ON_ERROR_KEY, DEFAULT_ABORT_ON_ERROR)) {
578
579 abortServer(env, e);
580 } else {
581
582 if(env instanceof RegionCoprocessorEnvironment) {
583 String tableName = ((RegionCoprocessorEnvironment)env).getRegionInfo().getTable().getNameAsString();
584 LOG.error("Removing coprocessor '" + env.toString() + "' from table '"+ tableName + "'", e);
585 } else {
586 LOG.error("Removing coprocessor '" + env.toString() + "' from " +
587 "environment",e);
588 }
589
590 coprocessors.remove(env);
591 try {
592 shutdown(env);
593 } catch (Exception x) {
594 LOG.error("Uncaught exception when shutting down coprocessor '"
595 + env.toString() + "'", x);
596 }
597 throw new DoNotRetryIOException("Coprocessor: '" + env.toString() +
598 "' threw: '" + e + "' and has been removed from the active " +
599 "coprocessor set.", e);
600 }
601 }
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623 @InterfaceAudience.Private
624 protected static boolean useLegacyMethod(final Class<? extends Coprocessor> clazz,
625 final String methodName, final Class<?>... parameterTypes) {
626 boolean useLegacy;
627
628 try {
629 clazz.getDeclaredMethod(methodName, parameterTypes);
630 LOG.debug("Found an implementation of '" + methodName + "' that uses updated method " +
631 "signature. Skipping legacy support for invocations in '" + clazz +"'.");
632 useLegacy = false;
633 } catch (NoSuchMethodException exception) {
634 useLegacy = true;
635 } catch (SecurityException exception) {
636 LOG.warn("The Security Manager denied our attempt to detect if the coprocessor '" + clazz +
637 "' requires legacy support; assuming it does. If you get later errors about legacy " +
638 "coprocessor use, consider updating your security policy to allow access to the package" +
639 " and declared members of your implementation.");
640 LOG.debug("Details of Security Manager rejection.", exception);
641 useLegacy = true;
642 }
643 return useLegacy;
644 }
645
646
647
648
649 private static final Set<Class<? extends Coprocessor>> legacyWarning =
650 new ConcurrentSkipListSet<Class<? extends Coprocessor>>(
651 new Comparator<Class<? extends Coprocessor>>() {
652 @Override
653 public int compare(Class<? extends Coprocessor> c1, Class<? extends Coprocessor> c2) {
654 if (c1.equals(c2)) {
655 return 0;
656 }
657 return c1.getName().compareTo(c2.getName());
658 }
659 });
660
661
662
663
664
665
666
667 @InterfaceAudience.Private
668 protected void legacyWarning(final Class<? extends Coprocessor> clazz, final String message) {
669 if(legacyWarning.add(clazz)) {
670 LOG.error("You have a legacy coprocessor loaded and there are events we can't map to the " +
671 " deprecated API. Your coprocessor will not see these events. Please update '" + clazz +
672 "'. Details of the problem: " + message);
673 }
674 }
675 }