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