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.zookeeper; 019 020import static org.apache.zookeeper.client.FourLetterWordMain.send4LetterWord; 021import java.io.File; 022import java.io.IOException; 023import java.io.InterruptedIOException; 024import java.io.PrintWriter; 025import java.io.StringWriter; 026import java.net.BindException; 027import java.net.ConnectException; 028import java.net.InetAddress; 029import java.net.InetSocketAddress; 030import java.util.ArrayList; 031import java.util.List; 032import java.util.Random; 033import org.apache.hadoop.conf.Configuration; 034import org.apache.hadoop.hbase.HConstants; 035import org.apache.hadoop.hbase.net.Address; 036import org.apache.hadoop.hbase.util.Threads; 037import org.apache.yetus.audience.InterfaceAudience; 038import org.apache.zookeeper.common.X509Exception; 039import org.apache.zookeeper.server.NIOServerCnxnFactory; 040import org.apache.zookeeper.server.ZooKeeperServer; 041import org.apache.zookeeper.server.persistence.FileTxnLog; 042import org.slf4j.Logger; 043import org.slf4j.LoggerFactory; 044import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting; 045 046/** 047 * TODO: Most of the code in this class is ripped from ZooKeeper tests. Instead 048 * of redoing it, we should contribute updates to their code which let us more 049 * easily access testing helper objects. 050 */ 051@InterfaceAudience.Public 052public class MiniZooKeeperCluster { 053 private static final Logger LOG = LoggerFactory.getLogger(MiniZooKeeperCluster.class); 054 private static final int TICK_TIME = 2000; 055 private static final int TIMEOUT = 1000; 056 private static final int DEFAULT_CONNECTION_TIMEOUT = 30000; 057 private int connectionTimeout; 058 public static final String LOOPBACK_HOST = InetAddress.getLoopbackAddress().getHostName(); 059 public static final String HOST = LOOPBACK_HOST; 060 061 private boolean started; 062 063 /** 064 * The default port. If zero, we use a random port. 065 */ 066 private int defaultClientPort = 0; 067 068 private final List<NIOServerCnxnFactory> standaloneServerFactoryList; 069 private final List<ZooKeeperServer> zooKeeperServers; 070 private final List<Integer> clientPortList; 071 072 private int activeZKServerIndex; 073 private int tickTime = 0; 074 075 private final Configuration configuration; 076 077 public MiniZooKeeperCluster() { 078 this(new Configuration()); 079 } 080 081 public MiniZooKeeperCluster(Configuration configuration) { 082 this.started = false; 083 this.configuration = configuration; 084 activeZKServerIndex = -1; 085 zooKeeperServers = new ArrayList<>(); 086 clientPortList = new ArrayList<>(); 087 standaloneServerFactoryList = new ArrayList<>(); 088 connectionTimeout = configuration 089 .getInt(HConstants.ZK_SESSION_TIMEOUT + ".localHBaseCluster", DEFAULT_CONNECTION_TIMEOUT); 090 } 091 092 /** 093 * Add a client port to the list. 094 * 095 * @param clientPort the specified port 096 */ 097 public void addClientPort(int clientPort) { 098 clientPortList.add(clientPort); 099 } 100 101 /** 102 * Get the list of client ports. 103 * 104 * @return clientPortList the client port list 105 */ 106 @VisibleForTesting public List<Integer> getClientPortList() { 107 return clientPortList; 108 } 109 110 /** 111 * Check whether the client port in a specific position of the client port list is valid. 112 * 113 * @param index the specified position 114 */ 115 private boolean hasValidClientPortInList(int index) { 116 return (clientPortList.size() > index && clientPortList.get(index) > 0); 117 } 118 119 public void setDefaultClientPort(int clientPort) { 120 if (clientPort <= 0) { 121 throw new IllegalArgumentException("Invalid default ZK client port: " + clientPort); 122 } 123 this.defaultClientPort = clientPort; 124 } 125 126 /** 127 * Selects a ZK client port. 128 * 129 * @param seedPort the seed port to start with; -1 means first time. 130 * @return a valid and unused client port 131 */ 132 private int selectClientPort(int seedPort) { 133 int i; 134 int returnClientPort = seedPort + 1; 135 if (returnClientPort == 0) { 136 // If the new port is invalid, find one - starting with the default client port. 137 // If the default client port is not specified, starting with a random port. 138 // The random port is selected from the range between 49152 to 65535. These ports cannot be 139 // registered with IANA and are intended for dynamic allocation (see http://bit.ly/dynports). 140 if (defaultClientPort > 0) { 141 returnClientPort = defaultClientPort; 142 } else { 143 returnClientPort = 0xc000 + new Random().nextInt(0x3f00); 144 } 145 } 146 // Make sure that the port is unused. 147 // break when an unused port is found 148 do { 149 for (i = 0; i < clientPortList.size(); i++) { 150 if (returnClientPort == clientPortList.get(i)) { 151 // Already used. Update the port and retry. 152 returnClientPort++; 153 break; 154 } 155 } 156 } while (i != clientPortList.size()); 157 return returnClientPort; 158 } 159 160 public void setTickTime(int tickTime) { 161 this.tickTime = tickTime; 162 } 163 164 public int getBackupZooKeeperServerNum() { 165 return zooKeeperServers.size() - 1; 166 } 167 168 public int getZooKeeperServerNum() { 169 return zooKeeperServers.size(); 170 } 171 172 // / XXX: From o.a.zk.t.ClientBase 173 private static void setupTestEnv() { 174 // during the tests we run with 100K prealloc in the logs. 175 // on windows systems prealloc of 64M was seen to take ~15seconds 176 // resulting in test failure (client timeout on first session). 177 // set env and directly in order to handle static init/gc issues 178 System.setProperty("zookeeper.preAllocSize", "100"); 179 FileTxnLog.setPreallocSize(100 * 1024); 180 // allow all 4 letter words 181 System.setProperty("zookeeper.4lw.commands.whitelist", "*"); 182 } 183 184 public int startup(File baseDir) throws IOException, InterruptedException { 185 int numZooKeeperServers = clientPortList.size(); 186 if (numZooKeeperServers == 0) { 187 numZooKeeperServers = 1; // need at least 1 ZK server for testing 188 } 189 return startup(baseDir, numZooKeeperServers); 190 } 191 192 /** 193 * @param baseDir the base directory to use 194 * @param numZooKeeperServers the number of ZooKeeper servers 195 * @return ClientPort server bound to, -1 if there was a binding problem and we couldn't pick 196 * another port. 197 * @throws IOException if an operation fails during the startup 198 * @throws InterruptedException if the startup fails 199 */ 200 public int startup(File baseDir, int numZooKeeperServers) 201 throws IOException, InterruptedException { 202 if (numZooKeeperServers <= 0) { 203 return -1; 204 } 205 206 setupTestEnv(); 207 shutdown(); 208 209 int tentativePort = -1; // the seed port 210 int currentClientPort; 211 212 // running all the ZK servers 213 for (int i = 0; i < numZooKeeperServers; i++) { 214 File dir = new File(baseDir, "zookeeper_" + i).getAbsoluteFile(); 215 createDir(dir); 216 int tickTimeToUse; 217 if (this.tickTime > 0) { 218 tickTimeToUse = this.tickTime; 219 } else { 220 tickTimeToUse = TICK_TIME; 221 } 222 223 // Set up client port - if we have already had a list of valid ports, use it. 224 if (hasValidClientPortInList(i)) { 225 currentClientPort = clientPortList.get(i); 226 } else { 227 tentativePort = selectClientPort(tentativePort); // update the seed 228 currentClientPort = tentativePort; 229 } 230 231 ZooKeeperServer server = new ZooKeeperServer(dir, dir, tickTimeToUse); 232 // Setting {min,max}SessionTimeout defaults to be the same as in Zookeeper 233 server.setMinSessionTimeout(configuration.getInt("hbase.zookeeper.property.minSessionTimeout", 234 -1)); 235 server.setMaxSessionTimeout(configuration.getInt("hbase.zookeeper.property.maxSessionTimeout", 236 -1)); 237 NIOServerCnxnFactory standaloneServerFactory; 238 while (true) { 239 try { 240 standaloneServerFactory = new NIOServerCnxnFactory(); 241 standaloneServerFactory.configure(new InetSocketAddress(LOOPBACK_HOST, currentClientPort), 242 configuration.getInt(HConstants.ZOOKEEPER_MAX_CLIENT_CNXNS, 243 HConstants.DEFAULT_ZOOKEEPER_MAX_CLIENT_CNXNS)); 244 } catch (BindException e) { 245 LOG.debug("Failed binding ZK Server to client port: " + currentClientPort, e); 246 // We're told to use some port but it's occupied, fail 247 if (hasValidClientPortInList(i)) { 248 return -1; 249 } 250 // This port is already in use, try to use another. 251 tentativePort = selectClientPort(tentativePort); 252 currentClientPort = tentativePort; 253 continue; 254 } 255 break; 256 } 257 258 // Start up this ZK server. Dump its stats. 259 standaloneServerFactory.startup(server); 260 LOG.info("Started connectionTimeout={}, dir={}, {}", connectionTimeout, dir, 261 getServerConfigurationOnOneLine(server)); 262 // Runs a 'stat' against the servers. 263 if (!waitForServerUp(currentClientPort, connectionTimeout)) { 264 Threads.printThreadInfo(System.out, 265 "Why is zk standalone server not coming up?"); 266 throw new IOException("Waiting for startup of standalone server; " + 267 "server isRunning=" + server.isRunning()); 268 } 269 270 // We have selected a port as a client port. Update clientPortList if necessary. 271 if (clientPortList.size() <= i) { // it is not in the list, add the port 272 clientPortList.add(currentClientPort); 273 } else if (clientPortList.get(i) <= 0) { // the list has invalid port, update with valid port 274 clientPortList.remove(i); 275 clientPortList.add(i, currentClientPort); 276 } 277 278 standaloneServerFactoryList.add(standaloneServerFactory); 279 zooKeeperServers.add(server); 280 } 281 282 // set the first one to be active ZK; Others are backups 283 activeZKServerIndex = 0; 284 started = true; 285 int clientPort = clientPortList.get(activeZKServerIndex); 286 LOG.info("Started MiniZooKeeperCluster and ran 'stat' on client port={}", clientPort); 287 return clientPort; 288 } 289 290 private String getServerConfigurationOnOneLine(ZooKeeperServer server) { 291 StringWriter sw = new StringWriter(); 292 try (PrintWriter pw = new PrintWriter(sw) { 293 @Override public void println(int x) { 294 super.print(x); 295 super.print(", "); 296 } 297 298 @Override public void println(String x) { 299 super.print(x); 300 super.print(", "); 301 } 302 }) { 303 server.dumpConf(pw); 304 } 305 return sw.toString(); 306 } 307 308 private void createDir(File dir) throws IOException { 309 try { 310 if (!dir.exists()) { 311 dir.mkdirs(); 312 } 313 } catch (SecurityException e) { 314 throw new IOException("creating dir: " + dir, e); 315 } 316 } 317 318 /** 319 * @throws IOException if waiting for the shutdown of a server fails 320 */ 321 public void shutdown() throws IOException { 322 // shut down all the zk servers 323 for (int i = 0; i < standaloneServerFactoryList.size(); i++) { 324 NIOServerCnxnFactory standaloneServerFactory = standaloneServerFactoryList.get(i); 325 int clientPort = clientPortList.get(i); 326 standaloneServerFactory.shutdown(); 327 if (!waitForServerDown(clientPort, connectionTimeout)) { 328 throw new IOException("Waiting for shutdown of standalone server at port=" + clientPort + 329 ", timeout=" + this.connectionTimeout); 330 } 331 } 332 standaloneServerFactoryList.clear(); 333 334 for (ZooKeeperServer zkServer: zooKeeperServers) { 335 // Explicitly close ZKDatabase since ZookeeperServer does not close them 336 zkServer.getZKDatabase().close(); 337 } 338 zooKeeperServers.clear(); 339 340 // clear everything 341 if (started) { 342 started = false; 343 activeZKServerIndex = 0; 344 clientPortList.clear(); 345 LOG.info("Shutdown MiniZK cluster with all ZK servers"); 346 } 347 } 348 349 /** 350 * @return clientPort return clientPort if there is another ZK backup can run 351 * when killing the current active; return -1, if there is no backups. 352 * @throws IOException if waiting for the shutdown of a server fails 353 */ 354 public int killCurrentActiveZooKeeperServer() throws IOException, InterruptedException { 355 if (!started || activeZKServerIndex < 0) { 356 return -1; 357 } 358 359 // Shutdown the current active one 360 NIOServerCnxnFactory standaloneServerFactory = 361 standaloneServerFactoryList.get(activeZKServerIndex); 362 int clientPort = clientPortList.get(activeZKServerIndex); 363 364 standaloneServerFactory.shutdown(); 365 if (!waitForServerDown(clientPort, connectionTimeout)) { 366 throw new IOException("Waiting for shutdown of standalone server"); 367 } 368 369 zooKeeperServers.get(activeZKServerIndex).getZKDatabase().close(); 370 371 // remove the current active zk server 372 standaloneServerFactoryList.remove(activeZKServerIndex); 373 clientPortList.remove(activeZKServerIndex); 374 zooKeeperServers.remove(activeZKServerIndex); 375 LOG.info("Kill the current active ZK servers in the cluster on client port: {}", clientPort); 376 377 if (standaloneServerFactoryList.isEmpty()) { 378 // there is no backup servers; 379 return -1; 380 } 381 clientPort = clientPortList.get(activeZKServerIndex); 382 LOG.info("Activate a backup zk server in the cluster on client port: {}", clientPort); 383 // return the next back zk server's port 384 return clientPort; 385 } 386 387 /** 388 * Kill one back up ZK servers. 389 * 390 * @throws IOException if waiting for the shutdown of a server fails 391 */ 392 public void killOneBackupZooKeeperServer() throws IOException, InterruptedException { 393 if (!started || activeZKServerIndex < 0 || standaloneServerFactoryList.size() <= 1) { 394 return ; 395 } 396 397 int backupZKServerIndex = activeZKServerIndex+1; 398 // Shutdown the current active one 399 NIOServerCnxnFactory standaloneServerFactory = 400 standaloneServerFactoryList.get(backupZKServerIndex); 401 int clientPort = clientPortList.get(backupZKServerIndex); 402 403 standaloneServerFactory.shutdown(); 404 if (!waitForServerDown(clientPort, connectionTimeout)) { 405 throw new IOException("Waiting for shutdown of standalone server"); 406 } 407 408 zooKeeperServers.get(backupZKServerIndex).getZKDatabase().close(); 409 410 // remove this backup zk server 411 standaloneServerFactoryList.remove(backupZKServerIndex); 412 clientPortList.remove(backupZKServerIndex); 413 zooKeeperServers.remove(backupZKServerIndex); 414 LOG.info("Kill one backup ZK servers in the cluster on client port: {}", clientPort); 415 } 416 417 // XXX: From o.a.zk.t.ClientBase. We just dropped the check for ssl/secure. 418 private static boolean waitForServerDown(int port, long timeout) throws IOException { 419 long start = System.currentTimeMillis(); 420 while (true) { 421 try { 422 send4LetterWord(HOST, port, "stat", false, (int)timeout); 423 } catch (IOException | X509Exception.SSLContextException e) { 424 return true; 425 } 426 427 if (System.currentTimeMillis() > start + timeout) { 428 break; 429 } 430 try { 431 Thread.sleep(TIMEOUT); 432 } catch (InterruptedException e) { 433 throw (InterruptedIOException)new InterruptedIOException().initCause(e); 434 } 435 } 436 return false; 437 } 438 439 // XXX: From o.a.zk.t.ClientBase. Its in the test jar but we don't depend on zk test jar. 440 // We remove the SSL/secure bit. Not used in here. 441 private static boolean waitForServerUp(int port, long timeout) throws IOException { 442 long start = System.currentTimeMillis(); 443 while (true) { 444 try { 445 String result = send4LetterWord(HOST, port, "stat", false, (int)timeout); 446 if (result.startsWith("Zookeeper version:") && !result.contains("READ-ONLY")) { 447 return true; 448 } else { 449 LOG.debug("Read {}", result); 450 } 451 } catch (ConnectException e) { 452 // ignore as this is expected, do not log stacktrace 453 LOG.info("{}:{} not up: {}", HOST, port, e.toString()); 454 } catch (IOException | X509Exception.SSLContextException e) { 455 // ignore as this is expected 456 LOG.info("{}:{} not up", HOST, port, e); 457 } 458 459 if (System.currentTimeMillis() > start + timeout) { 460 break; 461 } 462 try { 463 Thread.sleep(TIMEOUT); 464 } catch (InterruptedException e) { 465 throw (InterruptedIOException)new InterruptedIOException().initCause(e); 466 } 467 } 468 return false; 469 } 470 471 public int getClientPort() { 472 return activeZKServerIndex < 0 || activeZKServerIndex >= clientPortList.size() ? -1 473 : clientPortList.get(activeZKServerIndex); 474 } 475 476 /** 477 * @return Address for this cluster instance. 478 */ 479 public Address getAddress() { 480 return Address.fromParts(HOST, getClientPort()); 481 } 482 483 List<ZooKeeperServer> getZooKeeperServers() { 484 return zooKeeperServers; 485 } 486}