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