1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.zookeeper;
20
21 import java.io.Closeable;
22 import java.io.IOException;
23 import java.util.ArrayList;
24 import java.util.HashMap;
25 import java.util.List;
26 import java.util.Map;
27 import java.util.concurrent.CopyOnWriteArrayList;
28 import java.util.concurrent.CountDownLatch;
29
30 import org.apache.commons.logging.Log;
31 import org.apache.commons.logging.LogFactory;
32 import org.apache.hadoop.hbase.classification.InterfaceAudience;
33 import org.apache.hadoop.conf.Configuration;
34 import org.apache.hadoop.hbase.Abortable;
35 import org.apache.hadoop.hbase.HConstants;
36 import org.apache.hadoop.hbase.HRegionInfo;
37 import org.apache.hadoop.hbase.ZooKeeperConnectionException;
38 import org.apache.hadoop.security.UserGroupInformation;
39 import org.apache.zookeeper.KeeperException;
40 import org.apache.zookeeper.WatchedEvent;
41 import org.apache.zookeeper.Watcher;
42 import org.apache.zookeeper.ZooDefs;
43 import org.apache.zookeeper.ZooDefs.Ids;
44 import org.apache.zookeeper.ZooDefs.Perms;
45 import org.apache.zookeeper.data.ACL;
46 import org.apache.zookeeper.data.Id;
47 import org.apache.zookeeper.data.Stat;
48
49
50
51
52
53
54
55
56
57
58
59
60 @InterfaceAudience.Private
61 public class ZooKeeperWatcher implements Watcher, Abortable, Closeable {
62 private static final Log LOG = LogFactory.getLog(ZooKeeperWatcher.class);
63
64
65
66 private String prefix;
67 private String identifier;
68
69
70 private String quorum;
71
72
73 private RecoverableZooKeeper recoverableZooKeeper;
74
75
76 protected Abortable abortable;
77
78 private boolean aborted = false;
79
80
81 private final List<ZooKeeperListener> listeners =
82 new CopyOnWriteArrayList<ZooKeeperListener>();
83
84
85
86 public CountDownLatch saslLatch = new CountDownLatch(1);
87
88
89
90
91 public String baseZNode;
92
93 private Map<Integer,String> metaReplicaZnodes = new HashMap<Integer, String>();
94
95 public String rsZNode;
96
97 public String drainingZNode;
98
99 private String masterAddressZNode;
100
101 public String backupMasterAddressesZNode;
102
103 public String clusterStateZNode;
104
105 public String assignmentZNode;
106
107 public String tableZNode;
108
109 public String clusterIdZNode;
110
111 public String splitLogZNode;
112
113 public String balancerZNode;
114
115 public String tableLockZNode;
116
117 public String recoveringRegionsZNode;
118
119 public static String namespaceZNode = "namespace";
120
121
122 public static final ArrayList<ACL> CREATOR_ALL_AND_WORLD_READABLE =
123 new ArrayList<ACL>() { {
124 add(new ACL(ZooDefs.Perms.READ,ZooDefs.Ids.ANYONE_ID_UNSAFE));
125 add(new ACL(ZooDefs.Perms.ALL,ZooDefs.Ids.AUTH_IDS));
126 }};
127
128 public final static String META_ZNODE_PREFIX = "meta-region-server";
129
130 private final Configuration conf;
131
132 private final Exception constructorCaller;
133
134
135
136
137
138
139
140
141 public ZooKeeperWatcher(Configuration conf, String identifier,
142 Abortable abortable) throws ZooKeeperConnectionException, IOException {
143 this(conf, identifier, abortable, false);
144 }
145
146
147
148
149
150
151
152
153
154
155
156
157 public ZooKeeperWatcher(Configuration conf, String identifier,
158 Abortable abortable, boolean canCreateBaseZNode)
159 throws IOException, ZooKeeperConnectionException {
160 this.conf = conf;
161
162
163 try {
164 throw new Exception("ZKW CONSTRUCTOR STACK TRACE FOR DEBUGGING");
165 } catch (Exception e) {
166 this.constructorCaller = e;
167 }
168 this.quorum = ZKConfig.getZKQuorumServersString(conf);
169 this.prefix = identifier;
170
171
172 this.identifier = identifier + "0x0";
173 this.abortable = abortable;
174 setNodeNames(conf);
175 this.recoverableZooKeeper = ZKUtil.connect(conf, quorum, this, identifier);
176 if (canCreateBaseZNode) {
177 createBaseZNodes();
178 }
179 }
180
181 private void createBaseZNodes() throws ZooKeeperConnectionException {
182 try {
183
184 ZKUtil.createWithParents(this, baseZNode);
185 if (conf.getBoolean("hbase.assignment.usezk", true)) {
186 ZKUtil.createAndFailSilent(this, assignmentZNode);
187 }
188 ZKUtil.createAndFailSilent(this, rsZNode);
189 ZKUtil.createAndFailSilent(this, drainingZNode);
190 ZKUtil.createAndFailSilent(this, tableZNode);
191 ZKUtil.createAndFailSilent(this, splitLogZNode);
192 ZKUtil.createAndFailSilent(this, backupMasterAddressesZNode);
193 ZKUtil.createAndFailSilent(this, tableLockZNode);
194 ZKUtil.createAndFailSilent(this, recoveringRegionsZNode);
195 } catch (KeeperException e) {
196 throw new ZooKeeperConnectionException(
197 prefix("Unexpected KeeperException creating base node"), e);
198 }
199 }
200
201
202
203 public boolean isClientReadable(String node) {
204
205
206
207 return
208 node.equals(baseZNode) ||
209 isAnyMetaReplicaZnode(node) ||
210 node.equals(getMasterAddressZNode()) ||
211 node.equals(clusterIdZNode)||
212 node.equals(rsZNode) ||
213
214 node.equals(tableZNode) ||
215 node.startsWith(tableZNode + "/");
216 }
217
218
219
220
221
222
223
224 public void checkAndSetZNodeAcls() {
225 if (!ZKUtil.isSecureZooKeeper(getConfiguration())) {
226 return;
227 }
228
229
230
231 try {
232 List<ACL> actualAcls = recoverableZooKeeper.getAcl(baseZNode, new Stat());
233
234 if (!isBaseZnodeAclSetup(actualAcls)) {
235 LOG.info("setting znode ACLs");
236 setZnodeAclsRecursive(baseZNode);
237 }
238 } catch(KeeperException.NoNodeException nne) {
239 return;
240 } catch(InterruptedException ie) {
241 interruptedException(ie);
242 } catch (IOException|KeeperException e) {
243 LOG.warn("Received exception while checking and setting zookeeper ACLs", e);
244 }
245 }
246
247
248
249
250
251
252 private void setZnodeAclsRecursive(String znode) throws KeeperException, InterruptedException {
253 List<String> children = recoverableZooKeeper.getChildren(znode, false);
254
255 for (String child : children) {
256 setZnodeAclsRecursive(ZKUtil.joinZNode(znode, child));
257 }
258 List<ACL> acls = ZKUtil.createACL(this, znode, true);
259 LOG.info("Setting ACLs for znode:" + znode + " , acl:" + acls);
260 recoverableZooKeeper.setAcl(znode, acls, -1);
261 }
262
263
264
265
266
267
268
269 private boolean isBaseZnodeAclSetup(List<ACL> acls) throws IOException {
270 String superUser = conf.get("hbase.superuser");
271
272
273
274 String hbaseUser = UserGroupInformation.getCurrentUser().getShortUserName();
275
276 if (acls.isEmpty()) {
277 return false;
278 }
279
280 for (ACL acl : acls) {
281 int perms = acl.getPerms();
282 Id id = acl.getId();
283
284
285 if (Ids.ANYONE_ID_UNSAFE.equals(id)) {
286 if (perms != Perms.READ) {
287 return false;
288 }
289 } else if (superUser != null && new Id("sasl", superUser).equals(id)) {
290 if (perms != Perms.ALL) {
291 return false;
292 }
293 } else if (new Id("sasl", hbaseUser).equals(id)) {
294 if (perms != Perms.ALL) {
295 return false;
296 }
297 } else {
298 return false;
299 }
300 }
301 return true;
302 }
303
304 @Override
305 public String toString() {
306 return this.identifier + ", quorum=" + quorum + ", baseZNode=" + baseZNode;
307 }
308
309
310
311
312
313
314
315 public String prefix(final String str) {
316 return this.toString() + " " + str;
317 }
318
319
320
321
322 private void setNodeNames(Configuration conf) {
323 baseZNode = conf.get(HConstants.ZOOKEEPER_ZNODE_PARENT,
324 HConstants.DEFAULT_ZOOKEEPER_ZNODE_PARENT);
325 metaReplicaZnodes.put(0, ZKUtil.joinZNode(baseZNode,
326 conf.get("zookeeper.znode.metaserver", "meta-region-server")));
327 int numMetaReplicas = conf.getInt(HConstants.META_REPLICAS_NUM,
328 HConstants.DEFAULT_META_REPLICA_NUM);
329 for (int i = 1; i < numMetaReplicas; i++) {
330 String str = ZKUtil.joinZNode(baseZNode,
331 conf.get("zookeeper.znode.metaserver", "meta-region-server") + "-" + i);
332 metaReplicaZnodes.put(i, str);
333 }
334 rsZNode = ZKUtil.joinZNode(baseZNode,
335 conf.get("zookeeper.znode.rs", "rs"));
336 drainingZNode = ZKUtil.joinZNode(baseZNode,
337 conf.get("zookeeper.znode.draining.rs", "draining"));
338 masterAddressZNode = ZKUtil.joinZNode(baseZNode,
339 conf.get("zookeeper.znode.master", "master"));
340 backupMasterAddressesZNode = ZKUtil.joinZNode(baseZNode,
341 conf.get("zookeeper.znode.backup.masters", "backup-masters"));
342 clusterStateZNode = ZKUtil.joinZNode(baseZNode,
343 conf.get("zookeeper.znode.state", "running"));
344 assignmentZNode = ZKUtil.joinZNode(baseZNode,
345 conf.get("zookeeper.znode.unassigned", "region-in-transition"));
346 tableZNode = ZKUtil.joinZNode(baseZNode,
347 conf.get("zookeeper.znode.tableEnableDisable", "table"));
348 clusterIdZNode = ZKUtil.joinZNode(baseZNode,
349 conf.get("zookeeper.znode.clusterId", "hbaseid"));
350 splitLogZNode = ZKUtil.joinZNode(baseZNode,
351 conf.get("zookeeper.znode.splitlog", HConstants.SPLIT_LOGDIR_NAME));
352 balancerZNode = ZKUtil.joinZNode(baseZNode,
353 conf.get("zookeeper.znode.balancer", "balancer"));
354 tableLockZNode = ZKUtil.joinZNode(baseZNode,
355 conf.get("zookeeper.znode.tableLock", "table-lock"));
356 recoveringRegionsZNode = ZKUtil.joinZNode(baseZNode,
357 conf.get("zookeeper.znode.recovering.regions", "recovering-regions"));
358 namespaceZNode = ZKUtil.joinZNode(baseZNode,
359 conf.get("zookeeper.znode.namespace", "namespace"));
360 }
361
362
363
364
365
366
367 public boolean isAnyMetaReplicaZnode(String node) {
368 if (metaReplicaZnodes.values().contains(node)) {
369 return true;
370 }
371 return false;
372 }
373
374
375
376
377
378
379 public boolean isDefaultMetaReplicaZnode(String node) {
380 if (getZNodeForReplica(HRegionInfo.DEFAULT_REPLICA_ID).equals(node)) {
381 return true;
382 }
383 return false;
384 }
385
386
387
388
389
390
391 public List<String> getMetaReplicaNodes() throws KeeperException {
392 List<String> childrenOfBaseNode = ZKUtil.listChildrenNoWatch(this, baseZNode);
393 List<String> metaReplicaNodes = new ArrayList<String>(2);
394 String pattern = conf.get("zookeeper.znode.metaserver","meta-region-server");
395 for (String child : childrenOfBaseNode) {
396 if (child.startsWith(pattern)) metaReplicaNodes.add(child);
397 }
398 return metaReplicaNodes;
399 }
400
401
402
403
404
405
406 public String getZNodeForReplica(int replicaId) {
407 String str = metaReplicaZnodes.get(replicaId);
408
409
410
411 if (str == null) {
412 str = ZKUtil.joinZNode(baseZNode,
413 conf.get("zookeeper.znode.metaserver", "meta-region-server") + "-" + replicaId);
414 }
415 return str;
416 }
417
418
419
420
421
422
423 public int getMetaReplicaIdFromZnode(String znode) {
424 String pattern = conf.get("zookeeper.znode.metaserver","meta-region-server");
425 if (znode.equals(pattern)) return HRegionInfo.DEFAULT_REPLICA_ID;
426
427 String nonDefaultPattern = pattern + "-";
428 return Integer.parseInt(znode.substring(nonDefaultPattern.length()));
429 }
430
431
432
433
434
435 public void registerListener(ZooKeeperListener listener) {
436 listeners.add(listener);
437 }
438
439
440
441
442
443
444 public void registerListenerFirst(ZooKeeperListener listener) {
445 listeners.add(0, listener);
446 }
447
448 public void unregisterListener(ZooKeeperListener listener) {
449 listeners.remove(listener);
450 }
451
452
453
454
455 public void unregisterAllListeners() {
456 listeners.clear();
457 }
458
459
460
461
462 public List<ZooKeeperListener> getListeners() {
463 return new ArrayList<ZooKeeperListener>(listeners);
464 }
465
466
467
468
469 public int getNumberOfListeners() {
470 return listeners.size();
471 }
472
473
474
475
476
477 public RecoverableZooKeeper getRecoverableZooKeeper() {
478 return recoverableZooKeeper;
479 }
480
481 public void reconnectAfterExpiration() throws IOException, KeeperException, InterruptedException {
482 recoverableZooKeeper.reconnectAfterExpiration();
483 }
484
485
486
487
488
489 public String getQuorum() {
490 return quorum;
491 }
492
493
494
495
496 public String getBaseZNode() {
497 return baseZNode;
498 }
499
500
501
502
503
504
505
506 @Override
507 public void process(WatchedEvent event) {
508 LOG.debug(prefix("Received ZooKeeper Event, " +
509 "type=" + event.getType() + ", " +
510 "state=" + event.getState() + ", " +
511 "path=" + event.getPath()));
512
513 switch(event.getType()) {
514
515
516 case None: {
517 connectionEvent(event);
518 break;
519 }
520
521
522
523 case NodeCreated: {
524 for(ZooKeeperListener listener : listeners) {
525 listener.nodeCreated(event.getPath());
526 }
527 break;
528 }
529
530 case NodeDeleted: {
531 for(ZooKeeperListener listener : listeners) {
532 listener.nodeDeleted(event.getPath());
533 }
534 break;
535 }
536
537 case NodeDataChanged: {
538 for(ZooKeeperListener listener : listeners) {
539 listener.nodeDataChanged(event.getPath());
540 }
541 break;
542 }
543
544 case NodeChildrenChanged: {
545 for(ZooKeeperListener listener : listeners) {
546 listener.nodeChildrenChanged(event.getPath());
547 }
548 break;
549 }
550 }
551 }
552
553
554
555
556
557
558
559
560
561
562
563
564
565 private void connectionEvent(WatchedEvent event) {
566 switch(event.getState()) {
567 case SyncConnected:
568
569
570 long finished = System.currentTimeMillis() +
571 this.conf.getLong("hbase.zookeeper.watcher.sync.connected.wait", 2000);
572 while (System.currentTimeMillis() < finished) {
573 try {
574 Thread.sleep(1);
575 } catch (InterruptedException e) {
576 LOG.warn("Interrupted while sleeping");
577 throw new RuntimeException("Interrupted while waiting for" +
578 " recoverableZooKeeper is set");
579 }
580 if (this.recoverableZooKeeper != null) break;
581 }
582
583 if (this.recoverableZooKeeper == null) {
584 LOG.error("ZK is null on connection event -- see stack trace " +
585 "for the stack trace when constructor was called on this zkw",
586 this.constructorCaller);
587 throw new NullPointerException("ZK is null");
588 }
589 this.identifier = this.prefix + "-0x" +
590 Long.toHexString(this.recoverableZooKeeper.getSessionId());
591
592 LOG.debug(this.identifier + " connected");
593 break;
594
595
596 case Disconnected:
597 LOG.debug(prefix("Received Disconnected from ZooKeeper, ignoring"));
598 break;
599
600 case Expired:
601 String msg = prefix(this.identifier + " received expired from " +
602 "ZooKeeper, aborting");
603
604
605 if (this.abortable != null) {
606 this.abortable.abort(msg, new KeeperException.SessionExpiredException());
607 }
608 break;
609
610 case ConnectedReadOnly:
611 case SaslAuthenticated:
612 case AuthFailed:
613 break;
614
615 default:
616 throw new IllegalStateException("Received event is not valid: " + event.getState());
617 }
618 }
619
620
621
622
623
624
625
626
627
628
629
630
631
632 public void sync(String path) throws KeeperException {
633 this.recoverableZooKeeper.sync(path, null, null);
634 }
635
636
637
638
639
640
641
642
643
644
645
646 public void keeperException(KeeperException ke)
647 throws KeeperException {
648 LOG.error(prefix("Received unexpected KeeperException, re-throwing exception"), ke);
649 throw ke;
650 }
651
652
653
654
655
656
657
658
659
660
661
662
663 public void interruptedException(InterruptedException ie) {
664 LOG.debug(prefix("Received InterruptedException, doing nothing here"), ie);
665
666 Thread.currentThread().interrupt();
667
668 }
669
670
671
672
673
674
675 @Override
676 public void close() {
677 try {
678 if (recoverableZooKeeper != null) {
679 recoverableZooKeeper.close();
680 }
681 } catch (InterruptedException e) {
682 Thread.currentThread().interrupt();
683 }
684 }
685
686 public Configuration getConfiguration() {
687 return conf;
688 }
689
690 @Override
691 public void abort(String why, Throwable e) {
692 if (this.abortable != null) this.abortable.abort(why, e);
693 else this.aborted = true;
694 }
695
696 @Override
697 public boolean isAborted() {
698 return this.abortable == null? this.aborted: this.abortable.isAborted();
699 }
700
701
702
703
704 public String getMasterAddressZNode() {
705 return this.masterAddressZNode;
706 }
707
708 }