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