View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
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      // Log the state of coprocessor loading here; should appear only once or
65      // twice in the daemon log, depending on HBase version, because there is
66      // only one RegionServerCoprocessorHost instance in the RS process
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      // While stopping the region server all coprocessors method should be executed first then the
86      // coprocessor should be cleaned up.
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          // invoke coprocessor stop method
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    * RegionServer coprocessor classes can be configured in any order, based on that priority is set
281    * and chained in a sorted order. For preStop(), coprocessor methods are invoked in call() and
282    * environment is shutdown in postEnvCall(). <br>
283    * Need to execute all coprocessor methods first then postEnvCall(), otherwise some coprocessors
284    * may remain shutdown if any exception occurs during next coprocessor execution which prevent
285    * RegionServer stop. (Refer:
286    * <a href="https://issues.apache.org/jira/browse/HBASE-16663">HBASE-16663</a>
287    * @param ctx CoprocessorOperation
288    * @return true if bypaas coprocessor execution, false if not.
289    * @throws IOException
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     // Iterate the coprocessors and execute CoprocessorOperation's call()
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     // Iterate the coprocessors and execute CoprocessorOperation's postEnvCall()
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    * Coprocessor environment extension providing access to region server
328    * related services.
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)) {// FindBugs: BC_UNCONFIRMED_CAST
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    * Environment priority comparator. Coprocessors are chained in sorted
373    * order.
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 }