| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more contributor license |
| * agreements. See the NOTICE file distributed with this work for additional information regarding |
| * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance with the License. You may obtain a |
| * copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software distributed under the License |
| * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express |
| * or implied. See the License for the specific language governing permissions and limitations under |
| * the License. |
| */ |
| |
| package org.apache.geode.management.internal.cli.commands; |
| |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collections; |
| import java.util.HashSet; |
| import java.util.Iterator; |
| import java.util.List; |
| import java.util.Set; |
| import java.util.concurrent.Callable; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.Future; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.TimeoutException; |
| |
| import org.apache.commons.collections.CollectionUtils; |
| import org.apache.commons.lang3.ArrayUtils; |
| import org.apache.commons.lang3.StringUtils; |
| import org.springframework.shell.core.annotation.CliCommand; |
| import org.springframework.shell.core.annotation.CliOption; |
| |
| import org.apache.geode.cache.DataPolicy; |
| import org.apache.geode.cache.Region; |
| import org.apache.geode.cache.control.RebalanceFactory; |
| import org.apache.geode.cache.control.RebalanceOperation; |
| import org.apache.geode.cache.control.RebalanceResults; |
| import org.apache.geode.cache.control.ResourceManager; |
| import org.apache.geode.cache.execute.Function; |
| import org.apache.geode.cache.partition.PartitionRebalanceInfo; |
| import org.apache.geode.distributed.DistributedMember; |
| import org.apache.geode.internal.cache.InternalCache; |
| import org.apache.geode.logging.internal.executors.LoggingExecutors; |
| import org.apache.geode.management.DistributedRegionMXBean; |
| import org.apache.geode.management.ManagementService; |
| import org.apache.geode.management.cli.CliMetaData; |
| import org.apache.geode.management.cli.GfshCommand; |
| import org.apache.geode.management.cli.Result; |
| import org.apache.geode.management.internal.MBeanJMXAdapter; |
| import org.apache.geode.management.internal.cli.LogWrapper; |
| import org.apache.geode.management.internal.cli.functions.RebalanceFunction; |
| import org.apache.geode.management.internal.cli.i18n.CliStrings; |
| import org.apache.geode.management.internal.cli.result.model.InfoResultModel; |
| import org.apache.geode.management.internal.cli.result.model.ResultModel; |
| import org.apache.geode.management.internal.cli.result.model.TabularResultModel; |
| import org.apache.geode.management.internal.security.ResourceOperation; |
| import org.apache.geode.security.ResourcePermission; |
| |
| public class RebalanceCommand extends GfshCommand { |
| @CliCommand(value = CliStrings.REBALANCE, help = CliStrings.REBALANCE__HELP) |
| @CliMetaData(relatedTopic = {CliStrings.TOPIC_GEODE_DATA, CliStrings.TOPIC_GEODE_REGION}) |
| @ResourceOperation(resource = ResourcePermission.Resource.DATA, |
| operation = ResourcePermission.Operation.MANAGE) |
| public ResultModel rebalance( |
| @CliOption(key = CliStrings.REBALANCE__INCLUDEREGION, |
| help = CliStrings.REBALANCE__INCLUDEREGION__HELP) String[] includeRegions, |
| @CliOption(key = CliStrings.REBALANCE__EXCLUDEREGION, |
| help = CliStrings.REBALANCE__EXCLUDEREGION__HELP) String[] excludeRegions, |
| @CliOption(key = CliStrings.REBALANCE__TIMEOUT, unspecifiedDefaultValue = "-1", |
| help = CliStrings.REBALANCE__TIMEOUT__HELP) long timeout, |
| @CliOption(key = CliStrings.REBALANCE__SIMULATE, specifiedDefaultValue = "true", |
| unspecifiedDefaultValue = "false", |
| help = CliStrings.REBALANCE__SIMULATE__HELP) boolean simulate) |
| throws Exception { |
| |
| ExecutorService commandExecutors = |
| LoggingExecutors.newSingleThreadExecutor("RebalanceCommand", false); |
| List<Future<ResultModel>> commandResult = new ArrayList<>(); |
| ResultModel result; |
| try { |
| commandResult.add( |
| commandExecutors.submit(rebalanceCallable(includeRegions, excludeRegions, simulate))); |
| |
| Future<ResultModel> fs = commandResult.get(0); |
| if (timeout > 0) { |
| result = fs.get(timeout, TimeUnit.SECONDS); |
| } else { |
| result = fs.get(); |
| } |
| } catch (TimeoutException timeoutException) { |
| result = ResultModel.createInfo(CliStrings.REBALANCE__MSG__REBALANCE_WILL_CONTINUE); |
| } |
| |
| return result; |
| } |
| |
| public Callable<ResultModel> rebalanceCallable(String[] includeRegions, String[] excludeRegions, |
| boolean simulate) { |
| return new ExecuteRebalanceWithTimeout(includeRegions, excludeRegions, simulate, |
| (InternalCache) getCache()); |
| } |
| |
| private boolean checkResultList(InfoResultModel errors, List resultList, |
| DistributedMember member) { |
| boolean toContinueForOtherMembers = false; |
| if (CollectionUtils.isNotEmpty(resultList)) { |
| for (Object object : resultList) { |
| if (object instanceof Exception) { |
| errors.addLine( |
| CliStrings.format(CliStrings.REBALANCE__MSG__NO_EXECUTION, member.getId()) + ": " + |
| ((Exception) object).getMessage()); |
| |
| LogWrapper.getInstance(getCache()).info(CliStrings.REBALANCE__MSG__NO_EXECUTION |
| + member.getId() + " exception=" + ((Throwable) object).getMessage(), |
| ((Throwable) object)); |
| |
| toContinueForOtherMembers = true; |
| break; |
| } else if (object instanceof Throwable) { |
| errors.addLine( |
| CliStrings.format(CliStrings.REBALANCE__MSG__NO_EXECUTION, member.getId()) + ": " + |
| ((Throwable) object).getMessage()); |
| |
| LogWrapper.getInstance(getCache()).info(CliStrings.REBALANCE__MSG__NO_EXECUTION |
| + member.getId() + " exception=" + ((Throwable) object).getMessage(), |
| ((Throwable) object)); |
| |
| toContinueForOtherMembers = true; |
| break; |
| } |
| } |
| } else { |
| LogWrapper.getInstance(getCache()).info( |
| "Rebalancing for member=" + member.getId() + ", resultList is either null or empty"); |
| errors.addLine("Rebalancing for member=" + member.getId() + |
| ", resultList is either null or empty"); |
| toContinueForOtherMembers = true; |
| } |
| return toContinueForOtherMembers; |
| } |
| |
| private void toCompositeResultData(ResultModel result, |
| List<String> rstlist, int index, boolean simulate, InternalCache cache) { |
| final int resultItemCount = 9; |
| |
| if (rstlist.size() <= resultItemCount || StringUtils.isEmpty(rstlist.get(resultItemCount))) { |
| return; |
| } |
| |
| TabularResultModel table1 = result.addTable("Table" + index); |
| String newLine = System.getProperty("line.separator"); |
| StringBuilder resultStr = new StringBuilder(); |
| resultStr.append(newLine); |
| table1.accumulate("Rebalanced Stats", CliStrings.REBALANCE__MSG__TOTALBUCKETCREATEBYTES); |
| table1.accumulate("Value", rstlist.get(0)); |
| resultStr.append(CliStrings.REBALANCE__MSG__TOTALBUCKETCREATEBYTES).append(" = ") |
| .append(rstlist.get(0)).append(newLine); |
| table1.accumulate("Rebalanced Stats", CliStrings.REBALANCE__MSG__TOTALBUCKETCREATETIM); |
| table1.accumulate("Value", rstlist.get(1)); |
| resultStr.append(CliStrings.REBALANCE__MSG__TOTALBUCKETCREATETIM).append(" = ") |
| .append(rstlist.get(1)).append(newLine); |
| |
| table1.accumulate("Rebalanced Stats", CliStrings.REBALANCE__MSG__TOTALBUCKETCREATESCOMPLETED); |
| table1.accumulate("Value", rstlist.get(2)); |
| resultStr.append(CliStrings.REBALANCE__MSG__TOTALBUCKETCREATESCOMPLETED).append(" = ") |
| .append(rstlist.get(2)).append(newLine); |
| |
| table1.accumulate("Rebalanced Stats", CliStrings.REBALANCE__MSG__TOTALBUCKETTRANSFERBYTES); |
| table1.accumulate("Value", rstlist.get(3)); |
| resultStr.append(CliStrings.REBALANCE__MSG__TOTALBUCKETTRANSFERBYTES).append(" = ") |
| .append(rstlist.get(3)).append(newLine); |
| |
| table1.accumulate("Rebalanced Stats", CliStrings.REBALANCE__MSG__TOTALBUCKETTRANSFERTIME); |
| table1.accumulate("Value", rstlist.get(4)); |
| resultStr.append(CliStrings.REBALANCE__MSG__TOTALBUCKETTRANSFERTIME).append(" = ") |
| .append(rstlist.get(4)).append(newLine); |
| |
| table1.accumulate("Rebalanced Stats", |
| CliStrings.REBALANCE__MSG__TOTALBUCKETTRANSFERSCOMPLETED); |
| table1.accumulate("Value", rstlist.get(5)); |
| resultStr.append(CliStrings.REBALANCE__MSG__TOTALBUCKETTRANSFERSCOMPLETED).append(" = ") |
| .append(rstlist.get(5)).append(newLine); |
| |
| table1.accumulate("Rebalanced Stats", CliStrings.REBALANCE__MSG__TOTALPRIMARYTRANSFERTIME); |
| table1.accumulate("Value", rstlist.get(6)); |
| resultStr.append(CliStrings.REBALANCE__MSG__TOTALPRIMARYTRANSFERTIME).append(" = ") |
| .append(rstlist.get(6)).append(newLine); |
| |
| table1.accumulate("Rebalanced Stats", |
| CliStrings.REBALANCE__MSG__TOTALPRIMARYTRANSFERSCOMPLETED); |
| table1.accumulate("Value", rstlist.get(7)); |
| resultStr.append(CliStrings.REBALANCE__MSG__TOTALPRIMARYTRANSFERSCOMPLETED).append(" = ") |
| .append(rstlist.get(7)).append(newLine); |
| |
| table1.accumulate("Rebalanced Stats", CliStrings.REBALANCE__MSG__TOTALTIME); |
| table1.accumulate("Value", rstlist.get(8)); |
| resultStr.append(CliStrings.REBALANCE__MSG__TOTALTIME).append(" = ").append(rstlist.get(8)) |
| .append(newLine); |
| |
| String headerText; |
| if (simulate) { |
| headerText = "Simulated partition regions"; |
| } else { |
| headerText = "Rebalanced partition regions"; |
| } |
| for (int i = resultItemCount; i < rstlist.size(); i++) { |
| headerText = headerText + " " + rstlist.get(i); |
| } |
| table1.setHeader(headerText); |
| cache.getLogger().info(headerText + resultStr); |
| } |
| |
| |
| // TODO EY Move this to its own class |
| private class ExecuteRebalanceWithTimeout implements Callable<ResultModel> { |
| |
| String[] includeRegions = null; |
| String[] excludeRegions = null; |
| boolean simulate; |
| InternalCache cache = null; |
| |
| @Override |
| public ResultModel call() throws Exception { |
| ResultModel result = executeRebalanceWithTimeout(includeRegions, excludeRegions, simulate); |
| |
| // if the result contains only error section, i.e. no rebalance operation is done, mark this |
| // command result to be error. This would happy if user hasn't specified any valid region. If |
| // only one region specified is valid and rebalance is done, the result would be marked as |
| // success. |
| if (result.getSectionSize() == 1 && result.getInfoSection("error") != null) { |
| result.setStatus(Result.Status.ERROR); |
| } |
| |
| return result; |
| } |
| |
| ExecuteRebalanceWithTimeout(String[] includedRegions, String[] excludedRegions, |
| boolean toSimulate, InternalCache cache) { |
| includeRegions = includedRegions; |
| excludeRegions = excludedRegions; |
| simulate = toSimulate; |
| this.cache = cache; |
| } |
| |
| private ResultModel executeRebalanceWithTimeout(String[] includeRegions, |
| String[] excludeRegions, |
| boolean simulate) { |
| |
| ResultModel result = new ResultModel(); |
| try { |
| RebalanceOperation op; |
| |
| if (ArrayUtils.isNotEmpty(includeRegions)) { |
| InfoResultModel errors = result.addInfo("error"); |
| int index = 0; |
| |
| for (String regionName : includeRegions) { |
| |
| // To be removed after region Name specification with "/" is fixed |
| regionName = regionName.startsWith("/") ? regionName : ("/" + regionName); |
| Region region = cache.getRegion(regionName); |
| |
| if (region == null) { |
| DistributedMember member = getAssociatedMembers(regionName, cache); |
| |
| if (member == null) { |
| errors.addLine(CliStrings.format( |
| CliStrings.REBALANCE__MSG__NO_ASSOCIATED_DISTRIBUTED_MEMBER, regionName)); |
| continue; |
| } |
| |
| Function rebalanceFunction = new RebalanceFunction(); |
| Object[] functionArgs = new Object[3]; |
| functionArgs[0] = simulate ? "true" : "false"; |
| Set<String> setRegionName = new HashSet<>(); |
| setRegionName.add(regionName); |
| functionArgs[1] = setRegionName; |
| |
| Set<String> excludeRegionSet = new HashSet<>(); |
| if (ArrayUtils.isNotEmpty(excludeRegions)) { |
| Collections.addAll(excludeRegionSet, excludeRegions); |
| } |
| functionArgs[2] = excludeRegionSet; |
| |
| List resultList; |
| try { |
| resultList = (ArrayList) executeFunction(rebalanceFunction, functionArgs, member) |
| .getResult(); |
| } catch (Exception ex) { |
| LogWrapper.getInstance(cache) |
| .info(CliStrings.format( |
| CliStrings.REBALANCE__MSG__EXCEPTION_IN_REBALANCE_FOR_MEMBER_0_Exception_1, |
| member.getId(), ex.getMessage()), ex); |
| errors.addLine(CliStrings.format( |
| CliStrings.REBALANCE__MSG__EXCEPTION_IN_REBALANCE_FOR_MEMBER_0_Exception, |
| member.getId()) + ": " + ex.getMessage()); |
| continue; |
| } |
| |
| if (checkResultList(errors, resultList, member)) { |
| continue; |
| } |
| List<String> rstList = Arrays.asList(((String) resultList.get(0)).split(",")); |
| |
| toCompositeResultData(result, rstList, index, simulate, cache); |
| } else { |
| |
| ResourceManager manager = cache.getResourceManager(); |
| RebalanceFactory rbFactory = manager.createRebalanceFactory(); |
| Set<String> excludeRegionSet = new HashSet<>(); |
| if (excludeRegions != null) { |
| Collections.addAll(excludeRegionSet, excludeRegions); |
| } |
| rbFactory.excludeRegions(excludeRegionSet); |
| Set<String> includeRegionSet = new HashSet<>(); |
| includeRegionSet.add(regionName); |
| rbFactory.includeRegions(includeRegionSet); |
| |
| if (simulate) { |
| op = manager.createRebalanceFactory().simulate(); |
| } else { |
| op = manager.createRebalanceFactory().start(); |
| } |
| // Wait until the rebalance is complete and then get the results |
| buildResultForRebalance(result, op.getResults(), index, simulate, cache); |
| } |
| index++; |
| } |
| LogWrapper.getInstance(cache).info("Rebalance returning result " + result); |
| return result; |
| } else { |
| LogWrapper.getInstance(cache) |
| .info("Starting Rebalance simulate=" + simulate + " result >> " + result); |
| result = executeRebalanceOnDS(cache, String.valueOf(simulate), excludeRegions); |
| } |
| } catch (Exception e) { |
| result = ResultModel.createError(e.getMessage()); |
| } |
| LogWrapper.getInstance(cache).info("Rebalance returning result >>>" + result); |
| return result; |
| } |
| } |
| |
| private DistributedMember getAssociatedMembers(String region, final InternalCache cache) { |
| DistributedRegionMXBean bean = |
| ManagementService.getManagementService(cache).getDistributedRegionMXBean(region); |
| |
| DistributedMember member = null; |
| |
| if (bean == null) { |
| return null; |
| } |
| |
| String[] membersName = bean.getMembers(); |
| Set<DistributedMember> dsMembers = getAllMembers(); |
| Iterator it = dsMembers.iterator(); |
| |
| boolean matchFound = false; |
| |
| if (membersName.length > 1) { |
| while (it.hasNext() && !matchFound) { |
| DistributedMember dsmember = (DistributedMember) it.next(); |
| for (String memberName : membersName) { |
| if (MBeanJMXAdapter.getMemberNameOrUniqueId(dsmember).equals(memberName)) { |
| member = dsmember; |
| matchFound = true; |
| break; |
| } |
| } |
| } |
| } |
| return member; |
| } |
| |
| private void buildResultForRebalance(ResultModel resultModel, |
| RebalanceResults results, int index, boolean simulate, InternalCache cache) { |
| Set<PartitionRebalanceInfo> regions = results.getPartitionRebalanceDetails(); |
| Iterator iterator = regions.iterator(); |
| |
| // add only if there are valid number of regions |
| if (regions.size() > 0 |
| && StringUtils.isNotEmpty(((PartitionRebalanceInfo) iterator.next()).getRegionPath())) { |
| final TabularResultModel resultData = resultModel.addTable("Table" + index); |
| String newLine = System.getProperty("line.separator"); |
| StringBuilder resultStr = new StringBuilder(); |
| resultStr.append(newLine); |
| |
| resultData.accumulate("Rebalanced Stats", CliStrings.REBALANCE__MSG__TOTALBUCKETCREATEBYTES); |
| resultData.accumulate("Value", results.getTotalBucketCreateBytes() + ""); |
| resultStr.append(CliStrings.REBALANCE__MSG__TOTALBUCKETCREATEBYTES).append(" = ") |
| .append(results.getTotalBucketCreateBytes()).append(newLine); |
| |
| resultData.accumulate("Rebalanced Stats", CliStrings.REBALANCE__MSG__TOTALBUCKETCREATETIM); |
| resultData.accumulate("Value", results.getTotalBucketCreateTime() + ""); |
| resultStr.append(CliStrings.REBALANCE__MSG__TOTALBUCKETCREATETIM).append(" = ") |
| .append(results.getTotalBucketCreateTime()).append(newLine); |
| |
| resultData.accumulate("Rebalanced Stats", |
| CliStrings.REBALANCE__MSG__TOTALBUCKETCREATESCOMPLETED); |
| resultData.accumulate("Value", results.getTotalBucketCreatesCompleted() + ""); |
| resultStr.append(CliStrings.REBALANCE__MSG__TOTALBUCKETCREATESCOMPLETED).append(" = ") |
| .append(results.getTotalBucketCreatesCompleted()).append(newLine); |
| |
| resultData.accumulate("Rebalanced Stats", |
| CliStrings.REBALANCE__MSG__TOTALBUCKETTRANSFERBYTES); |
| resultData.accumulate("Value", results.getTotalBucketTransferBytes() + ""); |
| resultStr.append(CliStrings.REBALANCE__MSG__TOTALBUCKETTRANSFERBYTES).append(" = ") |
| .append(results.getTotalBucketTransferBytes()).append(newLine); |
| |
| resultData.accumulate("Rebalanced Stats", CliStrings.REBALANCE__MSG__TOTALBUCKETTRANSFERTIME); |
| resultData.accumulate("Value", results.getTotalBucketTransferTime() + ""); |
| resultStr.append(CliStrings.REBALANCE__MSG__TOTALBUCKETTRANSFERTIME).append(" = ") |
| .append(results.getTotalBucketTransferTime()).append(newLine); |
| |
| resultData.accumulate("Rebalanced Stats", |
| CliStrings.REBALANCE__MSG__TOTALBUCKETTRANSFERSCOMPLETED); |
| resultData.accumulate("Value", results.getTotalBucketTransfersCompleted() + ""); |
| resultStr.append(CliStrings.REBALANCE__MSG__TOTALBUCKETTRANSFERSCOMPLETED).append(" = ") |
| .append(results.getTotalBucketTransfersCompleted()).append(newLine); |
| |
| resultData.accumulate("Rebalanced Stats", |
| CliStrings.REBALANCE__MSG__TOTALPRIMARYTRANSFERTIME); |
| resultData.accumulate("Value", results.getTotalPrimaryTransferTime() + ""); |
| resultStr.append(CliStrings.REBALANCE__MSG__TOTALPRIMARYTRANSFERTIME).append(" = ") |
| .append(results.getTotalPrimaryTransferTime()).append(newLine); |
| |
| resultData.accumulate("Rebalanced Stats", |
| CliStrings.REBALANCE__MSG__TOTALPRIMARYTRANSFERSCOMPLETED); |
| resultData.accumulate("Value", results.getTotalPrimaryTransfersCompleted() + ""); |
| resultStr.append(CliStrings.REBALANCE__MSG__TOTALPRIMARYTRANSFERSCOMPLETED).append(" = ") |
| .append(results.getTotalPrimaryTransfersCompleted()).append(newLine); |
| |
| resultData.accumulate("Rebalanced Stats", CliStrings.REBALANCE__MSG__TOTALTIME); |
| resultData.accumulate("Value", results.getTotalTime() + ""); |
| resultStr.append(CliStrings.REBALANCE__MSG__TOTALTIME).append(" = ") |
| .append(results.getTotalTime()).append(newLine); |
| |
| Iterator<PartitionRebalanceInfo> it = regions.iterator(); |
| |
| String headerText; |
| |
| if (simulate) { |
| headerText = "Simulated partition regions "; |
| } else { |
| headerText = "Rebalanced partition regions "; |
| } |
| |
| while (it.hasNext()) { |
| PartitionRebalanceInfo rgn = it.next(); |
| headerText = headerText + " " + rgn.getRegionPath(); |
| } |
| resultData.setHeader(resultData.getHeader() + headerText); |
| |
| cache.getLogger().info(headerText + resultStr); |
| } |
| } |
| |
| private ResultModel executeRebalanceOnDS(InternalCache cache, String simulate, |
| String[] excludeRegionsList) { |
| ResultModel resultModel = new ResultModel(); |
| int index = 1; |
| InfoResultModel errors = resultModel.addInfo("errors"); |
| List<String> listExcludedRegion = new ArrayList<>(); |
| if (excludeRegionsList != null) { |
| Collections.addAll(listExcludedRegion, excludeRegionsList); |
| } |
| List<MemberPRInfo> listMemberRegion = getMemberRegionList(cache, listExcludedRegion); |
| |
| if (listMemberRegion.size() == 0) { |
| return ResultModel.createInfo(CliStrings.REBALANCE__MSG__NO_REBALANCING_REGIONS_ON_DS); |
| } |
| |
| Iterator<MemberPRInfo> iterator = listMemberRegion.iterator(); |
| boolean flagToContinueWithRebalance = false; |
| |
| // check if list has some members that can be rebalanced |
| while (iterator.hasNext()) { |
| if (iterator.next().dsMemberList.size() > 1) { |
| flagToContinueWithRebalance = true; |
| break; |
| } |
| } |
| |
| if (!flagToContinueWithRebalance) { |
| return ResultModel.createInfo(CliStrings.REBALANCE__MSG__NO_REBALANCING_REGIONS_ON_DS); |
| } |
| |
| for (MemberPRInfo memberPR : listMemberRegion) { |
| try { |
| // check if there are more than one members associated with region for rebalancing |
| if (memberPR.dsMemberList.size() > 1) { |
| for (int i = 0; i < memberPR.dsMemberList.size(); i++) { |
| DistributedMember dsMember = memberPR.dsMemberList.get(i); |
| Function rebalanceFunction = new RebalanceFunction(); |
| Object[] functionArgs = new Object[3]; |
| functionArgs[0] = simulate; |
| Set<String> regionSet = new HashSet<>(); |
| |
| regionSet.add(memberPR.region); |
| functionArgs[1] = regionSet; |
| |
| Set<String> excludeRegionSet = new HashSet<>(); |
| functionArgs[2] = excludeRegionSet; |
| |
| List resultList = null; |
| |
| try { |
| if (checkMemberPresence(dsMember)) { |
| resultList = (ArrayList) executeFunction(rebalanceFunction, functionArgs, dsMember) |
| .getResult(); |
| |
| if (checkResultList(errors, resultList, dsMember)) { |
| continue; |
| } |
| |
| List<String> rstList = Arrays.asList(((String) resultList.get(0)).split(",")); |
| toCompositeResultData(resultModel, rstList, index, simulate.equals("true"), cache); |
| index++; |
| |
| // Rebalancing for region is done so break and continue with other region |
| break; |
| } else { |
| if (i == memberPR.dsMemberList.size() - 1) { |
| errors.addLine( |
| CliStrings.format( |
| CliStrings.REBALANCE__MSG__NO_EXECUTION_FOR_REGION_0_ON_MEMBERS_1, |
| memberPR.region, listOfAllMembers(memberPR.dsMemberList)) + ", " + |
| CliStrings.REBALANCE__MSG__MEMBERS_MIGHT_BE_DEPARTED); |
| } else { |
| continue; |
| } |
| } |
| } catch (Exception ex) { |
| if (i == memberPR.dsMemberList.size() - 1) { |
| errors.addLine( |
| CliStrings.format( |
| CliStrings.REBALANCE__MSG__NO_EXECUTION_FOR_REGION_0_ON_MEMBERS_1, |
| memberPR.region, listOfAllMembers(memberPR.dsMemberList)) + ", " + |
| CliStrings.REBALANCE__MSG__REASON + ex.getMessage()); |
| } else { |
| continue; |
| } |
| } |
| |
| if (checkResultList(errors, resultList, dsMember)) { |
| continue; |
| } |
| |
| List<String> rstList = Arrays.asList(((String) resultList.get(0)).split(",")); |
| toCompositeResultData(resultModel, rstList, index, simulate.equals("true"), cache); |
| index++; |
| } |
| } |
| } catch (Exception e) { |
| return ResultModel.createError(e.getMessage()); |
| } |
| } |
| return resultModel; |
| } |
| |
| private static class MemberPRInfo { |
| ArrayList<DistributedMember> dsMemberList; |
| public String region; |
| |
| MemberPRInfo() { |
| region = ""; |
| dsMemberList = new ArrayList<>(); |
| } |
| |
| @Override |
| public boolean equals(Object o2) { |
| return o2 != null && this.region.equals(((MemberPRInfo) o2).region); |
| } |
| |
| @Override |
| public int hashCode() { |
| return this.region.hashCode(); |
| } |
| } |
| |
| private List<MemberPRInfo> getMemberRegionList(InternalCache cache, |
| List<String> listExcludedRegion) { |
| List<MemberPRInfo> listMemberPRInfo = new ArrayList<>(); |
| String[] listDSRegions = |
| ManagementService.getManagementService(cache).getDistributedSystemMXBean().listRegions(); |
| final Set<DistributedMember> dsMembers = getAllMembers(); |
| |
| for (String regionName : listDSRegions) { |
| // check for excluded regions |
| boolean excludedRegionMatch = false; |
| for (String aListExcludedRegion : listExcludedRegion) { |
| // this is needed since region name may start with / or without it |
| // also |
| String excludedRegion = aListExcludedRegion.trim(); |
| if (regionName.startsWith("/")) { |
| if (!excludedRegion.startsWith("/")) { |
| excludedRegion = "/" + excludedRegion; |
| } |
| } |
| if (excludedRegion.startsWith("/")) { |
| if (!regionName.startsWith("/")) { |
| regionName = "/" + regionName; |
| } |
| } |
| |
| if (excludedRegion.equals(regionName)) { |
| excludedRegionMatch = true; |
| break; |
| } |
| } |
| |
| if (excludedRegionMatch) { |
| // ignore this region |
| continue; |
| } |
| |
| if (!regionName.startsWith("/")) { |
| regionName = Region.SEPARATOR + regionName; |
| } |
| // remove this prefix / |
| DistributedRegionMXBean bean = |
| ManagementService.getManagementService(cache).getDistributedRegionMXBean(regionName); |
| |
| if (bean != null) { |
| if (bean.getRegionType().equals(DataPolicy.PARTITION.toString()) |
| || bean.getRegionType().equals(DataPolicy.PERSISTENT_PARTITION.toString())) { |
| |
| String[] memberNames = bean.getMembers(); |
| for (DistributedMember dsmember : dsMembers) { |
| for (String memberName : memberNames) { |
| if (MBeanJMXAdapter.getMemberNameOrUniqueId(dsmember).equals(memberName)) { |
| MemberPRInfo memberAndItsPRRegions = new MemberPRInfo(); |
| memberAndItsPRRegions.region = regionName; |
| memberAndItsPRRegions.dsMemberList.add(dsmember); |
| if (listMemberPRInfo.contains(memberAndItsPRRegions)) { |
| // add member for appropriate region |
| int index = listMemberPRInfo.indexOf(memberAndItsPRRegions); |
| MemberPRInfo listMember = listMemberPRInfo.get(index); |
| listMember.dsMemberList.add(dsmember); |
| } else { |
| listMemberPRInfo.add(memberAndItsPRRegions); |
| } |
| break; |
| } |
| } |
| } |
| } |
| } |
| } |
| |
| return listMemberPRInfo; |
| } |
| |
| private boolean checkMemberPresence(DistributedMember dsMember) { |
| // check if member's presence just before executing function |
| // this is to avoid running a function on departed members #47248 |
| Set<DistributedMember> dsMemberList = getAllNormalMembers(); |
| return dsMemberList.contains(dsMember); |
| } |
| |
| private String listOfAllMembers(ArrayList<DistributedMember> dsMemberList) { |
| StringBuilder listMembersId = new StringBuilder(); |
| for (int j = 0; j < dsMemberList.size() - 1; j++) { |
| listMembersId.append(dsMemberList.get(j).getId()); |
| listMembersId.append(" ; "); |
| } |
| return listMembersId.toString(); |
| } |
| } |
