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;
019
020import java.io.IOException;
021import java.io.InterruptedIOException;
022import java.util.HashSet;
023import java.util.Iterator;
024import java.util.List;
025import java.util.Set;
026import java.util.concurrent.ExecutorService;
027import java.util.concurrent.Executors;
028import java.util.stream.Collectors;
029import org.apache.hadoop.hbase.ServerMetrics;
030import org.apache.hadoop.hbase.ServerMetricsBuilder;
031import org.apache.hadoop.hbase.ServerName;
032import org.apache.hadoop.hbase.client.VersionInfoUtil;
033import org.apache.hadoop.hbase.master.procedure.ServerCrashProcedure;
034import org.apache.hadoop.hbase.util.Pair;
035import org.apache.hadoop.hbase.zookeeper.ZKListener;
036import org.apache.hadoop.hbase.zookeeper.ZKUtil;
037import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
038import org.apache.hadoop.hbase.zookeeper.ZNodePaths;
039import org.apache.yetus.audience.InterfaceAudience;
040import org.apache.zookeeper.KeeperException;
041import org.slf4j.Logger;
042import org.slf4j.LoggerFactory;
043
044import org.apache.hbase.thirdparty.com.google.common.util.concurrent.ThreadFactoryBuilder;
045
046import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
047import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.RegionServerInfo;
048
049/**
050 * Tracks the online region servers via ZK.
051 * <p/>
052 * Handling of new RSs checking in is done via RPC. This class is only responsible for watching for
053 * expired nodes. It handles listening for changes in the RS node list. The only exception is when
054 * master restart, we will use the list fetched from zk to construct the initial set of live region
055 * servers.
056 * <p/>
057 * If an RS node gets deleted, this automatically handles calling of
058 * {@link ServerManager#expireServer(ServerName)}
059 */
060@InterfaceAudience.Private
061public class RegionServerTracker extends ZKListener {
062  private static final Logger LOG = LoggerFactory.getLogger(RegionServerTracker.class);
063  private final Set<ServerName> regionServers = new HashSet<>();
064  private final ServerManager serverManager;
065  private final MasterServices server;
066  // As we need to send request to zk when processing the nodeChildrenChanged event, we'd better
067  // move the operation to a single threaded thread pool in order to not block the zk event
068  // processing since all the zk listener across HMaster will be called in one thread sequentially.
069  private final ExecutorService executor;
070
071  public RegionServerTracker(ZKWatcher watcher, MasterServices server,
072      ServerManager serverManager) {
073    super(watcher);
074    this.server = server;
075    this.serverManager = serverManager;
076    this.executor = Executors.newSingleThreadExecutor(
077      new ThreadFactoryBuilder().setDaemon(true).setNameFormat("RegionServerTracker-%d").build());
078  }
079
080  private Pair<ServerName, RegionServerInfo> getServerInfo(String name)
081      throws KeeperException, IOException {
082    ServerName serverName = ServerName.parseServerName(name);
083    String nodePath = ZNodePaths.joinZNode(watcher.getZNodePaths().rsZNode, name);
084    byte[] data;
085    try {
086      data = ZKUtil.getData(watcher, nodePath);
087    } catch (InterruptedException e) {
088      throw (InterruptedIOException) new InterruptedIOException().initCause(e);
089    }
090    if (data == null) {
091      // we should receive a children changed event later and then we will expire it, so we still
092      // need to add it to the region server set.
093      LOG.warn("Server node {} does not exist, already dead?", name);
094      return Pair.newPair(serverName, null);
095    }
096    if (data.length == 0 || !ProtobufUtil.isPBMagicPrefix(data)) {
097      // this should not happen actually, unless we have bugs or someone has messed zk up.
098      LOG.warn("Invalid data for region server node {} on zookeeper, data length = {}", name,
099        data.length);
100      return Pair.newPair(serverName, null);
101    }
102    RegionServerInfo.Builder builder = RegionServerInfo.newBuilder();
103    int magicLen = ProtobufUtil.lengthOfPBMagic();
104    ProtobufUtil.mergeFrom(builder, data, magicLen, data.length - magicLen);
105    return Pair.newPair(serverName, builder.build());
106  }
107
108  /**
109   * Starts the tracking of online RegionServers. All RSes will be tracked after this method is
110   * called.
111   * <p/>
112   * In this method, we will also construct the region server sets in {@link ServerManager}. If a
113   * region server is dead between the crash of the previous master instance and the start of the
114   * current master instance, we will schedule a SCP for it. This is done in
115   * {@link ServerManager#findDeadServersAndProcess(Set, Set)}, we call it here under the lock
116   * protection to prevent concurrency issues with server expiration operation.
117   * @param deadServersFromPE the region servers which already have SCP associated.
118   * @param liveServersFromWALDir the live region servers from wal directory.
119   * @param splittingServersFromWALDir Servers whose WALs are being actively 'split'.
120   */
121  public void start(Set<ServerCrashProcedure> deadServersFromPE, Set<ServerName> liveServersFromWALDir,
122      Set<ServerName> splittingServersFromWALDir)
123      throws KeeperException, IOException {
124    LOG.info("Starting RegionServerTracker; {} have existing ServerCrashProcedures, {} " +
125        "possibly 'live' servers, and {} 'splitting'.", deadServersFromPE.size(),
126        liveServersFromWALDir.size(), splittingServersFromWALDir.size());
127    // deadServersFromPE is made from a list of outstanding ServerCrashProcedures.
128    // splittingServersFromWALDir are being actively split -- the directory in the FS ends in
129    // '-SPLITTING'. Each splitting server should have a corresponding SCP. Log if not.
130    Set<ServerName> deadServerNames = deadServersFromPE.stream()
131        .map(s -> s.getServerName()).collect(Collectors.toSet());
132    splittingServersFromWALDir.stream().filter(s -> !deadServerNames.contains(s)).
133      forEach(s -> LOG.error("{} has no matching ServerCrashProcedure", s));
134    watcher.registerListener(this);
135    synchronized (this) {
136      List<String> servers =
137        ZKUtil.listChildrenAndWatchForNewChildren(watcher, watcher.getZNodePaths().rsZNode);
138      for (String n : servers) {
139        Pair<ServerName, RegionServerInfo> pair = getServerInfo(n);
140        ServerName serverName = pair.getFirst();
141        RegionServerInfo info = pair.getSecond();
142        regionServers.add(serverName);
143        ServerMetrics serverMetrics = info != null ? ServerMetricsBuilder.of(serverName,
144          VersionInfoUtil.getVersionNumber(info.getVersionInfo()),
145          info.getVersionInfo().getVersion()) : ServerMetricsBuilder.of(serverName);
146        serverManager.checkAndRecordNewServer(serverName, serverMetrics);
147      }
148      serverManager.findDeadServersAndProcess(deadServersFromPE, liveServersFromWALDir);
149    }
150  }
151
152  public void stop() {
153    executor.shutdownNow();
154  }
155
156  private synchronized void refresh() {
157    List<String> names;
158    try {
159      names = ZKUtil.listChildrenAndWatchForNewChildren(watcher, watcher.getZNodePaths().rsZNode);
160    } catch (KeeperException e) {
161      // here we need to abort as we failed to set watcher on the rs node which means that we can
162      // not track the node deleted evetnt any more.
163      server.abort("Unexpected zk exception getting RS nodes", e);
164      return;
165    }
166    Set<ServerName> servers =
167      names.stream().map(ServerName::parseServerName).collect(Collectors.toSet());
168    for (Iterator<ServerName> iter = regionServers.iterator(); iter.hasNext();) {
169      ServerName sn = iter.next();
170      if (!servers.contains(sn)) {
171        LOG.info("RegionServer ephemeral node deleted, processing expiration [{}]", sn);
172        serverManager.expireServer(sn);
173        iter.remove();
174      }
175    }
176    // here we do not need to parse the region server info as it is useless now, we only need the
177    // server name.
178    boolean newServerAdded = false;
179    for (ServerName sn : servers) {
180      if (regionServers.add(sn)) {
181        newServerAdded = true;
182        LOG.info("RegionServer ephemeral node created, adding [" + sn + "]");
183      }
184    }
185    if (newServerAdded && server.isInitialized()) {
186      // Only call the check to move servers if a RegionServer was added to the cluster; in this
187      // case it could be a server with a new version so it makes sense to run the check.
188      server.checkIfShouldMoveSystemRegionAsync();
189    }
190  }
191
192  @Override
193  public void nodeChildrenChanged(String path) {
194    if (path.equals(watcher.getZNodePaths().rsZNode) && !server.isAborted() &&
195      !server.isStopped()) {
196      executor.execute(this::refresh);
197    }
198  }
199}