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;
019
020import com.google.errorprone.annotations.RestrictedApi;
021import java.io.IOException;
022import java.lang.reflect.InvocationTargetException;
023import java.net.BindException;
024import java.net.InetSocketAddress;
025import java.util.LinkedHashMap;
026import java.util.List;
027import java.util.Map;
028import java.util.Optional;
029import org.apache.hadoop.conf.Configuration;
030import org.apache.hadoop.hbase.client.ConnectionUtils;
031import org.apache.hadoop.hbase.conf.ConfigurationObserver;
032import org.apache.hadoop.hbase.coprocessor.ClientMetaCoprocessorHost;
033import org.apache.hadoop.hbase.io.ByteBuffAllocator;
034import org.apache.hadoop.hbase.ipc.HBaseRPCErrorHandler;
035import org.apache.hadoop.hbase.ipc.PriorityFunction;
036import org.apache.hadoop.hbase.ipc.QosPriority;
037import org.apache.hadoop.hbase.ipc.RpcScheduler;
038import org.apache.hadoop.hbase.ipc.RpcServer;
039import org.apache.hadoop.hbase.ipc.RpcServer.BlockingServiceAndInterface;
040import org.apache.hadoop.hbase.ipc.RpcServerFactory;
041import org.apache.hadoop.hbase.ipc.RpcServerInterface;
042import org.apache.hadoop.hbase.namequeues.NamedQueuePayload;
043import org.apache.hadoop.hbase.namequeues.NamedQueueRecorder;
044import org.apache.hadoop.hbase.net.Address;
045import org.apache.hadoop.hbase.regionserver.RpcSchedulerFactory;
046import org.apache.hadoop.hbase.security.User;
047import org.apache.hadoop.hbase.security.access.AccessChecker;
048import org.apache.hadoop.hbase.security.access.NoopAccessChecker;
049import org.apache.hadoop.hbase.security.access.Permission;
050import org.apache.hadoop.hbase.security.access.ZKPermissionWatcher;
051import org.apache.hadoop.hbase.util.DNS;
052import org.apache.hadoop.hbase.util.OOMEChecker;
053import org.apache.hadoop.hbase.util.ReservoirSample;
054import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
055import org.apache.yetus.audience.InterfaceAudience;
056import org.apache.zookeeper.KeeperException;
057import org.slf4j.Logger;
058import org.slf4j.LoggerFactory;
059
060import org.apache.hbase.thirdparty.com.google.protobuf.Message;
061import org.apache.hbase.thirdparty.com.google.protobuf.RpcController;
062import org.apache.hbase.thirdparty.com.google.protobuf.ServiceException;
063
064import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
065import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.AdminService;
066import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearSlowLogResponseRequest;
067import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ClearSlowLogResponses;
068import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UpdateConfigurationRequest;
069import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UpdateConfigurationResponse;
070import org.apache.hadoop.hbase.shaded.protobuf.generated.RPCProtos.RequestHeader;
071import org.apache.hadoop.hbase.shaded.protobuf.generated.RegistryProtos.ClientMetaService;
072import org.apache.hadoop.hbase.shaded.protobuf.generated.RegistryProtos.GetActiveMasterRequest;
073import org.apache.hadoop.hbase.shaded.protobuf.generated.RegistryProtos.GetActiveMasterResponse;
074import org.apache.hadoop.hbase.shaded.protobuf.generated.RegistryProtos.GetBootstrapNodesRequest;
075import org.apache.hadoop.hbase.shaded.protobuf.generated.RegistryProtos.GetBootstrapNodesResponse;
076import org.apache.hadoop.hbase.shaded.protobuf.generated.RegistryProtos.GetClusterIdRequest;
077import org.apache.hadoop.hbase.shaded.protobuf.generated.RegistryProtos.GetClusterIdResponse;
078import org.apache.hadoop.hbase.shaded.protobuf.generated.RegistryProtos.GetMastersRequest;
079import org.apache.hadoop.hbase.shaded.protobuf.generated.RegistryProtos.GetMastersResponse;
080import org.apache.hadoop.hbase.shaded.protobuf.generated.RegistryProtos.GetMastersResponseEntry;
081import org.apache.hadoop.hbase.shaded.protobuf.generated.RegistryProtos.GetMetaRegionLocationsRequest;
082import org.apache.hadoop.hbase.shaded.protobuf.generated.RegistryProtos.GetMetaRegionLocationsResponse;
083
084/**
085 * Base class for Master and RegionServer RpcServices.
086 */
087@InterfaceAudience.Private
088public abstract class HBaseRpcServicesBase<S extends HBaseServerBase<?>>
089  implements ClientMetaService.BlockingInterface, AdminService.BlockingInterface,
090  HBaseRPCErrorHandler, PriorityFunction, ConfigurationObserver {
091
092  private static final Logger LOG = LoggerFactory.getLogger(HBaseRpcServicesBase.class);
093
094  public static final String CLIENT_BOOTSTRAP_NODE_LIMIT = "hbase.client.bootstrap.node.limit";
095
096  public static final int DEFAULT_CLIENT_BOOTSTRAP_NODE_LIMIT = 10;
097
098  protected final S server;
099
100  // Server to handle client requests.
101  protected final RpcServer rpcServer;
102
103  private final InetSocketAddress isa;
104
105  protected final PriorityFunction priority;
106
107  private ClientMetaCoprocessorHost clientMetaCoprocessorHost;
108
109  private AccessChecker accessChecker;
110
111  private ZKPermissionWatcher zkPermissionWatcher;
112
113  protected HBaseRpcServicesBase(S server, String processName) throws IOException {
114    this.server = server;
115    Configuration conf = server.getConfiguration();
116    final RpcSchedulerFactory rpcSchedulerFactory;
117    try {
118      rpcSchedulerFactory = getRpcSchedulerFactoryClass(conf).asSubclass(RpcSchedulerFactory.class)
119        .getDeclaredConstructor().newInstance();
120    } catch (NoSuchMethodException | InvocationTargetException | InstantiationException
121      | IllegalAccessException e) {
122      throw new IllegalArgumentException(e);
123    }
124    String hostname = DNS.getHostname(conf, getDNSServerType());
125    int port = conf.getInt(getPortConfigName(), getDefaultPort());
126    // Creation of a HSA will force a resolve.
127    final InetSocketAddress initialIsa = new InetSocketAddress(hostname, port);
128    final InetSocketAddress bindAddress = new InetSocketAddress(getHostname(conf, hostname), port);
129    if (initialIsa.getAddress() == null) {
130      throw new IllegalArgumentException("Failed resolve of " + initialIsa);
131    }
132    priority = createPriority();
133    // Using Address means we don't get the IP too. Shorten it more even to just the host name
134    // w/o the domain.
135    final String name = processName + "/"
136      + Address.fromParts(initialIsa.getHostName(), initialIsa.getPort()).toStringWithoutDomain();
137    server.setName(name);
138    // Set how many times to retry talking to another server over Connection.
139    ConnectionUtils.setServerSideHConnectionRetriesConfig(conf, name, LOG);
140    boolean reservoirEnabled =
141      conf.getBoolean(ByteBuffAllocator.ALLOCATOR_POOL_ENABLED_KEY, defaultReservoirEnabled());
142    try {
143      // use final bindAddress for this server.
144      rpcServer = RpcServerFactory.createRpcServer(server, name, getServices(), bindAddress, conf,
145        rpcSchedulerFactory.create(conf, this, server), reservoirEnabled);
146    } catch (BindException be) {
147      throw new IOException(be.getMessage() + ". To switch ports use the '" + getPortConfigName()
148        + "' configuration property.", be.getCause() != null ? be.getCause() : be);
149    }
150    final InetSocketAddress address = rpcServer.getListenerAddress();
151    if (address == null) {
152      throw new IOException("Listener channel is closed");
153    }
154    // Set our address, however we need the final port that was given to rpcServer
155    isa = new InetSocketAddress(initialIsa.getHostName(), address.getPort());
156    rpcServer.setErrorHandler(this);
157
158    clientMetaCoprocessorHost = new ClientMetaCoprocessorHost(conf);
159  }
160
161  protected abstract boolean defaultReservoirEnabled();
162
163  protected abstract DNS.ServerType getDNSServerType();
164
165  protected abstract String getHostname(Configuration conf, String defaultHostname);
166
167  protected abstract String getPortConfigName();
168
169  protected abstract int getDefaultPort();
170
171  protected abstract PriorityFunction createPriority();
172
173  protected abstract Class<?> getRpcSchedulerFactoryClass(Configuration conf);
174
175  protected abstract List<BlockingServiceAndInterface> getServices();
176
177  protected final void internalStart(ZKWatcher zkWatcher) {
178    if (AccessChecker.isAuthorizationSupported(getConfiguration())) {
179      accessChecker = new AccessChecker(getConfiguration());
180    } else {
181      accessChecker = new NoopAccessChecker(getConfiguration());
182    }
183    zkPermissionWatcher =
184      new ZKPermissionWatcher(zkWatcher, accessChecker.getAuthManager(), getConfiguration());
185    try {
186      zkPermissionWatcher.start();
187    } catch (KeeperException e) {
188      LOG.error("ZooKeeper permission watcher initialization failed", e);
189    }
190    rpcServer.start();
191  }
192
193  protected final void requirePermission(String request, Permission.Action perm)
194    throws IOException {
195    if (accessChecker != null) {
196      accessChecker.requirePermission(RpcServer.getRequestUser().orElse(null), request, null, perm);
197    }
198  }
199
200  @RestrictedApi(explanation = "Should only be called in tests", link = "",
201      allowedOnPath = ".*/src/test/.*")
202  public ClientMetaCoprocessorHost getClientMetaCoprocessorHost() {
203    return clientMetaCoprocessorHost;
204  }
205
206  public AccessChecker getAccessChecker() {
207    return accessChecker;
208  }
209
210  public ZKPermissionWatcher getZkPermissionWatcher() {
211    return zkPermissionWatcher;
212  }
213
214  protected final void internalStop() {
215    if (zkPermissionWatcher != null) {
216      zkPermissionWatcher.close();
217    }
218    rpcServer.stop();
219  }
220
221  public Configuration getConfiguration() {
222    return server.getConfiguration();
223  }
224
225  public S getServer() {
226    return server;
227  }
228
229  public InetSocketAddress getSocketAddress() {
230    return isa;
231  }
232
233  public RpcServerInterface getRpcServer() {
234    return rpcServer;
235  }
236
237  public RpcScheduler getRpcScheduler() {
238    return rpcServer.getScheduler();
239  }
240
241  @Override
242  public int getPriority(RequestHeader header, Message param, User user) {
243    return priority.getPriority(header, param, user);
244  }
245
246  @Override
247  public long getDeadline(RequestHeader header, Message param) {
248    return priority.getDeadline(header, param);
249  }
250
251  /**
252   * Check if an OOME and, if so, abort immediately to avoid creating more objects.
253   * @return True if we OOME'd and are aborting.
254   */
255  @Override
256  public boolean checkOOME(Throwable e) {
257    return OOMEChecker.exitIfOOME(e, getClass().getSimpleName());
258  }
259
260  @Override
261  public void onConfigurationChange(Configuration conf) {
262    rpcServer.onConfigurationChange(conf);
263  }
264
265  @Override
266  public GetClusterIdResponse getClusterId(RpcController controller, GetClusterIdRequest request)
267    throws ServiceException {
268    try {
269      clientMetaCoprocessorHost.preGetClusterId();
270
271      String clusterId = server.getClusterId();
272      String clusterIdReply = clientMetaCoprocessorHost.postGetClusterId(clusterId);
273
274      return GetClusterIdResponse.newBuilder().setClusterId(clusterIdReply).build();
275    } catch (IOException e) {
276      throw new ServiceException(e);
277    }
278  }
279
280  @Override
281  public GetActiveMasterResponse getActiveMaster(RpcController controller,
282    GetActiveMasterRequest request) throws ServiceException {
283    GetActiveMasterResponse.Builder builder = GetActiveMasterResponse.newBuilder();
284
285    try {
286      clientMetaCoprocessorHost.preGetActiveMaster();
287
288      ServerName serverName = server.getActiveMaster().orElse(null);
289      ServerName serverNameReply = clientMetaCoprocessorHost.postGetActiveMaster(serverName);
290
291      if (serverNameReply != null) {
292        builder.setServerName(ProtobufUtil.toServerName(serverNameReply));
293      }
294    } catch (IOException e) {
295      throw new ServiceException(e);
296    }
297
298    return builder.build();
299  }
300
301  @Override
302  public GetMastersResponse getMasters(RpcController controller, GetMastersRequest request)
303    throws ServiceException {
304    GetMastersResponse.Builder builder = GetMastersResponse.newBuilder();
305
306    try {
307      clientMetaCoprocessorHost.preGetMasters();
308
309      Map<ServerName, Boolean> serverNames = new LinkedHashMap<>();
310
311      server.getActiveMaster().ifPresent(serverName -> serverNames.put(serverName, Boolean.TRUE));
312      server.getBackupMasters().forEach(serverName -> serverNames.put(serverName, Boolean.FALSE));
313
314      Map<ServerName, Boolean> serverNamesReply =
315        clientMetaCoprocessorHost.postGetMasters(serverNames);
316
317      serverNamesReply
318        .forEach((serverName, active) -> builder.addMasterServers(GetMastersResponseEntry
319          .newBuilder().setServerName(ProtobufUtil.toServerName(serverName)).setIsActive(active)));
320    } catch (IOException e) {
321      throw new ServiceException(e);
322    }
323
324    return builder.build();
325  }
326
327  @Override
328  public GetMetaRegionLocationsResponse getMetaRegionLocations(RpcController controller,
329    GetMetaRegionLocationsRequest request) throws ServiceException {
330    GetMetaRegionLocationsResponse.Builder builder = GetMetaRegionLocationsResponse.newBuilder();
331
332    try {
333      clientMetaCoprocessorHost.preGetMetaLocations();
334
335      List<HRegionLocation> metaLocations = server.getMetaLocations();
336      List<HRegionLocation> metaLocationsReply =
337        clientMetaCoprocessorHost.postGetMetaLocations(metaLocations);
338
339      metaLocationsReply
340        .forEach(location -> builder.addMetaLocations(ProtobufUtil.toRegionLocation(location)));
341    } catch (IOException e) {
342      throw new ServiceException(e);
343    }
344
345    return builder.build();
346  }
347
348  @Override
349  public final GetBootstrapNodesResponse getBootstrapNodes(RpcController controller,
350    GetBootstrapNodesRequest request) throws ServiceException {
351    GetBootstrapNodesResponse.Builder builder = GetBootstrapNodesResponse.newBuilder();
352
353    try {
354      clientMetaCoprocessorHost.preGetBootstrapNodes();
355
356      int maxNodeCount = server.getConfiguration().getInt(CLIENT_BOOTSTRAP_NODE_LIMIT,
357        DEFAULT_CLIENT_BOOTSTRAP_NODE_LIMIT);
358      ReservoirSample<ServerName> sample = new ReservoirSample<>(maxNodeCount);
359      sample.add(server.getBootstrapNodes());
360
361      List<ServerName> bootstrapNodes = sample.getSamplingResult();
362      List<ServerName> bootstrapNodesReply =
363        clientMetaCoprocessorHost.postGetBootstrapNodes(bootstrapNodes);
364
365      bootstrapNodesReply
366        .forEach(serverName -> builder.addServerName(ProtobufUtil.toServerName(serverName)));
367    } catch (IOException e) {
368      throw new ServiceException(e);
369    }
370
371    return builder.build();
372  }
373
374  @Override
375  @QosPriority(priority = HConstants.ADMIN_QOS)
376  public UpdateConfigurationResponse updateConfiguration(RpcController controller,
377    UpdateConfigurationRequest request) throws ServiceException {
378    try {
379      requirePermission("updateConfiguration", Permission.Action.ADMIN);
380      this.server.updateConfiguration();
381
382      clientMetaCoprocessorHost = new ClientMetaCoprocessorHost(getConfiguration());
383    } catch (Exception e) {
384      throw new ServiceException(e);
385    }
386    return UpdateConfigurationResponse.getDefaultInstance();
387  }
388
389  @Override
390  @QosPriority(priority = HConstants.ADMIN_QOS)
391  public ClearSlowLogResponses clearSlowLogsResponses(final RpcController controller,
392    final ClearSlowLogResponseRequest request) throws ServiceException {
393    try {
394      requirePermission("clearSlowLogsResponses", Permission.Action.ADMIN);
395    } catch (IOException e) {
396      throw new ServiceException(e);
397    }
398    final NamedQueueRecorder namedQueueRecorder = this.server.getNamedQueueRecorder();
399    boolean slowLogsCleaned = Optional.ofNullable(namedQueueRecorder)
400      .map(
401        queueRecorder -> queueRecorder.clearNamedQueue(NamedQueuePayload.NamedQueueEvent.SLOW_LOG))
402      .orElse(false);
403    ClearSlowLogResponses clearSlowLogResponses =
404      ClearSlowLogResponses.newBuilder().setIsCleaned(slowLogsCleaned).build();
405    return clearSlowLogResponses;
406  }
407}