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