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 */ 018package org.apache.hadoop.hbase.coprocessor; 019 020import java.io.IOException; 021import java.util.ArrayList; 022import java.util.Collections; 023import java.util.Comparator; 024import java.util.HashSet; 025import java.util.List; 026import java.util.Optional; 027import java.util.Set; 028import java.util.TreeSet; 029import java.util.UUID; 030import java.util.concurrent.ConcurrentSkipListSet; 031import java.util.concurrent.atomic.AtomicInteger; 032import java.util.function.Function; 033import org.apache.hadoop.conf.Configuration; 034import org.apache.hadoop.fs.Path; 035import org.apache.hadoop.hbase.Abortable; 036import org.apache.hadoop.hbase.Coprocessor; 037import org.apache.hadoop.hbase.CoprocessorEnvironment; 038import org.apache.hadoop.hbase.DoNotRetryIOException; 039import org.apache.hadoop.hbase.HConstants; 040import org.apache.hadoop.hbase.ipc.RpcServer; 041import org.apache.hadoop.hbase.security.User; 042import org.apache.hadoop.hbase.util.CoprocessorClassLoader; 043import org.apache.hadoop.hbase.util.SortedList; 044import org.apache.yetus.audience.InterfaceAudience; 045import org.slf4j.Logger; 046import org.slf4j.LoggerFactory; 047 048import org.apache.hbase.thirdparty.com.google.common.base.Strings; 049 050/** 051 * Provides the common setup framework and runtime services for coprocessor invocation from HBase 052 * services. 053 * @param <C> type of specific coprocessor this host will handle 054 * @param <E> type of specific coprocessor environment this host requires. provides 055 */ 056@InterfaceAudience.Private 057public abstract class CoprocessorHost<C extends Coprocessor, E extends CoprocessorEnvironment<C>> { 058 public static final String REGION_COPROCESSOR_CONF_KEY = "hbase.coprocessor.region.classes"; 059 public static final String REGIONSERVER_COPROCESSOR_CONF_KEY = 060 "hbase.coprocessor.regionserver.classes"; 061 public static final String USER_REGION_COPROCESSOR_CONF_KEY = 062 "hbase.coprocessor.user.region.classes"; 063 public static final String MASTER_COPROCESSOR_CONF_KEY = "hbase.coprocessor.master.classes"; 064 public static final String WAL_COPROCESSOR_CONF_KEY = "hbase.coprocessor.wal.classes"; 065 public static final String RPC_COPROCESSOR_CONF_KEY = "hbase.coprocessor.rpc.classes"; 066 public static final String ABORT_ON_ERROR_KEY = "hbase.coprocessor.abortonerror"; 067 public static final boolean DEFAULT_ABORT_ON_ERROR = true; 068 public static final String COPROCESSORS_ENABLED_CONF_KEY = "hbase.coprocessor.enabled"; 069 public static final boolean DEFAULT_COPROCESSORS_ENABLED = true; 070 public static final String USER_COPROCESSORS_ENABLED_CONF_KEY = "hbase.coprocessor.user.enabled"; 071 public static final boolean DEFAULT_USER_COPROCESSORS_ENABLED = true; 072 public static final String SKIP_LOAD_DUPLICATE_TABLE_COPROCESSOR = 073 "hbase.skip.load.duplicate.table.coprocessor"; 074 public static final boolean DEFAULT_SKIP_LOAD_DUPLICATE_TABLE_COPROCESSOR = false; 075 076 private static final Logger LOG = LoggerFactory.getLogger(CoprocessorHost.class); 077 protected Abortable abortable; 078 /** Ordered set of loaded coprocessors with lock */ 079 protected final SortedList<E> coprocEnvironments = 080 new SortedList<>(new EnvironmentPriorityComparator()); 081 protected Configuration conf; 082 // unique file prefix to use for local copies of jars when classloading 083 protected String pathPrefix; 084 protected AtomicInteger loadSequence = new AtomicInteger(); 085 086 public CoprocessorHost(Abortable abortable) { 087 this.abortable = abortable; 088 this.pathPrefix = UUID.randomUUID().toString(); 089 } 090 091 /** 092 * Not to be confused with the per-object _coprocessors_ (above), coprocessorNames is static and 093 * stores the set of all coprocessors ever loaded by any thread in this JVM. It is strictly 094 * additive: coprocessors are added to coprocessorNames, by checkAndLoadInstance() but are never 095 * removed, since the intention is to preserve a history of all loaded coprocessors for diagnosis 096 * in case of server crash (HBASE-4014). 097 */ 098 private static Set<String> coprocessorNames = Collections.synchronizedSet(new HashSet<String>()); 099 100 public static Set<String> getLoadedCoprocessors() { 101 synchronized (coprocessorNames) { 102 return new HashSet(coprocessorNames); 103 } 104 } 105 106 /** 107 * Used to create a parameter to the HServerLoad constructor so that HServerLoad can provide 108 * information about the coprocessors loaded by this regionserver. (HBASE-4070: Improve region 109 * server metrics to report loaded coprocessors to master). 110 */ 111 public Set<String> getCoprocessors() { 112 Set<String> returnValue = new TreeSet<>(); 113 for (E e : coprocEnvironments) { 114 returnValue.add(e.getInstance().getClass().getSimpleName()); 115 } 116 return returnValue; 117 } 118 119 /** 120 * Get the full class names of all loaded coprocessors. This method returns the complete class 121 * names including package information, which is useful for precise coprocessor identification and 122 * comparison. 123 */ 124 public Set<String> getCoprocessorClassNames() { 125 Set<String> returnValue = new TreeSet<>(); 126 for (E e : coprocEnvironments) { 127 returnValue.add(e.getInstance().getClass().getName()); 128 } 129 return returnValue; 130 } 131 132 /** 133 * Load system coprocessors once only. Read the class names from configuration. Called by 134 * constructor. 135 */ 136 protected void loadSystemCoprocessors(Configuration conf, String confKey) { 137 boolean coprocessorsEnabled = 138 conf.getBoolean(COPROCESSORS_ENABLED_CONF_KEY, DEFAULT_COPROCESSORS_ENABLED); 139 if (!coprocessorsEnabled) { 140 return; 141 } 142 143 Class<?> implClass; 144 145 // load default coprocessors from configure file 146 String[] defaultCPClasses = conf.getStrings(confKey); 147 if (defaultCPClasses == null || defaultCPClasses.length == 0) return; 148 149 int currentSystemPriority = Coprocessor.PRIORITY_SYSTEM; 150 for (String className : defaultCPClasses) { 151 // After HBASE-23710 and HBASE-26714 when configuring for system coprocessor, we accept 152 // an optional format of className|priority|path 153 String[] classNameToken = className.split("\\|"); 154 boolean hasPriorityOverride = false; 155 boolean hasPath = false; 156 className = classNameToken[0]; 157 int overridePriority = Coprocessor.PRIORITY_SYSTEM; 158 Path path = null; 159 if (classNameToken.length > 1 && !Strings.isNullOrEmpty(classNameToken[1])) { 160 overridePriority = Integer.parseInt(classNameToken[1]); 161 hasPriorityOverride = true; 162 } 163 if (classNameToken.length > 2 && !Strings.isNullOrEmpty(classNameToken[2])) { 164 path = new Path(classNameToken[2].trim()); 165 hasPath = true; 166 } 167 className = className.trim(); 168 if (findCoprocessor(className) != null) { 169 // If already loaded will just continue 170 LOG.warn("Attempted duplicate loading of " + className + "; skipped"); 171 continue; 172 } 173 ClassLoader cl = this.getClass().getClassLoader(); 174 try { 175 // override the class loader if a path for the system coprocessor is provided. 176 if (hasPath) { 177 cl = CoprocessorClassLoader.getClassLoader(path, this.getClass().getClassLoader(), 178 pathPrefix, conf); 179 } 180 Thread.currentThread().setContextClassLoader(cl); 181 implClass = cl.loadClass(className); 182 int coprocPriority = hasPriorityOverride ? overridePriority : currentSystemPriority; 183 // Add coprocessors as we go to guard against case where a coprocessor is specified twice 184 // in the configuration 185 E env = checkAndLoadInstance(implClass, coprocPriority, conf); 186 if (env != null) { 187 this.coprocEnvironments.add(env); 188 LOG.info("System coprocessor {} loaded, priority={}.", className, coprocPriority); 189 if (!hasPriorityOverride) { 190 ++currentSystemPriority; 191 } 192 } 193 } catch (Throwable t) { 194 // We always abort if system coprocessors cannot be loaded 195 abortServer(className, t); 196 } 197 } 198 } 199 200 /** 201 * Load a coprocessor implementation into the host 202 * @param path path to implementation jar 203 * @param className the main class name 204 * @param priority chaining priority 205 * @param conf configuration for coprocessor 206 * @throws java.io.IOException Exception 207 */ 208 public E load(Path path, String className, int priority, Configuration conf) throws IOException { 209 String[] includedClassPrefixes = null; 210 if (conf.get(HConstants.CP_HTD_ATTR_INCLUSION_KEY) != null) { 211 String prefixes = conf.get(HConstants.CP_HTD_ATTR_INCLUSION_KEY); 212 includedClassPrefixes = prefixes.split(";"); 213 } 214 return load(path, className, priority, conf, includedClassPrefixes); 215 } 216 217 /** 218 * Load a coprocessor implementation into the host 219 * @param path path to implementation jar 220 * @param className the main class name 221 * @param priority chaining priority 222 * @param conf configuration for coprocessor 223 * @param includedClassPrefixes class name prefixes to include 224 * @throws java.io.IOException Exception 225 */ 226 public E load(Path path, String className, int priority, Configuration conf, 227 String[] includedClassPrefixes) throws IOException { 228 Class<?> implClass; 229 LOG.debug("Loading coprocessor class " + className + " with path " + path + " and priority " 230 + priority); 231 232 boolean skipLoadDuplicateCoprocessor = conf.getBoolean(SKIP_LOAD_DUPLICATE_TABLE_COPROCESSOR, 233 DEFAULT_SKIP_LOAD_DUPLICATE_TABLE_COPROCESSOR); 234 if (skipLoadDuplicateCoprocessor && findCoprocessor(className) != null) { 235 // If already loaded will just continue 236 LOG.warn("Attempted duplicate loading of {}; skipped", className); 237 return null; 238 } 239 240 ClassLoader cl = null; 241 if (path == null) { 242 try { 243 implClass = getClass().getClassLoader().loadClass(className); 244 } catch (ClassNotFoundException e) { 245 throw new IOException("No jar path specified for " + className); 246 } 247 } else { 248 cl = 249 CoprocessorClassLoader.getClassLoader(path, getClass().getClassLoader(), pathPrefix, conf); 250 try { 251 implClass = ((CoprocessorClassLoader) cl).loadClass(className, includedClassPrefixes); 252 } catch (ClassNotFoundException e) { 253 throw new IOException("Cannot load external coprocessor class " + className, e); 254 } 255 } 256 257 // load custom code for coprocessor 258 Thread currentThread = Thread.currentThread(); 259 ClassLoader hostClassLoader = currentThread.getContextClassLoader(); 260 try { 261 // switch temporarily to the thread classloader for custom CP 262 currentThread.setContextClassLoader(cl); 263 E cpInstance = checkAndLoadInstance(implClass, priority, conf); 264 return cpInstance; 265 } finally { 266 // restore the fresh (host) classloader 267 currentThread.setContextClassLoader(hostClassLoader); 268 } 269 } 270 271 public void load(Class<? extends C> implClass, int priority, Configuration conf) 272 throws IOException { 273 E env = checkAndLoadInstance(implClass, priority, conf); 274 coprocEnvironments.add(env); 275 } 276 277 /** 278 * @param implClass Implementation class 279 * @param priority priority 280 * @param conf configuration 281 * @throws java.io.IOException Exception 282 */ 283 public E checkAndLoadInstance(Class<?> implClass, int priority, Configuration conf) 284 throws IOException { 285 // create the instance 286 C impl; 287 try { 288 impl = checkAndGetInstance(implClass); 289 if (impl == null) { 290 LOG.error("Cannot load coprocessor " + implClass.getSimpleName()); 291 return null; 292 } 293 } catch (InstantiationException | IllegalAccessException e) { 294 throw new IOException(e); 295 } 296 // create the environment 297 E env = createEnvironment(impl, priority, loadSequence.incrementAndGet(), conf); 298 assert env instanceof BaseEnvironment; 299 ((BaseEnvironment<C>) env).startup(); 300 // HBASE-4014: maintain list of loaded coprocessors for later crash analysis 301 // if server (master or regionserver) aborts. 302 coprocessorNames.add(implClass.getName()); 303 return env; 304 } 305 306 /** 307 * Called when a new Coprocessor class is loaded 308 */ 309 public abstract E createEnvironment(C instance, int priority, int sequence, Configuration conf); 310 311 /** 312 * Called when a new Coprocessor class needs to be loaded. Checks if type of the given class is 313 * what the corresponding host implementation expects. If it is of correct type, returns an 314 * instance of the coprocessor to be loaded. If not, returns null. If an exception occurs when 315 * trying to create instance of a coprocessor, it's passed up and eventually results into server 316 * aborting. 317 */ 318 public abstract C checkAndGetInstance(Class<?> implClass) 319 throws InstantiationException, IllegalAccessException; 320 321 public void shutdown(E e) { 322 assert e instanceof BaseEnvironment; 323 if (LOG.isDebugEnabled()) { 324 LOG.debug("Stop coprocessor " + e.getInstance().getClass().getName()); 325 } 326 ((BaseEnvironment<C>) e).shutdown(); 327 } 328 329 /** 330 * Find coprocessors by full class name or simple name. 331 */ 332 public C findCoprocessor(String className) { 333 for (E env : coprocEnvironments) { 334 if ( 335 env.getInstance().getClass().getName().equals(className) 336 || env.getInstance().getClass().getSimpleName().equals(className) 337 ) { 338 return env.getInstance(); 339 } 340 } 341 return null; 342 } 343 344 public <T extends C> T findCoprocessor(Class<T> cls) { 345 for (E env : coprocEnvironments) { 346 if (cls.isAssignableFrom(env.getInstance().getClass())) { 347 return (T) env.getInstance(); 348 } 349 } 350 return null; 351 } 352 353 /** 354 * Find list of coprocessors that extend/implement the given class/interface 355 * @param cls the class/interface to look for 356 * @return the list of coprocessors, or null if not found 357 */ 358 public <T extends C> List<T> findCoprocessors(Class<T> cls) { 359 ArrayList<T> ret = new ArrayList<>(); 360 361 for (E env : coprocEnvironments) { 362 C cp = env.getInstance(); 363 364 if (cp != null) { 365 if (cls.isAssignableFrom(cp.getClass())) { 366 ret.add((T) cp); 367 } 368 } 369 } 370 return ret; 371 } 372 373 /** 374 * Find a coprocessor environment by class name 375 * @param className the class name 376 * @return the coprocessor, or null if not found 377 */ 378 public E findCoprocessorEnvironment(String className) { 379 for (E env : coprocEnvironments) { 380 if ( 381 env.getInstance().getClass().getName().equals(className) 382 || env.getInstance().getClass().getSimpleName().equals(className) 383 ) { 384 return env; 385 } 386 } 387 return null; 388 } 389 390 /** 391 * Retrieves the set of classloaders used to instantiate Coprocessor classes defined in external 392 * jar files. 393 * @return A set of ClassLoader instances 394 */ 395 Set<ClassLoader> getExternalClassLoaders() { 396 Set<ClassLoader> externalClassLoaders = new HashSet<>(); 397 final ClassLoader systemClassLoader = this.getClass().getClassLoader(); 398 for (E env : coprocEnvironments) { 399 ClassLoader cl = env.getInstance().getClass().getClassLoader(); 400 if (cl != systemClassLoader) { 401 // do not include system classloader 402 externalClassLoaders.add(cl); 403 } 404 } 405 return externalClassLoaders; 406 } 407 408 /** 409 * Environment priority comparator. Coprocessors are chained in sorted order. 410 */ 411 static class EnvironmentPriorityComparator implements Comparator<CoprocessorEnvironment> { 412 @Override 413 public int compare(final CoprocessorEnvironment env1, final CoprocessorEnvironment env2) { 414 if (env1.getPriority() < env2.getPriority()) { 415 return -1; 416 } else if (env1.getPriority() > env2.getPriority()) { 417 return 1; 418 } 419 if (env1.getLoadSequence() < env2.getLoadSequence()) { 420 return -1; 421 } else if (env1.getLoadSequence() > env2.getLoadSequence()) { 422 return 1; 423 } 424 return 0; 425 } 426 } 427 428 protected void abortServer(final E environment, final Throwable e) { 429 abortServer(environment.getInstance().getClass().getName(), e); 430 } 431 432 protected void abortServer(final String coprocessorName, final Throwable e) { 433 String message = "The coprocessor " + coprocessorName + " threw " + e.toString(); 434 LOG.error(message, e); 435 if (abortable != null) { 436 abortable.abort(message, e); 437 } else { 438 LOG.warn("No available Abortable, process was not aborted"); 439 } 440 } 441 442 /** 443 * This is used by coprocessor hooks which are declared to throw IOException (or its subtypes). 444 * For such hooks, we should handle throwable objects depending on the Throwable's type. Those 445 * which are instances of IOException should be passed on to the client. This is in conformance 446 * with the HBase idiom regarding IOException: that it represents a circumstance that should be 447 * passed along to the client for its own handling. For example, a coprocessor that implements 448 * access controls would throw a subclass of IOException, such as AccessDeniedException, in its 449 * preGet() method to prevent an unauthorized client's performing a Get on a particular table. 450 * @param env Coprocessor Environment 451 * @param e Throwable object thrown by coprocessor. 452 * @exception IOException Exception 453 */ 454 // Note to devs: Class comments of all observers ({@link MasterObserver}, {@link WALObserver}, 455 // etc) mention this nuance of our exception handling so that coprocessor can throw appropriate 456 // exceptions depending on situation. If any changes are made to this logic, make sure to 457 // update all classes' comments. 458 protected void handleCoprocessorThrowable(final E env, final Throwable e) throws IOException { 459 if (e instanceof IOException) { 460 throw (IOException) e; 461 } 462 // If we got here, e is not an IOException. A loaded coprocessor has a 463 // fatal bug, and the server (master or regionserver) should remove the 464 // faulty coprocessor from its set of active coprocessors. Setting 465 // 'hbase.coprocessor.abortonerror' to true will cause abortServer(), 466 // which may be useful in development and testing environments where 467 // 'failing fast' for error analysis is desired. 468 if (env.getConfiguration().getBoolean(ABORT_ON_ERROR_KEY, DEFAULT_ABORT_ON_ERROR)) { 469 // server is configured to abort. 470 abortServer(env, e); 471 } else { 472 // If available, pull a table name out of the environment 473 if (env instanceof RegionCoprocessorEnvironment) { 474 String tableName = 475 ((RegionCoprocessorEnvironment) env).getRegionInfo().getTable().getNameAsString(); 476 LOG.error("Removing coprocessor '" + env.toString() + "' from table '" + tableName + "'", 477 e); 478 } else { 479 LOG.error("Removing coprocessor '" + env.toString() + "' from " + "environment", e); 480 } 481 482 coprocEnvironments.remove(env); 483 try { 484 shutdown(env); 485 } catch (Exception x) { 486 LOG.error("Uncaught exception when shutting down coprocessor '" + env.toString() + "'", x); 487 } 488 throw new DoNotRetryIOException("Coprocessor: '" + env.toString() + "' threw: '" + e 489 + "' and has been removed from the active " + "coprocessor set.", e); 490 } 491 } 492 493 /** 494 * Used to limit legacy handling to once per Coprocessor class per classloader. 495 */ 496 private static final Set<Class<? extends Coprocessor>> legacyWarning = 497 new ConcurrentSkipListSet<>(new Comparator<Class<? extends Coprocessor>>() { 498 @Override 499 public int compare(Class<? extends Coprocessor> c1, Class<? extends Coprocessor> c2) { 500 if (c1.equals(c2)) { 501 return 0; 502 } 503 return c1.getName().compareTo(c2.getName()); 504 } 505 }); 506 507 /** 508 * Implementations defined function to get an observer of type {@code O} from a coprocessor of 509 * type {@code C}. Concrete implementations of CoprocessorHost define one getter for each observer 510 * they can handle. For e.g. RegionCoprocessorHost will use 3 getters, one for each of 511 * RegionObserver, EndpointObserver and BulkLoadObserver. These getters are used by 512 * {@code ObserverOperation} to get appropriate observer from the coprocessor. 513 */ 514 @FunctionalInterface 515 public interface ObserverGetter<C, O> extends Function<C, Optional<O>> { 516 } 517 518 private abstract class ObserverOperation<O> extends ObserverContextImpl<E> { 519 ObserverGetter<C, O> observerGetter; 520 521 ObserverOperation(ObserverGetter<C, O> observerGetter) { 522 this(observerGetter, null); 523 } 524 525 ObserverOperation(ObserverGetter<C, O> observerGetter, User user) { 526 this(observerGetter, user, false); 527 } 528 529 ObserverOperation(ObserverGetter<C, O> observerGetter, boolean bypassable) { 530 this(observerGetter, null, bypassable); 531 } 532 533 ObserverOperation(ObserverGetter<C, O> observerGetter, User user, boolean bypassable) { 534 super(user != null ? user : RpcServer.getRequestUser().orElse(null), bypassable); 535 this.observerGetter = observerGetter; 536 } 537 538 abstract void callObserver() throws IOException; 539 540 protected void postEnvCall() { 541 } 542 } 543 544 // Can't derive ObserverOperation from ObserverOperationWithResult (R = Void) because then all 545 // ObserverCaller implementations will have to have a return statement. 546 // O = observer, E = environment, C = coprocessor, R=result type 547 public abstract class ObserverOperationWithoutResult<O> extends ObserverOperation<O> { 548 protected abstract void call(O observer) throws IOException; 549 550 public ObserverOperationWithoutResult(ObserverGetter<C, O> observerGetter) { 551 super(observerGetter); 552 } 553 554 public ObserverOperationWithoutResult(ObserverGetter<C, O> observerGetter, User user) { 555 super(observerGetter, user); 556 } 557 558 public ObserverOperationWithoutResult(ObserverGetter<C, O> observerGetter, User user, 559 boolean bypassable) { 560 super(observerGetter, user, bypassable); 561 } 562 563 /** 564 * In case of coprocessors which have many kinds of observers (for eg, {@link RegionCoprocessor} 565 * has BulkLoadObserver, RegionObserver, etc), some implementations may not need all observers, 566 * in which case they will return null for that observer's getter. We simply ignore such cases. 567 */ 568 @Override 569 void callObserver() throws IOException { 570 Optional<O> observer = observerGetter.apply(getEnvironment().getInstance()); 571 if (observer.isPresent()) { 572 call(observer.get()); 573 } 574 } 575 } 576 577 public abstract class ObserverOperationWithResult<O, R> extends ObserverOperation<O> { 578 protected abstract R call(O observer) throws IOException; 579 580 private R result; 581 582 public ObserverOperationWithResult(ObserverGetter<C, O> observerGetter, R result) { 583 this(observerGetter, result, false); 584 } 585 586 public ObserverOperationWithResult(ObserverGetter<C, O> observerGetter, R result, 587 boolean bypassable) { 588 this(observerGetter, result, null, bypassable); 589 } 590 591 public ObserverOperationWithResult(ObserverGetter<C, O> observerGetter, R result, User user) { 592 this(observerGetter, result, user, false); 593 } 594 595 private ObserverOperationWithResult(ObserverGetter<C, O> observerGetter, R result, User user, 596 boolean bypassable) { 597 super(observerGetter, user, bypassable); 598 this.result = result; 599 } 600 601 protected R getResult() { 602 return this.result; 603 } 604 605 @Override 606 void callObserver() throws IOException { 607 Optional<O> observer = observerGetter.apply(getEnvironment().getInstance()); 608 if (observer.isPresent()) { 609 result = call(observer.get()); 610 } 611 } 612 } 613 614 ////////////////////////////////////////////////////////////////////////////////////////// 615 // Functions to execute observer hooks and handle results (if any) 616 ////////////////////////////////////////////////////////////////////////////////////////// 617 618 /** 619 * Do not call with an observerOperation that is null! Have the caller check. 620 */ 621 protected <O, R> R execOperationWithResult( 622 final ObserverOperationWithResult<O, R> observerOperation) throws IOException { 623 boolean bypass = execOperation(observerOperation); 624 R result = observerOperation.getResult(); 625 return bypass == observerOperation.isBypassable() ? result : null; 626 } 627 628 /** 629 * @return True if we are to bypass (Can only be <code>true</code> if 630 * ObserverOperation#isBypassable(). 631 */ 632 protected <O> boolean execOperation(final ObserverOperation<O> observerOperation) 633 throws IOException { 634 boolean bypass = false; 635 if (observerOperation == null) { 636 return bypass; 637 } 638 List<E> envs = coprocEnvironments.get(); 639 for (E env : envs) { 640 observerOperation.prepare(env); 641 Thread currentThread = Thread.currentThread(); 642 ClassLoader cl = currentThread.getContextClassLoader(); 643 try { 644 currentThread.setContextClassLoader(env.getClassLoader()); 645 observerOperation.callObserver(); 646 } catch (Throwable e) { 647 handleCoprocessorThrowable(env, e); 648 } finally { 649 currentThread.setContextClassLoader(cl); 650 } 651 // Internal to shouldBypass, it checks if obeserverOperation#isBypassable(). 652 bypass |= observerOperation.shouldBypass(); 653 observerOperation.postEnvCall(); 654 if (bypass) { 655 // If CP says bypass, skip out w/o calling any following CPs; they might ruin our response. 656 // In hbase1, this used to be called 'complete'. In hbase2, we unite bypass and 'complete'. 657 break; 658 } 659 } 660 return bypass; 661 } 662 663 /** 664 * Coprocessor classes can be configured in any order, based on that priority is set and chained 665 * in a sorted order. Should be used preStop*() hooks i.e. when master/regionserver is going down. 666 * This function first calls coprocessor methods (using ObserverOperation.call()) and then 667 * shutdowns the environment in postEnvCall(). <br> 668 * Need to execute all coprocessor methods first then postEnvCall(), otherwise some coprocessors 669 * may remain shutdown if any exception occurs during next coprocessor execution which prevent 670 * master/regionserver stop or cluster shutdown. (Refer: 671 * <a href="https://issues.apache.org/jira/browse/HBASE-16663">HBASE-16663</a> 672 * @return true if bypaas coprocessor execution, false if not. 673 */ 674 protected <O> boolean execShutdown(final ObserverOperation<O> observerOperation) 675 throws IOException { 676 if (observerOperation == null) return false; 677 boolean bypass = false; 678 List<E> envs = coprocEnvironments.get(); 679 // Iterate the coprocessors and execute ObserverOperation's call() 680 for (E env : envs) { 681 observerOperation.prepare(env); 682 Thread currentThread = Thread.currentThread(); 683 ClassLoader cl = currentThread.getContextClassLoader(); 684 try { 685 currentThread.setContextClassLoader(env.getClassLoader()); 686 observerOperation.callObserver(); 687 } catch (Throwable e) { 688 handleCoprocessorThrowable(env, e); 689 } finally { 690 currentThread.setContextClassLoader(cl); 691 } 692 bypass |= observerOperation.shouldBypass(); 693 } 694 695 // Iterate the coprocessors and execute ObserverOperation's postEnvCall() 696 for (E env : envs) { 697 observerOperation.prepare(env); 698 observerOperation.postEnvCall(); 699 } 700 return bypass; 701 } 702}