modules/core/client-api/src/main/java/org/rhq/core/clientapi/server/discovery/DiscoveryServerService.java | 17 - modules/core/plugin-container/src/main/java/org/rhq/core/pc/inventory/InventoryManager.java | 149 +++++++--- modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/discovery/DiscoveryServerServiceImpl.java | 32 +- 3 files changed, 161 insertions(+), 37 deletions(-)
New commits: commit 4d3f721141b42c1e1ac6eed3a93124f51fbae3e6 Author: Jay Shaughnessy jshaughn@jshaughn.csb Date: Thu Jan 31 15:43:28 2013 -0500
Try to batch unknown resource merge in the inventory sync to help with large inventories and an agent start --purgeData - also fix issues with server service annotations
diff --git a/modules/core/client-api/src/main/java/org/rhq/core/clientapi/server/discovery/DiscoveryServerService.java b/modules/core/client-api/src/main/java/org/rhq/core/clientapi/server/discovery/DiscoveryServerService.java index 6b2fba6..04915ae 100644 --- a/modules/core/client-api/src/main/java/org/rhq/core/clientapi/server/discovery/DiscoveryServerService.java +++ b/modules/core/client-api/src/main/java/org/rhq/core/clientapi/server/discovery/DiscoveryServerService.java @@ -22,6 +22,7 @@ */ package org.rhq.core.clientapi.server.discovery;
+import java.util.List; import java.util.Map; import java.util.Set;
@@ -63,8 +64,8 @@ public interface DiscoveryServerService { @LimitedConcurrency(CONCURRENCY_LIMIT_INVENTORY_REPORT) @Timeout(0L) // should be something like 1000L * 60 * 30 but until we can be assured we never take longer, disable timeout - ResourceSyncInfo mergeInventoryReport(InventoryReport inventoryReport) - throws InvalidInventoryReportException, StaleTypeException; + ResourceSyncInfo mergeInventoryReport(InventoryReport inventoryReport) throws InvalidInventoryReportException, + StaleTypeException;
/** * Merges a new availability report from the agent into the server. This updates the availability statuses of known @@ -90,9 +91,19 @@ public interface DiscoveryServerService { * @param includeDescendants * @return a tree of resources with the latest data */ + @LimitedConcurrency(CONCURRENCY_LIMIT_INVENTORY_SYNC) Set<Resource> getResources(Set<Integer> resourceIds, boolean includeDescendants);
/** + * Returns the Resources with the given id's. The children are not set. + * + * @param resourceIds + * @return a list of resources in the same order as the passed in ids, with the latest data + */ + @LimitedConcurrency(CONCURRENCY_LIMIT_INVENTORY_SYNC) + List<Resource> getResourcesAsList(List<Integer> resourceIds); + + /** * Set the specified resource enabled or disabled. The call has no effect if the resource is already * in the desired state. * @@ -158,7 +169,7 @@ public interface DiscoveryServerService { * @return details on what resources have been upgraded with what data. */ Set<ResourceUpgradeResponse> upgradeResources(Set<ResourceUpgradeRequest> upgradeRequests); - + /** * Gives the server a chance to apply any necessary post-processing that's needed for newly committed resources * that have been successfully synchronized on the agent. diff --git a/modules/core/plugin-container/src/main/java/org/rhq/core/pc/inventory/InventoryManager.java b/modules/core/plugin-container/src/main/java/org/rhq/core/pc/inventory/InventoryManager.java index eaca661..780eaf8 100644 --- a/modules/core/plugin-container/src/main/java/org/rhq/core/pc/inventory/InventoryManager.java +++ b/modules/core/plugin-container/src/main/java/org/rhq/core/pc/inventory/InventoryManager.java @@ -120,6 +120,7 @@ import org.rhq.core.pluginapi.upgrade.ResourceUpgradeContext; import org.rhq.core.pluginapi.upgrade.ResourceUpgradeFacet; import org.rhq.core.system.SystemInfo; import org.rhq.core.system.SystemInfoFactory; +import org.rhq.core.util.StopWatch; import org.rhq.core.util.exception.ThrowableUtil; import org.rhq.core.util.exception.WrappedRemotingException;
@@ -851,8 +852,7 @@ public class InventoryManager extends AgentService implements ContainerService, // it will be accessible and editable by the user. Report the start exception at the end. handleInvalidPluginConfigurationResourceError(resource, t); throw new PluginContainerException("The resource [" + resource - + "] has been added but could not be started. Verify the supplied configuration values: ", - t); + + "] has been added but could not be started. Verify the supplied configuration values: ", t); } }
@@ -1108,7 +1108,7 @@ public class InventoryManager extends AgentService implements ContainerService, log.info("Syncing local inventory with Server inventory..."); long startTime = System.currentTimeMillis(); Set<Resource> syncedResources = new LinkedHashSet<Resource>(); - Set<Integer> unknownResourceIds = new LinkedHashSet<Integer>(); + Set<ResourceSyncInfo> unknownResourceSyncInfos = new LinkedHashSet<ResourceSyncInfo>(); Set<Integer> modifiedResourceIds = new LinkedHashSet<Integer>(); Set<Integer> deletedResourceIds = new LinkedHashSet<Integer>(); Set<Resource> newlyCommittedResources = new LinkedHashSet<Resource>(); @@ -1124,16 +1124,16 @@ public class InventoryManager extends AgentService implements ContainerService, }
log.debug("Processing Server sync info..."); - processSyncInfo(syncInfo, syncedResources, unknownResourceIds, modifiedResourceIds, deletedResourceIds, - newlyCommittedResources); + processSyncInfo(syncInfo, syncedResources, unknownResourceSyncInfos, modifiedResourceIds, + deletedResourceIds, newlyCommittedResources); if (log.isDebugEnabled()) { log.debug(String.format("DONE Processing sync info - took [%d] ms - synced [%d] Resources " + "- found [%d] unknown Resources and [%d] modified Resources.", - (System.currentTimeMillis() - startTime), syncedResources.size(), unknownResourceIds.size(), + (System.currentTimeMillis() - startTime), syncedResources.size(), unknownResourceSyncInfos.size(), modifiedResourceIds.size())); }
- mergeUnknownResources(unknownResourceIds); + mergeUnknownResources(unknownResourceSyncInfos); mergeModifiedResources(modifiedResourceIds); if (!partialInventory) { purgeObsoleteResources(allServerSideUuids); @@ -1157,7 +1157,7 @@ public class InventoryManager extends AgentService implements ContainerService, // the upgrade phase. Not to mention the fact that no thread pools are initialized yet by the // time the upgrade kicks in.. if (!isResourceUpgradeActive() - && (!syncedResources.isEmpty() || !unknownResourceIds.isEmpty() || !modifiedResourceIds.isEmpty())) { + && (!syncedResources.isEmpty() || !unknownResourceSyncInfos.isEmpty() || !modifiedResourceIds.isEmpty())) { performAvailabilityChecks(true); this.inventoryThreadPoolExecutor.schedule((Callable<? extends Object>) this.serviceScanExecutor, configuration.getChildResourceDiscoveryDelay(), TimeUnit.SECONDS); @@ -1610,8 +1610,9 @@ public class InventoryManager extends AgentService implements ContainerService, * @throws PluginContainerException * @return true the resource has been successfully prepared and can be started. False if the resource should not be started. */ - private boolean prepareResourceForActivation(Resource resource, @NotNull ResourceContainer container, - boolean forceReinitialization) throws InvalidPluginConfigurationException, PluginContainerException { + private boolean prepareResourceForActivation(Resource resource, @NotNull + ResourceContainer container, boolean forceReinitialization) throws InvalidPluginConfigurationException, + PluginContainerException {
if (resourceUpgradeDelegate.hasUpgradeFailed(resource)) { if (log.isTraceEnabled()) { @@ -1753,8 +1754,9 @@ public class InventoryManager extends AgentService implements ContainerService, * plugin configuration * @throws PluginContainerException for all other errors */ - public void activateResource(Resource resource, @NotNull ResourceContainer container, boolean updatedPluginConfig) - throws InvalidPluginConfigurationException, PluginContainerException { + public void activateResource(Resource resource, @NotNull + ResourceContainer container, boolean updatedPluginConfig) throws InvalidPluginConfigurationException, + PluginContainerException {
if (resourceUpgradeDelegate.hasUpgradeFailed(resource)) { if (log.isTraceEnabled()) { @@ -1831,8 +1833,7 @@ public class InventoryManager extends AgentService implements ContainerService, getOperationContext(resource), // for operation manager access getContentContext(resource), // for content manager access getAvailabilityContext(resource, this.availabilityCollectors), // for components that want to perform async avail checking - getInventoryContext(resource), - this.configuration.getPluginContainerDeployment()); // helps components make determinations of what to do + getInventoryContext(resource), this.configuration.getPluginContainerDeployment()); // helps components make determinations of what to do }
public <T extends ResourceComponent<?>> ResourceUpgradeContext<T> createResourceUpgradeContext(Resource resource, @@ -1851,8 +1852,7 @@ public class InventoryManager extends AgentService implements ContainerService, getOperationContext(resource), // for operation manager access getContentContext(resource), // for content manager access getAvailabilityContext(resource, this.availabilityCollectors), // for components that want avail manager access - getInventoryContext(resource), - this.configuration.getPluginContainerDeployment()); // helps components make determinations of what to do + getInventoryContext(resource), this.configuration.getPluginContainerDeployment()); // helps components make determinations of what to do }
/** @@ -2753,8 +2753,8 @@ public class InventoryManager extends AgentService implements ContainerService, }
private void processSyncInfo(ResourceSyncInfo syncInfo, Set<Resource> syncedResources, - Set<Integer> unknownResourceIds, Set<Integer> modifiedResourceIds, Set<Integer> deletedResourceIds, - Set<Resource> newlyCommittedResources) { + Set<ResourceSyncInfo> unknownResourceSyncInfos, Set<Integer> modifiedResourceIds, + Set<Integer> deletedResourceIds, Set<Resource> newlyCommittedResources) { if (InventoryStatus.DELETED == syncInfo.getInventoryStatus()) { // A previously deleted resource still being reported by the server. Support for this option can // be removed if the server is ever modified to not report deleted resources. It is happening currently @@ -2765,7 +2765,7 @@ public class InventoryManager extends AgentService implements ContainerService, ResourceContainer container = this.resourceContainers.get(syncInfo.getUuid()); if (container == null) { // Either a manually added Resource or just something we haven't discovered. - unknownResourceIds.add(syncInfo.getId()); + unknownResourceSyncInfos.add(syncInfo); log.info("Got unknown resource: " + syncInfo.getId()); } else { Resource resource = container.getResource(); @@ -2811,7 +2811,7 @@ public class InventoryManager extends AgentService implements ContainerService,
// Recurse... for (ResourceSyncInfo childSyncInfo : syncInfo.getChildSyncInfos()) { - processSyncInfo(childSyncInfo, syncedResources, unknownResourceIds, modifiedResourceIds, + processSyncInfo(childSyncInfo, syncedResources, unknownResourceSyncInfos, modifiedResourceIds, deletedResourceIds, newlyCommittedResources); } } @@ -2831,18 +2831,16 @@ public class InventoryManager extends AgentService implements ContainerService, } }
- private void mergeUnknownResources(Set<Integer> unknownResourceIds) { - if (log.isDebugEnabled()) { - log.debug("Merging [" + unknownResourceIds.size() - + "] unknown Resources and their descendants into local inventory..."); - } + private void mergeUnknownResources(Set<ResourceSyncInfo> unknownResourceSyncInfos) { + //TODO if (log.isDebugEnabled()) { + log.info("Merging [" + unknownResourceSyncInfos.size() + + "] unknown Resources and their descendants into local inventory..."); + //}
- if (!unknownResourceIds.isEmpty()) { + if (!unknownResourceSyncInfos.isEmpty()) { PluginMetadataManager pmm = this.pluginManager.getMetadataManager();
- Set<Resource> unknownResources = configuration.getServerServices().getDiscoveryServerService() - .getResources(unknownResourceIds, true); - + Set<Resource> unknownResources = getResourcesFromSyncInfos(unknownResourceSyncInfos); Set<Integer> toBeIgnored = new HashSet<Integer>();
for (Resource unknownResource : unknownResources) { @@ -2860,11 +2858,102 @@ public class InventoryManager extends AgentService implements ContainerService, } }
- unknownResourceIds.removeAll(toBeIgnored); + unknownResourceSyncInfos.removeAll(toBeIgnored); } return; }
+ private Set<Resource> getResourcesFromSyncInfos(Set<ResourceSyncInfo> syncInfos) { + + Set<Resource> result = new HashSet<Resource>(syncInfos.size()); + + for (ResourceSyncInfo syncInfo : syncInfos) { + Resource resource = getResourceFromSyncInfo(syncInfo); + result.add(resource); + } + + return result; + } + + static final int BATCH_SIZE = 500; + + private Resource getResourceFromSyncInfo(ResourceSyncInfo syncInfo) { + Resource result; + + List<Integer> resourceIdList = treeToBreadthFirstList(syncInfo); + + Map<Integer, Resource> resourceMap = new HashMap<Integer, Resource>(resourceIdList.size()); + + while (!resourceIdList.isEmpty()) { + int size = resourceIdList.size(); + int end = (BATCH_SIZE < size) ? BATCH_SIZE : size; + + List<Integer> resourceIdBatch = resourceIdList.subList(0, end); + List<Resource> resourceBatch = configuration.getServerServices().getDiscoveryServerService() + .getResourcesAsList(resourceIdBatch); + + // Advance our progress and possibly help GC. This will remove the processed resources from the backing list + resourceIdBatch.clear(); + + // add the newly fetched resources to the end of our master list + for (Resource r : resourceBatch) { + resourceMap.put(r.getId(), r); + } + + // Again, help the GC + resourceBatch.clear(); + } + + if (resourceIdList.size() != resourceMap.size()) { + log.warn("Expected [" + resourceIdList.size() + "] but found [" + resourceMap.size() + + "] resources when fetching from server"); + } + + StopWatch stopWatch = new StopWatch(); + + result = syncInfoTreeToResourceTree(syncInfo, null, resourceMap); + + //TODO if (log.isDebugEnabled()) { + log.info("syncInfoTreeToResourceTree time=[" + stopWatch + "]"); + //} + + return result; + } + + private Resource syncInfoTreeToResourceTree(ResourceSyncInfo syncInfo, Resource parentResource, + Map<Integer, Resource> resourceMap) { + Resource result = resourceMap.get(syncInfo.getId()); + + if (null == result || null == syncInfo.getChildSyncInfos()) { + return result; + } + + for (ResourceSyncInfo child : syncInfo.getChildSyncInfos()) { + Resource childResource = syncInfoTreeToResourceTree(child, result, resourceMap); + if (null != childResource) { + result.addChildResource(childResource); + } + } + + return result; + } + + private List<Integer> treeToBreadthFirstList(ResourceSyncInfo syncInfo) { + List<Integer> result = new ArrayList<Integer>(); + + LinkedList<ResourceSyncInfo> queue = new LinkedList<ResourceSyncInfo>(); + queue.add(syncInfo); + while (!queue.isEmpty()) { + ResourceSyncInfo node = queue.remove(); + result.add(node.getId()); + for (ResourceSyncInfo child : node.getChildSyncInfos()) { + queue.add(child); + } + } + + return result; + } + // private void print(Resource resourceTreeNode, int level) { // StringBuilder builder = new StringBuilder(); // for (int i = 0; i < level; i++) { diff --git a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/discovery/DiscoveryServerServiceImpl.java b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/discovery/DiscoveryServerServiceImpl.java index ebc0cd1..fe03746 100644 --- a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/discovery/DiscoveryServerServiceImpl.java +++ b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/discovery/DiscoveryServerServiceImpl.java @@ -18,7 +18,9 @@ */ package org.rhq.enterprise.server.discovery;
+import java.util.ArrayList; import java.util.HashSet; +import java.util.List; import java.util.Map; import java.util.Set;
@@ -130,10 +132,12 @@ public class DiscoveryServerServiceImpl implements DiscoveryServerService {
long elapsed = (System.currentTimeMillis() - start); if (elapsed > 20000L) { - log.warn("Performance: processed " + reportToString + " - needFull=[" + !ok + "] in (" + elapsed + ")ms"); + log.warn("Performance: processed " + reportToString + " - needFull=[" + !ok + "] in (" + elapsed + + ")ms"); } else { if (log.isDebugEnabled()) { - log.debug("Performance: processed " + reportToString + " - needFull=[" + !ok + "] in (" + elapsed + ")ms"); + log.debug("Performance: processed " + reportToString + " - needFull=[" + !ok + "] in (" + elapsed + + ")ms"); } }
@@ -167,6 +171,26 @@ public class DiscoveryServerServiceImpl implements DiscoveryServerService { }
@Override + public List<Resource> getResourcesAsList(List<Integer> resourceIds) { + long start = System.currentTimeMillis(); + ResourceManagerLocal resourceManager = LookupUtil.getResourceManager(); + List<Resource> result = new ArrayList<Resource>(resourceIds.size()); + for (Integer resourceId : resourceIds) { + //TODO: This can probably be one call to resource criteria fetch + Resource resource = resourceManager.getResourceTree(resourceId, false); + if (isVisibleInInventory(resource)) { + resource = convertToPojoResource(resource, false); + result.add(resource); + } + } + if (log.isDebugEnabled()) { + log.debug("Performance: get ResourcesAsList [" + resourceIds + "], timing (" + + (System.currentTimeMillis() - start) + ")ms"); + } + return result; + } + + @Override public Map<Integer, InventoryStatus> getInventoryStatus(int rootResourceId, boolean descendents) { long start = System.currentTimeMillis(); ResourceManagerLocal resourceManager = LookupUtil.getResourceManager(); @@ -244,8 +268,8 @@ public class DiscoveryServerServiceImpl implements DiscoveryServerService { }
private static boolean isVisibleInInventory(Resource resource) { - return resource.getInventoryStatus() != InventoryStatus.DELETED && - resource.getInventoryStatus() != InventoryStatus.UNINVENTORIED; + return resource.getInventoryStatus() != InventoryStatus.DELETED + && resource.getInventoryStatus() != InventoryStatus.UNINVENTORIED; }
@Override
rhq-commits@lists.fedorahosted.org