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