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.master.procedure;
019
020import java.io.IOException;
021import java.lang.Thread.UncaughtExceptionHandler;
022import java.util.List;
023import java.util.Set;
024import java.util.concurrent.TimeUnit;
025import javax.security.sasl.SaslException;
026import org.apache.hadoop.hbase.CallQueueTooBigException;
027import org.apache.hadoop.hbase.DoNotRetryIOException;
028import org.apache.hadoop.hbase.ServerName;
029import org.apache.hadoop.hbase.client.AsyncRegionServerAdmin;
030import org.apache.hadoop.hbase.client.RegionInfo;
031import org.apache.hadoop.hbase.ipc.RpcConnectionConstants;
032import org.apache.hadoop.hbase.ipc.ServerNotRunningYetException;
033import org.apache.hadoop.hbase.master.MasterServices;
034import org.apache.hadoop.hbase.master.ServerListener;
035import org.apache.hadoop.hbase.master.ServerManager;
036import org.apache.hadoop.hbase.procedure2.ProcedureExecutor;
037import org.apache.hadoop.hbase.procedure2.RemoteProcedureDispatcher;
038import org.apache.hadoop.hbase.regionserver.RegionServerAbortedException;
039import org.apache.hadoop.hbase.regionserver.RegionServerStoppedException;
040import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
041import org.apache.hadoop.hbase.util.FutureUtils;
042import org.apache.hadoop.ipc.RemoteException;
043import org.apache.yetus.audience.InterfaceAudience;
044import org.slf4j.Logger;
045import org.slf4j.LoggerFactory;
046
047import org.apache.hbase.thirdparty.com.google.common.collect.ArrayListMultimap;
048import org.apache.hbase.thirdparty.com.google.protobuf.ByteString;
049
050import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
051import org.apache.hadoop.hbase.shaded.protobuf.RequestConverter;
052import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.CloseRegionRequest;
053import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ExecuteProceduresRequest;
054import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.ExecuteProceduresResponse;
055import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.OpenRegionRequest;
056import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.RemoteProcedureRequest;
057
058/**
059 * A remote procecdure dispatcher for regionservers.
060 */
061@InterfaceAudience.Private
062public class RSProcedureDispatcher extends RemoteProcedureDispatcher<MasterProcedureEnv, ServerName>
063  implements ServerListener {
064  private static final Logger LOG = LoggerFactory.getLogger(RSProcedureDispatcher.class);
065
066  public static final String RS_RPC_STARTUP_WAIT_TIME_CONF_KEY =
067    "hbase.regionserver.rpc.startup.waittime";
068  private static final int DEFAULT_RS_RPC_STARTUP_WAIT_TIME = 60000;
069
070  protected final MasterServices master;
071  private final long rsStartupWaitTime;
072  private MasterProcedureEnv procedureEnv;
073
074  public RSProcedureDispatcher(final MasterServices master) {
075    super(master.getConfiguration());
076
077    this.master = master;
078    this.rsStartupWaitTime = master.getConfiguration().getLong(RS_RPC_STARTUP_WAIT_TIME_CONF_KEY,
079      DEFAULT_RS_RPC_STARTUP_WAIT_TIME);
080  }
081
082  @Override
083  protected UncaughtExceptionHandler getUncaughtExceptionHandler() {
084    return new UncaughtExceptionHandler() {
085
086      @Override
087      public void uncaughtException(Thread t, Throwable e) {
088        LOG.error("Unexpected error caught, this may cause the procedure to hang forever", e);
089      }
090    };
091  }
092
093  @Override
094  public boolean start() {
095    if (!super.start()) {
096      return false;
097    }
098    setTimeoutExecutorUncaughtExceptionHandler(this::abort);
099    if (master.isStopped()) {
100      LOG.debug("Stopped");
101      return false;
102    }
103    // Around startup, if failed, some of the below may be set back to null so NPE is possible.
104    ServerManager sm = master.getServerManager();
105    if (sm == null) {
106      LOG.debug("ServerManager is null");
107      return false;
108    }
109    sm.registerListener(this);
110    ProcedureExecutor<MasterProcedureEnv> pe = master.getMasterProcedureExecutor();
111    if (pe == null) {
112      LOG.debug("ProcedureExecutor is null");
113      return false;
114    }
115    this.procedureEnv = pe.getEnvironment();
116    if (this.procedureEnv == null) {
117      LOG.debug("ProcedureEnv is null; stopping={}", master.isStopping());
118      return false;
119    }
120    try {
121      for (ServerName serverName : sm.getOnlineServersList()) {
122        addNode(serverName);
123      }
124    } catch (Exception e) {
125      LOG.info("Failed start", e);
126      return false;
127    }
128    return true;
129  }
130
131  private void abort(Thread t, Throwable e) {
132    LOG.error("Caught error", e);
133    if (!master.isStopped() && !master.isStopping() && !master.isAborted()) {
134      master.abort("Aborting master", e);
135    }
136  }
137
138  @Override
139  public boolean stop() {
140    if (!super.stop()) {
141      return false;
142    }
143
144    master.getServerManager().unregisterListener(this);
145    return true;
146  }
147
148  @Override
149  protected void remoteDispatch(final ServerName serverName,
150    final Set<RemoteProcedure> remoteProcedures) {
151    if (!master.getServerManager().isServerOnline(serverName)) {
152      // fail fast
153      submitTask(new DeadRSRemoteCall(serverName, remoteProcedures));
154    } else {
155      submitTask(new ExecuteProceduresRemoteCall(serverName, remoteProcedures));
156    }
157  }
158
159  @Override
160  protected void abortPendingOperations(final ServerName serverName,
161    final Set<RemoteProcedure> operations) {
162    // TODO: Replace with a ServerNotOnlineException()
163    final IOException e = new DoNotRetryIOException("server not online " + serverName);
164    for (RemoteProcedure proc : operations) {
165      proc.remoteCallFailed(procedureEnv, serverName, e);
166    }
167  }
168
169  @Override
170  public void serverAdded(final ServerName serverName) {
171    addNode(serverName);
172  }
173
174  @Override
175  public void serverRemoved(final ServerName serverName) {
176    removeNode(serverName);
177  }
178
179  private interface RemoteProcedureResolver {
180    void dispatchOpenRequests(MasterProcedureEnv env, List<RegionOpenOperation> operations);
181
182    void dispatchCloseRequests(MasterProcedureEnv env, List<RegionCloseOperation> operations);
183
184    void dispatchServerOperations(MasterProcedureEnv env, List<ServerOperation> operations);
185  }
186
187  /**
188   * Fetches {@link org.apache.hadoop.hbase.procedure2.RemoteProcedureDispatcher.RemoteOperation}s
189   * from the given {@code remoteProcedures} and groups them by class of the returned operation.
190   * Then {@code resolver} is used to dispatch {@link RegionOpenOperation}s and
191   * {@link RegionCloseOperation}s.
192   * @param serverName RegionServer to which the remote operations are sent
193   * @param operations Remote procedures which are dispatched to the given server
194   * @param resolver   Used to dispatch remote procedures to given server.
195   */
196  public void splitAndResolveOperation(ServerName serverName, Set<RemoteProcedure> operations,
197    RemoteProcedureResolver resolver) {
198    MasterProcedureEnv env = master.getMasterProcedureExecutor().getEnvironment();
199    ArrayListMultimap<Class<?>, RemoteOperation> reqsByType =
200      buildAndGroupRequestByType(env, serverName, operations);
201
202    List<RegionOpenOperation> openOps = fetchType(reqsByType, RegionOpenOperation.class);
203    if (!openOps.isEmpty()) {
204      resolver.dispatchOpenRequests(env, openOps);
205    }
206
207    List<RegionCloseOperation> closeOps = fetchType(reqsByType, RegionCloseOperation.class);
208    if (!closeOps.isEmpty()) {
209      resolver.dispatchCloseRequests(env, closeOps);
210    }
211
212    List<ServerOperation> refreshOps = fetchType(reqsByType, ServerOperation.class);
213    if (!refreshOps.isEmpty()) {
214      resolver.dispatchServerOperations(env, refreshOps);
215    }
216
217    if (!reqsByType.isEmpty()) {
218      LOG.warn("unknown request type in the queue: " + reqsByType);
219    }
220  }
221
222  private class DeadRSRemoteCall extends ExecuteProceduresRemoteCall {
223
224    public DeadRSRemoteCall(ServerName serverName, Set<RemoteProcedure> remoteProcedures) {
225      super(serverName, remoteProcedures);
226    }
227
228    @Override
229    public void run() {
230      remoteCallFailed(procedureEnv,
231        new RegionServerStoppedException("Server " + getServerName() + " is not online"));
232    }
233  }
234
235  // ==========================================================================
236  // Compatibility calls
237  // ==========================================================================
238  protected class ExecuteProceduresRemoteCall implements RemoteProcedureResolver, Runnable {
239
240    private final ServerName serverName;
241
242    private final Set<RemoteProcedure> remoteProcedures;
243
244    private int numberOfAttemptsSoFar = 0;
245    private long maxWaitTime = -1;
246
247    private final long rsRpcRetryInterval;
248    private static final String RS_RPC_RETRY_INTERVAL_CONF_KEY =
249      "hbase.regionserver.rpc.retry.interval";
250    private static final int DEFAULT_RS_RPC_RETRY_INTERVAL = 100;
251
252    private ExecuteProceduresRequest.Builder request = null;
253
254    public ExecuteProceduresRemoteCall(final ServerName serverName,
255      final Set<RemoteProcedure> remoteProcedures) {
256      this.serverName = serverName;
257      this.remoteProcedures = remoteProcedures;
258      this.rsRpcRetryInterval = master.getConfiguration().getLong(RS_RPC_RETRY_INTERVAL_CONF_KEY,
259        DEFAULT_RS_RPC_RETRY_INTERVAL);
260    }
261
262    private AsyncRegionServerAdmin getRsAdmin() throws IOException {
263      return master.getAsyncClusterConnection().getRegionServerAdmin(serverName);
264    }
265
266    protected final ServerName getServerName() {
267      return serverName;
268    }
269
270    private boolean scheduleForRetry(IOException e) {
271      LOG.debug("Request to {} failed, try={}", serverName, numberOfAttemptsSoFar, e);
272      // Should we wait a little before retrying? If the server is starting it's yes.
273      if (e instanceof ServerNotRunningYetException) {
274        long remainingTime = getMaxWaitTime() - EnvironmentEdgeManager.currentTime();
275        if (remainingTime > 0) {
276          LOG.warn("Waiting a little before retrying {}, try={}, can wait up to {}ms", serverName,
277            numberOfAttemptsSoFar, remainingTime);
278          numberOfAttemptsSoFar++;
279          // Retry every rsRpcRetryInterval millis up to maximum wait time.
280          submitTask(this, rsRpcRetryInterval, TimeUnit.MILLISECONDS);
281          return true;
282        }
283        LOG.warn("{} is throwing ServerNotRunningYetException for {}ms; trying another server",
284          serverName, getMaxWaitTime());
285        return false;
286      }
287      if (e instanceof DoNotRetryIOException) {
288        LOG.warn("{} tells us DoNotRetry due to {}, try={}, give up", serverName, e.toString(),
289          numberOfAttemptsSoFar);
290        return false;
291      }
292      // This category of exceptions is thrown in the rpc framework, where we can make sure
293      // that the call has not been executed yet, so it is safe to mark it as fail.
294      // Especially for open a region, we'd better choose another region server.
295      // Notice that, it is safe to quit only if this is the first time we send request to region
296      // server. Maybe the region server has accepted our request the first time, and then there is
297      // a network error which prevents we receive the response, and the second time we hit
298      // this category of exceptions, obviously it is not safe to quit here, otherwise it may lead
299      // to a double assign...
300      if (numberOfAttemptsSoFar == 0 && unableToConnectToServer(e)) {
301        return false;
302      }
303      // Always retry for other exception types if the region server is not dead yet.
304      if (!master.getServerManager().isServerOnline(serverName)) {
305        LOG.warn("Request to {} failed due to {}, try={} and the server is not online, give up",
306          serverName, e.toString(), numberOfAttemptsSoFar);
307        return false;
308      }
309      if (e instanceof RegionServerAbortedException || e instanceof RegionServerStoppedException) {
310        // A better way is to return true here to let the upper layer quit, and then schedule a
311        // background task to check whether the region server is dead. And if it is dead, call
312        // remoteCallFailed to tell the upper layer. Keep retrying here does not lead to incorrect
313        // result, but waste some resources.
314        LOG.warn("{} is aborted or stopped, for safety we still need to"
315          + " wait until it is fully dead, try={}", serverName, numberOfAttemptsSoFar);
316      } else {
317        LOG.warn("request to {} failed due to {}, try={}, retrying... , request params: {}",
318          serverName, e.toString(), numberOfAttemptsSoFar, request.build());
319      }
320      numberOfAttemptsSoFar++;
321      // Add some backoff here as the attempts rise otherwise if a stuck condition, will fill logs
322      // with failed attempts. None of our backoff classes -- RetryCounter or ClientBackoffPolicy
323      // -- fit here nicely so just do something simple; increment by rsRpcRetryInterval millis *
324      // retry^2 on each try
325      // up to max of 10 seconds (don't want to back off too much in case of situation change).
326      submitTask(this,
327        Math.min(rsRpcRetryInterval * (this.numberOfAttemptsSoFar * this.numberOfAttemptsSoFar),
328          10 * 1000),
329        TimeUnit.MILLISECONDS);
330      return true;
331    }
332
333    /**
334     * The category of exceptions where we can ensure that the request has not yet been received
335     * and/or processed by the target regionserver yet and hence we can determine whether it is safe
336     * to choose different regionserver as the target.
337     * @param e IOException thrown by the underlying rpc framework.
338     * @return true if the exception belongs to the category where the regionserver has not yet
339     *         received the request yet.
340     */
341    private boolean unableToConnectToServer(IOException e) {
342      if (e instanceof CallQueueTooBigException) {
343        LOG.warn("request to {} failed due to {}, try={}, this usually because"
344          + " server is overloaded, give up", serverName, e, numberOfAttemptsSoFar);
345        return true;
346      }
347      if (isSaslError(e)) {
348        LOG.warn("{} is not reachable; give up after first attempt", serverName, e);
349        return true;
350      }
351      return false;
352    }
353
354    private boolean isSaslError(IOException e) {
355      Throwable cause = e;
356      while (true) {
357        if (cause instanceof IOException) {
358          IOException unwrappedCause = unwrapException((IOException) cause);
359          if (
360            unwrappedCause instanceof SaslException
361              || (unwrappedCause.getMessage() != null && unwrappedCause.getMessage()
362                .contains(RpcConnectionConstants.RELOGIN_IS_IN_PROGRESS))
363          ) {
364            return true;
365          }
366        }
367        cause = cause.getCause();
368        if (cause == null) {
369          return false;
370        }
371      }
372    }
373
374    private long getMaxWaitTime() {
375      if (this.maxWaitTime < 0) {
376        // This is the max attempts, not retries, so it should be at least 1.
377        this.maxWaitTime = EnvironmentEdgeManager.currentTime() + rsStartupWaitTime;
378      }
379      return this.maxWaitTime;
380    }
381
382    private IOException unwrapException(IOException e) {
383      if (e instanceof RemoteException) {
384        e = ((RemoteException) e).unwrapRemoteException();
385      }
386      return e;
387    }
388
389    @Override
390    public void run() {
391      request = ExecuteProceduresRequest.newBuilder();
392      if (LOG.isTraceEnabled()) {
393        LOG.trace("Building request with operations count=" + remoteProcedures.size());
394      }
395      splitAndResolveOperation(getServerName(), remoteProcedures, this);
396
397      try {
398        sendRequest(getServerName(), request.build());
399      } catch (IOException e) {
400        e = unwrapException(e);
401        // TODO: In the future some operation may want to bail out early.
402        // TODO: How many times should we retry (use numberOfAttemptsSoFar)
403        if (!scheduleForRetry(e)) {
404          remoteCallFailed(procedureEnv, e);
405        }
406      }
407    }
408
409    @Override
410    public void dispatchOpenRequests(final MasterProcedureEnv env,
411      final List<RegionOpenOperation> operations) {
412      request.addOpenRegion(buildOpenRegionRequest(env, getServerName(), operations));
413    }
414
415    @Override
416    public void dispatchCloseRequests(final MasterProcedureEnv env,
417      final List<RegionCloseOperation> operations) {
418      for (RegionCloseOperation op : operations) {
419        request.addCloseRegion(op.buildCloseRegionRequest(getServerName()));
420      }
421    }
422
423    @Override
424    public void dispatchServerOperations(MasterProcedureEnv env, List<ServerOperation> operations) {
425      operations.stream().map(ServerOperation::buildRequest).forEachOrdered(request::addProc);
426    }
427
428    // will be overridden in test.
429    protected ExecuteProceduresResponse sendRequest(final ServerName serverName,
430      final ExecuteProceduresRequest request) throws IOException {
431      return FutureUtils.get(getRsAdmin().executeProcedures(request));
432    }
433
434    protected final void remoteCallFailed(final MasterProcedureEnv env, final IOException e) {
435      for (RemoteProcedure proc : remoteProcedures) {
436        proc.remoteCallFailed(env, getServerName(), e);
437      }
438    }
439  }
440
441  private static OpenRegionRequest buildOpenRegionRequest(final MasterProcedureEnv env,
442    final ServerName serverName, final List<RegionOpenOperation> operations) {
443    final OpenRegionRequest.Builder builder = OpenRegionRequest.newBuilder();
444    builder.setServerStartCode(serverName.getStartCode());
445    operations.stream().map(RemoteOperation::getInitiatingMasterActiveTime).findAny()
446      .ifPresent(builder::setInitiatingMasterActiveTime);
447    builder.setMasterSystemTime(EnvironmentEdgeManager.currentTime());
448    for (RegionOpenOperation op : operations) {
449      builder.addOpenInfo(op.buildRegionOpenInfoRequest(env));
450    }
451    return builder.build();
452  }
453
454  // ==========================================================================
455  // RPC Messages
456  // - ServerOperation: refreshConfig, grant, revoke, ... (TODO)
457  // - RegionOperation: open, close, flush, snapshot, ...
458  // ==========================================================================
459
460  public static final class ServerOperation extends RemoteOperation {
461
462    private final long procId;
463
464    private final Class<?> rsProcClass;
465
466    private final byte[] rsProcData;
467
468    public ServerOperation(RemoteProcedure remoteProcedure, long procId, Class<?> rsProcClass,
469      byte[] rsProcData, long initiatingMasterActiveTime) {
470      super(remoteProcedure, initiatingMasterActiveTime);
471      this.procId = procId;
472      this.rsProcClass = rsProcClass;
473      this.rsProcData = rsProcData;
474    }
475
476    public RemoteProcedureRequest buildRequest() {
477      return RemoteProcedureRequest.newBuilder().setProcId(procId)
478        .setProcClass(rsProcClass.getName()).setProcData(ByteString.copyFrom(rsProcData))
479        .setInitiatingMasterActiveTime(getInitiatingMasterActiveTime()).build();
480    }
481  }
482
483  public static abstract class RegionOperation extends RemoteOperation {
484    protected final RegionInfo regionInfo;
485    protected final long procId;
486
487    protected RegionOperation(RemoteProcedure remoteProcedure, RegionInfo regionInfo, long procId,
488      long initiatingMasterActiveTime) {
489      super(remoteProcedure, initiatingMasterActiveTime);
490      this.regionInfo = regionInfo;
491      this.procId = procId;
492    }
493  }
494
495  public static class RegionOpenOperation extends RegionOperation {
496
497    public RegionOpenOperation(RemoteProcedure remoteProcedure, RegionInfo regionInfo, long procId,
498      long initiatingMasterActiveTime) {
499      super(remoteProcedure, regionInfo, procId, initiatingMasterActiveTime);
500    }
501
502    public OpenRegionRequest.RegionOpenInfo
503      buildRegionOpenInfoRequest(final MasterProcedureEnv env) {
504      return RequestConverter.buildRegionOpenInfo(regionInfo,
505        env.getAssignmentManager().getFavoredNodes(regionInfo), procId);
506    }
507  }
508
509  public static class RegionCloseOperation extends RegionOperation {
510    private final ServerName destinationServer;
511    private boolean evictCache;
512
513    public RegionCloseOperation(RemoteProcedure remoteProcedure, RegionInfo regionInfo, long procId,
514      ServerName destinationServer, boolean evictCache, long initiatingMasterActiveTime) {
515      super(remoteProcedure, regionInfo, procId, initiatingMasterActiveTime);
516      this.destinationServer = destinationServer;
517      this.evictCache = evictCache;
518    }
519
520    public ServerName getDestinationServer() {
521      return destinationServer;
522    }
523
524    public CloseRegionRequest buildCloseRegionRequest(final ServerName serverName) {
525      return ProtobufUtil.buildCloseRegionRequest(serverName, regionInfo.getRegionName(),
526        getDestinationServer(), procId, evictCache, getInitiatingMasterActiveTime());
527    }
528  }
529}