1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase.replication;
20
21 import java.io.IOException;
22 import java.util.ArrayList;
23 import java.util.List;
24 import java.util.Map;
25 import java.util.Set;
26 import java.util.TreeMap;
27 import java.util.concurrent.ConcurrentHashMap;
28 import java.util.concurrent.ConcurrentMap;
29
30 import org.apache.commons.logging.Log;
31 import org.apache.commons.logging.LogFactory;
32 import org.apache.hadoop.hbase.HBaseConfiguration;
33 import org.apache.hadoop.hbase.classification.InterfaceAudience;
34 import org.apache.hadoop.conf.Configuration;
35 import org.apache.hadoop.hbase.Abortable;
36 import org.apache.hadoop.hbase.CompoundConfiguration;
37 import org.apache.hadoop.hbase.TableName;
38 import org.apache.hadoop.hbase.exceptions.DeserializationException;
39 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
40 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.BytesBytesPair;
41 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.NameStringPair;
42 import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos;
43 import org.apache.hadoop.hbase.replication.ReplicationPeer.PeerState;
44 import org.apache.hadoop.hbase.util.Bytes;
45 import org.apache.hadoop.hbase.util.Pair;
46 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
47 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
48 import org.apache.hadoop.hbase.zookeeper.ZKUtil.ZKUtilOp;
49 import org.apache.zookeeper.KeeperException;
50 import org.apache.zookeeper.KeeperException.NoNodeException;
51
52 import com.google.protobuf.ByteString;
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80 @InterfaceAudience.Private
81 public class ReplicationPeersZKImpl extends ReplicationStateZKBase implements ReplicationPeers {
82
83
84 private Map<String, ReplicationPeerZKImpl> peerClusters;
85 private final String tableCFsNodeName;
86 private final ReplicationQueuesClient queuesClient;
87
88 private static final Log LOG = LogFactory.getLog(ReplicationPeersZKImpl.class);
89
90 public ReplicationPeersZKImpl(final ZooKeeperWatcher zk, final Configuration conf,
91 final ReplicationQueuesClient queuesClient, Abortable abortable) {
92 super(zk, conf, abortable);
93 this.tableCFsNodeName = conf.get("zookeeper.znode.replication.peers.tableCFs", "tableCFs");
94 this.peerClusters = new ConcurrentHashMap<String, ReplicationPeerZKImpl>();
95 this.queuesClient = queuesClient;
96 }
97
98 @Override
99 public void init() throws ReplicationException {
100 try {
101 if (ZKUtil.checkExists(this.zookeeper, this.peersZNode) < 0) {
102 ZKUtil.createWithParents(this.zookeeper, this.peersZNode);
103 }
104 } catch (KeeperException e) {
105 throw new ReplicationException("Could not initialize replication peers", e);
106 }
107 addExistingPeers();
108 }
109
110 @Override
111 public void addPeer(String id, ReplicationPeerConfig peerConfig, String tableCFs)
112 throws ReplicationException {
113 try {
114 if (peerExists(id)) {
115 throw new IllegalArgumentException("Cannot add a peer with id=" + id
116 + " because that id already exists.");
117 }
118
119 if(id.contains("-") || id.equals(ReplicationQueuesZKImpl.RS_LOCK_ZNODE)) {
120 throw new IllegalArgumentException("Found invalid peer name:" + id);
121 }
122
123 checkQueuesDeleted(id);
124
125 ZKUtil.createWithParents(this.zookeeper, this.peersZNode);
126
127 List<ZKUtilOp> listOfOps = new ArrayList<ZKUtil.ZKUtilOp>();
128 ZKUtilOp op1 = ZKUtilOp.createAndFailSilent(ZKUtil.joinZNode(this.peersZNode, id),
129 toByteArray(peerConfig));
130
131
132
133
134 ZKUtilOp op2 = ZKUtilOp.createAndFailSilent(getPeerStateNode(id), ENABLED_ZNODE_BYTES);
135 String tableCFsStr = (tableCFs == null) ? "" : tableCFs;
136 ZKUtilOp op3 = ZKUtilOp.createAndFailSilent(getTableCFsNode(id), Bytes.toBytes(tableCFsStr));
137 listOfOps.add(op1);
138 listOfOps.add(op2);
139 listOfOps.add(op3);
140 ZKUtil.multiOrSequential(this.zookeeper, listOfOps, false);
141
142 } catch (KeeperException e) {
143 throw new ReplicationException("Could not add peer with id=" + id
144 + ", peerConfif=>" + peerConfig, e);
145 }
146 }
147
148 @Override
149 public void removePeer(String id) throws ReplicationException {
150 try {
151 if (!peerExists(id)) {
152 throw new IllegalArgumentException("Cannot remove peer with id=" + id
153 + " because that id does not exist.");
154 }
155 ZKUtil.deleteNodeRecursively(this.zookeeper, ZKUtil.joinZNode(this.peersZNode, id));
156 } catch (KeeperException e) {
157 throw new ReplicationException("Could not remove peer with id=" + id, e);
158 }
159 }
160
161 @Override
162 public void enablePeer(String id) throws ReplicationException {
163 changePeerState(id, ZooKeeperProtos.ReplicationState.State.ENABLED);
164 LOG.info("peer " + id + " is enabled");
165 }
166
167 @Override
168 public void disablePeer(String id) throws ReplicationException {
169 changePeerState(id, ZooKeeperProtos.ReplicationState.State.DISABLED);
170 LOG.info("peer " + id + " is disabled");
171 }
172
173 @Override
174 public String getPeerTableCFsConfig(String id) throws ReplicationException {
175 try {
176 if (!peerExists(id)) {
177 throw new IllegalArgumentException("peer " + id + " doesn't exist");
178 }
179 try {
180 return Bytes.toString(ZKUtil.getData(this.zookeeper, getTableCFsNode(id)));
181 } catch (Exception e) {
182 throw new ReplicationException(e);
183 }
184 } catch (KeeperException e) {
185 throw new ReplicationException("Unable to get tableCFs of the peer with id=" + id, e);
186 }
187 }
188
189 @Override
190 public void setPeerTableCFsConfig(String id, String tableCFsStr) throws ReplicationException {
191 try {
192 if (!peerExists(id)) {
193 throw new IllegalArgumentException("Cannot set peer tableCFs because id=" + id
194 + " does not exist.");
195 }
196 String tableCFsZKNode = getTableCFsNode(id);
197 byte[] tableCFs = Bytes.toBytes(tableCFsStr);
198 if (ZKUtil.checkExists(this.zookeeper, tableCFsZKNode) != -1) {
199 ZKUtil.setData(this.zookeeper, tableCFsZKNode, tableCFs);
200 } else {
201 ZKUtil.createAndWatch(this.zookeeper, tableCFsZKNode, tableCFs);
202 }
203 LOG.info("Peer tableCFs with id= " + id + " is now " + tableCFsStr);
204 } catch (KeeperException e) {
205 throw new ReplicationException("Unable to change tableCFs of the peer with id=" + id, e);
206 }
207 }
208
209 @Override
210 public Map<TableName, List<String>> getTableCFs(String id) throws IllegalArgumentException {
211 ReplicationPeer replicationPeer = this.peerClusters.get(id);
212 if (replicationPeer == null) {
213 throw new IllegalArgumentException("Peer with id= " + id + " is not connected");
214 }
215 return replicationPeer.getTableCFs();
216 }
217
218 @Override
219 public boolean getStatusOfPeer(String id) {
220 ReplicationPeer replicationPeer = this.peerClusters.get(id);
221 if (replicationPeer == null) {
222 throw new IllegalArgumentException("Peer with id= " + id + " is not connected");
223 }
224 return replicationPeer.getPeerState() == PeerState.ENABLED;
225 }
226
227 @Override
228 public boolean getStatusOfPeerFromBackingStore(String id) throws ReplicationException {
229 try {
230 if (!peerExists(id)) {
231 throw new IllegalArgumentException("peer " + id + " doesn't exist");
232 }
233 String peerStateZNode = getPeerStateNode(id);
234 try {
235 return ReplicationPeerZKImpl.isStateEnabled(ZKUtil.getData(this.zookeeper, peerStateZNode));
236 } catch (KeeperException e) {
237 throw new ReplicationException(e);
238 } catch (DeserializationException e) {
239 throw new ReplicationException(e);
240 }
241 } catch (KeeperException e) {
242 throw new ReplicationException("Unable to get status of the peer with id=" + id +
243 " from backing store", e);
244 } catch (InterruptedException e) {
245 throw new ReplicationException(e);
246 }
247 }
248
249 @Override
250 public Map<String, ReplicationPeerConfig> getAllPeerConfigs() {
251 Map<String, ReplicationPeerConfig> peers = new TreeMap<String, ReplicationPeerConfig>();
252 List<String> ids = null;
253 try {
254 ids = ZKUtil.listChildrenNoWatch(this.zookeeper, this.peersZNode);
255 for (String id : ids) {
256 ReplicationPeerConfig peerConfig = getReplicationPeerConfig(id);
257 if (peerConfig == null) {
258 LOG.warn("Failed to get replication peer configuration of clusterid=" + id
259 + " znode content, continuing.");
260 continue;
261 }
262 peers.put(id, peerConfig);
263 }
264 } catch (KeeperException e) {
265 this.abortable.abort("Cannot get the list of peers ", e);
266 } catch (ReplicationException e) {
267 this.abortable.abort("Cannot get the list of peers ", e);
268 }
269 return peers;
270 }
271
272 @Override
273 public ReplicationPeer getPeer(String peerId) {
274 return peerClusters.get(peerId);
275 }
276
277 @Override
278 public Set<String> getPeerIds() {
279 return peerClusters.keySet();
280 }
281
282
283
284
285 @Override
286 public ReplicationPeerConfig getReplicationPeerConfig(String peerId)
287 throws ReplicationException {
288 String znode = ZKUtil.joinZNode(this.peersZNode, peerId);
289 byte[] data = null;
290 try {
291 data = ZKUtil.getData(this.zookeeper, znode);
292 } catch (InterruptedException e) {
293 LOG.warn("Could not get configuration for peer because the thread " +
294 "was interrupted. peerId=" + peerId);
295 Thread.currentThread().interrupt();
296 return null;
297 } catch (KeeperException e) {
298 throw new ReplicationException("Error getting configuration for peer with id="
299 + peerId, e);
300 }
301 if (data == null) {
302 LOG.error("Could not get configuration for peer because it doesn't exist. peerId=" + peerId);
303 return null;
304 }
305
306 try {
307 return parsePeerFrom(data);
308 } catch (DeserializationException e) {
309 LOG.warn("Failed to parse cluster key from peerId=" + peerId
310 + ", specifically the content from the following znode: " + znode);
311 return null;
312 }
313 }
314
315 @Override
316 public Pair<ReplicationPeerConfig, Configuration> getPeerConf(String peerId)
317 throws ReplicationException {
318 ReplicationPeerConfig peerConfig = getReplicationPeerConfig(peerId);
319
320 if (peerConfig == null) {
321 return null;
322 }
323
324 Configuration otherConf;
325 try {
326 otherConf = HBaseConfiguration.createClusterConf(this.conf, peerConfig.getClusterKey());
327 } catch (IOException e) {
328 LOG.error("Can't get peer configuration for peerId=" + peerId + " because:", e);
329 return null;
330 }
331
332 if (!peerConfig.getConfiguration().isEmpty()) {
333 CompoundConfiguration compound = new CompoundConfiguration();
334 compound.add(otherConf);
335 compound.addStringMap(peerConfig.getConfiguration());
336 return new Pair<ReplicationPeerConfig, Configuration>(peerConfig, compound);
337 }
338
339 return new Pair<ReplicationPeerConfig, Configuration>(peerConfig, otherConf);
340 }
341
342
343
344
345 @Override
346 public List<String> getAllPeerIds() {
347 List<String> ids = null;
348 try {
349 ids = ZKUtil.listChildrenAndWatchThem(this.zookeeper, this.peersZNode);
350 } catch (KeeperException e) {
351 this.abortable.abort("Cannot get the list of peers ", e);
352 }
353 return ids;
354 }
355
356
357
358
359
360 private void addExistingPeers() throws ReplicationException {
361 List<String> znodes = null;
362 try {
363 znodes = ZKUtil.listChildrenNoWatch(this.zookeeper, this.peersZNode);
364 } catch (KeeperException e) {
365 throw new ReplicationException("Error getting the list of peer clusters.", e);
366 }
367 if (znodes != null) {
368 for (String z : znodes) {
369 createAndAddPeer(z);
370 }
371 }
372 }
373
374 @Override
375 public boolean peerAdded(String peerId) throws ReplicationException {
376 return createAndAddPeer(peerId);
377 }
378
379 @Override
380 public void peerRemoved(String peerId) {
381 ReplicationPeer rp = this.peerClusters.get(peerId);
382 if (rp != null) {
383 ((ConcurrentMap<String, ReplicationPeerZKImpl>) peerClusters).remove(peerId, rp);
384 }
385 }
386
387
388
389
390
391
392 public boolean createAndAddPeer(String peerId) throws ReplicationException {
393 if (peerClusters == null) {
394 return false;
395 }
396 if (this.peerClusters.containsKey(peerId)) {
397 return false;
398 }
399
400 ReplicationPeerZKImpl peer = null;
401 try {
402 peer = createPeer(peerId);
403 } catch (Exception e) {
404 throw new ReplicationException("Error adding peer with id=" + peerId, e);
405 }
406 if (peer == null) {
407 return false;
408 }
409 ReplicationPeerZKImpl previous =
410 ((ConcurrentMap<String, ReplicationPeerZKImpl>) peerClusters).putIfAbsent(peerId, peer);
411 if (previous == null) {
412 LOG.info("Added new peer cluster=" + peer.getPeerConfig().getClusterKey());
413 } else {
414 LOG.info("Peer already present, " + previous.getPeerConfig().getClusterKey() +
415 ", new cluster=" + peer.getPeerConfig().getClusterKey());
416 }
417 return true;
418 }
419
420 private String getTableCFsNode(String id) {
421 return ZKUtil.joinZNode(this.peersZNode, ZKUtil.joinZNode(id, this.tableCFsNodeName));
422 }
423
424 private String getPeerStateNode(String id) {
425 return ZKUtil.joinZNode(this.peersZNode, ZKUtil.joinZNode(id, this.peerStateNodeName));
426 }
427
428
429
430
431
432
433 private void changePeerState(String id, ZooKeeperProtos.ReplicationState.State state)
434 throws ReplicationException {
435 try {
436 if (!peerExists(id)) {
437 throw new IllegalArgumentException("Cannot enable/disable peer because id=" + id
438 + " does not exist.");
439 }
440 String peerStateZNode = getPeerStateNode(id);
441 byte[] stateBytes =
442 (state == ZooKeeperProtos.ReplicationState.State.ENABLED) ? ENABLED_ZNODE_BYTES
443 : DISABLED_ZNODE_BYTES;
444 if (ZKUtil.checkExists(this.zookeeper, peerStateZNode) != -1) {
445 ZKUtil.setData(this.zookeeper, peerStateZNode, stateBytes);
446 } else {
447 ZKUtil.createAndWatch(this.zookeeper, peerStateZNode, stateBytes);
448 }
449 LOG.info("Peer with id= " + id + " is now " + state.name());
450 } catch (KeeperException e) {
451 throw new ReplicationException("Unable to change state of the peer with id=" + id, e);
452 }
453 }
454
455
456
457
458
459
460
461 private ReplicationPeerZKImpl createPeer(String peerId) throws ReplicationException {
462 Pair<ReplicationPeerConfig, Configuration> pair = getPeerConf(peerId);
463 if (pair == null) {
464 return null;
465 }
466 Configuration peerConf = pair.getSecond();
467
468 ReplicationPeerZKImpl peer = new ReplicationPeerZKImpl(peerConf, peerId, pair.getFirst());
469 try {
470 peer.startStateTracker(this.zookeeper, this.getPeerStateNode(peerId));
471 } catch (KeeperException e) {
472 throw new ReplicationException("Error starting the peer state tracker for peerId=" +
473 peerId, e);
474 }
475
476 try {
477 peer.startTableCFsTracker(this.zookeeper, this.getTableCFsNode(peerId));
478 } catch (KeeperException e) {
479 throw new ReplicationException("Error starting the peer tableCFs tracker for peerId=" +
480 peerId, e);
481 }
482
483 return peer;
484 }
485
486
487
488
489
490
491 private static ReplicationPeerConfig parsePeerFrom(final byte[] bytes)
492 throws DeserializationException {
493 if (ProtobufUtil.isPBMagicPrefix(bytes)) {
494 int pblen = ProtobufUtil.lengthOfPBMagic();
495 ZooKeeperProtos.ReplicationPeer.Builder builder =
496 ZooKeeperProtos.ReplicationPeer.newBuilder();
497 ZooKeeperProtos.ReplicationPeer peer;
498 try {
499 ProtobufUtil.mergeFrom(builder, bytes, pblen, bytes.length - pblen);
500 peer = builder.build();
501 } catch (IOException e) {
502 throw new DeserializationException(e);
503 }
504 return convert(peer);
505 } else {
506 if (bytes.length > 0) {
507 return new ReplicationPeerConfig().setClusterKey(Bytes.toString(bytes));
508 }
509 return new ReplicationPeerConfig().setClusterKey("");
510 }
511 }
512
513 private static ReplicationPeerConfig convert(ZooKeeperProtos.ReplicationPeer peer) {
514 ReplicationPeerConfig peerConfig = new ReplicationPeerConfig();
515 if (peer.hasClusterkey()) {
516 peerConfig.setClusterKey(peer.getClusterkey());
517 }
518 if (peer.hasReplicationEndpointImpl()) {
519 peerConfig.setReplicationEndpointImpl(peer.getReplicationEndpointImpl());
520 }
521
522 for (BytesBytesPair pair : peer.getDataList()) {
523 peerConfig.getPeerData().put(pair.getFirst().toByteArray(), pair.getSecond().toByteArray());
524 }
525
526 for (NameStringPair pair : peer.getConfigurationList()) {
527 peerConfig.getConfiguration().put(pair.getName(), pair.getValue());
528 }
529 return peerConfig;
530 }
531
532 private static ZooKeeperProtos.ReplicationPeer convert(ReplicationPeerConfig peerConfig) {
533 ZooKeeperProtos.ReplicationPeer.Builder builder = ZooKeeperProtos.ReplicationPeer.newBuilder();
534 if (peerConfig.getClusterKey() != null) {
535 builder.setClusterkey(peerConfig.getClusterKey());
536 }
537 if (peerConfig.getReplicationEndpointImpl() != null) {
538 builder.setReplicationEndpointImpl(peerConfig.getReplicationEndpointImpl());
539 }
540
541 for (Map.Entry<byte[], byte[]> entry : peerConfig.getPeerData().entrySet()) {
542 builder.addData(BytesBytesPair.newBuilder()
543 .setFirst(ByteString.copyFrom(entry.getKey()))
544 .setSecond(ByteString.copyFrom(entry.getValue()))
545 .build());
546 }
547
548 for (Map.Entry<String, String> entry : peerConfig.getConfiguration().entrySet()) {
549 builder.addConfiguration(NameStringPair.newBuilder()
550 .setName(entry.getKey())
551 .setValue(entry.getValue())
552 .build());
553 }
554
555 return builder.build();
556 }
557
558
559
560
561
562
563
564 private static byte[] toByteArray(final ReplicationPeerConfig peerConfig) {
565 byte[] bytes = convert(peerConfig).toByteArray();
566 return ProtobufUtil.prependPBMagic(bytes);
567 }
568
569 private void checkQueuesDeleted(String peerId) throws ReplicationException {
570 if (queuesClient == null) return;
571 try {
572 List<String> replicators = queuesClient.getListOfReplicators();
573 if (replicators == null || replicators.isEmpty()) {
574 return;
575 }
576 for (String replicator : replicators) {
577 List<String> queueIds = queuesClient.getAllQueues(replicator);
578 for (String queueId : queueIds) {
579 ReplicationQueueInfo queueInfo = new ReplicationQueueInfo(queueId);
580 if (queueInfo.getPeerId().equals(peerId)) {
581 throw new ReplicationException("undeleted queue for peerId: " + peerId
582 + ", replicator: " + replicator + ", queueId: " + queueId);
583 }
584 }
585 }
586
587 if (-1 != ZKUtil.checkExists(zookeeper, hfileRefsZNode)
588 && queuesClient.getAllPeersFromHFileRefsQueue().contains(peerId)) {
589 throw new ReplicationException("Undeleted queue for peerId: " + peerId
590 + ", found in hfile-refs node path " + hfileRefsZNode);
591 }
592 } catch (KeeperException e) {
593 throw new ReplicationException("Could not check queues deleted with id=" + peerId, e);
594 }
595 }
596 }