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