1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21 package org.apache.hadoop.hbase.rsgroup;
22
23 import com.google.common.collect.Lists;
24 import com.google.common.collect.Maps;
25
26 import com.google.common.collect.Sets;
27 import com.google.protobuf.ServiceException;
28
29 import java.io.IOException;
30 import java.util.ArrayList;
31 import java.util.Collections;
32 import java.util.Comparator;
33 import java.util.HashMap;
34 import java.util.HashSet;
35 import java.util.LinkedList;
36 import java.util.List;
37 import java.util.Map;
38 import java.util.NavigableSet;
39 import java.util.Set;
40 import java.util.TreeSet;
41 import java.util.concurrent.atomic.AtomicBoolean;
42
43 import org.apache.commons.logging.Log;
44 import org.apache.commons.logging.LogFactory;
45 import org.apache.hadoop.hbase.Cell;
46 import org.apache.hadoop.hbase.CellUtil;
47 import org.apache.hadoop.hbase.Coprocessor;
48 import org.apache.hadoop.hbase.DoNotRetryIOException;
49 import org.apache.hadoop.hbase.HColumnDescriptor;
50 import org.apache.hadoop.hbase.HConstants;
51 import org.apache.hadoop.hbase.HRegionInfo;
52 import org.apache.hadoop.hbase.HTableDescriptor;
53 import org.apache.hadoop.hbase.MetaTableAccessor;
54 import org.apache.hadoop.hbase.ServerName;
55 import org.apache.hadoop.hbase.TableName;
56 import org.apache.hadoop.hbase.TableStateManager;
57 import org.apache.hadoop.hbase.client.ClusterConnection;
58 import org.apache.hadoop.hbase.client.Delete;
59 import org.apache.hadoop.hbase.client.Get;
60 import org.apache.hadoop.hbase.client.Mutation;
61 import org.apache.hadoop.hbase.client.Put;
62 import org.apache.hadoop.hbase.client.Result;
63 import org.apache.hadoop.hbase.client.Table;
64 import org.apache.hadoop.hbase.constraint.ConstraintException;
65 import org.apache.hadoop.hbase.coprocessor.MultiRowMutationEndpoint;
66 import org.apache.hadoop.hbase.ipc.CoprocessorRpcChannel;
67 import org.apache.hadoop.hbase.master.MasterServices;
68 import org.apache.hadoop.hbase.master.RegionState;
69 import org.apache.hadoop.hbase.master.ServerListener;
70 import org.apache.hadoop.hbase.master.procedure.CreateTableProcedure;
71 import org.apache.hadoop.hbase.master.procedure.ProcedurePrepareLatch;
72 import org.apache.hadoop.hbase.net.Address;
73 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
74 import org.apache.hadoop.hbase.protobuf.RequestConverter;
75 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
76 import org.apache.hadoop.hbase.protobuf.generated.MultiRowMutationProtos;
77 import org.apache.hadoop.hbase.protobuf.generated.RSGroupProtos;
78 import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos;
79 import org.apache.hadoop.hbase.protobuf.generated.MultiRowMutationProtos.MutateRowsRequest;
80 import org.apache.hadoop.hbase.regionserver.DisabledRegionSplitPolicy;
81 import org.apache.hadoop.hbase.util.Bytes;
82 import org.apache.hadoop.hbase.util.ModifyRegionUtils;
83 import org.apache.hadoop.hbase.util.Threads;
84 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
85 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
86 import org.apache.zookeeper.KeeperException;
87
88
89
90
91
92
93
94 public class RSGroupInfoManagerImpl implements RSGroupInfoManager, ServerListener {
95 private static final Log LOG = LogFactory.getLog(RSGroupInfoManagerImpl.class);
96
97
98 private final static HTableDescriptor RSGROUP_TABLE_DESC;
99 static {
100 RSGROUP_TABLE_DESC = new HTableDescriptor(RSGROUP_TABLE_NAME);
101 RSGROUP_TABLE_DESC.addFamily(new HColumnDescriptor(META_FAMILY_BYTES));
102 RSGROUP_TABLE_DESC.setRegionSplitPolicyClassName(DisabledRegionSplitPolicy.class.getName());
103 try {
104 RSGROUP_TABLE_DESC.addCoprocessor(
105 MultiRowMutationEndpoint.class.getName(),
106 null, Coprocessor.PRIORITY_SYSTEM, null);
107 } catch (IOException ex) {
108 throw new RuntimeException(ex);
109 }
110 }
111
112 private volatile Map<String, RSGroupInfo> rsGroupMap;
113 private volatile Map<TableName, String> tableMap;
114 private MasterServices master;
115 private ClusterConnection conn;
116 private ZooKeeperWatcher watcher;
117 private RSGroupStartupWorker rsGroupStartupWorker;
118
119 private volatile Set<String> prevRSGroups;
120 private RSGroupSerDe rsGroupSerDe;
121 private DefaultServerUpdater defaultServerUpdater;
122 private FailedOpenUpdater failedOpenUpdater;
123 private boolean isInit = false;
124
125 public RSGroupInfoManagerImpl(MasterServices master) throws IOException {
126 this.rsGroupMap = Collections.emptyMap();
127 this.tableMap = Collections.emptyMap();
128 rsGroupSerDe = new RSGroupSerDe();
129 this.master = master;
130 this.watcher = master.getZooKeeper();
131 this.conn = master.getConnection();
132 prevRSGroups = new HashSet<String>();
133 }
134
135 public void init() throws IOException{
136 rsGroupStartupWorker = new RSGroupStartupWorker(this, master, conn);
137 refresh();
138 defaultServerUpdater = new DefaultServerUpdater(this);
139 Threads.setDaemonThreadRunning(defaultServerUpdater);
140 failedOpenUpdater = new FailedOpenUpdater(this);
141 Threads.setDaemonThreadRunning(failedOpenUpdater);
142 master.getServerManager().registerListener(this);
143 isInit = true;
144 }
145
146 boolean isInit() {
147 return isInit;
148 }
149
150 public void start(){
151
152 rsGroupStartupWorker.start();
153 }
154
155
156
157
158
159
160 @Override
161 public synchronized void addRSGroup(RSGroupInfo rsGroupInfo) throws IOException {
162 checkGroupName(rsGroupInfo.getName());
163 if (rsGroupMap.get(rsGroupInfo.getName()) != null ||
164 rsGroupInfo.getName().equals(RSGroupInfo.DEFAULT_GROUP)) {
165 throw new DoNotRetryIOException("Group already exists: "+ rsGroupInfo.getName());
166 }
167 Map<String, RSGroupInfo> newGroupMap = Maps.newHashMap(rsGroupMap);
168 newGroupMap.put(rsGroupInfo.getName(), rsGroupInfo);
169 flushConfig(newGroupMap);
170 }
171
172 @Override
173 public synchronized Set<Address> moveServers(Set<Address> servers, String srcGroup,
174 String dstGroup) throws IOException {
175 if (servers == null) {
176 throw new ConstraintException("The list of servers to move cannot be null.");
177 }
178 Set<Address> movedServers = Sets.newHashSet();
179 if (!rsGroupMap.containsKey(srcGroup)) {
180 throw new DoNotRetryIOException("Group "+srcGroup+" does not exist");
181 }
182 if (!rsGroupMap.containsKey(dstGroup)) {
183 throw new DoNotRetryIOException("Group "+dstGroup+" does not exist");
184 }
185
186 RSGroupInfo src = new RSGroupInfo(getRSGroup(srcGroup));
187 RSGroupInfo dst = new RSGroupInfo(getRSGroup(dstGroup));
188 for(Address el: servers) {
189 if (src.removeServer(el)) {
190 movedServers.add(el);
191 }
192 dst.addServer(el);
193 }
194
195 Map<String,RSGroupInfo> newGroupMap = Maps.newHashMap(rsGroupMap);
196 newGroupMap.put(src.getName(), src);
197 newGroupMap.put(dst.getName(), dst);
198
199 flushConfig(newGroupMap);
200 return movedServers;
201 }
202
203
204
205
206
207
208
209 @Override
210 public RSGroupInfo getRSGroupOfServer(Address server) throws IOException {
211 for (RSGroupInfo info : rsGroupMap.values()) {
212 if (info.containsServer(server)){
213 return info;
214 }
215 }
216 return null;
217 }
218
219
220
221
222
223
224
225
226 @Override
227 public RSGroupInfo getRSGroup(String groupName) throws IOException {
228 return rsGroupMap.get(groupName);
229 }
230
231
232
233
234
235
236
237 @Override
238 public String getRSGroupOfTable(TableName tableName) throws IOException {
239 return tableMap.get(tableName);
240 }
241
242 @Override
243 public synchronized void moveTables(
244 Set<TableName> tableNames, String groupName) throws IOException {
245 if (groupName != null && !rsGroupMap.containsKey(groupName)) {
246 throw new DoNotRetryIOException("Group "+groupName+" does not exist");
247 }
248
249 Map<String,RSGroupInfo> newGroupMap = Maps.newHashMap(rsGroupMap);
250 for(TableName tableName: tableNames) {
251 if (tableMap.containsKey(tableName)) {
252 RSGroupInfo src = new RSGroupInfo(newGroupMap.get(tableMap.get(tableName)));
253 src.removeTable(tableName);
254 newGroupMap.put(src.getName(), src);
255 }
256 if(groupName != null) {
257 RSGroupInfo dst = new RSGroupInfo(newGroupMap.get(groupName));
258 dst.addTable(tableName);
259 newGroupMap.put(dst.getName(), dst);
260 }
261 }
262
263 flushConfig(newGroupMap);
264 }
265
266
267
268
269
270
271
272
273 @Override
274 public synchronized void removeRSGroup(String groupName) throws IOException {
275 if (!rsGroupMap.containsKey(groupName) || groupName.equals(RSGroupInfo.DEFAULT_GROUP)) {
276 throw new DoNotRetryIOException("Group "+groupName+" does not exist or is a reserved group");
277 }
278 Map<String,RSGroupInfo> newGroupMap = Maps.newHashMap(rsGroupMap);
279 newGroupMap.remove(groupName);
280 flushConfig(newGroupMap);
281 }
282
283 @Override
284 public List<RSGroupInfo> listRSGroups() throws IOException {
285 List<RSGroupInfo> list = Lists.newLinkedList(rsGroupMap.values());
286 return list;
287 }
288
289 @Override
290 public boolean isOnline() {
291 return rsGroupStartupWorker.isOnline();
292 }
293
294 @Override
295 public synchronized void refresh() throws IOException {
296 refresh(false);
297 }
298
299 private synchronized void refresh(boolean forceOnline) throws IOException {
300 List<RSGroupInfo> groupList = new LinkedList<RSGroupInfo>();
301
302
303
304 if (forceOnline || isOnline()) {
305 LOG.debug("Refreshing in Online mode.");
306 try (Table rsGroupTable = conn.getTable(RSGROUP_TABLE_NAME)) {
307 groupList.addAll(rsGroupSerDe.retrieveGroupList(rsGroupTable));
308 }
309 } else {
310 LOG.debug("Refreshing in Offline mode.");
311 String groupBasePath = ZKUtil.joinZNode(watcher.baseZNode, rsGroupZNode);
312 groupList.addAll(rsGroupSerDe.retrieveGroupList(watcher, groupBasePath));
313 }
314
315
316 NavigableSet<TableName> orphanTables = new TreeSet<TableName>();
317 for(String entry: master.getTableDescriptors().getAll().keySet()) {
318 orphanTables.add(TableName.valueOf(entry));
319 }
320
321 for (RSGroupInfo group: groupList) {
322 if(!group.getName().equals(RSGroupInfo.DEFAULT_GROUP)) {
323 orphanTables.removeAll(group.getTables());
324 }
325 }
326
327
328
329
330 groupList.add(new RSGroupInfo(RSGroupInfo.DEFAULT_GROUP,
331 Sets.newHashSet(getDefaultServers()),
332 orphanTables));
333
334
335 HashMap<String, RSGroupInfo> newGroupMap = Maps.newHashMap();
336 HashMap<TableName, String> newTableMap = Maps.newHashMap();
337 for (RSGroupInfo group : groupList) {
338 newGroupMap.put(group.getName(), group);
339 for(TableName table: group.getTables()) {
340 newTableMap.put(table, group.getName());
341 }
342 }
343 rsGroupMap = Collections.unmodifiableMap(newGroupMap);
344 tableMap = Collections.unmodifiableMap(newTableMap);
345
346 prevRSGroups.clear();
347 prevRSGroups.addAll(rsGroupMap.keySet());
348 }
349
350 private synchronized Map<TableName,String> flushConfigTable(Map<String,RSGroupInfo> newGroupMap)
351 throws IOException {
352 Map<TableName,String> newTableMap = Maps.newHashMap();
353 List<Mutation> mutations = Lists.newArrayList();
354
355
356 for(String groupName : prevRSGroups) {
357 if(!newGroupMap.containsKey(groupName)) {
358 Delete d = new Delete(Bytes.toBytes(groupName));
359 mutations.add(d);
360 }
361 }
362
363
364 for(RSGroupInfo RSGroupInfo : newGroupMap.values()) {
365 RSGroupProtos.RSGroupInfo proto = RSGroupProtobufUtil.toProtoGroupInfo(RSGroupInfo);
366 Put p = new Put(Bytes.toBytes(RSGroupInfo.getName()));
367 p.addColumn(META_FAMILY_BYTES,
368 META_QUALIFIER_BYTES,
369 proto.toByteArray());
370 mutations.add(p);
371 for(TableName entry: RSGroupInfo.getTables()) {
372 newTableMap.put(entry, RSGroupInfo.getName());
373 }
374 }
375
376 if(mutations.size() > 0) {
377 multiMutate(mutations);
378 }
379 return newTableMap;
380 }
381
382 private synchronized void flushConfig(Map<String, RSGroupInfo> newGroupMap) throws IOException {
383 Map<TableName, String> newTableMap;
384
385
386
387 if (!isOnline()) {
388 Map<String, RSGroupInfo> m = Maps.newHashMap(rsGroupMap);
389 RSGroupInfo oldDefaultGroup = m.remove(RSGroupInfo.DEFAULT_GROUP);
390 RSGroupInfo newDefaultGroup = newGroupMap.remove(RSGroupInfo.DEFAULT_GROUP);
391 if (!m.equals(newGroupMap) ||
392 !oldDefaultGroup.getTables().equals(newDefaultGroup.getTables())) {
393 throw new IOException("Only default servers can be updated during offline mode");
394 }
395 newGroupMap.put(RSGroupInfo.DEFAULT_GROUP, newDefaultGroup);
396 rsGroupMap = newGroupMap;
397 return;
398 }
399
400 newTableMap = flushConfigTable(newGroupMap);
401
402
403
404 rsGroupMap = Collections.unmodifiableMap(newGroupMap);
405 tableMap = Collections.unmodifiableMap(newTableMap);
406
407
408 try {
409 String groupBasePath = ZKUtil.joinZNode(watcher.baseZNode, rsGroupZNode);
410 ZKUtil.createAndFailSilent(watcher, groupBasePath, ProtobufUtil.PB_MAGIC);
411
412 List<ZKUtil.ZKUtilOp> zkOps = new ArrayList<ZKUtil.ZKUtilOp>(newGroupMap.size());
413 for(String groupName : prevRSGroups) {
414 if(!newGroupMap.containsKey(groupName)) {
415 String znode = ZKUtil.joinZNode(groupBasePath, groupName);
416 zkOps.add(ZKUtil.ZKUtilOp.deleteNodeFailSilent(znode));
417 }
418 }
419
420
421 for(RSGroupInfo RSGroupInfo : newGroupMap.values()) {
422 String znode = ZKUtil.joinZNode(groupBasePath, RSGroupInfo.getName());
423 RSGroupProtos.RSGroupInfo proto = RSGroupProtobufUtil.toProtoGroupInfo(RSGroupInfo);
424 LOG.debug("Updating znode: "+znode);
425 ZKUtil.createAndFailSilent(watcher, znode);
426 zkOps.add(ZKUtil.ZKUtilOp.deleteNodeFailSilent(znode));
427 zkOps.add(ZKUtil.ZKUtilOp.createAndFailSilent(znode,
428 ProtobufUtil.prependPBMagic(proto.toByteArray())));
429 }
430 LOG.debug("Writing ZK GroupInfo count: " + zkOps.size());
431
432 ZKUtil.multiOrSequential(watcher, zkOps, false);
433 } catch (KeeperException e) {
434 LOG.error("Failed to write to rsGroupZNode", e);
435 master.abort("Failed to write to rsGroupZNode", e);
436 throw new IOException("Failed to write to rsGroupZNode",e);
437 }
438
439 prevRSGroups.clear();
440 prevRSGroups.addAll(newGroupMap.keySet());
441 }
442
443 private List<ServerName> getOnlineRS() throws IOException {
444 if (master != null) {
445 return master.getServerManager().getOnlineServersList();
446 }
447 try {
448 LOG.debug("Reading online RS from zookeeper");
449 List<ServerName> servers = new LinkedList<ServerName>();
450 for (String el: ZKUtil.listChildrenNoWatch(watcher, watcher.rsZNode)) {
451 servers.add(ServerName.parseServerName(el));
452 }
453 return servers;
454 } catch (KeeperException e) {
455 throw new IOException("Failed to retrieve server list from zookeeper", e);
456 }
457 }
458
459 private List<Address> getDefaultServers() throws IOException {
460 List<Address> defaultServers = new LinkedList<Address>();
461 for(ServerName server : getOnlineRS()) {
462 Address address = Address.fromParts(server.getHostname(), server.getPort());
463 boolean found = false;
464 for(RSGroupInfo info : rsGroupMap.values()) {
465 if(!RSGroupInfo.DEFAULT_GROUP.equals(info.getName()) &&
466 info.containsServer(address)) {
467 found = true;
468 break;
469 }
470 }
471 if(!found) {
472 defaultServers.add(address);
473 }
474 }
475 return defaultServers;
476 }
477
478 private synchronized void updateDefaultServers(
479 Set<Address> server) throws IOException {
480 RSGroupInfo info = rsGroupMap.get(RSGroupInfo.DEFAULT_GROUP);
481 RSGroupInfo newInfo = new RSGroupInfo(info.getName(), server, info.getTables());
482 HashMap<String, RSGroupInfo> newGroupMap = Maps.newHashMap(rsGroupMap);
483 newGroupMap.put(newInfo.getName(), newInfo);
484 flushConfig(newGroupMap);
485 }
486
487 @Override
488 public void serverAdded(ServerName serverName) {
489 defaultServerUpdater.serverChanged();
490 failedOpenUpdater.serverChanged();
491 }
492
493 @Override
494 public void serverRemoved(ServerName serverName) {
495 defaultServerUpdater.serverChanged();
496 }
497
498 private static class DefaultServerUpdater extends Thread {
499 private static final Log LOG = LogFactory.getLog(DefaultServerUpdater.class);
500 private RSGroupInfoManagerImpl mgr;
501 private volatile boolean hasChanged = false;
502
503 public DefaultServerUpdater(RSGroupInfoManagerImpl mgr) {
504 this.mgr = mgr;
505 setName(DefaultServerUpdater.class.getName()+"-" + mgr.master.getServerName());
506 setDaemon(true);
507 }
508
509 @Override
510 public void run() {
511 List<Address> prevDefaultServers = new LinkedList<Address>();
512 while (!mgr.master.isAborted() && !mgr.master.isStopped()) {
513 try {
514 if (LOG.isDebugEnabled()) {
515 LOG.debug("Updating default servers");
516 }
517 List<Address> servers = mgr.getDefaultServers();
518 Collections.sort(servers, new Comparator<Address>() {
519 @Override
520 public int compare(Address o1, Address o2) {
521 int diff = o1.getHostname().compareTo(o2.getHostname());
522 if (diff != 0) {
523 return diff;
524 }
525 return o1.getPort() - o2.getPort();
526 }
527 });
528 if(!servers.equals(prevDefaultServers)) {
529 mgr.updateDefaultServers(Sets.<Address>newHashSet(servers));
530 prevDefaultServers = servers;
531 LOG.info("Updated with servers: "+servers.size());
532 }
533 try {
534 synchronized (this) {
535 while (!hasChanged) {
536 wait();
537 }
538 hasChanged = false;
539 }
540 } catch (InterruptedException e) {
541 LOG.warn("Interrupted", e);
542 }
543 } catch (IOException e) {
544 LOG.warn("Failed to update default servers", e);
545 }
546 }
547 }
548
549
550 public void serverChanged() {
551 synchronized (this) {
552 hasChanged = true;
553 this.notify();
554 }
555 }
556 }
557
558 private static class FailedOpenUpdater extends Thread {
559 private static final Log LOG = LogFactory.getLog(FailedOpenUpdater.class);
560
561 private final RSGroupInfoManagerImpl mgr;
562 private final long waitInterval;
563 private volatile boolean hasChanged = false;
564
565 public FailedOpenUpdater(RSGroupInfoManagerImpl mgr) {
566 this.mgr = mgr;
567 this.waitInterval = mgr.master.getConfiguration().getLong(REASSIGN_WAIT_INTERVAL_KEY,
568 DEFAULT_REASSIGN_WAIT_INTERVAL);
569 setName(FailedOpenUpdater.class.getName()+"-" + mgr.master.getServerName());
570 setDaemon(true);
571 }
572
573 @Override
574 public void run() {
575 while (!mgr.master.isAborted() && !mgr.master.isStopped()) {
576 boolean interrupted = false;
577 try {
578 synchronized (this) {
579 while (!hasChanged) {
580 wait();
581 }
582 hasChanged = false;
583 }
584 } catch (InterruptedException e) {
585 LOG.warn("Interrupted", e);
586 interrupted = true;
587 }
588 if (mgr.master.isAborted() || mgr.master.isStopped() || interrupted) {
589 continue;
590 }
591
592
593 try {
594 Thread.sleep(waitInterval);
595 } catch (InterruptedException e) {
596 LOG.warn("Interrupted", e);
597 }
598 if (mgr.master.isAborted() || mgr.master.isStopped()) {
599 continue;
600 }
601
602
603 List<HRegionInfo> failedAssignments = Lists.newArrayList();
604 for (RegionState state:
605 mgr.master.getAssignmentManager().getRegionStates().getRegionsInTransition()) {
606 if (state.isFailedOpen()) {
607 failedAssignments.add(state.getRegion());
608 }
609 }
610 for (HRegionInfo region: failedAssignments) {
611 LOG.info("Retrying assignment of " + region);
612 mgr.master.getAssignmentManager().unassign(region);
613 }
614 }
615 }
616
617
618 public void serverChanged() {
619 synchronized (this) {
620 hasChanged = true;
621 this.notify();
622 }
623 }
624 }
625
626 @Override
627 public void waiting() {
628
629 }
630
631 private static class RSGroupStartupWorker extends Thread {
632 private static final Log LOG = LogFactory.getLog(RSGroupStartupWorker.class);
633
634 private volatile boolean isOnline = false;
635 private MasterServices masterServices;
636 private RSGroupInfoManagerImpl groupInfoManager;
637 private ClusterConnection conn;
638
639 public RSGroupStartupWorker(RSGroupInfoManagerImpl groupInfoManager,
640 MasterServices masterServices,
641 ClusterConnection conn) {
642 this.masterServices = masterServices;
643 this.groupInfoManager = groupInfoManager;
644 this.conn = conn;
645 setName(RSGroupStartupWorker.class.getName()+"-"+masterServices.getServerName());
646 setDaemon(true);
647 }
648
649 @Override
650 public void run() {
651 if(waitForGroupTableOnline()) {
652 LOG.info("GroupBasedLoadBalancer is now online");
653 }
654 }
655
656 public boolean waitForGroupTableOnline() {
657 final List<HRegionInfo> foundRegions = new LinkedList<HRegionInfo>();
658 final List<HRegionInfo> assignedRegions = new LinkedList<HRegionInfo>();
659 final AtomicBoolean found = new AtomicBoolean(false);
660 final TableStateManager tsm =
661 masterServices.getAssignmentManager().getTableStateManager();
662 boolean createSent = false;
663 while (!found.get() && isMasterRunning()) {
664 foundRegions.clear();
665 assignedRegions.clear();
666 found.set(true);
667 try {
668 boolean rootMetaFound =
669 masterServices.getMetaTableLocator().verifyMetaRegionLocation(
670 conn,
671 masterServices.getZooKeeper(),
672 1);
673 final AtomicBoolean nsFound = new AtomicBoolean(false);
674 if (rootMetaFound) {
675
676 MetaTableAccessor.Visitor visitor = new MetaTableAccessor.Visitor() {
677 @Override
678 public boolean visit(Result row) throws IOException {
679
680 HRegionInfo info = MetaTableAccessor.getHRegionInfo(row);
681 if (info != null) {
682 Cell serverCell =
683 row.getColumnLatestCell(HConstants.CATALOG_FAMILY,
684 HConstants.SERVER_QUALIFIER);
685 if (RSGROUP_TABLE_NAME.equals(info.getTable()) && serverCell != null) {
686 ServerName sn =
687 ServerName.parseVersionedServerName(CellUtil.cloneValue(serverCell));
688 if (sn == null) {
689 found.set(false);
690 } else if (tsm.isTableState(RSGROUP_TABLE_NAME,
691 ZooKeeperProtos.Table.State.ENABLED)) {
692 try {
693 ClientProtos.ClientService.BlockingInterface rs = conn.getClient(sn);
694 ClientProtos.GetRequest request =
695 RequestConverter.buildGetRequest(info.getRegionName(),
696 new Get(ROW_KEY));
697 rs.get(null, request);
698 assignedRegions.add(info);
699 } catch(Exception ex) {
700 LOG.debug("Caught exception while verifying group region", ex);
701 }
702 }
703 foundRegions.add(info);
704 }
705 if (TableName.NAMESPACE_TABLE_NAME.equals(info.getTable())) {
706 Cell cell = row.getColumnLatestCell(HConstants.CATALOG_FAMILY,
707 HConstants.SERVER_QUALIFIER);
708 ServerName sn = null;
709 if(cell != null) {
710 sn = ServerName.parseVersionedServerName(CellUtil.cloneValue(cell));
711 }
712 if (sn == null) {
713 nsFound.set(false);
714 } else if (tsm.isTableState(TableName.NAMESPACE_TABLE_NAME,
715 ZooKeeperProtos.Table.State.ENABLED)) {
716 try {
717 ClientProtos.ClientService.BlockingInterface rs = conn.getClient(sn);
718 ClientProtos.GetRequest request =
719 RequestConverter.buildGetRequest(info.getRegionName(),
720 new Get(ROW_KEY));
721 rs.get(null, request);
722 nsFound.set(true);
723 } catch(Exception ex) {
724 LOG.debug("Caught exception while verifying group region", ex);
725 }
726 }
727 }
728 }
729 return true;
730 }
731 };
732 MetaTableAccessor.fullScan(conn, visitor);
733
734 if (foundRegions.size() < 1 && rootMetaFound && !createSent && nsFound.get()) {
735 groupInfoManager.createGroupTable(masterServices);
736 createSent = true;
737 }
738 LOG.info("Group table: " + RSGROUP_TABLE_NAME + " isOnline: " + found.get()
739 + ", regionCount: " + foundRegions.size() + ", assignCount: "
740 + assignedRegions.size() + ", rootMetaFound: "+rootMetaFound);
741 found.set(found.get() && assignedRegions.size() == foundRegions.size()
742 && foundRegions.size() > 0);
743 } else {
744 LOG.info("Waiting for catalog tables to come online");
745 found.set(false);
746 }
747 if (found.get()) {
748 LOG.debug("With group table online, refreshing cached information.");
749 groupInfoManager.refresh(true);
750 isOnline = true;
751
752 groupInfoManager.flushConfig(groupInfoManager.rsGroupMap);
753 }
754 } catch (RuntimeException e) {
755 throw e;
756 } catch(Exception e) {
757 found.set(false);
758 LOG.warn("Failed to perform check", e);
759 }
760 try {
761 Thread.sleep(100);
762 } catch (InterruptedException e) {
763 LOG.info("Sleep interrupted", e);
764 }
765 }
766 return found.get();
767 }
768
769 public boolean isOnline() {
770 return isOnline;
771 }
772
773 private boolean isMasterRunning() {
774 return !masterServices.isAborted() && !masterServices.isStopped();
775 }
776 }
777
778 private void createGroupTable(MasterServices masterServices) throws IOException {
779 HRegionInfo[] newRegions =
780 ModifyRegionUtils.createHRegionInfos(RSGROUP_TABLE_DESC, null);
781 ProcedurePrepareLatch latch = ProcedurePrepareLatch.createLatch();
782 masterServices.getMasterProcedureExecutor().submitProcedure(
783 new CreateTableProcedure(
784 masterServices.getMasterProcedureExecutor().getEnvironment(),
785 RSGROUP_TABLE_DESC,
786 newRegions,
787 latch));
788 latch.await();
789
790 int tries = 600;
791 while(masterServices.getAssignmentManager().getRegionStates()
792 .getRegionServerOfRegion(newRegions[0]) == null && tries > 0) {
793 try {
794 Thread.sleep(100);
795 } catch (InterruptedException e) {
796 throw new IOException("Wait interrupted", e);
797 }
798 tries--;
799 }
800 if(tries <= 0) {
801 throw new IOException("Failed to create group table.");
802 }
803 }
804
805 private void multiMutate(List<Mutation> mutations)
806 throws IOException {
807 MutateRowsRequest.Builder mrmBuilder = MutateRowsRequest.newBuilder();
808 for (Mutation mutation : mutations) {
809 if (mutation instanceof Put) {
810 mrmBuilder.addMutationRequest(ProtobufUtil.toMutation(
811 ClientProtos.MutationProto.MutationType.PUT, mutation));
812 } else if (mutation instanceof Delete) {
813 mrmBuilder.addMutationRequest(ProtobufUtil.toMutation(
814 ClientProtos.MutationProto.MutationType.DELETE, mutation));
815 } else {
816 throw new DoNotRetryIOException("multiMutate doesn't support "
817 + mutation.getClass().getName());
818 }
819 }
820 MutateRowsRequest mrm = mrmBuilder.build();
821
822
823 conn.clearRegionCache(RSGROUP_TABLE_NAME);
824 try (Table rsGroupTable = conn.getTable(RSGROUP_TABLE_NAME)) {
825 CoprocessorRpcChannel channel = rsGroupTable.coprocessorService(ROW_KEY);
826 MultiRowMutationProtos.MultiRowMutationService.BlockingInterface service =
827 MultiRowMutationProtos.MultiRowMutationService.newBlockingStub(channel);
828 try {
829 service.mutateRows(null, mrm);
830 } catch (ServiceException ex) {
831 ProtobufUtil.toIOException(ex);
832 }
833 }
834 }
835
836 private void checkGroupName(String groupName) throws ConstraintException {
837 if(!groupName.matches("[a-zA-Z0-9_]+")) {
838 throw new ConstraintException("Group name should only contain alphanumeric characters");
839 }
840 }
841
842 @Override
843 public void moveServersAndTables(Set<Address> servers, Set<TableName> tables, String srcGroup,
844 String dstGroup) throws IOException {
845
846 RSGroupInfo srcGroupInfo = getRSGroup(srcGroup);
847 RSGroupInfo dstGroupInfo = getRSGroup(dstGroup);
848
849
850 for (Address el: servers) {
851 srcGroupInfo.removeServer(el);
852 dstGroupInfo.addServer(el);
853 }
854
855 for(TableName tableName: tables) {
856 srcGroupInfo.removeTable(tableName);
857 dstGroupInfo.addTable(tableName);
858 }
859
860
861 Map<String,RSGroupInfo> newGroupMap = Maps.newHashMap(rsGroupMap);
862 newGroupMap.put(srcGroupInfo.getName(), srcGroupInfo);
863 newGroupMap.put(dstGroupInfo.getName(), dstGroupInfo);
864 flushConfig(newGroupMap);
865 }
866
867 @Override
868 public synchronized void removeServers(Set<Address> servers) throws IOException {
869 Map<String, RSGroupInfo> rsGroupInfos = new HashMap<String, RSGroupInfo>();
870 for (Address el: servers) {
871 RSGroupInfo rsGroupInfo = getRSGroupOfServer(el);
872 if (rsGroupInfo != null) {
873 RSGroupInfo newRsGroupInfo = rsGroupInfos.get(rsGroupInfo.getName());
874 if (newRsGroupInfo == null) {
875 rsGroupInfo.removeServer(el);
876 rsGroupInfos.put(rsGroupInfo.getName(), rsGroupInfo);
877 } else {
878 newRsGroupInfo.removeServer(el);
879 rsGroupInfos.put(newRsGroupInfo.getName(), newRsGroupInfo);
880 }
881 } else {
882 LOG.warn("Server " + el + " does not belong to any rsgroup.");
883 }
884 }
885 if (rsGroupInfos.size() > 0) {
886 Map<String, RSGroupInfo> newGroupMap = Maps.newHashMap(rsGroupMap);
887 newGroupMap.putAll(rsGroupInfos);
888 flushConfig(newGroupMap);
889 }
890 }
891 }