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.BufferedReader;
22 import java.io.IOException;
23 import java.io.InputStreamReader;
24 import java.io.PrintWriter;
25 import java.net.InetSocketAddress;
26 import java.net.Socket;
27 import java.util.ArrayList;
28 import java.util.Arrays;
29 import java.util.Deque;
30 import java.util.HashMap;
31 import java.util.LinkedList;
32 import java.util.List;
33 import java.util.Map;
34
35 import javax.security.auth.login.AppConfigurationEntry;
36 import javax.security.auth.login.AppConfigurationEntry.LoginModuleControlFlag;
37
38 import org.apache.commons.lang.StringUtils;
39 import org.apache.commons.logging.Log;
40 import org.apache.commons.logging.LogFactory;
41 import org.apache.hadoop.conf.Configuration;
42 import org.apache.hadoop.hbase.AuthUtil;
43 import org.apache.hadoop.hbase.HBaseConfiguration;
44 import org.apache.hadoop.hbase.HConstants;
45 import org.apache.hadoop.hbase.ServerName;
46 import org.apache.hadoop.hbase.classification.InterfaceAudience;
47 import org.apache.hadoop.hbase.exceptions.DeserializationException;
48 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
49 import org.apache.hadoop.hbase.protobuf.generated.ClusterStatusProtos;
50 import org.apache.hadoop.hbase.protobuf.generated.ClusterStatusProtos.RegionStoreSequenceIds;
51 import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos;
52 import org.apache.hadoop.hbase.replication.ReplicationStateZKBase;
53 import org.apache.hadoop.hbase.security.Superusers;
54 import org.apache.hadoop.hbase.util.ByteStringer;
55 import org.apache.hadoop.hbase.util.Bytes;
56 import org.apache.hadoop.hbase.util.Threads;
57 import org.apache.hadoop.hbase.zookeeper.ZKUtil.ZKUtilOp.CreateAndFailSilent;
58 import org.apache.hadoop.hbase.zookeeper.ZKUtil.ZKUtilOp.DeleteNodeFailSilent;
59 import org.apache.hadoop.hbase.zookeeper.ZKUtil.ZKUtilOp.SetData;
60 import org.apache.hadoop.security.SecurityUtil;
61 import org.apache.hadoop.security.authentication.util.KerberosUtil;
62 import org.apache.zookeeper.AsyncCallback;
63 import org.apache.zookeeper.CreateMode;
64 import org.apache.zookeeper.KeeperException;
65 import org.apache.zookeeper.KeeperException.NoNodeException;
66 import org.apache.zookeeper.Op;
67 import org.apache.zookeeper.Watcher;
68 import org.apache.zookeeper.ZooDefs.Ids;
69 import org.apache.zookeeper.ZooDefs.Perms;
70 import org.apache.zookeeper.ZooKeeper;
71 import org.apache.zookeeper.client.ZooKeeperSaslClient;
72 import org.apache.zookeeper.data.ACL;
73 import org.apache.zookeeper.data.Id;
74 import org.apache.zookeeper.data.Stat;
75 import org.apache.zookeeper.proto.CreateRequest;
76 import org.apache.zookeeper.proto.DeleteRequest;
77 import org.apache.zookeeper.proto.SetDataRequest;
78 import org.apache.zookeeper.server.ZooKeeperSaslServer;
79
80 import com.google.protobuf.InvalidProtocolBufferException;
81
82
83
84
85
86
87
88
89
90
91 @InterfaceAudience.Private
92 public class ZKUtil {
93 private static final Log LOG = LogFactory.getLog(ZKUtil.class);
94
95
96 public static final char ZNODE_PATH_SEPARATOR = '/';
97 private static int zkDumpConnectionTimeOut;
98
99
100
101
102
103
104
105
106
107
108
109
110 public static RecoverableZooKeeper connect(Configuration conf, Watcher watcher)
111 throws IOException {
112 String ensemble = ZKConfig.getZKQuorumServersString(conf);
113 return connect(conf, ensemble, watcher);
114 }
115
116 public static RecoverableZooKeeper connect(Configuration conf, String ensemble,
117 Watcher watcher)
118 throws IOException {
119 return connect(conf, ensemble, watcher, null);
120 }
121
122 public static RecoverableZooKeeper connect(Configuration conf, String ensemble,
123 Watcher watcher, final String identifier)
124 throws IOException {
125 if(ensemble == null) {
126 throw new IOException("Unable to determine ZooKeeper ensemble");
127 }
128 int timeout = conf.getInt(HConstants.ZK_SESSION_TIMEOUT,
129 HConstants.DEFAULT_ZK_SESSION_TIMEOUT);
130 if (LOG.isTraceEnabled()) {
131 LOG.trace(identifier + " opening connection to ZooKeeper ensemble=" + ensemble);
132 }
133 int retry = conf.getInt("zookeeper.recovery.retry", 3);
134 int retryIntervalMillis =
135 conf.getInt("zookeeper.recovery.retry.intervalmill", 1000);
136 int maxSleepTime = conf.getInt("zookeeper.recovery.retry.maxsleeptime", 60000);
137 zkDumpConnectionTimeOut = conf.getInt("zookeeper.dump.connection.timeout",
138 1000);
139 return new RecoverableZooKeeper(ensemble, timeout, watcher,
140 retry, retryIntervalMillis, maxSleepTime, identifier);
141 }
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157 public static void loginServer(Configuration conf, String keytabFileKey,
158 String userNameKey, String hostname) throws IOException {
159 login(conf, keytabFileKey, userNameKey, hostname,
160 ZooKeeperSaslServer.LOGIN_CONTEXT_NAME_KEY,
161 JaasConfiguration.SERVER_KEYTAB_KERBEROS_CONFIG_NAME);
162 }
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178 public static void loginClient(Configuration conf, String keytabFileKey,
179 String userNameKey, String hostname) throws IOException {
180 login(conf, keytabFileKey, userNameKey, hostname,
181 ZooKeeperSaslClient.LOGIN_CONTEXT_NAME_KEY,
182 JaasConfiguration.CLIENT_KEYTAB_KERBEROS_CONFIG_NAME);
183 }
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201 private static void login(Configuration conf, String keytabFileKey,
202 String userNameKey, String hostname,
203 String loginContextProperty, String loginContextName)
204 throws IOException {
205 if (!isSecureZooKeeper(conf))
206 return;
207
208
209
210 if (System.getProperty("java.security.auth.login.config") != null)
211 return;
212
213
214 String keytabFilename = conf.get(keytabFileKey);
215 if (keytabFilename == null) {
216 LOG.warn("no keytab specified for: " + keytabFileKey);
217 return;
218 }
219
220 String principalConfig = conf.get(userNameKey, System.getProperty("user.name"));
221 String principalName = SecurityUtil.getServerPrincipal(principalConfig, hostname);
222
223
224
225
226 JaasConfiguration jaasConf = new JaasConfiguration(loginContextName,
227 principalName, keytabFilename);
228 javax.security.auth.login.Configuration.setConfiguration(jaasConf);
229 System.setProperty(loginContextProperty, loginContextName);
230 }
231
232
233
234
235 private static class JaasConfiguration extends javax.security.auth.login.Configuration {
236 private static final String SERVER_KEYTAB_KERBEROS_CONFIG_NAME =
237 "zookeeper-server-keytab-kerberos";
238 private static final String CLIENT_KEYTAB_KERBEROS_CONFIG_NAME =
239 "zookeeper-client-keytab-kerberos";
240
241 private static final Map<String, String> BASIC_JAAS_OPTIONS =
242 new HashMap<String,String>();
243 static {
244 String jaasEnvVar = System.getenv("HBASE_JAAS_DEBUG");
245 if (jaasEnvVar != null && "true".equalsIgnoreCase(jaasEnvVar)) {
246 BASIC_JAAS_OPTIONS.put("debug", "true");
247 }
248 }
249
250 private static final Map<String,String> KEYTAB_KERBEROS_OPTIONS =
251 new HashMap<String,String>();
252 static {
253 KEYTAB_KERBEROS_OPTIONS.put("doNotPrompt", "true");
254 KEYTAB_KERBEROS_OPTIONS.put("storeKey", "true");
255 KEYTAB_KERBEROS_OPTIONS.put("refreshKrb5Config", "true");
256 KEYTAB_KERBEROS_OPTIONS.putAll(BASIC_JAAS_OPTIONS);
257 }
258
259 private static final AppConfigurationEntry KEYTAB_KERBEROS_LOGIN =
260 new AppConfigurationEntry(KerberosUtil.getKrb5LoginModuleName(),
261 LoginModuleControlFlag.REQUIRED,
262 KEYTAB_KERBEROS_OPTIONS);
263
264 private static final AppConfigurationEntry[] KEYTAB_KERBEROS_CONF =
265 new AppConfigurationEntry[]{KEYTAB_KERBEROS_LOGIN};
266
267 private javax.security.auth.login.Configuration baseConfig;
268 private final String loginContextName;
269 private final boolean useTicketCache;
270 private final String keytabFile;
271 private final String principal;
272
273 public JaasConfiguration(String loginContextName, String principal) {
274 this(loginContextName, principal, null, true);
275 }
276
277 public JaasConfiguration(String loginContextName, String principal, String keytabFile) {
278 this(loginContextName, principal, keytabFile, keytabFile == null || keytabFile.length() == 0);
279 }
280
281 private JaasConfiguration(String loginContextName, String principal,
282 String keytabFile, boolean useTicketCache) {
283 try {
284 this.baseConfig = javax.security.auth.login.Configuration.getConfiguration();
285 } catch (SecurityException e) {
286 this.baseConfig = null;
287 }
288 this.loginContextName = loginContextName;
289 this.useTicketCache = useTicketCache;
290 this.keytabFile = keytabFile;
291 this.principal = principal;
292 LOG.info("JaasConfiguration loginContextName=" + loginContextName +
293 " principal=" + principal + " useTicketCache=" + useTicketCache +
294 " keytabFile=" + keytabFile);
295 }
296
297 @Override
298 public AppConfigurationEntry[] getAppConfigurationEntry(String appName) {
299 if (loginContextName.equals(appName)) {
300 if (!useTicketCache) {
301 KEYTAB_KERBEROS_OPTIONS.put("keyTab", keytabFile);
302 KEYTAB_KERBEROS_OPTIONS.put("useKeyTab", "true");
303 }
304 KEYTAB_KERBEROS_OPTIONS.put("principal", principal);
305 KEYTAB_KERBEROS_OPTIONS.put("useTicketCache", useTicketCache ? "true" : "false");
306 return KEYTAB_KERBEROS_CONF;
307 }
308 if (baseConfig != null) return baseConfig.getAppConfigurationEntry(appName);
309 return(null);
310 }
311 }
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327 public static String joinZNode(String prefix, String suffix) {
328 return prefix + ZNODE_PATH_SEPARATOR + suffix;
329 }
330
331
332
333
334
335
336 public static String getParent(String node) {
337 int idx = node.lastIndexOf(ZNODE_PATH_SEPARATOR);
338 return idx <= 0 ? null : node.substring(0, idx);
339 }
340
341
342
343
344
345
346 public static String getNodeName(String path) {
347 return path.substring(path.lastIndexOf("/")+1);
348 }
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364 public static boolean watchAndCheckExists(ZooKeeperWatcher zkw, String znode)
365 throws KeeperException {
366 try {
367 Stat s = zkw.getRecoverableZooKeeper().exists(znode, zkw);
368 boolean exists = s != null ? true : false;
369 if (exists) {
370 LOG.debug(zkw.prefix("Set watcher on existing znode=" + znode));
371 } else {
372 LOG.debug(zkw.prefix("Set watcher on znode that does not yet exist, " + znode));
373 }
374 return exists;
375 } catch (KeeperException e) {
376 LOG.warn(zkw.prefix("Unable to set watcher on znode " + znode), e);
377 zkw.keeperException(e);
378 return false;
379 } catch (InterruptedException e) {
380 LOG.warn(zkw.prefix("Unable to set watcher on znode " + znode), e);
381 zkw.interruptedException(e);
382 return false;
383 }
384 }
385
386
387
388
389
390
391
392
393
394
395
396 public static boolean setWatchIfNodeExists(ZooKeeperWatcher zkw, String znode)
397 throws KeeperException {
398 try {
399 zkw.getRecoverableZooKeeper().getData(znode, true, null);
400 return true;
401 } catch (NoNodeException e) {
402 return false;
403 } catch (InterruptedException e) {
404 LOG.warn(zkw.prefix("Unable to set watcher on znode " + znode), e);
405 zkw.interruptedException(e);
406 return false;
407 }
408 }
409
410
411
412
413
414
415
416
417
418 public static int checkExists(ZooKeeperWatcher zkw, String znode)
419 throws KeeperException {
420 try {
421 Stat s = zkw.getRecoverableZooKeeper().exists(znode, null);
422 return s != null ? s.getVersion() : -1;
423 } catch (KeeperException e) {
424 LOG.warn(zkw.prefix("Unable to set watcher on znode (" + znode + ")"), e);
425 zkw.keeperException(e);
426 return -1;
427 } catch (InterruptedException e) {
428 LOG.warn(zkw.prefix("Unable to set watcher on znode (" + znode + ")"), e);
429 zkw.interruptedException(e);
430 return -1;
431 }
432 }
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454 public static List<String> listChildrenAndWatchForNewChildren(
455 ZooKeeperWatcher zkw, String znode)
456 throws KeeperException {
457 try {
458 List<String> children = zkw.getRecoverableZooKeeper().getChildren(znode, zkw);
459 return children;
460 } catch(KeeperException.NoNodeException ke) {
461 LOG.debug(zkw.prefix("Unable to list children of znode " + znode + " " +
462 "because node does not exist (not an error)"));
463 return null;
464 } catch (KeeperException e) {
465 LOG.warn(zkw.prefix("Unable to list children of znode " + znode + " "), e);
466 zkw.keeperException(e);
467 return null;
468 } catch (InterruptedException e) {
469 LOG.warn(zkw.prefix("Unable to list children of znode " + znode + " "), e);
470 zkw.interruptedException(e);
471 return null;
472 }
473 }
474
475
476
477
478
479
480
481
482
483
484 public static List<String> listChildrenAndWatchThem(ZooKeeperWatcher zkw,
485 String znode) throws KeeperException {
486 List<String> children = listChildrenAndWatchForNewChildren(zkw, znode);
487 if (children == null) {
488 return null;
489 }
490 for (String child : children) {
491 watchAndCheckExists(zkw, joinZNode(znode, child));
492 }
493 return children;
494 }
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510 public static List<String> listChildrenNoWatch(ZooKeeperWatcher zkw, String znode)
511 throws KeeperException {
512 List<String> children = null;
513 try {
514
515 children = zkw.getRecoverableZooKeeper().getChildren(znode, null);
516 } catch(KeeperException.NoNodeException nne) {
517 return null;
518 } catch(InterruptedException ie) {
519 zkw.interruptedException(ie);
520 }
521 return children;
522 }
523
524
525
526
527
528 @Deprecated
529 public static class NodeAndData {
530 private String node;
531 private byte [] data;
532 public NodeAndData(String node, byte [] data) {
533 this.node = node;
534 this.data = data;
535 }
536 public String getNode() {
537 return node;
538 }
539 public byte [] getData() {
540 return data;
541 }
542 @Override
543 public String toString() {
544 return node;
545 }
546 public boolean isEmpty() {
547 return (data == null || data.length == 0);
548 }
549 }
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567 public static boolean nodeHasChildren(ZooKeeperWatcher zkw, String znode)
568 throws KeeperException {
569 try {
570 return !zkw.getRecoverableZooKeeper().getChildren(znode, null).isEmpty();
571 } catch(KeeperException.NoNodeException ke) {
572 LOG.debug(zkw.prefix("Unable to list children of znode " + znode + " " +
573 "because node does not exist (not an error)"));
574 return false;
575 } catch (KeeperException e) {
576 LOG.warn(zkw.prefix("Unable to list children of znode " + znode), e);
577 zkw.keeperException(e);
578 return false;
579 } catch (InterruptedException e) {
580 LOG.warn(zkw.prefix("Unable to list children of znode " + znode), e);
581 zkw.interruptedException(e);
582 return false;
583 }
584 }
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599 public static int getNumberOfChildren(ZooKeeperWatcher zkw, String znode)
600 throws KeeperException {
601 try {
602 Stat stat = zkw.getRecoverableZooKeeper().exists(znode, null);
603 return stat == null ? 0 : stat.getNumChildren();
604 } catch(KeeperException e) {
605 LOG.warn(zkw.prefix("Unable to get children of node " + znode));
606 zkw.keeperException(e);
607 } catch(InterruptedException e) {
608 zkw.interruptedException(e);
609 }
610 return 0;
611 }
612
613
614
615
616
617
618
619
620
621
622 public static byte [] getData(ZooKeeperWatcher zkw, String znode)
623 throws KeeperException, InterruptedException {
624 try {
625 byte [] data = zkw.getRecoverableZooKeeper().getData(znode, null, null);
626 logRetrievedMsg(zkw, znode, data, false);
627 return data;
628 } catch (KeeperException.NoNodeException e) {
629 LOG.debug(zkw.prefix("Unable to get data of znode " + znode + " " +
630 "because node does not exist (not an error)"));
631 return null;
632 } catch (KeeperException e) {
633 LOG.warn(zkw.prefix("Unable to get data of znode " + znode), e);
634 zkw.keeperException(e);
635 return null;
636 }
637 }
638
639
640
641
642
643
644
645
646
647
648
649
650 public static byte [] getDataAndWatch(ZooKeeperWatcher zkw, String znode)
651 throws KeeperException {
652 return getDataInternal(zkw, znode, null, true);
653 }
654
655
656
657
658
659
660
661
662
663
664
665
666
667 public static byte[] getDataAndWatch(ZooKeeperWatcher zkw, String znode,
668 Stat stat) throws KeeperException {
669 return getDataInternal(zkw, znode, stat, true);
670 }
671
672 private static byte[] getDataInternal(ZooKeeperWatcher zkw, String znode, Stat stat,
673 boolean watcherSet)
674 throws KeeperException {
675 try {
676 byte [] data = zkw.getRecoverableZooKeeper().getData(znode, zkw, stat);
677 logRetrievedMsg(zkw, znode, data, watcherSet);
678 return data;
679 } catch (KeeperException.NoNodeException e) {
680
681
682 LOG.trace(zkw.prefix("Unable to get data of znode " + znode + " " +
683 "because node does not exist (not an error)"));
684 return null;
685 } catch (KeeperException e) {
686 LOG.warn(zkw.prefix("Unable to get data of znode " + znode), e);
687 zkw.keeperException(e);
688 return null;
689 } catch (InterruptedException e) {
690 LOG.warn(zkw.prefix("Unable to get data of znode " + znode), e);
691 zkw.interruptedException(e);
692 return null;
693 }
694 }
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711 public static byte [] getDataNoWatch(ZooKeeperWatcher zkw, String znode,
712 Stat stat)
713 throws KeeperException {
714 try {
715 byte [] data = zkw.getRecoverableZooKeeper().getData(znode, null, stat);
716 logRetrievedMsg(zkw, znode, data, false);
717 return data;
718 } catch (KeeperException.NoNodeException e) {
719 LOG.debug(zkw.prefix("Unable to get data of znode " + znode + " " +
720 "because node does not exist (not necessarily an error)"));
721 return null;
722 } catch (KeeperException e) {
723 LOG.warn(zkw.prefix("Unable to get data of znode " + znode), e);
724 zkw.keeperException(e);
725 return null;
726 } catch (InterruptedException e) {
727 LOG.warn(zkw.prefix("Unable to get data of znode " + znode), e);
728 zkw.interruptedException(e);
729 return null;
730 }
731 }
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750 public static List<NodeAndData> getChildDataAndWatchForNewChildren(
751 ZooKeeperWatcher zkw, String baseNode) throws KeeperException {
752 List<String> nodes =
753 ZKUtil.listChildrenAndWatchForNewChildren(zkw, baseNode);
754 if (nodes != null) {
755 List<NodeAndData> newNodes = new ArrayList<NodeAndData>();
756 for (String node : nodes) {
757 String nodePath = ZKUtil.joinZNode(baseNode, node);
758 byte[] data = ZKUtil.getDataAndWatch(zkw, nodePath);
759 newNodes.add(new NodeAndData(nodePath, data));
760 }
761 return newNodes;
762 }
763 return null;
764 }
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782 public static void updateExistingNodeData(ZooKeeperWatcher zkw, String znode,
783 byte [] data, int expectedVersion)
784 throws KeeperException {
785 try {
786 zkw.getRecoverableZooKeeper().setData(znode, data, expectedVersion);
787 } catch(InterruptedException ie) {
788 zkw.interruptedException(ie);
789 }
790 }
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816 public static boolean setData(ZooKeeperWatcher zkw, String znode,
817 byte [] data, int expectedVersion)
818 throws KeeperException, KeeperException.NoNodeException {
819 try {
820 return zkw.getRecoverableZooKeeper().setData(znode, data, expectedVersion) != null;
821 } catch (InterruptedException e) {
822 zkw.interruptedException(e);
823 return false;
824 }
825 }
826
827
828
829
830
831
832
833
834
835
836 public static void createSetData(final ZooKeeperWatcher zkw, final String znode,
837 final byte [] data)
838 throws KeeperException {
839 if (checkExists(zkw, znode) == -1) {
840 ZKUtil.createWithParents(zkw, znode, data);
841 } else {
842 ZKUtil.setData(zkw, znode, data);
843 }
844 }
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862 public static void setData(ZooKeeperWatcher zkw, String znode, byte [] data)
863 throws KeeperException, KeeperException.NoNodeException {
864 setData(zkw, (SetData)ZKUtilOp.setData(znode, data));
865 }
866
867 private static void setData(ZooKeeperWatcher zkw, SetData setData)
868 throws KeeperException, KeeperException.NoNodeException {
869 SetDataRequest sd = (SetDataRequest)toZooKeeperOp(zkw, setData).toRequestRecord();
870 setData(zkw, sd.getPath(), sd.getData(), sd.getVersion());
871 }
872
873
874
875
876
877
878 public static boolean isSecureZooKeeper(Configuration conf) {
879
880
881 try {
882 javax.security.auth.login.Configuration testConfig =
883 javax.security.auth.login.Configuration.getConfiguration();
884 if (testConfig.getAppConfigurationEntry("Client") == null
885 && testConfig.getAppConfigurationEntry(
886 JaasConfiguration.CLIENT_KEYTAB_KERBEROS_CONFIG_NAME) == null
887 && testConfig.getAppConfigurationEntry(
888 JaasConfiguration.SERVER_KEYTAB_KERBEROS_CONFIG_NAME) == null
889 && conf.get(HConstants.ZK_CLIENT_KERBEROS_PRINCIPAL) == null
890 && conf.get(HConstants.ZK_SERVER_KERBEROS_PRINCIPAL) == null) {
891
892 return false;
893 }
894 } catch(Exception e) {
895
896 return false;
897 }
898
899
900 return "kerberos".equalsIgnoreCase(conf.get("hbase.security.authentication"));
901 }
902
903 private static ArrayList<ACL> createACL(ZooKeeperWatcher zkw, String node) {
904 return createACL(zkw, node, isSecureZooKeeper(zkw.getConfiguration()));
905 }
906
907 public static ArrayList<ACL> createACL(ZooKeeperWatcher zkw, String node,
908 boolean isSecureZooKeeper) {
909 if (!node.startsWith(zkw.baseZNode)) {
910 return Ids.OPEN_ACL_UNSAFE;
911 }
912 if (isSecureZooKeeper) {
913 ArrayList<ACL> acls = new ArrayList<ACL>();
914
915 String[] superUsers = zkw.getConfiguration().getStrings(Superusers.SUPERUSER_CONF_KEY);
916 if (superUsers != null) {
917 List<String> groups = new ArrayList<String>();
918 for (String user : superUsers) {
919 if (AuthUtil.isGroupPrincipal(user)) {
920
921 groups.add(user);
922 } else {
923 acls.add(new ACL(Perms.ALL, new Id("sasl", user)));
924 }
925 }
926 if (!groups.isEmpty()) {
927 LOG.warn("Znode ACL setting for group " + groups
928 + " is skipped, Zookeeper doesn't support this feature presently.");
929 }
930 }
931
932
933 if (zkw.isClientReadable(node)) {
934 acls.addAll(Ids.CREATOR_ALL_ACL);
935 acls.addAll(Ids.READ_ACL_UNSAFE);
936 } else {
937 acls.addAll(Ids.CREATOR_ALL_ACL);
938 }
939 return acls;
940 } else {
941 return Ids.OPEN_ACL_UNSAFE;
942 }
943 }
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967 public static boolean createEphemeralNodeAndWatch(ZooKeeperWatcher zkw,
968 String znode, byte [] data)
969 throws KeeperException {
970 boolean ret = true;
971 try {
972 zkw.getRecoverableZooKeeper().create(znode, data, createACL(zkw, znode),
973 CreateMode.EPHEMERAL);
974 } catch (KeeperException.NodeExistsException nee) {
975 ret = false;
976 } catch (InterruptedException e) {
977 LOG.info("Interrupted", e);
978 Thread.currentThread().interrupt();
979 }
980 if(!watchAndCheckExists(zkw, znode)) {
981
982 return createEphemeralNodeAndWatch(zkw, znode, data);
983 }
984 return ret;
985 }
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
1001
1002
1003
1004
1005
1006
1007 public static boolean createNodeIfNotExistsAndWatch(
1008 ZooKeeperWatcher zkw, String znode, byte [] data)
1009 throws KeeperException {
1010 boolean ret = true;
1011 try {
1012 zkw.getRecoverableZooKeeper().create(znode, data, createACL(zkw, znode),
1013 CreateMode.PERSISTENT);
1014 } catch (KeeperException.NodeExistsException nee) {
1015 ret = false;
1016 } catch (InterruptedException e) {
1017 zkw.interruptedException(e);
1018 return false;
1019 }
1020 try {
1021 zkw.getRecoverableZooKeeper().exists(znode, zkw);
1022 } catch (InterruptedException e) {
1023 zkw.interruptedException(e);
1024 return false;
1025 }
1026 return ret;
1027 }
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043 public static String createNodeIfNotExistsNoWatch(ZooKeeperWatcher zkw, String znode,
1044 byte[] data, CreateMode createMode) throws KeeperException {
1045
1046 String createdZNode = null;
1047 try {
1048 createdZNode = zkw.getRecoverableZooKeeper().create(znode, data,
1049 createACL(zkw, znode), createMode);
1050 } catch (KeeperException.NodeExistsException nee) {
1051 return znode;
1052 } catch (InterruptedException e) {
1053 zkw.interruptedException(e);
1054 return null;
1055 }
1056 return createdZNode;
1057 }
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075 public static int createAndWatch(ZooKeeperWatcher zkw,
1076 String znode, byte [] data)
1077 throws KeeperException, KeeperException.NodeExistsException {
1078 try {
1079 zkw.getRecoverableZooKeeper().create(znode, data, createACL(zkw, znode),
1080 CreateMode.PERSISTENT);
1081 Stat stat = zkw.getRecoverableZooKeeper().exists(znode, zkw);
1082 if (stat == null){
1083
1084 throw KeeperException.create(KeeperException.Code.SYSTEMERROR,
1085 "ZK.exists returned null (i.e.: znode does not exist) for znode=" + znode);
1086 }
1087 return stat.getVersion();
1088 } catch (InterruptedException e) {
1089 zkw.interruptedException(e);
1090 return -1;
1091 }
1092 }
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107 public static void asyncCreate(ZooKeeperWatcher zkw,
1108 String znode, byte [] data, final AsyncCallback.StringCallback cb,
1109 final Object ctx) {
1110 zkw.getRecoverableZooKeeper().getZooKeeper().create(znode, data,
1111 createACL(zkw, znode), CreateMode.PERSISTENT, cb, ctx);
1112 }
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124 public static void createAndFailSilent(ZooKeeperWatcher zkw,
1125 String znode) throws KeeperException {
1126 createAndFailSilent(zkw, znode, new byte[0]);
1127 }
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140 public static void createAndFailSilent(ZooKeeperWatcher zkw,
1141 String znode, byte[] data)
1142 throws KeeperException {
1143 createAndFailSilent(zkw,
1144 (CreateAndFailSilent)ZKUtilOp.createAndFailSilent(znode, data));
1145 }
1146
1147 private static void createAndFailSilent(ZooKeeperWatcher zkw, CreateAndFailSilent cafs)
1148 throws KeeperException {
1149 CreateRequest create = (CreateRequest)toZooKeeperOp(zkw, cafs).toRequestRecord();
1150 String znode = create.getPath();
1151 try {
1152 RecoverableZooKeeper zk = zkw.getRecoverableZooKeeper();
1153 if (zk.exists(znode, false) == null) {
1154 zk.create(znode, create.getData(), create.getAcl(), CreateMode.fromFlag(create.getFlags()));
1155 }
1156 } catch(KeeperException.NodeExistsException nee) {
1157 } catch(KeeperException.NoAuthException nee){
1158 try {
1159 if (null == zkw.getRecoverableZooKeeper().exists(znode, false)) {
1160
1161 throw(nee);
1162 }
1163 } catch (InterruptedException ie) {
1164 zkw.interruptedException(ie);
1165 }
1166 } catch(InterruptedException ie) {
1167 zkw.interruptedException(ie);
1168 }
1169 }
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182 public static void createWithParents(ZooKeeperWatcher zkw, String znode)
1183 throws KeeperException {
1184 createWithParents(zkw, znode, new byte[0]);
1185 }
1186
1187
1188
1189
1190
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200 public static void createWithParents(ZooKeeperWatcher zkw, String znode, byte[] data)
1201 throws KeeperException {
1202 try {
1203 if(znode == null) {
1204 return;
1205 }
1206 zkw.getRecoverableZooKeeper().create(znode, data, createACL(zkw, znode),
1207 CreateMode.PERSISTENT);
1208 } catch(KeeperException.NodeExistsException nee) {
1209 return;
1210 } catch(KeeperException.NoNodeException nne) {
1211 createWithParents(zkw, getParent(znode));
1212 createWithParents(zkw, znode, data);
1213 } catch(InterruptedException ie) {
1214 zkw.interruptedException(ie);
1215 }
1216 }
1217
1218
1219
1220
1221
1222
1223
1224
1225 public static void deleteNode(ZooKeeperWatcher zkw, String node)
1226 throws KeeperException {
1227 deleteNode(zkw, node, -1);
1228 }
1229
1230
1231
1232
1233
1234 public static boolean deleteNode(ZooKeeperWatcher zkw, String node,
1235 int version)
1236 throws KeeperException {
1237 try {
1238 zkw.getRecoverableZooKeeper().delete(node, version);
1239 return true;
1240 } catch(KeeperException.BadVersionException bve) {
1241 return false;
1242 } catch(InterruptedException ie) {
1243 zkw.interruptedException(ie);
1244 return false;
1245 }
1246 }
1247
1248
1249
1250
1251
1252
1253
1254 public static void deleteNodeFailSilent(ZooKeeperWatcher zkw, String node)
1255 throws KeeperException {
1256 deleteNodeFailSilent(zkw,
1257 (DeleteNodeFailSilent)ZKUtilOp.deleteNodeFailSilent(node));
1258 }
1259
1260 private static void deleteNodeFailSilent(ZooKeeperWatcher zkw,
1261 DeleteNodeFailSilent dnfs) throws KeeperException {
1262 DeleteRequest delete = (DeleteRequest)toZooKeeperOp(zkw, dnfs).toRequestRecord();
1263 try {
1264 zkw.getRecoverableZooKeeper().delete(delete.getPath(), delete.getVersion());
1265 } catch(KeeperException.NoNodeException nne) {
1266 } catch(InterruptedException ie) {
1267 zkw.interruptedException(ie);
1268 }
1269 }
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280 public static void deleteNodeRecursively(ZooKeeperWatcher zkw, String node)
1281 throws KeeperException {
1282 deleteNodeRecursivelyMultiOrSequential(zkw, true, node);
1283 }
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296 public static void deleteChildrenRecursively(ZooKeeperWatcher zkw, String node)
1297 throws KeeperException {
1298 deleteChildrenRecursivelyMultiOrSequential(zkw, true, node);
1299 }
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
1324
1325
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335 public static void deleteChildrenRecursivelyMultiOrSequential(
1336 ZooKeeperWatcher zkw, boolean runSequentialOnMultiFailure,
1337 String... pathRoots) throws KeeperException {
1338 if (pathRoots == null || pathRoots.length <= 0) {
1339 LOG.warn("Given path is not valid!");
1340 return;
1341 }
1342 List<ZKUtilOp> ops = new ArrayList<ZKUtil.ZKUtilOp>();
1343 for (String eachRoot : pathRoots) {
1344 List<String> children = listChildrenBFSNoWatch(zkw, eachRoot);
1345
1346 for (int i = children.size() - 1; i >= 0; --i) {
1347 ops.add(ZKUtilOp.deleteNodeFailSilent(children.get(i)));
1348 }
1349 }
1350
1351 if (ops.size() > 0) {
1352 multiOrSequential(zkw, ops, runSequentialOnMultiFailure);
1353 }
1354 }
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388
1389
1390 public static void deleteNodeRecursivelyMultiOrSequential(ZooKeeperWatcher zkw,
1391 boolean runSequentialOnMultiFailure, String... pathRoots) throws KeeperException {
1392 if (pathRoots == null || pathRoots.length <= 0) {
1393 LOG.warn("Given path is not valid!");
1394 return;
1395 }
1396 List<ZKUtilOp> ops = new ArrayList<ZKUtil.ZKUtilOp>();
1397 for (String eachRoot : pathRoots) {
1398
1399
1400 List<String> children = listChildrenBFSAndWatchThem(zkw, eachRoot);
1401
1402 for (int i = children.size() - 1; i >= 0; --i) {
1403 ops.add(ZKUtilOp.deleteNodeFailSilent(children.get(i)));
1404 }
1405 try {
1406 if (zkw.getRecoverableZooKeeper().exists(eachRoot, zkw) != null) {
1407 ops.add(ZKUtilOp.deleteNodeFailSilent(eachRoot));
1408 }
1409 } catch (InterruptedException e) {
1410 zkw.interruptedException(e);
1411 }
1412 }
1413
1414 if (ops.size() > 0) {
1415 multiOrSequential(zkw, ops, runSequentialOnMultiFailure);
1416 }
1417 }
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432 private static List<String> listChildrenBFSNoWatch(ZooKeeperWatcher zkw,
1433 final String znode) throws KeeperException {
1434 Deque<String> queue = new LinkedList<String>();
1435 List<String> tree = new ArrayList<String>();
1436 queue.add(znode);
1437 while (true) {
1438 String node = queue.pollFirst();
1439 if (node == null) {
1440 break;
1441 }
1442 List<String> children = listChildrenNoWatch(zkw, node);
1443 if (children == null) {
1444 continue;
1445 }
1446 for (final String child : children) {
1447 final String childPath = node + "/" + child;
1448 queue.add(childPath);
1449 tree.add(childPath);
1450 }
1451 }
1452 return tree;
1453 }
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468 private static List<String> listChildrenBFSAndWatchThem(ZooKeeperWatcher zkw, final String znode)
1469 throws KeeperException {
1470 Deque<String> queue = new LinkedList<String>();
1471 List<String> tree = new ArrayList<String>();
1472 queue.add(znode);
1473 while (true) {
1474 String node = queue.pollFirst();
1475 if (node == null) {
1476 break;
1477 }
1478 List<String> children = listChildrenAndWatchThem(zkw, node);
1479 if (children == null) {
1480 continue;
1481 }
1482 for (final String child : children) {
1483 final String childPath = node + "/" + child;
1484 queue.add(childPath);
1485 tree.add(childPath);
1486 }
1487 }
1488 return tree;
1489 }
1490
1491
1492
1493
1494
1495
1496 public abstract static class ZKUtilOp {
1497 private String path;
1498
1499 private ZKUtilOp(String path) {
1500 this.path = path;
1501 }
1502
1503
1504
1505
1506 public static ZKUtilOp createAndFailSilent(String path, byte[] data) {
1507 return new CreateAndFailSilent(path, data);
1508 }
1509
1510
1511
1512
1513 public static ZKUtilOp deleteNodeFailSilent(String path) {
1514 return new DeleteNodeFailSilent(path);
1515 }
1516
1517
1518
1519
1520 public static ZKUtilOp setData(String path, byte [] data) {
1521 return new SetData(path, data);
1522 }
1523
1524
1525
1526
1527 public String getPath() {
1528 return path;
1529 }
1530
1531
1532
1533
1534
1535 public static class CreateAndFailSilent extends ZKUtilOp {
1536 private byte [] data;
1537
1538 private CreateAndFailSilent(String path, byte [] data) {
1539 super(path);
1540 this.data = data;
1541 }
1542
1543 public byte[] getData() {
1544 return data;
1545 }
1546
1547 @Override
1548 public boolean equals(Object o) {
1549 if (this == o) return true;
1550 if (!(o instanceof CreateAndFailSilent)) return false;
1551
1552 CreateAndFailSilent op = (CreateAndFailSilent) o;
1553 return getPath().equals(op.getPath()) && Arrays.equals(data, op.data);
1554 }
1555
1556 @Override
1557 public int hashCode() {
1558 int ret = 17 + getPath().hashCode() * 31;
1559 return ret * 31 + Bytes.hashCode(data);
1560 }
1561 }
1562
1563
1564
1565
1566
1567 public static class DeleteNodeFailSilent extends ZKUtilOp {
1568 private DeleteNodeFailSilent(String path) {
1569 super(path);
1570 }
1571
1572 @Override
1573 public boolean equals(Object o) {
1574 if (this == o) return true;
1575 if (!(o instanceof DeleteNodeFailSilent)) return false;
1576
1577 return super.equals(o);
1578 }
1579
1580 @Override
1581 public int hashCode() {
1582 return getPath().hashCode();
1583 }
1584 }
1585
1586
1587
1588
1589 public static class SetData extends ZKUtilOp {
1590 private byte [] data;
1591
1592 private SetData(String path, byte [] data) {
1593 super(path);
1594 this.data = data;
1595 }
1596
1597 public byte[] getData() {
1598 return data;
1599 }
1600
1601 @Override
1602 public boolean equals(Object o) {
1603 if (this == o) return true;
1604 if (!(o instanceof SetData)) return false;
1605
1606 SetData op = (SetData) o;
1607 return getPath().equals(op.getPath()) && Arrays.equals(data, op.data);
1608 }
1609
1610 @Override
1611 public int hashCode() {
1612 int ret = getPath().hashCode();
1613 return ret * 31 + Bytes.hashCode(data);
1614 }
1615 }
1616 }
1617
1618
1619
1620
1621 private static Op toZooKeeperOp(ZooKeeperWatcher zkw, ZKUtilOp op)
1622 throws UnsupportedOperationException {
1623 if(op == null) return null;
1624
1625 if (op instanceof CreateAndFailSilent) {
1626 CreateAndFailSilent cafs = (CreateAndFailSilent)op;
1627 return Op.create(cafs.getPath(), cafs.getData(), createACL(zkw, cafs.getPath()),
1628 CreateMode.PERSISTENT);
1629 } else if (op instanceof DeleteNodeFailSilent) {
1630 DeleteNodeFailSilent dnfs = (DeleteNodeFailSilent)op;
1631 return Op.delete(dnfs.getPath(), -1);
1632 } else if (op instanceof SetData) {
1633 SetData sd = (SetData)op;
1634 return Op.setData(sd.getPath(), sd.getData(), -1);
1635 } else {
1636 throw new UnsupportedOperationException("Unexpected ZKUtilOp type: "
1637 + op.getClass().getName());
1638 }
1639 }
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
1653
1654
1655
1656
1657
1658
1659
1660
1661
1662 public static void multiOrSequential(ZooKeeperWatcher zkw, List<ZKUtilOp> ops,
1663 boolean runSequentialOnMultiFailure) throws KeeperException {
1664 if (ops == null) return;
1665 boolean useMulti = zkw.getConfiguration().getBoolean(HConstants.ZOOKEEPER_USEMULTI, false);
1666
1667 if (useMulti) {
1668 List<Op> zkOps = new LinkedList<Op>();
1669 for (ZKUtilOp op : ops) {
1670 zkOps.add(toZooKeeperOp(zkw, op));
1671 }
1672 try {
1673 zkw.getRecoverableZooKeeper().multi(zkOps);
1674 } catch (KeeperException ke) {
1675 switch (ke.code()) {
1676 case NODEEXISTS:
1677 case NONODE:
1678 case BADVERSION:
1679 case NOAUTH:
1680
1681
1682 if (runSequentialOnMultiFailure) {
1683 LOG.info("On call to ZK.multi, received exception: " + ke.toString() + "."
1684 + " Attempting to run operations sequentially because"
1685 + " runSequentialOnMultiFailure is: " + runSequentialOnMultiFailure + ".");
1686 processSequentially(zkw, ops);
1687 break;
1688 }
1689 default:
1690 throw ke;
1691 }
1692 } catch (InterruptedException ie) {
1693 zkw.interruptedException(ie);
1694 }
1695 } else {
1696
1697 processSequentially(zkw, ops);
1698 }
1699
1700 }
1701
1702 private static void processSequentially(ZooKeeperWatcher zkw, List<ZKUtilOp> ops)
1703 throws KeeperException, NoNodeException {
1704 for (ZKUtilOp op : ops) {
1705 if (op instanceof CreateAndFailSilent) {
1706 createAndFailSilent(zkw, (CreateAndFailSilent) op);
1707 } else if (op instanceof DeleteNodeFailSilent) {
1708 deleteNodeFailSilent(zkw, (DeleteNodeFailSilent) op);
1709 } else if (op instanceof SetData) {
1710 setData(zkw, (SetData) op);
1711 } else {
1712 throw new UnsupportedOperationException("Unexpected ZKUtilOp type: "
1713 + op.getClass().getName());
1714 }
1715 }
1716 }
1717
1718
1719
1720
1721
1722
1723 public static String dump(ZooKeeperWatcher zkw) {
1724 StringBuilder sb = new StringBuilder();
1725 try {
1726 sb.append("HBase is rooted at ").append(zkw.baseZNode);
1727 sb.append("\nActive master address: ");
1728 try {
1729 sb.append(MasterAddressTracker.getMasterAddress(zkw));
1730 } catch (IOException e) {
1731 sb.append("<<FAILED LOOKUP: " + e.getMessage() + ">>");
1732 }
1733 sb.append("\nBackup master addresses:");
1734 for (String child : listChildrenNoWatch(zkw,
1735 zkw.backupMasterAddressesZNode)) {
1736 sb.append("\n ").append(child);
1737 }
1738 sb.append("\nRegion server holding hbase:meta: "
1739 + new MetaTableLocator().getMetaRegionLocation(zkw));
1740 Configuration conf = HBaseConfiguration.create();
1741 int numMetaReplicas = conf.getInt(HConstants.META_REPLICAS_NUM,
1742 HConstants.DEFAULT_META_REPLICA_NUM);
1743 for (int i = 1; i < numMetaReplicas; i++) {
1744 sb.append("\nRegion server holding hbase:meta, replicaId " + i + " "
1745 + new MetaTableLocator().getMetaRegionLocation(zkw, i));
1746 }
1747 sb.append("\nRegion servers:");
1748 for (String child : listChildrenNoWatch(zkw, zkw.rsZNode)) {
1749 sb.append("\n ").append(child);
1750 }
1751 try {
1752 getReplicationZnodesDump(zkw, sb);
1753 } catch (KeeperException ke) {
1754 LOG.warn("Couldn't get the replication znode dump", ke);
1755 }
1756 sb.append("\nQuorum Server Statistics:");
1757 String[] servers = zkw.getQuorum().split(",");
1758 for (String server : servers) {
1759 sb.append("\n ").append(server);
1760 try {
1761 String[] stat = getServerStats(server, ZKUtil.zkDumpConnectionTimeOut);
1762
1763 if (stat == null) {
1764 sb.append("[Error] invalid quorum server: " + server);
1765 break;
1766 }
1767
1768 for (String s : stat) {
1769 sb.append("\n ").append(s);
1770 }
1771 } catch (Exception e) {
1772 sb.append("\n ERROR: ").append(e.getMessage());
1773 }
1774 }
1775 } catch (KeeperException ke) {
1776 sb.append("\nFATAL ZooKeeper Exception!\n");
1777 sb.append("\n" + ke.getMessage());
1778 }
1779 return sb.toString();
1780 }
1781
1782
1783
1784
1785
1786
1787
1788 private static void getReplicationZnodesDump(ZooKeeperWatcher zkw, StringBuilder sb)
1789 throws KeeperException {
1790 String replicationZNodeName = zkw.getConfiguration().get("zookeeper.znode.replication",
1791 "replication");
1792 String replicationZnode = joinZNode(zkw.baseZNode, replicationZNodeName);
1793 if (ZKUtil.checkExists(zkw, replicationZnode) == -1) return;
1794
1795 sb.append("\n").append(replicationZnode).append(": ");
1796 List<String> children = ZKUtil.listChildrenNoWatch(zkw, replicationZnode);
1797 for (String child : children) {
1798 String znode = joinZNode(replicationZnode, child);
1799 if (child.equals(zkw.getConfiguration().get("zookeeper.znode.replication.peers", "peers"))) {
1800 appendPeersZnodes(zkw, znode, sb);
1801 } else if (child.equals(zkw.getConfiguration().
1802 get("zookeeper.znode.replication.rs", "rs"))) {
1803 appendRSZnodes(zkw, znode, sb);
1804 } else if (child.equals(zkw.getConfiguration().get(
1805 ReplicationStateZKBase.ZOOKEEPER_ZNODE_REPLICATION_HFILE_REFS_KEY,
1806 ReplicationStateZKBase.ZOOKEEPER_ZNODE_REPLICATION_HFILE_REFS_DEFAULT))) {
1807 appendHFileRefsZnodes(zkw, znode, sb);
1808 }
1809 }
1810 }
1811
1812 private static void appendHFileRefsZnodes(ZooKeeperWatcher zkw, String hfileRefsZnode,
1813 StringBuilder sb) throws KeeperException {
1814 sb.append("\n").append(hfileRefsZnode).append(": ");
1815 for (String peerIdZnode : ZKUtil.listChildrenNoWatch(zkw, hfileRefsZnode)) {
1816 String znodeToProcess = ZKUtil.joinZNode(hfileRefsZnode, peerIdZnode);
1817 sb.append("\n").append(znodeToProcess).append(": ");
1818 List<String> peerHFileRefsZnodes = ZKUtil.listChildrenNoWatch(zkw, znodeToProcess);
1819 int size = peerHFileRefsZnodes.size();
1820 for (int i = 0; i < size; i++) {
1821 sb.append(peerHFileRefsZnodes.get(i));
1822 if (i != size - 1) {
1823 sb.append(", ");
1824 }
1825 }
1826 }
1827 }
1828
1829
1830
1831
1832
1833
1834 public static String getReplicationZnodesDump(ZooKeeperWatcher zkw) throws KeeperException {
1835 StringBuilder sb = new StringBuilder();
1836 getReplicationZnodesDump(zkw, sb);
1837 return sb.toString();
1838 }
1839
1840 private static void appendRSZnodes(ZooKeeperWatcher zkw, String znode, StringBuilder sb)
1841 throws KeeperException {
1842 List<String> stack = new LinkedList<String>();
1843 stack.add(znode);
1844 do {
1845 String znodeToProcess = stack.remove(stack.size() - 1);
1846 sb.append("\n").append(znodeToProcess).append(": ");
1847 byte[] data;
1848 try {
1849 data = ZKUtil.getData(zkw, znodeToProcess);
1850 } catch (InterruptedException e) {
1851 zkw.interruptedException(e);
1852 return;
1853 }
1854 if (data != null && data.length > 0) {
1855 long position = 0;
1856 try {
1857 position = ZKUtil.parseWALPositionFrom(ZKUtil.getData(zkw, znodeToProcess));
1858 sb.append(position);
1859 } catch (DeserializationException ignored) {
1860 } catch (InterruptedException e) {
1861 zkw.interruptedException(e);
1862 return;
1863 }
1864 }
1865 for (String zNodeChild : ZKUtil.listChildrenNoWatch(zkw, znodeToProcess)) {
1866 stack.add(ZKUtil.joinZNode(znodeToProcess, zNodeChild));
1867 }
1868 } while (stack.size() > 0);
1869 }
1870
1871 private static void appendPeersZnodes(ZooKeeperWatcher zkw, String peersZnode,
1872 StringBuilder sb) throws KeeperException {
1873 int pblen = ProtobufUtil.lengthOfPBMagic();
1874 sb.append("\n").append(peersZnode).append(": ");
1875 for (String peerIdZnode : ZKUtil.listChildrenNoWatch(zkw, peersZnode)) {
1876 String znodeToProcess = ZKUtil.joinZNode(peersZnode, peerIdZnode);
1877 byte[] data;
1878 try {
1879 data = ZKUtil.getData(zkw, znodeToProcess);
1880 } catch (InterruptedException e) {
1881 zkw.interruptedException(e);
1882 return;
1883 }
1884
1885 try {
1886 ZooKeeperProtos.ReplicationPeer.Builder builder =
1887 ZooKeeperProtos.ReplicationPeer.newBuilder();
1888 ProtobufUtil.mergeFrom(builder, data, pblen, data.length - pblen);
1889 String clusterKey = builder.getClusterkey();
1890 sb.append("\n").append(znodeToProcess).append(": ").append(clusterKey);
1891
1892 appendPeerState(zkw, znodeToProcess, sb);
1893 } catch (IOException ipbe) {
1894 LOG.warn("Got Exception while parsing peer: " + znodeToProcess, ipbe);
1895 }
1896 }
1897 }
1898
1899 private static void appendPeerState(ZooKeeperWatcher zkw, String znodeToProcess,
1900 StringBuilder sb) throws KeeperException, InvalidProtocolBufferException {
1901 String peerState = zkw.getConfiguration().get("zookeeper.znode.replication.peers.state",
1902 "peer-state");
1903 int pblen = ProtobufUtil.lengthOfPBMagic();
1904 for (String child : ZKUtil.listChildrenNoWatch(zkw, znodeToProcess)) {
1905 if (!child.equals(peerState)) continue;
1906 String peerStateZnode = ZKUtil.joinZNode(znodeToProcess, child);
1907 sb.append("\n").append(peerStateZnode).append(": ");
1908 byte[] peerStateData;
1909 try {
1910 peerStateData = ZKUtil.getData(zkw, peerStateZnode);
1911 ZooKeeperProtos.ReplicationState.Builder builder =
1912 ZooKeeperProtos.ReplicationState.newBuilder();
1913 ProtobufUtil.mergeFrom(builder, peerStateData, pblen, peerStateData.length - pblen);
1914 sb.append(builder.getState().name());
1915 } catch (IOException ipbe) {
1916 LOG.warn("Got Exception while parsing peer: " + znodeToProcess, ipbe);
1917 } catch (InterruptedException e) {
1918 zkw.interruptedException(e);
1919 return;
1920 }
1921 }
1922 }
1923
1924
1925
1926
1927
1928
1929
1930
1931
1932 public static String[] getServerStats(String server, int timeout)
1933 throws IOException {
1934 String[] sp = server.split(":");
1935 if (sp == null || sp.length == 0) {
1936 return null;
1937 }
1938
1939 String host = sp[0];
1940 int port = sp.length > 1 ? Integer.parseInt(sp[1])
1941 : HConstants.DEFAULT_ZOOKEPER_CLIENT_PORT;
1942
1943 Socket socket = new Socket();
1944 InetSocketAddress sockAddr = new InetSocketAddress(host, port);
1945 socket.connect(sockAddr, timeout);
1946
1947 socket.setSoTimeout(timeout);
1948 PrintWriter out = new PrintWriter(socket.getOutputStream(), true);
1949 BufferedReader in = new BufferedReader(new InputStreamReader(
1950 socket.getInputStream()));
1951 out.println("stat");
1952 out.flush();
1953 ArrayList<String> res = new ArrayList<String>();
1954 while (true) {
1955 String line = in.readLine();
1956 if (line != null) {
1957 res.add(line);
1958 } else {
1959 break;
1960 }
1961 }
1962 socket.close();
1963 return res.toArray(new String[res.size()]);
1964 }
1965
1966 private static void logRetrievedMsg(final ZooKeeperWatcher zkw,
1967 final String znode, final byte [] data, final boolean watcherSet) {
1968 if (!LOG.isTraceEnabled()) return;
1969 LOG.trace(zkw.prefix("Retrieved " + ((data == null)? 0: data.length) +
1970 " byte(s) of data from znode " + znode +
1971 (watcherSet? " and set watcher; ": "; data=") +
1972 (data == null? "null": data.length == 0? "empty": (
1973 znode.startsWith(zkw.assignmentZNode)?
1974 ZKAssign.toString(data):
1975 znode.startsWith(ZooKeeperWatcher.META_ZNODE_PREFIX)?
1976 getServerNameOrEmptyString(data):
1977 znode.startsWith(zkw.backupMasterAddressesZNode)?
1978 getServerNameOrEmptyString(data):
1979 StringUtils.abbreviate(Bytes.toStringBinary(data), 32)))));
1980 }
1981
1982 private static String getServerNameOrEmptyString(final byte [] data) {
1983 try {
1984 return ServerName.parseFrom(data).toString();
1985 } catch (DeserializationException e) {
1986 return "";
1987 }
1988 }
1989
1990
1991
1992
1993
1994 public static void waitForBaseZNode(Configuration conf) throws IOException {
1995 LOG.info("Waiting until the base znode is available");
1996 String parentZNode = conf.get(HConstants.ZOOKEEPER_ZNODE_PARENT,
1997 HConstants.DEFAULT_ZOOKEEPER_ZNODE_PARENT);
1998 ZooKeeper zk = new ZooKeeper(ZKConfig.getZKQuorumServersString(conf),
1999 conf.getInt(HConstants.ZK_SESSION_TIMEOUT,
2000 HConstants.DEFAULT_ZK_SESSION_TIMEOUT), EmptyWatcher.instance);
2001
2002 final int maxTimeMs = 10000;
2003 final int maxNumAttempts = maxTimeMs / HConstants.SOCKET_RETRY_WAIT_MS;
2004
2005 KeeperException keeperEx = null;
2006 try {
2007 try {
2008 for (int attempt = 0; attempt < maxNumAttempts; ++attempt) {
2009 try {
2010 if (zk.exists(parentZNode, false) != null) {
2011 LOG.info("Parent znode exists: " + parentZNode);
2012 keeperEx = null;
2013 break;
2014 }
2015 } catch (KeeperException e) {
2016 keeperEx = e;
2017 }
2018 Threads.sleepWithoutInterrupt(HConstants.SOCKET_RETRY_WAIT_MS);
2019 }
2020 } finally {
2021 zk.close();
2022 }
2023 } catch (InterruptedException ex) {
2024 Thread.currentThread().interrupt();
2025 }
2026
2027 if (keeperEx != null) {
2028 throw new IOException(keeperEx);
2029 }
2030 }
2031
2032
2033 public static byte[] blockUntilAvailable(
2034 final ZooKeeperWatcher zkw, final String znode, final long timeout)
2035 throws InterruptedException {
2036 if (timeout < 0) throw new IllegalArgumentException();
2037 if (zkw == null) throw new IllegalArgumentException();
2038 if (znode == null) throw new IllegalArgumentException();
2039
2040 byte[] data = null;
2041 boolean finished = false;
2042 final long endTime = System.currentTimeMillis() + timeout;
2043 while (!finished) {
2044 try {
2045 data = ZKUtil.getData(zkw, znode);
2046 } catch(KeeperException e) {
2047 if (e instanceof KeeperException.SessionExpiredException
2048 || e instanceof KeeperException.AuthFailedException) {
2049
2050 throw new InterruptedException("interrupted due to " + e);
2051 }
2052 LOG.warn("Unexpected exception handling blockUntilAvailable", e);
2053 }
2054
2055 if (data == null && (System.currentTimeMillis() +
2056 HConstants.SOCKET_RETRY_WAIT_MS < endTime)) {
2057 Thread.sleep(HConstants.SOCKET_RETRY_WAIT_MS);
2058 } else {
2059 finished = true;
2060 }
2061 }
2062
2063 return data;
2064 }
2065
2066
2067
2068
2069
2070
2071
2072
2073 public static KeeperException convert(final DeserializationException e) {
2074 KeeperException ke = new KeeperException.DataInconsistencyException();
2075 ke.initCause(e);
2076 return ke;
2077 }
2078
2079
2080
2081
2082
2083 public static void logZKTree(ZooKeeperWatcher zkw, String root) {
2084 if (!LOG.isDebugEnabled()) return;
2085 LOG.debug("Current zk system:");
2086 String prefix = "|-";
2087 LOG.debug(prefix + root);
2088 try {
2089 logZKTree(zkw, root, prefix);
2090 } catch (KeeperException e) {
2091 throw new RuntimeException(e);
2092 }
2093 }
2094
2095
2096
2097
2098
2099
2100 protected static void logZKTree(ZooKeeperWatcher zkw, String root, String prefix)
2101 throws KeeperException {
2102 List<String> children = ZKUtil.listChildrenNoWatch(zkw, root);
2103 if (children == null) return;
2104 for (String child : children) {
2105 LOG.debug(prefix + child);
2106 String node = ZKUtil.joinZNode(root.equals("/") ? "" : root, child);
2107 logZKTree(zkw, node, prefix + "---");
2108 }
2109 }
2110
2111
2112
2113
2114
2115
2116 public static byte[] positionToByteArray(final long position) {
2117 byte[] bytes = ZooKeeperProtos.ReplicationHLogPosition.newBuilder().setPosition(position)
2118 .build().toByteArray();
2119 return ProtobufUtil.prependPBMagic(bytes);
2120 }
2121
2122
2123
2124
2125
2126
2127 public static long parseWALPositionFrom(final byte[] bytes) throws DeserializationException {
2128 if (bytes == null) {
2129 throw new DeserializationException("Unable to parse null WAL position.");
2130 }
2131 if (ProtobufUtil.isPBMagicPrefix(bytes)) {
2132 int pblen = ProtobufUtil.lengthOfPBMagic();
2133 ZooKeeperProtos.ReplicationHLogPosition.Builder builder =
2134 ZooKeeperProtos.ReplicationHLogPosition.newBuilder();
2135 ZooKeeperProtos.ReplicationHLogPosition position;
2136 try {
2137 ProtobufUtil.mergeFrom(builder, bytes, pblen, bytes.length - pblen);
2138 position = builder.build();
2139 } catch (IOException e) {
2140 throw new DeserializationException(e);
2141 }
2142 return position.getPosition();
2143 } else {
2144 if (bytes.length > 0) {
2145 return Bytes.toLong(bytes);
2146 }
2147 return 0;
2148 }
2149 }
2150
2151
2152
2153
2154
2155
2156
2157
2158 public static byte[] regionSequenceIdsToByteArray(final Long regionLastFlushedSequenceId,
2159 final Map<byte[], Long> storeSequenceIds) {
2160 ClusterStatusProtos.RegionStoreSequenceIds.Builder regionSequenceIdsBuilder =
2161 ClusterStatusProtos.RegionStoreSequenceIds.newBuilder();
2162 ClusterStatusProtos.StoreSequenceId.Builder storeSequenceIdBuilder =
2163 ClusterStatusProtos.StoreSequenceId.newBuilder();
2164 if (storeSequenceIds != null) {
2165 for (Map.Entry<byte[], Long> e : storeSequenceIds.entrySet()){
2166 byte[] columnFamilyName = e.getKey();
2167 Long curSeqId = e.getValue();
2168 storeSequenceIdBuilder.setFamilyName(ByteStringer.wrap(columnFamilyName));
2169 storeSequenceIdBuilder.setSequenceId(curSeqId);
2170 regionSequenceIdsBuilder.addStoreSequenceId(storeSequenceIdBuilder.build());
2171 storeSequenceIdBuilder.clear();
2172 }
2173 }
2174 regionSequenceIdsBuilder.setLastFlushedSequenceId(regionLastFlushedSequenceId);
2175 byte[] result = regionSequenceIdsBuilder.build().toByteArray();
2176 return ProtobufUtil.prependPBMagic(result);
2177 }
2178
2179
2180
2181
2182
2183
2184 public static RegionStoreSequenceIds parseRegionStoreSequenceIds(final byte[] bytes)
2185 throws DeserializationException {
2186 if (bytes == null || !ProtobufUtil.isPBMagicPrefix(bytes)) {
2187 throw new DeserializationException("Unable to parse RegionStoreSequenceIds.");
2188 }
2189 RegionStoreSequenceIds.Builder regionSequenceIdsBuilder =
2190 ClusterStatusProtos.RegionStoreSequenceIds.newBuilder();
2191 int pblen = ProtobufUtil.lengthOfPBMagic();
2192 RegionStoreSequenceIds storeIds = null;
2193 try {
2194 ProtobufUtil.mergeFrom(regionSequenceIdsBuilder, bytes, pblen, bytes.length - pblen);
2195 storeIds = regionSequenceIdsBuilder.build();
2196 } catch (IOException e) {
2197 throw new DeserializationException(e);
2198 }
2199 return storeIds;
2200 }
2201 }