[rhq] modules/enterprise
by lkrejci
modules/enterprise/binding/src/main/java/org/rhq/bindings/script/BaseRhqSchemeScriptSourceProvider.java | 7 -
modules/enterprise/scripting/api/src/main/java/org/rhq/scripting/ScriptSourceProvider.java | 16 ++
modules/enterprise/server/client-api/src/main/java/org/rhq/enterprise/client/RhqDownloadsScriptSourceProvider.java | 66 ++++++++--
modules/enterprise/server/client-api/src/test/java/org/rhq/enterprise/client/test/RhqDownloadsScriptSourceProviderTest.java | 4
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/core/CoreServer.java | 27 +++-
5 files changed, 102 insertions(+), 18 deletions(-)
New commits:
commit eeffaed93e68612eb6a88226479702d4f373e510
Author: Lukas Krejci <lkrejci(a)redhat.com>
Date: Tue May 28 03:11:56 2013 +0200
[BZ 967622 - Server CLI script can't require modules from rhq://downloads]
This was primarily caused by a new permission required by EAP 6.1 we've
switched to. To access the ModelControllerClient, the code now needs a
RuntimePermission("canAccessModelController").
We need the ModelControllerClient when determining the EAR installation
dir, which we in turn need when determining where to locate the scripts
in the "rhq://downloads" location. This is done while running a script,
which is done in a restricted access control context which does not and
should not have that permission.
The minimal fix is to wrap the getting of the ModelControllerClient in a
privileged action but some more "defensive" code was added in the
RhqDownloadScriptSourceProvider, too, to guard against different times it
might get instantiated during the script execution.
Javadocs were updated to warn about the security considerations when
writing a script source provider available on the server side.
diff --git a/modules/enterprise/binding/src/main/java/org/rhq/bindings/script/BaseRhqSchemeScriptSourceProvider.java b/modules/enterprise/binding/src/main/java/org/rhq/bindings/script/BaseRhqSchemeScriptSourceProvider.java
index 4c6756f..87241ef 100644
--- a/modules/enterprise/binding/src/main/java/org/rhq/bindings/script/BaseRhqSchemeScriptSourceProvider.java
+++ b/modules/enterprise/binding/src/main/java/org/rhq/bindings/script/BaseRhqSchemeScriptSourceProvider.java
@@ -55,8 +55,11 @@ public abstract class BaseRhqSchemeScriptSourceProvider implements ScriptSourceP
* The base implementation of the {@link #getScriptSource(URI)} method
* only checks that the scheme of the URI is "rhq" and that the scheme
* specific part starts with "//".
- *
- * @param scriptUri
+ * <p/>
+ * Please follow the general suggestions mentioned in {@link ScriptSourceProvider#getScriptSource(java.net.URI)}
+ * docs.
+ *
+ * @param scriptUri the URI to load the script from
* @return the reader of the script or null if the script could not be
* found using the URI
*
diff --git a/modules/enterprise/scripting/api/src/main/java/org/rhq/scripting/ScriptSourceProvider.java b/modules/enterprise/scripting/api/src/main/java/org/rhq/scripting/ScriptSourceProvider.java
index e65f793..fb0f606 100644
--- a/modules/enterprise/scripting/api/src/main/java/org/rhq/scripting/ScriptSourceProvider.java
+++ b/modules/enterprise/scripting/api/src/main/java/org/rhq/scripting/ScriptSourceProvider.java
@@ -29,14 +29,26 @@ import java.net.URI;
* <p>
* Implementations of this interface can be located using the {@link ScriptSourceProviderFactory}
* if they are registered in META-INF/services.
- *
+ * <p/>
+ * Note that instances of this class can be created and called in an access control context with limited privileges.
+ * If you need to make safe calls that require privileges not granted to a script run in the RHQ server (by default this
+ * is determined by the {@code org.rhq.bindings.StandardScriptPermissions} class), make sure to call such actions with
+ * elevated permissions through
+ * {@link java.security.AccessController#doPrivileged(java.security.PrivilegedExceptionAction)} or any of its
+ * derivatives.
+ * <p/>
+ * For example JNDI look-ups are not allowed by default for the scripts, so if your provider needs to perform some
+ * JNDI lookups to locate the script to include, you need to wrap any code that does a JNDI look-up as above.
+ *
* @author Lukas Krejci
*/
public interface ScriptSourceProvider {
/**
* Returns the reader of the source of the script specified by given location.
- *
+ * <p/>
+ * Review the class description for the security considerations.
+ *
* @param location the location of the script
* @return the reader of the script source or null if it could not be found
*/
diff --git a/modules/enterprise/server/client-api/src/main/java/org/rhq/enterprise/client/RhqDownloadsScriptSourceProvider.java b/modules/enterprise/server/client-api/src/main/java/org/rhq/enterprise/client/RhqDownloadsScriptSourceProvider.java
index 1cbbe39..23e8fb5 100644
--- a/modules/enterprise/server/client-api/src/main/java/org/rhq/enterprise/client/RhqDownloadsScriptSourceProvider.java
+++ b/modules/enterprise/server/client-api/src/main/java/org/rhq/enterprise/client/RhqDownloadsScriptSourceProvider.java
@@ -26,6 +26,11 @@ import java.io.FileNotFoundException;
import java.io.InputStreamReader;
import java.io.Reader;
import java.net.URI;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
+import java.security.PrivilegedActionException;
+import java.security.PrivilegedExceptionAction;
+import java.util.concurrent.Callable;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -45,20 +50,67 @@ public class RhqDownloadsScriptSourceProvider extends BaseRhqSchemeScriptSourceP
private CoreServerMBean coreServer;
+ /**
+ * A bunch of methods optionally enclosed in privileged actions to ensure that we can access the information
+ * inside RHQ server even though it might not be accessible in the access control context the scripts run in
+ * in the RHQ server.
+ */
+ private static class SecurityActions {
+ private SecurityActions() {
+
+ }
+
+ private static CoreServerMBean lookupCoreServer() {
+ if (System.getSecurityManager() == null) {
+ return LookupUtil.getCoreServer();
+ } else {
+ return AccessController.doPrivileged(new PrivilegedAction<CoreServerMBean>() {
+ @Override
+ public CoreServerMBean run() {
+ return LookupUtil.getCoreServer();
+ }
+ });
+ }
+ }
+
+ private static File getDownloadDir(final CoreServerMBean coreServer) {
+ File earDeployDir;
+ if (System.getSecurityManager() == null) {
+ earDeployDir = coreServer.getEarDeploymentDir();
+ } else {
+ earDeployDir = AccessController.doPrivileged(new PrivilegedAction<File>() {
+ @Override
+ public File run() {
+ return coreServer.getEarDeploymentDir();
+ }
+ });
+ }
+ File downloadDir = new File(earDeployDir, "rhq-downloads");
+ return downloadDir;
+ }
+ }
+
public RhqDownloadsScriptSourceProvider() {
- this(LookupUtil.getCoreServer());
+ //we need to do a safe JNDI lookup, but we may run in unprivileged context...
+ this(SecurityActions.lookupCoreServer());
}
-
+
+ /**
+ * This is meant only for testing purposes.
+ *
+ * @param coreServerMBean the CoreServer MBean to use
+ */
public RhqDownloadsScriptSourceProvider(CoreServerMBean coreServerMBean) {
super(AUTHORITY);
this.coreServer = coreServerMBean;
}
-
+
@Override
protected Reader doGetScriptSource(URI scriptUri) {
String path = scriptUri.getPath().substring(1); //remove the leading /
- File downloadsDir = getDownloadHomeDir();
+ //We're going to be doing an MBean call here which the scripts don't have privs for by default
+ File downloadsDir = SecurityActions.getDownloadDir(coreServer);
File scriptDownloads = new File(downloadsDir, "script-modules");
File file = new File(scriptDownloads, path);
@@ -69,10 +121,4 @@ public class RhqDownloadsScriptSourceProvider extends BaseRhqSchemeScriptSourceP
return null;
}
}
-
- private File getDownloadHomeDir() {
- File earDeployDir = this.coreServer.getEarDeploymentDir();
- File downloadDir = new File(earDeployDir, "rhq-downloads");
- return downloadDir;
- }
}
diff --git a/modules/enterprise/server/client-api/src/test/java/org/rhq/enterprise/client/test/RhqDownloadsScriptSourceProviderTest.java b/modules/enterprise/server/client-api/src/test/java/org/rhq/enterprise/client/test/RhqDownloadsScriptSourceProviderTest.java
index e62dd32..1b366ba 100644
--- a/modules/enterprise/server/client-api/src/test/java/org/rhq/enterprise/client/test/RhqDownloadsScriptSourceProviderTest.java
+++ b/modules/enterprise/server/client-api/src/test/java/org/rhq/enterprise/client/test/RhqDownloadsScriptSourceProviderTest.java
@@ -85,8 +85,8 @@ public class RhqDownloadsScriptSourceProviderTest {
try {
String contents = StreamUtil.slurp(rdr);
assertEquals(contents, EXPECTED_CONTENTS, "Unexpected script loaded");
- } finally {
- rdr.close();
+ } finally {
+ StreamUtil.safeClose(rdr);
}
}
}
diff --git a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/core/CoreServer.java b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/core/CoreServer.java
index 9a60f12..b74b704 100644
--- a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/core/CoreServer.java
+++ b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/core/CoreServer.java
@@ -22,6 +22,8 @@ import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.lang.management.ManagementFactory;
+import java.security.AccessController;
+import java.security.PrivilegedAction;
import java.util.Date;
import java.util.HashMap;
import java.util.Properties;
@@ -69,6 +71,24 @@ public class CoreServer implements CoreServerMBean {
*/
private static final String PROP_BUILD_DATE = "Build-Date";
+ private static class SecurityActions {
+ private SecurityActions() {
+
+ }
+
+ private static ModelControllerClient getModelControllerClient() {
+ if (System.getSecurityManager() == null) {
+ return ManagementService.getClient();
+ } else {
+ return AccessController.doPrivileged(new PrivilegedAction<ModelControllerClient>() {
+ @Override
+ public ModelControllerClient run() {
+ return ManagementService.getClient();
+ }
+ });
+ }
+ }
+ }
private Properties buildProps;
private Date bootTime;
@@ -136,7 +156,10 @@ public class CoreServer implements CoreServerMBean {
@Override
public File getEarDeploymentDir() {
- ModelControllerClient mcc = ManagementService.getClient();
+ //Getting model controller client requires privs our callers might not have, but we want to provide this
+ //function even to them.
+ ModelControllerClient mcc = SecurityActions.getModelControllerClient();
+
try {
DeploymentJBossASClient client = new DeploymentJBossASClient(mcc);
String earPath = client.getDeploymentPath(RHQConstants.EAR_FILE_NAME);
@@ -229,4 +252,4 @@ public class CoreServer implements CoreServerMBean {
return buildProps;
}
-}
\ No newline at end of file
+}
10 years, 11 months
[rhq] Branch 'feature/cassandra-backend' - modules/core modules/enterprise
by Jiri Kremser
modules/core/domain/src/main/java/org/rhq/core/domain/cloud/StorageNode.java | 24 +
modules/core/domain/src/main/java/org/rhq/core/domain/cloud/StorageNodeLoadComposite.java | 23 -
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/StorageNodeManagerBean.java | 150 +++++-----
3 files changed, 120 insertions(+), 77 deletions(-)
New commits:
commit 5a5a8c931a96fcd5f02a3a51c542afc2afd121b4
Author: Jirka Kremser <jkremser(a)redhat.com>
Date: Mon May 27 23:34:31 2013 +0200
Making StorageNodeManagerBean.getLoad() method more efficient using named queries. Support for correct unit formatting of StorageNodeLoadComposite.toString().
diff --git a/modules/core/domain/src/main/java/org/rhq/core/domain/cloud/StorageNode.java b/modules/core/domain/src/main/java/org/rhq/core/domain/cloud/StorageNode.java
index 1970349..e967549 100644
--- a/modules/core/domain/src/main/java/org/rhq/core/domain/cloud/StorageNode.java
+++ b/modules/core/domain/src/main/java/org/rhq/core/domain/cloud/StorageNode.java
@@ -40,6 +40,7 @@ import javax.persistence.PrePersist;
import javax.persistence.SequenceGenerator;
import javax.persistence.Table;
+import org.rhq.core.domain.measurement.MeasurementSchedule;
import org.rhq.core.domain.resource.Resource;
/**
@@ -59,7 +60,26 @@ import org.rhq.core.domain.resource.Resource;
@NamedQuery(name = StorageNode.QUERY_FIND_ALL_NOT_INSTALLED, query = "SELECT s FROM StorageNode s WHERE NOT s.operationMode = 'INSTALLED'"),
@NamedQuery(name = StorageNode.QUERY_FIND_ALL_NORMAL, query = "SELECT s FROM StorageNode s WHERE s.operationMode = 'NORMAL'"),
@NamedQuery(name = StorageNode.QUERY_DELETE_BY_ID, query = "" //
- + "DELETE FROM StorageNode s WHERE s.id = :storageNodeId ") })
+ + "DELETE FROM StorageNode s WHERE s.id = :storageNodeId "),
+
+ @NamedQuery(name = StorageNode.QUERY_FIND_SCHEDULE_IDS_BY_PARENT_RESOURCE_ID_AND_MEASUREMENT_DEFINITION_NAMES, query = "" //
+ + " SELECT def.name, ms.id FROM MeasurementSchedule ms " //
+ + " JOIN ms.definition def " //
+ + " JOIN ms.resource res " //
+ + " WHERE ms.definition = def " //
+ + " AND res.parentResource.id = :parrentId " //
+ + " AND ms.enabled = true" //
+ + " AND def.name IN (:metricNames)"), //
+
+ @NamedQuery(name = StorageNode.QUERY_FIND_SCHEDULE_IDS_BY_GRANDPARENT_RESOURCE_ID_AND_MEASUREMENT_DEFINITION_NAMES, query = "" //
+ + " SELECT def.name, ms.id FROM MeasurementSchedule ms " //
+ + " JOIN ms.definition def " //
+ + " JOIN ms.resource res " //
+ + " WHERE ms.definition = def " //
+ + " AND res.parentResource.parentResource.id = :grandparrentId " //
+ + " AND ms.enabled = true" //
+ + " AND def.name IN (:metricNames)") //
+})
@SequenceGenerator(allocationSize = org.rhq.core.domain.util.Constants.ALLOCATION_SIZE, name = "RHQ_STORAGE_NODE_ID_SEQ", sequenceName = "RHQ_STORAGE_NODE_ID_SEQ")
@Table(name = "RHQ_STORAGE_NODE")
public class StorageNode implements Serializable {
@@ -71,6 +91,8 @@ public class StorageNode implements Serializable {
public static final String QUERY_FIND_ALL_NOT_INSTALLED = "StorageNode.findAllCloudMembers";
public static final String QUERY_DELETE_BY_ID = "StorageNode.deleteById";
public static final String QUERY_FIND_ALL_NORMAL = "StorageNode.findAllNormalCloudMembers";
+ public static final String QUERY_FIND_SCHEDULE_IDS_BY_PARENT_RESOURCE_ID_AND_MEASUREMENT_DEFINITION_NAMES = "StorageNode.findScheduleIdsByParentResourceIdAndMeasurementDefinitionNames";
+ public static final String QUERY_FIND_SCHEDULE_IDS_BY_GRANDPARENT_RESOURCE_ID_AND_MEASUREMENT_DEFINITION_NAMES = "StorageNode.findScheduleIdsByGrandparentResourceIdAndMeasurementDefinitionNames";
private static final String JMX_CONNECTION_STRING = "service:jmx:rmi:///jndi/rmi://%s:%s/jmxrmi";
diff --git a/modules/core/domain/src/main/java/org/rhq/core/domain/cloud/StorageNodeLoadComposite.java b/modules/core/domain/src/main/java/org/rhq/core/domain/cloud/StorageNodeLoadComposite.java
index b7766e9..e93670e 100644
--- a/modules/core/domain/src/main/java/org/rhq/core/domain/cloud/StorageNodeLoadComposite.java
+++ b/modules/core/domain/src/main/java/org/rhq/core/domain/cloud/StorageNodeLoadComposite.java
@@ -23,7 +23,6 @@
package org.rhq.core.domain.cloud;
import java.io.Serializable;
-import java.util.concurrent.TimeUnit;
import org.rhq.core.domain.measurement.MeasurementAggregate;
import org.rhq.core.domain.measurement.MeasurementUnits;
@@ -38,15 +37,11 @@ public class StorageNodeLoadComposite implements Serializable {
private long beginTime;
private long endTime;
- private MeasurementAggregateWithUnits heapCommited; // cassandra server jvm / memory subsystem resource
-
- private MeasurementAggregateWithUnits heapUsed; // cassandra server jvm / memory subsystem resource
-
- private MeasurementAggregateWithUnits load; // database management services / storage service resource
-
- private MeasurementAggregate tokens; // ~ jmx op - getTokens(hostname).size() or jmx attribute (StorageService/tokens).size()
-
- private MeasurementAggregateWithUnits actuallyOwns; // up to date # of tokens for this node (tokenToEndpointMap jmx attr.) can be taken from the associated resource's configuration or I can provide this as a metric in C* plugin
+ private MeasurementAggregateWithUnits heapCommited;
+ private MeasurementAggregateWithUnits heapUsed;
+ private MeasurementAggregateWithUnits load;
+ private MeasurementAggregate tokens;
+ private MeasurementAggregateWithUnits actuallyOwns;
public StorageNodeLoadComposite() {
// GWT needs this
@@ -150,6 +145,7 @@ public class StorageNodeLoadComposite implements Serializable {
private final MeasurementAggregate aggregate;
private final MeasurementUnits units;
+ private String formattedValue;
public MeasurementAggregateWithUnits(MeasurementAggregate aggregate, MeasurementUnits units) {
this.aggregate = aggregate;
@@ -164,8 +160,15 @@ public class StorageNodeLoadComposite implements Serializable {
return units;
}
+ public void setFormattedValue(String formattedValue) {
+ this.formattedValue = formattedValue;
+ }
+
@Override
public String toString() {
+ if (formattedValue != null) {
+ return formattedValue;
+ }
return aggregate.toString() + " (" + units.toString() + ")";
}
}
diff --git a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/StorageNodeManagerBean.java b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/StorageNodeManagerBean.java
index 8b874ec..89ffe11 100644
--- a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/StorageNodeManagerBean.java
+++ b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/StorageNodeManagerBean.java
@@ -25,12 +25,11 @@
package org.rhq.enterprise.server.cloud;
import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collection;
-import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
-import java.util.Set;
import javax.annotation.PostConstruct;
import javax.ejb.EJB;
@@ -48,13 +47,9 @@ import org.rhq.core.domain.cloud.StorageNode;
import org.rhq.core.domain.cloud.StorageNode.OperationMode;
import org.rhq.core.domain.cloud.StorageNodeLoadComposite;
import org.rhq.core.domain.configuration.Configuration;
-import org.rhq.core.domain.criteria.MeasurementDefinitionCriteria;
import org.rhq.core.domain.criteria.StorageNodeCriteria;
import org.rhq.core.domain.measurement.AvailabilityType;
-import org.rhq.core.domain.measurement.DataType;
import org.rhq.core.domain.measurement.MeasurementAggregate;
-import org.rhq.core.domain.measurement.MeasurementDefinition;
-import org.rhq.core.domain.measurement.MeasurementSchedule;
import org.rhq.core.domain.measurement.MeasurementUnits;
import org.rhq.core.domain.resource.InventoryStatus;
import org.rhq.core.domain.resource.Resource;
@@ -65,8 +60,7 @@ import org.rhq.enterprise.server.authz.RequiredPermission;
import org.rhq.enterprise.server.measurement.MeasurementDataManagerLocal;
import org.rhq.enterprise.server.measurement.MeasurementDefinitionManagerLocal;
import org.rhq.enterprise.server.measurement.MeasurementScheduleManagerLocal;
-import org.rhq.enterprise.server.resource.ResourceManagerLocal;
-import org.rhq.enterprise.server.resource.ResourceTypeManagerLocal;
+import org.rhq.enterprise.server.rest.reporting.MeasurementConverter;
import org.rhq.enterprise.server.util.CriteriaQueryGenerator;
import org.rhq.enterprise.server.util.CriteriaQueryRunner;
import org.rhq.server.metrics.CQLException;
@@ -171,71 +165,85 @@ public class StorageNodeManagerBean implements StorageNodeManagerLocal, StorageN
}
}
+ @RequiredPermission(Permission.MANAGE_SETTINGS)
public StorageNodeLoadComposite getLoad(Subject subject, StorageNode node, long beginTime, long endTime) {
- //TODO: check the rights of subject
StorageNodeLoadComposite result = new StorageNodeLoadComposite(node, beginTime, endTime);
+ final String tokensMetric = "Tokens", ownershipMetric = "Ownership", loadMetric = "Load";
+ final String heapComittedMetric = "{HeapMemoryUsage.committed}", heapUsedMetric = "{HeapMemoryUsage.used}";
+
+ int resourceId;
+ if (node.getResource() == null) {
+ resourceId = entityManager.merge(node).getResource().getId();
+ } else {
+ resourceId = node.getResource().getId();
+ }
- StorageNode mergedNode = entityManager.merge(node);
- Set<ResourceType> childResourceTypes = mergedNode.getResource().getResourceType().getChildResourceTypes();
-
- // ResourceType memorySubsystemType = typeManager.getResourceTypeByNameAndPlugin(subject, "RHQ Storage Node",
- // PLUGIN_NAME);
- // ResourceType storageServiceType = typeManager.getResourceTypeByNameAndPlugin(subject, "RHQ Storage Node",
- // PLUGIN_NAME);
-
- Set<Resource> childResources = mergedNode.getResource().getChildResources();
- int storageServiceResourceId = -1;
- int memorySubsystemResourceId = -1;
- for (Resource res : childResources) {
- if ("Storage Service".equals(res.getName())) {
- storageServiceResourceId = res.getId();
- } else if ("Cassandra Server JVM".equals(res.getName())) {
- Set<Resource> childJVMResources = res.getChildResources();
- for (Resource resJVM : childJVMResources) {
- if ("Memory Subsystem".equals(resJVM.getName())) {
- memorySubsystemResourceId = resJVM.getId();
- break;
- }
- }
- }
+ // get the schedule ids for Storage Service resource
+ TypedQuery<Object[]> query = entityManager.<Object[]> createNamedQuery(
+ StorageNode.QUERY_FIND_SCHEDULE_IDS_BY_PARENT_RESOURCE_ID_AND_MEASUREMENT_DEFINITION_NAMES, Object[].class);
+ query.setParameter("parrentId", resourceId).setParameter("metricNames",
+ Arrays.asList(tokensMetric, ownershipMetric, loadMetric));
+ List<Object[]> scheduleIds = query.getResultList();
+ Map<String, Integer> scheduleIdsMap = new HashMap<String, Integer>(4);
+ for (Object[] pair : scheduleIds) {
+ scheduleIdsMap.put((String) pair[0], (Integer) pair[1]);
}
- // TODO: perhaps crating a new named query will be more efficient than this
- if (storageServiceResourceId != -1) {
- MeasurementDefinitionCriteria criteria = new MeasurementDefinitionCriteria();
- criteria.addFilterName("Tokens");
- criteria.addFilterResourceTypeName("StorageService");
- PageList<MeasurementDefinition> measDefinition1 = measurementDefinitionManager
- .findMeasurementDefinitionsByCriteria(subject, criteria);
-
- criteria = new MeasurementDefinitionCriteria();
- criteria.addFilterName("Ownership");
- criteria.addFilterResourceTypeName("StorageService");
- PageList<MeasurementDefinition> measDefinition2 = measurementDefinitionManager
- .findMeasurementDefinitionsByCriteria(subject, criteria);
-
- if (!measDefinition1.isEmpty() && !measDefinition2.isEmpty()) {
- List<MeasurementSchedule> schedules = scheduleManager.findSchedulesByResourceIdsAndDefinitionIds(
- new int[] { storageServiceResourceId }, new int[] { measDefinition1.get(1).getId(),
- measDefinition2.get(1).getId() });
- if (!schedules.isEmpty()) {
- MeasurementAggregate tokensAggregate = measurementManager.getAggregate(subject, schedules.get(0)
- .getId(), beginTime, endTime);
- result.setTokens(tokensAggregate);
- MeasurementAggregate ovnershipAggregate = measurementManager.getAggregate(subject, schedules.get(1)
- .getId(), beginTime, endTime);
- StorageNodeLoadComposite.MeasurementAggregateWithUnits ovnershipAggregateWithUnits = new StorageNodeLoadComposite.MeasurementAggregateWithUnits(
- ovnershipAggregate, MeasurementUnits.PERCENTAGE);
- result.setActuallyOwns(ovnershipAggregateWithUnits);
- }
- }
+ // get the schedule ids for Memory Subsystem resource
+ query = entityManager.<Object[]> createNamedQuery(
+ StorageNode.QUERY_FIND_SCHEDULE_IDS_BY_GRANDPARENT_RESOURCE_ID_AND_MEASUREMENT_DEFINITION_NAMES,
+ Object[].class);
+ query.setParameter("grandparrentId", resourceId).setParameter("metricNames",
+ Arrays.asList(heapComittedMetric, heapUsedMetric));
+ scheduleIds = query.getResultList();
+ for (Object[] pair : scheduleIds) {
+ scheduleIdsMap.put((String) pair[0], (Integer) pair[1]);
}
- // private MeasurementAggregateWithUnits heapCommited; // cassandra server jvm / memory subsystem resource
- // private MeasurementAggregateWithUnits heapUsed; // cassandra server jvm / memory subsystem resource
- // private MeasurementAggregateWithUnits load; // database management services / storage service
- // DONE -- private MeasurementAggregate tokens; // ~ jmx op - getTokens(hostname).size() or jmx attribute (StorageService/tokens).size()
- // DONE --private MeasurementAggregate actuallyOwns; // up to date value (tokenToEndpointMap) can be taken from the associated resource's configuration
+ // find the aggregates and enrich the result instance
+ if (!scheduleIdsMap.isEmpty()) {
+ if (scheduleIdsMap.get(tokensMetric) != null) {
+ MeasurementAggregate tokensAggregate = measurementManager.getAggregate(subject,
+ scheduleIdsMap.get(tokensMetric), beginTime, endTime);
+ result.setTokens(tokensAggregate);
+ }
+ if (scheduleIdsMap.get(ownershipMetric) != null) {
+ MeasurementAggregate ownershipAggregate = measurementManager.getAggregate(subject,
+ scheduleIdsMap.get(ownershipMetric), beginTime, endTime);
+ StorageNodeLoadComposite.MeasurementAggregateWithUnits ovnershipAggregateWithUnits = new StorageNodeLoadComposite.MeasurementAggregateWithUnits(
+ ownershipAggregate, MeasurementUnits.PERCENTAGE);
+ ovnershipAggregateWithUnits.setFormattedValue(getSummaryString(ownershipAggregate,
+ MeasurementUnits.PERCENTAGE));
+ result.setActuallyOwns(ovnershipAggregateWithUnits);
+ }
+ if (scheduleIdsMap.get(loadMetric) != null) {
+ MeasurementAggregate loadAggregate = measurementManager.getAggregate(subject,
+ scheduleIdsMap.get(loadMetric), beginTime, endTime);
+ StorageNodeLoadComposite.MeasurementAggregateWithUnits loadAggregateWithUnits = new StorageNodeLoadComposite.MeasurementAggregateWithUnits(
+ loadAggregate, MeasurementUnits.BYTES);
+ loadAggregateWithUnits.setFormattedValue(getSummaryString(loadAggregate, MeasurementUnits.BYTES));
+ result.setLoad(loadAggregateWithUnits);
+ }
+
+ if (scheduleIdsMap.get(heapComittedMetric) != null) {
+ MeasurementAggregate heapCommitedAggregate = measurementManager.getAggregate(subject,
+ scheduleIdsMap.get(heapComittedMetric), beginTime, endTime);
+ StorageNodeLoadComposite.MeasurementAggregateWithUnits heapCommitedAggregateWithUnits = new StorageNodeLoadComposite.MeasurementAggregateWithUnits(
+ heapCommitedAggregate, MeasurementUnits.BYTES);
+ heapCommitedAggregateWithUnits.setFormattedValue(getSummaryString(heapCommitedAggregate,
+ MeasurementUnits.BYTES));
+ result.setHeapCommited(heapCommitedAggregateWithUnits);
+ }
+ if (scheduleIdsMap.get(heapUsedMetric) != null) {
+ MeasurementAggregate heapUsedAggregate = measurementManager.getAggregate(subject,
+ scheduleIdsMap.get(heapUsedMetric), beginTime, endTime);
+ StorageNodeLoadComposite.MeasurementAggregateWithUnits heapUsedAggregateWithUnits = new StorageNodeLoadComposite.MeasurementAggregateWithUnits(
+ heapUsedAggregate, MeasurementUnits.BYTES);
+ heapUsedAggregateWithUnits
+ .setFormattedValue(getSummaryString(heapUsedAggregate, MeasurementUnits.BYTES));
+ result.setHeapUsed(heapUsedAggregateWithUnits);
+ }
+ }
return result;
}
@@ -261,4 +269,14 @@ public class StorageNodeManagerBean implements StorageNodeManagerLocal, StorageN
entityManager);
return runner.execute();
}
+
+ private String getSummaryString(MeasurementAggregate aggregate, MeasurementUnits units) {
+ String formattedValue = "Min: "
+ + MeasurementConverter.format(aggregate.getMin(), units, true)
+ + ", Max: "
+ + MeasurementConverter.format(aggregate.getMax(), units, true)
+ + ", Avg: "
+ + MeasurementConverter.format(aggregate.getAvg(), units, true);
+ return formattedValue;
+ }
}
10 years, 11 months
[rhq] Branch 'feature/cassandra-backend' - modules/core
by Jiri Kremser
modules/core/domain/src/main/java/org/rhq/core/domain/criteria/Criteria.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
New commits:
commit 113ff6638ccfae48ef60ee282eb5d85a849685fe
Author: Jirka Kremser <jkremser(a)redhat.com>
Date: Mon May 27 14:53:11 2013 +0200
just a typo
diff --git a/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/Criteria.java b/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/Criteria.java
index f398020..e224fca 100644
--- a/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/Criteria.java
+++ b/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/Criteria.java
@@ -179,7 +179,7 @@ public abstract class Criteria implements Serializable, BaseCriteria {
this.sortId = sortId;
} else {
- throw new UnsupportedOperationException("ID sort is not supported by supported by this class");
+ throw new UnsupportedOperationException("ID sort is not supported by this class");
}
}
10 years, 11 months
[rhq] Branch 'feature/cassandra-backend' - modules/enterprise
by John Sanda
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerBean.java | 12 ++++++++--
1 file changed, 10 insertions(+), 2 deletions(-)
New commits:
commit 9a09c435eeebc9d46271fafb8ff0b7be3c7a0c1e
Author: John Sanda <jsanda(a)redhat.com>
Date: Sun May 26 11:59:54 2013 -0400
handle race condition during raw data inserts
MeasurementDataManagerBean passes a callback to MetricsServer.addData(). That
callback is invoked once for each raw data object that is persisted and then
after all of the data has been persisted. The callback populates a list with
the inserted raw data, and that list is then used to update the alert
notification cache. I was seeing NPEs in AgentConditionCache.checkConditions
because the collection passed to it contained null elements. Through some
tedious debugging I beleive that this is a result of concurrent inserts to the
list in the aforementione callback. A lock is now used to serialize inserts in
that list.
diff --git a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerBean.java b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerBean.java
index a8457d1..77a8904 100644
--- a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerBean.java
+++ b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerBean.java
@@ -33,6 +33,7 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeoutException;
+import java.util.concurrent.locks.ReentrantLock;
import javax.ejb.Asynchronous;
import javax.ejb.EJB;
@@ -230,7 +231,9 @@ public class MeasurementDataManagerBean implements MeasurementDataManagerLocal,
MetricsServer metricsServer = sessionManagerBean.getMetricsServer();
metricsServer.addNumericData(data, new RawDataInsertedCallback() {
- private List<MeasurementData> insertedData = new ArrayList<MeasurementData>();
+ private ReentrantLock lock = new ReentrantLock();
+
+ private List<MeasurementData> insertedData = new ArrayList<MeasurementData>(data.size());
@Override
public void onFinish() {
@@ -240,7 +243,12 @@ public class MeasurementDataManagerBean implements MeasurementDataManagerLocal,
@Override
public void onSuccess(MeasurementDataNumeric measurementDataNumeric) {
- insertedData.add(measurementDataNumeric);
+ try {
+ lock.lock();
+ insertedData.add(measurementDataNumeric);
+ } finally {
+ lock.unlock();
+ }
}
@Override
10 years, 11 months
[rhq] Branch 'feature/cassandra-backend' - modules/common
by John Sanda
modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/SchemaManager.java | 42 ++++++++--
1 file changed, 35 insertions(+), 7 deletions(-)
New commits:
commit a4bf6ed6445c590f7e5ced8e01c05d19c857c1e8
Author: John Sanda <jsanda(a)redhat.com>
Date: Sat May 25 07:47:35 2013 -0400
Apply schema changes based based on cluster size
This commit includes the following changes to schema creation:
* configure system_auth keyspace RF based on cluster size
* configure rhq keyspace RF based on cluster size
* set gc_grace_seconds to 0 for all rhq tables if it single node cluster
* set gc_grace_seconds to 8 days if it is multi-node cluster
diff --git a/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/SchemaManager.java b/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/SchemaManager.java
index 0eec414..c03dc58 100644
--- a/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/SchemaManager.java
+++ b/modules/common/cassandra-schema/src/main/java/org/rhq/cassandra/schema/SchemaManager.java
@@ -115,8 +115,23 @@ public class SchemaManager {
log.debug("Creating user [rhqadmin]");
session.execute("CREATE USER rhqadmin WITH PASSWORD 'rhqadmin' SUPERUSER");
log.debug("Creating keyspace [" + RHQ_KEYSPACE + "]");
- session
- .execute("CREATE KEYSPACE rhq WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 1};");
+
+ int replicationFactor = 1;
+ if (nodes.size() == 1) {
+ log.debug("Setting replication_factor to 1 for rhq keyspace since this is a single node cluster.");
+ replicationFactor = 1;
+ } else if (nodes.size() < 4) {
+ log.debug("Setting replication_factor to 2 for rhq keyspace since this is a " + nodes.size() +
+ " node cluster");
+ replicationFactor = 2;
+ } else {
+ log.debug("Setting replication_factor to 3 for rhq keyspace since this is a " + nodes.size() +
+ " node cluster");
+ replicationFactor = 3;
+ }
+
+ session.execute("CREATE KEYSPACE rhq WITH replication = {'class': 'SimpleStrategy', " +
+ "'replication_factor': " + replicationFactor + "};");
// Note that once we have a schema management tool back in place, the call
// to createTables will be moved back to the updateSchema method as it
@@ -171,7 +186,20 @@ public class SchemaManager {
}
private void createTables() {
+ int gcGraceSeconds = 864000;
+ if (nodes.size() == 1) {
+ gcGraceSeconds = 0;
+ } else {
+ gcGraceSeconds = 691200; // 8 days
+ }
+
try {
+ log.debug("Setting system_auth keyspace replication_factor to " + nodes.size());
+ session.execute(
+ "ALTER KEYSPACE system_auth WITH replication = " +
+ "{'class' : 'SimpleStrategy', 'replication_factor' : " + nodes.size() + "};"
+ );
+
log.debug("Creating table raw_metrics");
session.execute(
"CREATE TABLE rhq.raw_metrics (" +
@@ -179,7 +207,7 @@ public class SchemaManager {
"time timestamp, " +
"value double, " +
"PRIMARY KEY (schedule_id, time) " +
- ") WITH COMPACT STORAGE"
+ ") WITH COMPACT STORAGE AND gc_grace_seconds = " + gcGraceSeconds + ";"
);
log.debug("Creating table one_hour_metrics");
session.execute(
@@ -189,7 +217,7 @@ public class SchemaManager {
"type int, " +
"value double, " +
"PRIMARY KEY (schedule_id, time, type) " +
- ") WITH COMPACT STORAGE"
+ ") WITH COMPACT STORAGE AND gc_grace_seconds = " + gcGraceSeconds + ";"
);
log.debug("Creating table six_hour_metrics");
session.execute(
@@ -199,7 +227,7 @@ public class SchemaManager {
"type int, " +
"value double, " +
"PRIMARY KEY (schedule_id, time, type) " +
- ") WITH COMPACT STORAGE;"
+ ") WITH COMPACT STORAGE AND gc_grace_seconds = " + gcGraceSeconds + ";"
);
log.debug("Creating table twenty_four_hour_metrics");
session.execute(
@@ -209,7 +237,7 @@ public class SchemaManager {
"type int, " +
"value double, " +
"PRIMARY KEY (schedule_id, time, type) " +
- ") WITH COMPACT STORAGE;"
+ ") WITH COMPACT STORAGE AND gc_grace_seconds = " + gcGraceSeconds + ";"
);
log.debug("Creating table metrics_index");
session.execute(
@@ -218,7 +246,7 @@ public class SchemaManager {
"time timestamp, " +
"schedule_id int, " +
"PRIMARY KEY ((bucket, time), schedule_id) " +
- ") WITH COMPACT STORAGE;"
+ ") WITH COMPACT STORAGE AND gc_grace_seconds = " + gcGraceSeconds + ";"
);
} catch (NoHostAvailableException e) {
throw new RuntimeException(e);
10 years, 12 months
[rhq] Branch 'feature/cassandra-backend' - modules/enterprise
by John Sanda
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/LoadInfo.java | 75 ----------
1 file changed, 75 deletions(-)
New commits:
commit 8c6fc939213b1302e0107a95a33d7ff17e4bfc69
Author: John Sanda <jsanda(a)redhat.com>
Date: Sat May 25 01:09:30 2013 -0400
remove class that was just for prototyping
diff --git a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/LoadInfo.java b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/LoadInfo.java
deleted file mode 100644
index 5cf05c2..0000000
--- a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/LoadInfo.java
+++ /dev/null
@@ -1,75 +0,0 @@
-/*
- *
- * * RHQ Management Platform
- * * Copyright (C) 2005-2012 Red Hat, Inc.
- * * All rights reserved.
- * *
- * * This program is free software; you can redistribute it and/or modify
- * * it under the terms of the GNU General Public License, version 2, as
- * * published by the Free Software Foundation, and/or the GNU Lesser
- * * General Public License, version 2.1, also as published by the Free
- * * Software Foundation.
- * *
- * * This program is distributed in the hope that it will be useful,
- * * but WITHOUT ANY WARRANTY; without even the implied warranty of
- * * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
- * * GNU General Public License and the GNU Lesser General Public License
- * * for more details.
- * *
- * * You should have received a copy of the GNU General Public License
- * * and the GNU Lesser General Public License along with this program;
- * * if not, write to the Free Software Foundation, Inc.,
- * * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
- *
- */
-
-package org.rhq.enterprise.server.cloud;
-
-import org.rhq.enterprise.server.measurement.MeasurementAggregate;
-
-/**
- * @author John Sanda
- */
-public class LoadInfo {
-
- private String diskSpaceUsed;
-
- private String ringOwnership;
-
- private MeasurementAggregate heapCommitted;
-
- private MeasurementAggregate heapUsage;
-
- public String getDiskSpaceUsed() {
- return diskSpaceUsed;
- }
-
- public void setDiskSpaceUsed(String diskSpaceUsed) {
- this.diskSpaceUsed = diskSpaceUsed;
- }
-
- public String getRingOwnership() {
- return ringOwnership;
- }
-
- public void setRingOwnership(String ringOwnership) {
- this.ringOwnership = ringOwnership;
- }
-
- public MeasurementAggregate getHeapCommitted() {
- return heapCommitted;
- }
-
- public void setHeapCommitted(MeasurementAggregate heapCommitted) {
- this.heapCommitted = heapCommitted;
- }
-
- public MeasurementAggregate getHeapUsage() {
- return heapUsage;
- }
-
- public void setHeapUsage(MeasurementAggregate heapUsage) {
- this.heapUsage = heapUsage;
- }
-
-}
10 years, 12 months
[rhq] modules/core
by Jay Shaughnessy
modules/core/domain/src/main/java/org/rhq/core/domain/criteria/Criteria.java | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
New commits:
commit ff37cd971a73a9fefc4024f5022aff85b7b90b90
Author: Jay Shaughnessy <jshaughn(a)redhat.com>
Date: Fri May 24 23:29:46 2013 -0400
remove use of method not supported by GWT
diff --git a/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/Criteria.java b/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/Criteria.java
index d8552de..f398020 100644
--- a/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/Criteria.java
+++ b/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/Criteria.java
@@ -179,7 +179,7 @@ public abstract class Criteria implements Serializable, BaseCriteria {
this.sortId = sortId;
} else {
- throw new UnsupportedOperationException("ID sort is not supported by " + this.getClass().getSimpleName());
+ throw new UnsupportedOperationException("ID sort is not supported by supported by this class");
}
}
@@ -188,7 +188,7 @@ public abstract class Criteria implements Serializable, BaseCriteria {
this.filterId = filterId;
} else {
- throw new UnsupportedOperationException("ID filter is not supported by " + this.getClass().getSimpleName());
+ throw new UnsupportedOperationException("ID filter is not supported by this class");
}
}
@@ -197,7 +197,7 @@ public abstract class Criteria implements Serializable, BaseCriteria {
this.filterIds = CriteriaUtils.getListIgnoringNulls(filterIds);
} else {
- throw new UnsupportedOperationException("IDS filter is not supported by " + this.getClass().getSimpleName());
+ throw new UnsupportedOperationException("IDS filter is not supported by this class");
}
}
10 years, 12 months
[rhq] Branch 'feature/cassandra-backend' - modules/core
by Jay Shaughnessy
modules/core/domain/src/main/java/org/rhq/core/domain/criteria/Criteria.java | 6 +++---
1 file changed, 3 insertions(+), 3 deletions(-)
New commits:
commit 400f15ee5d0c93c0e0d5e78e4b30b2d8b5952f41
Author: Jay Shaughnessy <jshaughn(a)redhat.com>
Date: Fri May 24 23:29:46 2013 -0400
remove use of method not supported by GWT
diff --git a/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/Criteria.java b/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/Criteria.java
index d8552de..f398020 100644
--- a/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/Criteria.java
+++ b/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/Criteria.java
@@ -179,7 +179,7 @@ public abstract class Criteria implements Serializable, BaseCriteria {
this.sortId = sortId;
} else {
- throw new UnsupportedOperationException("ID sort is not supported by " + this.getClass().getSimpleName());
+ throw new UnsupportedOperationException("ID sort is not supported by supported by this class");
}
}
@@ -188,7 +188,7 @@ public abstract class Criteria implements Serializable, BaseCriteria {
this.filterId = filterId;
} else {
- throw new UnsupportedOperationException("ID filter is not supported by " + this.getClass().getSimpleName());
+ throw new UnsupportedOperationException("ID filter is not supported by this class");
}
}
@@ -197,7 +197,7 @@ public abstract class Criteria implements Serializable, BaseCriteria {
this.filterIds = CriteriaUtils.getListIgnoringNulls(filterIds);
} else {
- throw new UnsupportedOperationException("IDS filter is not supported by " + this.getClass().getSimpleName());
+ throw new UnsupportedOperationException("IDS filter is not supported by this class");
}
}
10 years, 12 months
[rhq] Branch 'feature/cassandra-backend' - 8 commits - modules/common modules/enterprise modules/helpers
by John Sanda
modules/common/cassandra-util/src/main/java/org/rhq/cassandra/util/ClusterBuilder.java | 8
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cassandra/SessionManagerBean.java | 5
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/LoadInfo.java | 75 ++++++
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerBean.java | 38 ++-
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerLocal.java | 2
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MetricsManagerBean.java | 4
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MetricsManagerLocal.java | 2
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsDAO.java | 10
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java | 119 +++++++--
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/RawDataInsertedCallback.java | 39 +++
modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/CassandraIntegrationTest.java | 40 +++
modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsDAOTest.java | 59 ++++
modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsServerTest.java | 113 +++++++--
modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java | 123 ++++++++++
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/MeasurementCollector.java | 19 +
modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/Simulator.java | 3
16 files changed, 604 insertions(+), 55 deletions(-)
New commits:
commit cc2d6c042251ff58a1ebc8d0f4df091d02fae0eb
Merge: 39ad031 f0a6431
Author: John Sanda <jsanda(a)redhat.com>
Date: Fri May 24 21:51:16 2013 -0400
Merge branch 'async-cql' into feature/cassandra-backend
Conflicts:
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsBaselineCalculator.java
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
commit f0a643191697efc5183450b8a35603aaec08a64d
Author: John Sanda <jsanda(a)redhat.com>
Date: Fri May 24 21:39:03 2013 -0400
removing sync version of MetricsServer.updateIndex
This commit also updates tsts in MetricsServerTest to block on async calls in
order to make the test logic simpler.
diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
index 880f21b..66d9a4b 100644
--- a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
+++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
@@ -26,8 +26,6 @@
package org.rhq.server.metrics;
import java.util.ArrayList;
-import java.util.Date;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
@@ -309,29 +307,7 @@ public class MetricsServer {
}
}
- private void persistRawData(final Set<MeasurementDataNumeric> dataSet, final RawDataInsertedCallback callback) {
- final long startTime = System.currentTimeMillis();
- final AtomicInteger remainingInserts = new AtomicInteger(dataSet.size());
-
- for (final MeasurementDataNumeric data : dataSet) {
- ResultSetFuture resultSetFuture = dao.insertRawData(data);
- Futures.addCallback(resultSetFuture, new FutureCallback<ResultSet>() {
- @Override
- public void onSuccess(ResultSet rows) {
- updateMetricsIndex(data, dataSet.size(), remainingInserts, startTime, callback);
- }
-
- @Override
- public void onFailure(Throwable throwable) {
- log.error("An error occurred while inserting raw data " + data, throwable);
- callback.onFailure(throwable);
- semaphore.release();
- }
- });
- }
- }
-
- private void updateMetricsIndex(final MeasurementDataNumeric rawData, final int total,
+ void updateMetricsIndex(final MeasurementDataNumeric rawData, final int total,
final AtomicInteger remainingInserts, final long startTime, final RawDataInsertedCallback callback) {
long timeSlice = dateTimeService.getTimeSlice(new DateTime(rawData.getTimestamp()),
@@ -362,20 +338,6 @@ public class MetricsServer {
});
}
- void updateMetricsIndex(Set<MeasurementDataNumeric> rawMetrics) {
- Map<Integer, Long> updates = new TreeMap<Integer, Long>();
- for (MeasurementDataNumeric rawMetric : rawMetrics) {
- updates.put(rawMetric.getScheduleId(), dateTimeService.getTimeSlice(
- new DateTime(rawMetric.getTimestamp()), configuration.getRawTimeSliceDuration()).getMillis());
- }
- Set<Date> dates = new HashSet<Date>();
- for (Long ts : updates.values()) {
- dates.add(new Date(ts));
- }
- log.debug("Updating one hour index wtih time slices " + dates);
- dao.updateMetricsIndex(MetricsTable.ONE_HOUR, updates);
- }
-
/**
* Computes and stores aggregate for all time slices that are ready to be aggregated.
* This is includes raw, 1hr, 6hr, and 24hr data.
diff --git a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/CassandraIntegrationTest.java b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/CassandraIntegrationTest.java
index 9bec034..8e311e1 100644
--- a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/CassandraIntegrationTest.java
+++ b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/CassandraIntegrationTest.java
@@ -25,14 +25,21 @@
package org.rhq.server.metrics;
+import static org.testng.Assert.fail;
+
import java.util.Date;
+import java.util.concurrent.CountDownLatch;
import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.Cluster;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.exceptions.NoHostAvailableException;
+import com.google.common.base.Throwables;
+import com.google.common.util.concurrent.FutureCallback;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.joda.time.DateTime;
import org.testng.annotations.AfterSuite;
import org.testng.annotations.BeforeSuite;
@@ -58,7 +65,7 @@ public class CassandraIntegrationTest {
private static DateTimeService dateTimeService;
@BeforeSuite
- @DeployCluster(numNodes = 2, username = "cassandra", password = "cassandra")
+ @DeployCluster(numNodes = 1, username = "cassandra", password = "cassandra")
public void deployCluster() throws Exception {
dateTimeService = new DateTimeService();
@@ -108,4 +115,37 @@ public class CassandraIntegrationTest {
throw new CQLException(e);
}
}
+
+ protected static class WaitForResults<ResultSetFuture> implements FutureCallback<ResultSetFuture> {
+
+ private final Log log = LogFactory.getLog(WaitForResults.class);
+
+ private CountDownLatch latch;
+
+ private Throwable throwable;
+
+ public WaitForResults(int numResults) {
+ latch = new CountDownLatch(numResults);
+ }
+
+ @Override
+ public void onSuccess(ResultSetFuture resultSetFuture) {
+ latch.countDown();
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ latch.countDown();
+ this.throwable = throwable;
+ log.error("An async operation failed", throwable);
+ }
+
+ public void await(String errorMsg) throws InterruptedException {
+ latch.await();
+ if (throwable != null) {
+ fail(errorMsg, Throwables.getRootCause(throwable));
+ }
+ }
+
+ }
}
diff --git a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsDAOTest.java b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsDAOTest.java
index c522316..6195016 100644
--- a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsDAOTest.java
+++ b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsDAOTest.java
@@ -29,7 +29,6 @@ import static java.util.Arrays.asList;
import static org.rhq.test.AssertUtils.assertCollectionMatchesNoOrder;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
-import static org.testng.Assert.fail;
import java.util.ArrayList;
import java.util.HashMap;
@@ -38,12 +37,9 @@ import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
-import java.util.concurrent.CountDownLatch;
import com.datastax.driver.core.ResultSetFuture;
-import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
-import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import org.apache.commons.logging.Log;
@@ -672,37 +668,4 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
}
}
- private static class WaitForResults<ResultSetFuture> implements FutureCallback<ResultSetFuture> {
-
- private final Log log = LogFactory.getLog(WaitForResults.class);
-
- private CountDownLatch latch;
-
- private Throwable throwable;
-
- public WaitForResults(int numResults) {
- latch = new CountDownLatch(numResults);
- }
-
- @Override
- public void onSuccess(ResultSetFuture resultSetFuture) {
- latch.countDown();
- }
-
- @Override
- public void onFailure(Throwable throwable) {
- latch.countDown();
- this.throwable = throwable;
- log.error("An async operation failed", throwable);
- }
-
- public void await(String errorMsg) throws InterruptedException {
- latch.await();
- if (throwable != null) {
- fail(errorMsg, Throwables.getRootCause(throwable));
- }
- }
-
- }
-
}
diff --git a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsServerTest.java b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsServerTest.java
index cf74fcd..d89b2d9 100644
--- a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsServerTest.java
+++ b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsServerTest.java
@@ -41,9 +41,12 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicInteger;
+import com.datastax.driver.core.ResultSetFuture;
import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.Futures;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -187,10 +190,10 @@ public class MetricsServerTest extends CassandraIntegrationTest {
metricsServer.setCurrentHour(hour6);
metricsServer.addNumericData(data, waitForRawInserts);
- metricsServer.calculateAggregates(hour6.getMillis());
-
waitForRawInserts.await("Failed to insert raw data");
+ metricsServer.calculateAggregates(hour6.getMillis());
+
// verify that one hour metric data is updated
List<AggregateNumericMetric> expected = asList(new AggregateNumericMetric(scheduleId,
divide((3.9 + 3.2 + 2.6), 3), 2.6, 3.9, lastHour.getMillis()));
@@ -205,7 +208,7 @@ public class MetricsServerTest extends CassandraIntegrationTest {
}
@Test//(enabled = ENABLED)
- public void aggregateRawDataDuring9thHour() {
+ public void aggregateRawDataDuring9thHour() throws Exception {
int scheduleId = 123;
DateTime hour0 = hour0();
@@ -225,10 +228,21 @@ public class MetricsServerTest extends CassandraIntegrationTest {
rawMetrics.add(new MeasurementDataNumeric(secondMetricTime.getMillis(), scheduleId, secondValue));
rawMetrics.add(new MeasurementDataNumeric(thirdMetricTime.getMillis(), scheduleId, thirdValue));
+ WaitForResults waitForRawInserts = new WaitForResults(rawMetrics.size());
+
+ for (MeasurementDataNumeric raw : rawMetrics) {
+ ResultSetFuture resultSetFuture = dao.insertRawData(raw);
+ Futures.addCallback(resultSetFuture, waitForRawInserts);
+ }
+ waitForRawInserts.await("Failed to insert raw data");
+
+ WaitForRawInserts waitForIndexUpdates = new WaitForRawInserts(rawMetrics.size());
+ AtomicInteger remainingUpdates = new AtomicInteger(rawMetrics.size());
for (MeasurementDataNumeric raw : rawMetrics) {
- dao.insertRawData(raw);
+ metricsServer.updateMetricsIndex(raw, rawMetrics.size(), remainingUpdates, hour8.getMillis(),
+ waitForIndexUpdates);
}
- metricsServer.updateMetricsIndex(rawMetrics);
+ waitForIndexUpdates.await("Failed to update metrics index for raw data");
metricsServer.setCurrentHour(hour9);
metricsServer.calculateAggregates(hour9.getMillis());
@@ -382,11 +396,11 @@ public class MetricsServerTest extends CassandraIntegrationTest {
WaitForRawInserts waitForRawInserts = new WaitForRawInserts(data.size());
metricsServer.addNumericData(data, waitForRawInserts);
+ waitForRawInserts.await("Failed to insert raw data");
+
List<MeasurementDataNumericHighLowComposite> actualData = Lists.newArrayList(metricsServer.findDataForResource(
scheduleId, beginTime.getMillis(), endTime.getMillis()));
- waitForRawInserts.await("Faile to insert raw data");
-
assertEquals(actualData.size(), buckets.getNumDataPoints(), "Expected to get back 60 data points.");
MeasurementDataNumericHighLowComposite expectedBucket0Data = new MeasurementDataNumericHighLowComposite(
diff --git a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
index 2f960fe..b3c30a5 100644
--- a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
+++ b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
@@ -60,25 +60,25 @@ public class TimeoutTest extends CassandraIntegrationTest {
long time = hour0().getMillis();
Set<MeasurementDataNumeric> data1 = new HashSet<MeasurementDataNumeric>();
- for (int i = 0; i < 500000; ++i) {
+ for (int i = 0; i < 50000; ++i) {
data1.add(new MeasurementDataNumeric(time, i, (double) i));
}
- Set<MeasurementDataNumeric> data2 = new HashSet<MeasurementDataNumeric>();
- for (int i = 0; i < 500000; ++i) {
- data2.add(new MeasurementDataNumeric(time, i + data1.size(), (double) i));
- }
+// Set<MeasurementDataNumeric> data2 = new HashSet<MeasurementDataNumeric>();
+// for (int i = 0; i < 50000; ++i) {
+// data2.add(new MeasurementDataNumeric(time, i + data1.size(), (double) i));
+// }
WaitForRawInserts waitForRawInserts1 = new WaitForRawInserts(data1.size());
- WaitForRawInserts waitForRawInserts2 = new WaitForRawInserts(data2.size());
+// WaitForRawInserts waitForRawInserts2 = new WaitForRawInserts(data2.size());
long start = System.currentTimeMillis();
metricsServer.addNumericData(data1, waitForRawInserts1);
waitForRawInserts1.await("Failed to insert raw metrics");
- log.info("Sleep before second round of inserts...");
- Thread.sleep(3000);
-
- metricsServer.addNumericData(data2, waitForRawInserts2);
+// log.info("Sleep before second round of inserts...");
+// Thread.sleep(3000);
+//
+// metricsServer.addNumericData(data2, waitForRawInserts2);
long end = System.currentTimeMillis();
log.info("Inserted " + data1.size() + " raw metrics in " + (end - start) + " ms");
commit 3ddfe5272d9ad38ffc3006ec7f8c2a5afe46a4f8
Author: John Sanda <jsanda(a)redhat.com>
Date: Fri May 24 13:20:20 2013 -0400
remove extra call to updateMetricsIndex and break up test into 2 large inserts
diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
index 6f1187c..880f21b 100644
--- a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
+++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
@@ -303,7 +303,6 @@ public class MetricsServer {
}
});
}
- updateMetricsIndex(dataSet);
} catch (Exception e) {
log.error("An error occurred while inserting raw numeric data", e);
throw new RuntimeException(e);
diff --git a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/CassandraIntegrationTest.java b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/CassandraIntegrationTest.java
index f59324d..9bec034 100644
--- a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/CassandraIntegrationTest.java
+++ b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/CassandraIntegrationTest.java
@@ -58,7 +58,7 @@ public class CassandraIntegrationTest {
private static DateTimeService dateTimeService;
@BeforeSuite
- @DeployCluster(numNodes = 1, username = "cassandra", password = "cassandra")
+ @DeployCluster(numNodes = 2, username = "cassandra", password = "cassandra")
public void deployCluster() throws Exception {
dateTimeService = new DateTimeService();
diff --git a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
index 4496d33..2f960fe 100644
--- a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
+++ b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
@@ -59,19 +59,29 @@ public class TimeoutTest extends CassandraIntegrationTest {
metricsServer.setDAO(dao);
long time = hour0().getMillis();
- Set<MeasurementDataNumeric> data = new HashSet<MeasurementDataNumeric>();
- for (int i = 0; i < 10000; ++i) {
- data.add(new MeasurementDataNumeric(time, i, (double) i));
+ Set<MeasurementDataNumeric> data1 = new HashSet<MeasurementDataNumeric>();
+ for (int i = 0; i < 500000; ++i) {
+ data1.add(new MeasurementDataNumeric(time, i, (double) i));
}
- WaitForRawInserts waitForRawInserts = new WaitForRawInserts(data.size());
+ Set<MeasurementDataNumeric> data2 = new HashSet<MeasurementDataNumeric>();
+ for (int i = 0; i < 500000; ++i) {
+ data2.add(new MeasurementDataNumeric(time, i + data1.size(), (double) i));
+ }
+
+ WaitForRawInserts waitForRawInserts1 = new WaitForRawInserts(data1.size());
+ WaitForRawInserts waitForRawInserts2 = new WaitForRawInserts(data2.size());
long start = System.currentTimeMillis();
- metricsServer.addNumericData(data, waitForRawInserts);
+ metricsServer.addNumericData(data1, waitForRawInserts1);
+ waitForRawInserts1.await("Failed to insert raw metrics");
+
+ log.info("Sleep before second round of inserts...");
+ Thread.sleep(3000);
- waitForRawInserts.await("Failed to insert raw metrics");
+ metricsServer.addNumericData(data2, waitForRawInserts2);
long end = System.currentTimeMillis();
- log.info("Inserted " + data.size() + " raw metrics in " + (end - start) + " ms");
+ log.info("Inserted " + data1.size() + " raw metrics in " + (end - start) + " ms");
}
private static class WaitForRawInserts implements RawDataInsertedCallback {
commit 70f3fa68b0f8ab9aafe520df938de0e277365fde
Author: John Sanda <jsanda(a)redhat.com>
Date: Fri May 24 09:00:45 2013 -0400
throttle async inserts with a semaphore
diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
index 5e49d45..6f1187c 100644
--- a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
+++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
@@ -32,7 +32,9 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
+import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Executors;
+import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicInteger;
import com.datastax.driver.core.ResultSet;
@@ -78,6 +80,32 @@ public class MetricsServer {
private ListeningExecutorService listeningExecutorService = MoreExecutors.listeningDecorator(
Executors.newFixedThreadPool(5));
+ private ConcurrentLinkedQueue<MeasurementDataNumeric> rawDataQueue =
+ new ConcurrentLinkedQueue<MeasurementDataNumeric>();
+
+ private Semaphore semaphore = new Semaphore(100);
+
+ private boolean shutdown = false;
+
+ public MetricsServer() {
+// listeningExecutorService.submit(new Runnable() {
+// @Override
+// public void run() {
+// while (!shutdown) {
+// try {
+// if (!rawDataQueue.isEmpty()) {
+// semaphore.acquire();
+// MeasurementDataNumeric rawData = rawDataQueue.poll();
+// persistRawData();
+// }
+// } catch (InterruptedException e) {
+//
+// }
+// }
+// }
+// });
+ }
+
public void setSession(Session session) {
this.session = session;
}
@@ -94,6 +122,10 @@ public class MetricsServer {
this.dateTimeService = dateTimeService;
}
+ public void shutdown() {
+ shutdown = true;
+ }
+
public RawNumericMetric findLatestValueForResource(int scheduleId) {
return dao.findLatestRawMetric(scheduleId);
}
@@ -246,33 +278,58 @@ public class MetricsServer {
public void addNumericData(final Set<MeasurementDataNumeric> dataSet,
final RawDataInsertedCallback callback) {
-// try {
+ try {
if (log.isDebugEnabled()) {
log.debug("Inserting " + dataSet.size() + " raw metrics");
}
- final long startTime = System.currentTimeMillis();
- final AtomicInteger remainingInserts = new AtomicInteger(dataSet.size());
-
- for (final MeasurementDataNumeric data : dataSet) {
- ResultSetFuture resultSetFuture = dao.insertRawData(data);
- Futures.addCallback(resultSetFuture, new FutureCallback<ResultSet>() {
- @Override
- public void onSuccess(ResultSet rows) {
- updateMetricsIndex(data, dataSet.size(), remainingInserts, startTime, callback);
- }
- @Override
- public void onFailure(Throwable throwable) {
- log.error("An error occurred while inserting raw data " + data, throwable);
- callback.onFailure(throwable);
- }
- });
- }
-// updateMetricsIndex(dataSet);
-// } catch (Exception e) {
-// log.error("An error occurred while inserting raw numeric data", e);
-// throw new RuntimeException(e);
-// }
+ final long startTime = System.currentTimeMillis();
+ final AtomicInteger remainingInserts = new AtomicInteger(dataSet.size());
+
+ for (final MeasurementDataNumeric data : dataSet) {
+ semaphore.acquire();
+ ResultSetFuture resultSetFuture = dao.insertRawData(data);
+ Futures.addCallback(resultSetFuture, new FutureCallback<ResultSet>() {
+ @Override
+ public void onSuccess(ResultSet rows) {
+ updateMetricsIndex(data, dataSet.size(), remainingInserts, startTime, callback);
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ log.error("An error occurred while inserting raw data " + data, throwable);
+ callback.onFailure(throwable);
+ semaphore.release();
+ }
+ });
+ }
+ updateMetricsIndex(dataSet);
+ } catch (Exception e) {
+ log.error("An error occurred while inserting raw numeric data", e);
+ throw new RuntimeException(e);
+ }
+ }
+
+ private void persistRawData(final Set<MeasurementDataNumeric> dataSet, final RawDataInsertedCallback callback) {
+ final long startTime = System.currentTimeMillis();
+ final AtomicInteger remainingInserts = new AtomicInteger(dataSet.size());
+
+ for (final MeasurementDataNumeric data : dataSet) {
+ ResultSetFuture resultSetFuture = dao.insertRawData(data);
+ Futures.addCallback(resultSetFuture, new FutureCallback<ResultSet>() {
+ @Override
+ public void onSuccess(ResultSet rows) {
+ updateMetricsIndex(data, dataSet.size(), remainingInserts, startTime, callback);
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ log.error("An error occurred while inserting raw data " + data, throwable);
+ callback.onFailure(throwable);
+ semaphore.release();
+ }
+ });
+ }
}
private void updateMetricsIndex(final MeasurementDataNumeric rawData, final int total,
@@ -293,6 +350,7 @@ public class MetricsServer {
}
callback.onFinish();
}
+ semaphore.release();
}
@Override
@@ -300,6 +358,7 @@ public class MetricsServer {
log.error("An error occurred while trying to update " + MetricsTable.INDEX + " for raw data " +
rawData);
callback.onFailure(throwable);
+ semaphore.release();
}
});
}
commit 0c4801d681161aa6edf16bf998a22bb8319d254c
Author: John Sanda <jsanda(a)redhat.com>
Date: Fri May 24 08:59:01 2013 -0400
log total insertion time
diff --git a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
index f32fd66..4496d33 100644
--- a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
+++ b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
@@ -41,6 +41,8 @@ import org.rhq.core.domain.measurement.MeasurementDataNumeric;
public class TimeoutTest extends CassandraIntegrationTest {
+ private final Log log = LogFactory.getLog(TimeoutTest.class);
+
@Test
public void generateTimeout() throws Exception {
MetricsConfiguration configuration = new MetricsConfiguration();
@@ -58,14 +60,18 @@ public class TimeoutTest extends CassandraIntegrationTest {
long time = hour0().getMillis();
Set<MeasurementDataNumeric> data = new HashSet<MeasurementDataNumeric>();
- for (int i = 0; i < 5000; ++i) {
+ for (int i = 0; i < 10000; ++i) {
data.add(new MeasurementDataNumeric(time, i, (double) i));
}
WaitForRawInserts waitForRawInserts = new WaitForRawInserts(data.size());
+ long start = System.currentTimeMillis();
metricsServer.addNumericData(data, waitForRawInserts);
waitForRawInserts.await("Failed to insert raw metrics");
+ long end = System.currentTimeMillis();
+
+ log.info("Inserted " + data.size() + " raw metrics in " + (end - start) + " ms");
}
private static class WaitForRawInserts implements RawDataInsertedCallback {
commit bd117a2bb465a879f57223958d8430c914be7b0d
Author: John Sanda <jsanda(a)redhat.com>
Date: Fri May 24 07:37:35 2013 -0400
adding test to generate timeout exception in driver due to large number of inserts
diff --git a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
new file mode 100644
index 0000000..f32fd66
--- /dev/null
+++ b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/TimeoutTest.java
@@ -0,0 +1,107 @@
+/*
+ *
+ * * RHQ Management Platform
+ * * Copyright (C) 2005-2012 Red Hat, Inc.
+ * * All rights reserved.
+ * *
+ * * This program is free software; you can redistribute it and/or modify
+ * * it under the terms of the GNU General Public License, version 2, as
+ * * published by the Free Software Foundation, and/or the GNU Lesser
+ * * General Public License, version 2.1, also as published by the Free
+ * * Software Foundation.
+ * *
+ * * This program is distributed in the hope that it will be useful,
+ * * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * * GNU General Public License and the GNU Lesser General Public License
+ * * for more details.
+ * *
+ * * You should have received a copy of the GNU General Public License
+ * * and the GNU Lesser General Public License along with this program;
+ * * if not, write to the Free Software Foundation, Inc.,
+ * * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
+ *
+ */
+
+package org.rhq.server.metrics;
+
+import static org.testng.Assert.fail;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+
+import com.google.common.base.Throwables;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.testng.annotations.Test;
+
+import org.rhq.core.domain.measurement.MeasurementDataNumeric;
+
+public class TimeoutTest extends CassandraIntegrationTest {
+
+ @Test
+ public void generateTimeout() throws Exception {
+ MetricsConfiguration configuration = new MetricsConfiguration();
+
+ MetricsServer metricsServer = new MetricsServer();
+ metricsServer.setSession(session);
+ metricsServer.setConfiguration(configuration);
+
+ DateTimeService dateTimeService = new DateTimeService();
+ dateTimeService.setConfiguration(configuration);
+ metricsServer.setDateTimeService(dateTimeService);
+
+ MetricsDAO dao = new MetricsDAO(session, configuration);
+ metricsServer.setDAO(dao);
+
+ long time = hour0().getMillis();
+ Set<MeasurementDataNumeric> data = new HashSet<MeasurementDataNumeric>();
+ for (int i = 0; i < 5000; ++i) {
+ data.add(new MeasurementDataNumeric(time, i, (double) i));
+ }
+
+ WaitForRawInserts waitForRawInserts = new WaitForRawInserts(data.size());
+ metricsServer.addNumericData(data, waitForRawInserts);
+
+ waitForRawInserts.await("Failed to insert raw metrics");
+ }
+
+ private static class WaitForRawInserts implements RawDataInsertedCallback {
+
+ private final Log log = LogFactory.getLog(WaitForRawInserts.class);
+
+ private CountDownLatch latch;
+
+ private Throwable throwable;
+
+ public WaitForRawInserts(int numInserts) {
+ latch = new CountDownLatch(numInserts);
+ }
+
+ @Override
+ public void onFinish() {
+ }
+
+ @Override
+ public void onSuccess(MeasurementDataNumeric measurementDataNumeric) {
+ latch.countDown();
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ latch.countDown();
+ this.throwable = throwable;
+ log.error("An async operation failed", throwable);
+ }
+
+ public void await(String errorMsg) throws InterruptedException {
+ latch.await();
+ if (throwable != null) {
+ fail(errorMsg, Throwables.getRootCause(throwable));
+ }
+ }
+ }
+
+}
commit e03eb8443c1dad9628452505295813ad3f3bb074
Author: John Sanda <jsanda(a)redhat.com>
Date: Thu May 23 20:13:59 2013 -0400
initial implementation for using the CQL driver's async API for inserting raw metrics
diff --git a/modules/common/cassandra-util/src/main/java/org/rhq/cassandra/util/ClusterBuilder.java b/modules/common/cassandra-util/src/main/java/org/rhq/cassandra/util/ClusterBuilder.java
index 6587c71..4687b4b 100644
--- a/modules/common/cassandra-util/src/main/java/org/rhq/cassandra/util/ClusterBuilder.java
+++ b/modules/common/cassandra-util/src/main/java/org/rhq/cassandra/util/ClusterBuilder.java
@@ -26,6 +26,7 @@
package org.rhq.cassandra.util;
import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.PoolingOptions;
import com.datastax.driver.core.ProtocolOptions;
/**
@@ -50,6 +51,13 @@ public class ClusterBuilder {
}
/**
+ * @see com.datastax.driver.core.Cluster.Builder#poolingOptions()
+ */
+ public PoolingOptions poolingOptions() {
+ return builder.poolingOptions();
+ }
+
+ /**
* @see Cluster.Builder#withCredentials(String, String)
*/
public ClusterBuilder withCredentials(String username, String password) {
diff --git a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cassandra/SessionManagerBean.java b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cassandra/SessionManagerBean.java
index 6af6ea8..911261e 100644
--- a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cassandra/SessionManagerBean.java
+++ b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cassandra/SessionManagerBean.java
@@ -33,6 +33,8 @@ import javax.ejb.EJB;
import javax.ejb.Singleton;
import com.datastax.driver.core.Cluster;
+import com.datastax.driver.core.HostDistance;
+import com.datastax.driver.core.PoolingOptions;
import com.datastax.driver.core.Session;
import org.rhq.cassandra.util.ClusterBuilder;
@@ -90,6 +92,9 @@ public class SessionManagerBean {
.withCredentials(username, password)
.withPort(port)
.build();
+ PoolingOptions poolingOptions = cluster.getConfiguration().getPoolingOptions();
+ poolingOptions.setMaxConnectionsPerHost(HostDistance.LOCAL, 16);
+ poolingOptions.setMaxConnectionsPerHost(HostDistance.REMOTE, 16);
session = cluster.connect("rhq");
metricsDAO = new MetricsDAO(session, metricsConfiguration);
diff --git a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/LoadInfo.java b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/LoadInfo.java
new file mode 100644
index 0000000..5cf05c2
--- /dev/null
+++ b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/cloud/LoadInfo.java
@@ -0,0 +1,75 @@
+/*
+ *
+ * * RHQ Management Platform
+ * * Copyright (C) 2005-2012 Red Hat, Inc.
+ * * All rights reserved.
+ * *
+ * * This program is free software; you can redistribute it and/or modify
+ * * it under the terms of the GNU General Public License, version 2, as
+ * * published by the Free Software Foundation, and/or the GNU Lesser
+ * * General Public License, version 2.1, also as published by the Free
+ * * Software Foundation.
+ * *
+ * * This program is distributed in the hope that it will be useful,
+ * * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * * GNU General Public License and the GNU Lesser General Public License
+ * * for more details.
+ * *
+ * * You should have received a copy of the GNU General Public License
+ * * and the GNU Lesser General Public License along with this program;
+ * * if not, write to the Free Software Foundation, Inc.,
+ * * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
+ *
+ */
+
+package org.rhq.enterprise.server.cloud;
+
+import org.rhq.enterprise.server.measurement.MeasurementAggregate;
+
+/**
+ * @author John Sanda
+ */
+public class LoadInfo {
+
+ private String diskSpaceUsed;
+
+ private String ringOwnership;
+
+ private MeasurementAggregate heapCommitted;
+
+ private MeasurementAggregate heapUsage;
+
+ public String getDiskSpaceUsed() {
+ return diskSpaceUsed;
+ }
+
+ public void setDiskSpaceUsed(String diskSpaceUsed) {
+ this.diskSpaceUsed = diskSpaceUsed;
+ }
+
+ public String getRingOwnership() {
+ return ringOwnership;
+ }
+
+ public void setRingOwnership(String ringOwnership) {
+ this.ringOwnership = ringOwnership;
+ }
+
+ public MeasurementAggregate getHeapCommitted() {
+ return heapCommitted;
+ }
+
+ public void setHeapCommitted(MeasurementAggregate heapCommitted) {
+ this.heapCommitted = heapCommitted;
+ }
+
+ public MeasurementAggregate getHeapUsage() {
+ return heapUsage;
+ }
+
+ public void setHeapUsage(MeasurementAggregate heapUsage) {
+ this.heapUsage = heapUsage;
+ }
+
+}
diff --git a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerBean.java b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerBean.java
index 2f300c3..9686dea 100644
--- a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerBean.java
+++ b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerBean.java
@@ -34,6 +34,7 @@ import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeoutException;
+import javax.ejb.Asynchronous;
import javax.ejb.EJB;
import javax.ejb.Stateless;
import javax.ejb.TransactionAttribute;
@@ -87,12 +88,15 @@ import org.rhq.enterprise.server.alert.engine.AlertConditionCacheManagerLocal;
import org.rhq.enterprise.server.alert.engine.AlertConditionCacheStats;
import org.rhq.enterprise.server.authz.AuthorizationManagerLocal;
import org.rhq.enterprise.server.authz.PermissionException;
+import org.rhq.enterprise.server.cassandra.SessionManagerBean;
import org.rhq.enterprise.server.core.AgentManagerLocal;
import org.rhq.enterprise.server.measurement.instrumentation.MeasurementMonitor;
import org.rhq.enterprise.server.measurement.util.MeasurementDataManagerUtility;
import org.rhq.enterprise.server.resource.group.ResourceGroupManagerLocal;
import org.rhq.enterprise.server.util.CriteriaQueryGenerator;
import org.rhq.enterprise.server.util.CriteriaQueryRunner;
+import org.rhq.server.metrics.MetricsServer;
+import org.rhq.server.metrics.RawDataInsertedCallback;
/**
* A manager for {@link MeasurementData}s.
@@ -152,6 +156,9 @@ public class MeasurementDataManagerBean implements MeasurementDataManagerLocal,
private MetricsManagerLocal metricsManager;
@EJB
+ private SessionManagerBean sessionManagerBean;
+
+ @EJB
private MeasurementScheduleManagerLocal measurementScheduleManager;
// doing a bulk delete in here, need to be in its own tx
@@ -214,13 +221,31 @@ public class MeasurementDataManagerBean implements MeasurementDataManagerLocal,
* @param data the actual data points
*/
@TransactionAttribute(TransactionAttributeType.NOT_SUPPORTED)
- public void addNumericData(Set<MeasurementDataNumeric> data) {
+ public void addNumericData(final Set<MeasurementDataNumeric> data) {
if ((data == null) || (data.isEmpty())) {
return;
}
- metricsManager.addNumericData(data);
- notifyAlertConditionCacheManager("mergeMeasurementReport", data.toArray(new MeasurementData[data.size()]));
+ MetricsServer metricsServer = sessionManagerBean.getMetricsServer();
+ metricsServer.addNumericData(data, new RawDataInsertedCallback() {
+
+ private List<MeasurementData> insertedData = new ArrayList<MeasurementData>();
+
+ @Override
+ public void onFinish() {
+ measurementDataManager.updateAlertConditionCache("mergeMeasurementReport",
+ insertedData.toArray(new MeasurementData[insertedData.size()]));
+ }
+
+ @Override
+ public void onSuccess(MeasurementDataNumeric measurementDataNumeric) {
+ insertedData.add(measurementDataNumeric);
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ }
+ });
}
@TransactionAttribute(TransactionAttributeType.REQUIRES_NEW)
@@ -548,6 +573,13 @@ public class MeasurementDataManagerBean implements MeasurementDataManagerLocal,
return metricsManager.findLatestValueForResource(scheduleId);
}
+ @Asynchronous
+ @Override
+ public void updateAlertConditionCache(String callingMethod, MeasurementData[] data) {
+ AlertConditionCacheStats stats = alertConditionCacheManager.checkConditions(data);
+ log.debug(callingMethod + ": " + stats.toString());
+ }
+
private void notifyAlertConditionCacheManager(String callingMethod, MeasurementData[] data) {
AlertConditionCacheStats stats = alertConditionCacheManager.checkConditions(data);
diff --git a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerLocal.java b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerLocal.java
index 644107d..cc4a86e 100644
--- a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerLocal.java
+++ b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementDataManagerLocal.java
@@ -140,4 +140,6 @@ public interface MeasurementDataManagerLocal extends MeasurementDataManagerRemot
MeasurementAggregate getAggregate(Subject subject, int groupId, int definitionId, long startTime, long endTime);
+ void updateAlertConditionCache(String callingMethod, MeasurementData[] data);
+
}
\ No newline at end of file
diff --git a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MetricsManagerBean.java b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MetricsManagerBean.java
index 26d580b..c77698b 100644
--- a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MetricsManagerBean.java
+++ b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MetricsManagerBean.java
@@ -51,9 +51,9 @@ public class MetricsManagerBean implements MetricsManagerLocal {
@Override
@TransactionAttribute(TransactionAttributeType.NOT_SUPPORTED)
- public void addNumericData(Set<MeasurementDataNumeric> data) {
+ public void addNumericData(Set<MeasurementDataNumeric> data, Runnable callback) {
MetricsServer metricsServer = getMetricsServer();
- metricsServer.addNumericData(data);
+ metricsServer.addNumericData(data, null);
}
@Override
diff --git a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MetricsManagerLocal.java b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MetricsManagerLocal.java
index 9880d1f..c49dd75 100644
--- a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MetricsManagerLocal.java
+++ b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MetricsManagerLocal.java
@@ -45,7 +45,7 @@ import org.rhq.server.metrics.domain.AggregateNumericMetric;
@Local
public interface MetricsManagerLocal {
- void addNumericData(Set<MeasurementDataNumeric> data);
+ void addNumericData(Set<MeasurementDataNumeric> data, Runnable callback);
Iterable<AggregateNumericMetric> calculateAggregates();
diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsDAO.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsDAO.java
index dd5c44c..6ad9755 100644
--- a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsDAO.java
+++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsDAO.java
@@ -37,6 +37,7 @@ import java.util.Set;
import com.datastax.driver.core.BoundStatement;
import com.datastax.driver.core.PreparedStatement;
import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.ResultSetFuture;
import com.datastax.driver.core.Session;
import com.datastax.driver.core.Statement;
import com.datastax.driver.core.exceptions.NoHostAvailableException;
@@ -141,10 +142,10 @@ public class MetricsDAO {
deleteIndexEntries = session.prepare("DELETE FROM " + MetricsTable.INDEX + " WHERE bucket = ? AND time = ?");
}
- public ResultSet insertRawData(MeasurementDataNumeric data) {
+ public ResultSetFuture insertRawData(MeasurementDataNumeric data) {
BoundStatement statement = insertRawData.bind(data.getScheduleId(), new Date(data.getTimestamp()),
data.getValue());
- return session.execute(statement);
+ return session.executeAsync(statement);
}
public List<MetricResultFuture<MeasurementDataNumeric>> insertRawMetricsAsync(Set<MeasurementDataNumeric> dataSet,
@@ -301,6 +302,11 @@ public class MetricsDAO {
}
}
+ public ResultSetFuture updateMetricsIndex(MetricsTable table, int scheduleId, long timestamp) {
+ BoundStatement statement = updateMetricsIndex.bind(table.getTableName(), new Date(timestamp), scheduleId);
+ return session.executeAsync(statement);
+ }
+
public void deleteMetricsIndexEntries(MetricsTable table, long timestamp) {
BoundStatement statement = deleteIndexEntries.bind(table.getTableName(), new Date(timestamp));
session.execute(statement);
diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
index fd285ee..5e49d45 100644
--- a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
+++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsServer.java
@@ -32,8 +32,16 @@ import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeMap;
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicInteger;
+import com.datastax.driver.core.ResultSet;
+import com.datastax.driver.core.ResultSetFuture;
import com.datastax.driver.core.Session;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import com.google.common.util.concurrent.ListeningExecutorService;
+import com.google.common.util.concurrent.MoreExecutors;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
@@ -67,6 +75,9 @@ public class MetricsServer {
private MetricsConfiguration configuration;
+ private ListeningExecutorService listeningExecutorService = MoreExecutors.listeningDecorator(
+ Executors.newFixedThreadPool(5));
+
public void setSession(Session session) {
this.session = session;
}
@@ -233,16 +244,64 @@ public class MetricsServer {
return data;
}
- public void addNumericData(Set<MeasurementDataNumeric> dataSet) {
- try {
- for (MeasurementDataNumeric data : dataSet) {
- dao.insertRawData(data);
+ public void addNumericData(final Set<MeasurementDataNumeric> dataSet,
+ final RawDataInsertedCallback callback) {
+// try {
+ if (log.isDebugEnabled()) {
+ log.debug("Inserting " + dataSet.size() + " raw metrics");
}
- updateMetricsIndex(dataSet);
- } catch (Exception e) {
- log.error("An error occurred while inserting raw numeric data", e);
- throw new RuntimeException(e);
- }
+ final long startTime = System.currentTimeMillis();
+ final AtomicInteger remainingInserts = new AtomicInteger(dataSet.size());
+
+ for (final MeasurementDataNumeric data : dataSet) {
+ ResultSetFuture resultSetFuture = dao.insertRawData(data);
+ Futures.addCallback(resultSetFuture, new FutureCallback<ResultSet>() {
+ @Override
+ public void onSuccess(ResultSet rows) {
+ updateMetricsIndex(data, dataSet.size(), remainingInserts, startTime, callback);
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ log.error("An error occurred while inserting raw data " + data, throwable);
+ callback.onFailure(throwable);
+ }
+ });
+ }
+// updateMetricsIndex(dataSet);
+// } catch (Exception e) {
+// log.error("An error occurred while inserting raw numeric data", e);
+// throw new RuntimeException(e);
+// }
+ }
+
+ private void updateMetricsIndex(final MeasurementDataNumeric rawData, final int total,
+ final AtomicInteger remainingInserts, final long startTime, final RawDataInsertedCallback callback) {
+
+ long timeSlice = dateTimeService.getTimeSlice(new DateTime(rawData.getTimestamp()),
+ configuration.getRawTimeSliceDuration()).getMillis();
+ ResultSetFuture resultSetFuture = dao.updateMetricsIndex(MetricsTable.ONE_HOUR, rawData.getScheduleId(),
+ timeSlice);
+ Futures.addCallback(resultSetFuture, new FutureCallback<ResultSet>() {
+ @Override
+ public void onSuccess(ResultSet rows) {
+ callback.onSuccess(rawData);
+ if (remainingInserts.decrementAndGet() == 0) {
+ long endTime = System.currentTimeMillis();
+ if (log.isDebugEnabled()) {
+ log.debug("Finished inserting " + total + " raw metrics in " + (endTime - startTime) + " ms");
+ }
+ callback.onFinish();
+ }
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ log.error("An error occurred while trying to update " + MetricsTable.INDEX + " for raw data " +
+ rawData);
+ callback.onFailure(throwable);
+ }
+ });
}
void updateMetricsIndex(Set<MeasurementDataNumeric> rawMetrics) {
@@ -255,7 +314,7 @@ public class MetricsServer {
for (Long ts : updates.values()) {
dates.add(new Date(ts));
}
- log.info("Updating one hour index wtih time slices " + dates);
+ log.debug("Updating one hour index wtih time slices " + dates);
dao.updateMetricsIndex(MetricsTable.ONE_HOUR, updates);
}
diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/RawDataInsertedCallback.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/RawDataInsertedCallback.java
new file mode 100644
index 0000000..ab9e50a
--- /dev/null
+++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/RawDataInsertedCallback.java
@@ -0,0 +1,39 @@
+/*
+ *
+ * * RHQ Management Platform
+ * * Copyright (C) 2005-2012 Red Hat, Inc.
+ * * All rights reserved.
+ * *
+ * * This program is free software; you can redistribute it and/or modify
+ * * it under the terms of the GNU General Public License, version 2, as
+ * * published by the Free Software Foundation, and/or the GNU Lesser
+ * * General Public License, version 2.1, also as published by the Free
+ * * Software Foundation.
+ * *
+ * * This program is distributed in the hope that it will be useful,
+ * * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ * * GNU General Public License and the GNU Lesser General Public License
+ * * for more details.
+ * *
+ * * You should have received a copy of the GNU General Public License
+ * * and the GNU Lesser General Public License along with this program;
+ * * if not, write to the Free Software Foundation, Inc.,
+ * * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
+ *
+ */
+
+package org.rhq.server.metrics;
+
+import com.google.common.util.concurrent.FutureCallback;
+
+import org.rhq.core.domain.measurement.MeasurementDataNumeric;
+
+/**
+ * @author John Sanda
+ */
+public interface RawDataInsertedCallback extends FutureCallback<MeasurementDataNumeric> {
+
+ void onFinish();
+
+}
diff --git a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsDAOTest.java b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsDAOTest.java
index 8f35486..c522316 100644
--- a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsDAOTest.java
+++ b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsDAOTest.java
@@ -29,6 +29,7 @@ import static java.util.Arrays.asList;
import static org.rhq.test.AssertUtils.assertCollectionMatchesNoOrder;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.fail;
import java.util.ArrayList;
import java.util.HashMap;
@@ -37,9 +38,16 @@ import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import com.datastax.driver.core.ResultSetFuture;
+import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
+import com.google.common.util.concurrent.FutureCallback;
+import com.google.common.util.concurrent.Futures;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
import org.joda.time.DateTime;
import org.joda.time.Hours;
import org.testng.annotations.BeforeClass;
@@ -59,6 +67,8 @@ import org.rhq.server.metrics.domain.RawNumericMetric;
*/
public class MetricsDAOTest extends CassandraIntegrationTest {
+ private final Log log = LogFactory.getLog(MetricsDAOTest.class);
+
private static final boolean ENABLED = true;
private final long SECOND = 1000;
@@ -84,16 +94,19 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
}
@Test(enabled = ENABLED)
- public void insertAndFindRawData() {
+ public void insertAndFindRawData() throws Exception {
DateTime hour0 = hour0();
DateTime currentTime = hour0.plusHours(4).plusMinutes(44);
- DateTime currentHour = currentTime.hourOfDay().roundFloorCopy();
DateTime threeMinutesAgo = currentTime.minusMinutes(3);
int scheduleId = 1;
MeasurementDataNumeric expected = new MeasurementDataNumeric(threeMinutesAgo.getMillis(), scheduleId, 1.23);
- dao.insertRawData(expected);
+ WaitForResults waitForResults = new WaitForResults(1);
+
+ ResultSetFuture resultSetFuture = dao.insertRawData(expected);
+ Futures.addCallback(resultSetFuture, waitForResults);
+ waitForResults.await("Failed to insert raw data");
List<RawNumericMetric> actualMetrics = Lists.newArrayList(dao.findRawMetrics(scheduleId,
threeMinutesAgo.minusSeconds(1).getMillis(), threeMinutesAgo.plusSeconds(1).getMillis()));
@@ -104,7 +117,7 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
}
@Test(enabled = ENABLED)
- public void findLatestRawMetric() {
+ public void findLatestRawMetric() throws Exception {
DateTime hour0 = hour0();
DateTime currentTime = hour0.plusHours(4).plusMinutes(44);
DateTime threeMinutesAgo = currentTime.minusMinutes(3);
@@ -118,9 +131,13 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
data.add(new MeasurementDataNumeric(twoMinutesAgo.getMillis(), scheduleId, 3.9));
data.add(new MeasurementDataNumeric(oneMinuteAgo.getMillis(), scheduleId, 2.6));
+ WaitForResults waitForResults = new WaitForResults(data.size());
+
for (MeasurementDataNumeric raw : data) {
- dao.insertRawData(raw);
+ ResultSetFuture resultSetFuture = dao.insertRawData(raw);
+ Futures.addCallback(resultSetFuture, waitForResults);
}
+ waitForResults.await("Failed to insert raw data");
RawNumericMetric actual = dao.findLatestRawMetric(scheduleId);
RawNumericMetric expected = new RawNumericMetric(scheduleId, oneMinuteAgo.getMillis(), 2.6);
@@ -129,7 +146,7 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
}
@Test(enabled = ENABLED)
- public void findRawMetricsForMultipleSchedules() {
+ public void findRawMetricsForMultipleSchedules() throws Exception {
DateTime currentTime = hour0().plusHours(4).plusMinutes(44);
DateTime currentHour = currentTime.hourOfDay().roundFloorCopy();
DateTime threeMinutesAgo = currentTime.minusMinutes(3);
@@ -147,9 +164,13 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
data.add(new MeasurementDataNumeric(oneMinuteAgo.getMillis(), scheduleId1, 3.1));
data.add(new MeasurementDataNumeric(oneMinuteAgo.getMillis(), scheduleId2, 3.2));
+ WaitForResults waitForResults = new WaitForResults(data.size());
+
for (MeasurementDataNumeric raw : data) {
- dao.insertRawData(raw);
+ ResultSetFuture resultSetFuture = dao.insertRawData(raw);
+ Futures.addCallback(resultSetFuture, waitForResults);
}
+ waitForResults.await("Failed to insert raw data");
List<RawNumericMetric> actualMetrics = Lists.newArrayList(dao.findRawMetrics(asList(scheduleId1, scheduleId2),
currentHour.getMillis(), currentHour.plusHours(1).getMillis()));
@@ -261,6 +282,34 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
}
@Test(enabled = ENABLED)
+ public void updateSixHourMetricsIndex() throws Exception {
+ int scheduleId1 = 1;
+ int scheduleId2= 2;
+
+ WaitForResults waitForResults = new WaitForResults(2);
+
+ ResultSetFuture resultSetFuture1 = dao.updateMetricsIndex(MetricsTable.TWENTY_FOUR_HOUR, scheduleId1,
+ hour0().getMillis());
+ ResultSetFuture resultSetFuture2 = dao.updateMetricsIndex(MetricsTable.TWENTY_FOUR_HOUR, scheduleId2,
+ hour0().getMillis());
+
+ Futures.addCallback(resultSetFuture1, waitForResults);
+ Futures.addCallback(resultSetFuture2, waitForResults);
+
+ waitForResults.await("Failed to update metrics index");
+
+ List<MetricsIndexEntry> expected = asList(
+ new MetricsIndexEntry(MetricsTable.TWENTY_FOUR_HOUR, hour0(), scheduleId1),
+ new MetricsIndexEntry(MetricsTable.TWENTY_FOUR_HOUR, hour0(), scheduleId2)
+ );
+
+ List<MetricsIndexEntry> actual = Lists.newArrayList(dao.findMetricsIndexEntries(MetricsTable.TWENTY_FOUR_HOUR,
+ hour0().getMillis()));
+
+ assertCollectionMatchesNoOrder(expected, actual, "Failed to update or retrieve metrics index entries");
+ }
+
+ @Test(enabled = ENABLED)
public void updateAndFindOneHourIndexEntries() {
DateTime hour0 = hour0();
int scheduleId1 = 1;
@@ -623,4 +672,37 @@ public class MetricsDAOTest extends CassandraIntegrationTest {
}
}
+ private static class WaitForResults<ResultSetFuture> implements FutureCallback<ResultSetFuture> {
+
+ private final Log log = LogFactory.getLog(WaitForResults.class);
+
+ private CountDownLatch latch;
+
+ private Throwable throwable;
+
+ public WaitForResults(int numResults) {
+ latch = new CountDownLatch(numResults);
+ }
+
+ @Override
+ public void onSuccess(ResultSetFuture resultSetFuture) {
+ latch.countDown();
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ latch.countDown();
+ this.throwable = throwable;
+ log.error("An async operation failed", throwable);
+ }
+
+ public void await(String errorMsg) throws InterruptedException {
+ latch.await();
+ if (throwable != null) {
+ fail(errorMsg, Throwables.getRootCause(throwable));
+ }
+ }
+
+ }
+
}
diff --git a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsServerTest.java b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsServerTest.java
index d895753..cf74fcd 100644
--- a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsServerTest.java
+++ b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/MetricsServerTest.java
@@ -31,6 +31,7 @@ import static org.rhq.test.AssertUtils.assertCollectionMatchesNoOrder;
import static org.rhq.test.AssertUtils.assertPropertiesMatch;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
import java.math.BigDecimal;
import java.math.MathContext;
@@ -39,7 +40,9 @@ import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import com.google.common.base.Throwables;
import com.google.common.collect.Lists;
import org.apache.commons.logging.Log;
@@ -137,8 +140,12 @@ public class MetricsServerTest extends CassandraIntegrationTest {
data.add(new MeasurementDataNumeric(twoMinutesAgo.getMillis(), scheduleId, 3.9));
data.add(new MeasurementDataNumeric(oneMinuteAgo.getMillis(), scheduleId, 2.6));
+ WaitForRawInserts waitForRawInserts = new WaitForRawInserts(data.size());
+
long timestamp = System.currentTimeMillis();
- metricsServer.addNumericData(data);
+ metricsServer.addNumericData(data, waitForRawInserts);
+
+ waitForRawInserts.await("Failed to insert raw data");
List<RawNumericMetric> actual = Lists.newArrayList(dao.findRawMetrics(scheduleId, hour0.plusHours(4)
.getMillis(), hour0.plusHours(5).getMillis()));
@@ -161,7 +168,7 @@ public class MetricsServerTest extends CassandraIntegrationTest {
}
@Test//(enabled = ENABLED)
- public void calculateAggregatesForOneScheduleWhenDBIsEmpty() {
+ public void calculateAggregatesForOneScheduleWhenDBIsEmpty() throws Exception {
int scheduleId = 123;
DateTime hour0 = hour0();
@@ -176,10 +183,14 @@ public class MetricsServerTest extends CassandraIntegrationTest {
data.add(new MeasurementDataNumeric(secondMetricTime.getMillis(), scheduleId, 3.9));
data.add(new MeasurementDataNumeric(thirdMetricTime.getMillis(), scheduleId, 2.6));
+ WaitForRawInserts waitForRawInserts = new WaitForRawInserts(data.size());
+
metricsServer.setCurrentHour(hour6);
- metricsServer.addNumericData(data);
+ metricsServer.addNumericData(data, waitForRawInserts);
metricsServer.calculateAggregates(hour6.getMillis());
+ waitForRawInserts.await("Failed to insert raw data");
+
// verify that one hour metric data is updated
List<AggregateNumericMetric> expected = asList(new AggregateNumericMetric(scheduleId,
divide((3.9 + 3.2 + 2.6), 3), 2.6, 3.9, lastHour.getMillis()));
@@ -349,7 +360,7 @@ public class MetricsServerTest extends CassandraIntegrationTest {
}
@Test//(enabled = ENABLED)
- public void findRawDataCompositesForResource() {
+ public void findRawDataCompositesForResource() throws Exception {
DateTime beginTime = now().minusHours(4);
DateTime endTime = now();
Buckets buckets = new Buckets(beginTime, endTime);
@@ -368,10 +379,14 @@ public class MetricsServerTest extends CassandraIntegrationTest {
data.add(new MeasurementDataNumeric(buckets.get(59).getStartTime() + buckets.getInterval() + 50, scheduleId,
4.56));
- metricsServer.addNumericData(data);
+ WaitForRawInserts waitForRawInserts = new WaitForRawInserts(data.size());
+
+ metricsServer.addNumericData(data, waitForRawInserts);
List<MeasurementDataNumericHighLowComposite> actualData = Lists.newArrayList(metricsServer.findDataForResource(
scheduleId, beginTime.getMillis(), endTime.getMillis()));
+ waitForRawInserts.await("Faile to insert raw data");
+
assertEquals(actualData.size(), buckets.getNumDataPoints(), "Expected to get back 60 data points.");
MeasurementDataNumericHighLowComposite expectedBucket0Data = new MeasurementDataNumericHighLowComposite(
@@ -390,7 +405,7 @@ public class MetricsServerTest extends CassandraIntegrationTest {
}
@Test
- public void findLatestValueForResource() {
+ public void findLatestValueForResource() throws Exception {
int scheduleId = 123;
DateTime fifteenMinutesAgo = now().minusMinutes(15);
@@ -402,7 +417,11 @@ public class MetricsServerTest extends CassandraIntegrationTest {
data.add(new MeasurementDataNumeric(tenMinutesAgo.getMillis(), scheduleId, 2.2));
data.add(new MeasurementDataNumeric(fiveMinutesAgo.getMillis(), scheduleId, 3.3));
- metricsServer.addNumericData(data);
+ WaitForRawInserts waitForRawInserts = new WaitForRawInserts(data.size());
+
+ metricsServer.addNumericData(data, waitForRawInserts);
+
+ waitForRawInserts.await("Failed to insert raw data");
RawNumericMetric actual = metricsServer.findLatestValueForResource(scheduleId);
RawNumericMetric expected = new RawNumericMetric(scheduleId, fiveMinutesAgo.getMillis(), 3.3);
@@ -411,7 +430,7 @@ public class MetricsServerTest extends CassandraIntegrationTest {
}
@Test
- public void getSummaryRawAggregateForResource() {
+ public void getSummaryRawAggregateForResource() throws Exception {
DateTime beginTime = now().minusHours(4);
DateTime endTime = now();
Buckets buckets = new Buckets(beginTime, endTime);
@@ -430,7 +449,11 @@ public class MetricsServerTest extends CassandraIntegrationTest {
data.add(new MeasurementDataNumeric(buckets.get(59).getStartTime() + buckets.getInterval() + 50, scheduleId,
4.56));
- metricsServer.addNumericData(data);
+ WaitForRawInserts waitForRawInserts = new WaitForRawInserts(data.size());
+
+ metricsServer.addNumericData(data, waitForRawInserts);
+
+ waitForRawInserts.await("Failed to insert raw data");
AggregateNumericMetric actual = metricsServer.getSummaryAggregate(scheduleId, beginTime.getMillis(),
endTime.getMillis());
@@ -511,7 +534,7 @@ public class MetricsServerTest extends CassandraIntegrationTest {
}
@Test
- public void getSummaryRawAggregateForGroup() {
+ public void getSummaryRawAggregateForGroup() throws Exception {
DateTime beginTime = now().minusHours(4);
DateTime endTime = now();
Buckets buckets = new Buckets(beginTime, endTime);
@@ -545,7 +568,11 @@ public class MetricsServerTest extends CassandraIntegrationTest {
data.add(new MeasurementDataNumeric(buckets.get(59).getStartTime() + buckets.getInterval() + 50, scheduleId2,
4.56));
- metricsServer.addNumericData(data);
+ WaitForRawInserts waitForRawInserts = new WaitForRawInserts(data.size());
+
+ metricsServer.addNumericData(data, waitForRawInserts);
+
+ waitForRawInserts.await("Failed to insert raw data");
AggregateNumericMetric actual = metricsServer.getSummaryAggregate(asList(scheduleId1, scheduleId2),
beginTime.getMillis(), endTime.getMillis());
@@ -559,7 +586,7 @@ public class MetricsServerTest extends CassandraIntegrationTest {
}
@Test
- public void findRawDataCompositesForGroup() {
+ public void findRawDataCompositesForGroup() throws Exception {
DateTime beginTime = now().minusHours(4);
DateTime endTime = now();
Buckets buckets = new Buckets(beginTime, endTime);
@@ -593,7 +620,11 @@ public class MetricsServerTest extends CassandraIntegrationTest {
data.add(new MeasurementDataNumeric(buckets.get(59).getStartTime() + buckets.getInterval() + 50, scheduleId2,
4.56));
- metricsServer.addNumericData(data);
+ WaitForRawInserts waitForRawInserts = new WaitForRawInserts(data.size());
+
+ metricsServer.addNumericData(data, waitForRawInserts);
+
+ waitForRawInserts.await("Failed to insert raw data");
List<MeasurementDataNumericHighLowComposite> actualData = metricsServer.findDataForGroup(
asList(scheduleId1, scheduleId2), beginTime.getMillis(), endTime.getMillis());
@@ -829,4 +860,40 @@ public class MetricsServerTest extends CassandraIntegrationTest {
"WHERE schedule_id = " + scheduleId + " AND time >= " + startTime + " AND time < " + endTime;
return new SimplePagedResult<RawNumericMetric>(cql, new RawNumericMetricMapper(true), session);
}
+
+ private static class WaitForRawInserts implements RawDataInsertedCallback {
+
+ private final Log log = LogFactory.getLog(WaitForRawInserts.class);
+
+ private CountDownLatch latch;
+
+ private Throwable throwable;
+
+ public WaitForRawInserts(int numInserts) {
+ latch = new CountDownLatch(numInserts);
+ }
+
+ @Override
+ public void onFinish() {
+ }
+
+ @Override
+ public void onSuccess(MeasurementDataNumeric measurementDataNumeric) {
+ latch.countDown();
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ latch.countDown();
+ this.throwable = throwable;
+ log.error("An async operation failed", throwable);
+ }
+
+ public void await(String errorMsg) throws InterruptedException {
+ latch.await();
+ if (throwable != null) {
+ fail(errorMsg, Throwables.getRootCause(throwable));
+ }
+ }
+ }
}
diff --git a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/MeasurementCollector.java b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/MeasurementCollector.java
index 35f575f..b0a67c8 100644
--- a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/MeasurementCollector.java
+++ b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/MeasurementCollector.java
@@ -36,6 +36,7 @@ import org.apache.commons.logging.LogFactory;
import org.rhq.core.domain.measurement.MeasurementDataNumeric;
import org.rhq.metrics.simulator.stats.Stats;
import org.rhq.server.metrics.MetricsServer;
+import org.rhq.server.metrics.RawDataInsertedCallback;
/**
* @author John Sanda
@@ -56,6 +57,8 @@ public class MeasurementCollector implements Runnable {
private int batchSize = 500;
+ private NoOpCallback rawInsertsCallback = new NoOpCallback();
+
public void setQueue(PriorityQueue<Schedule> queue) {
this.queue = queue;
}
@@ -113,7 +116,7 @@ public class MeasurementCollector implements Runnable {
}
metricsCollected = data.size();
try {
- metricsServer.addNumericData(data);
+ metricsServer.addNumericData(data, rawInsertsCallback);
} catch (Exception e) {
log.error("An error occurred while trying to store raw metrics", e);
log.error("Requesting simulation shutdown...");
@@ -135,4 +138,18 @@ public class MeasurementCollector implements Runnable {
log.info("Finished collecting and storing " + metricsCollected + " raw metric in " +totalTime + " ms.");
}
}
+
+ private static class NoOpCallback implements RawDataInsertedCallback {
+ @Override
+ public void onFinish() {
+ }
+
+ @Override
+ public void onSuccess(MeasurementDataNumeric measurementDataNumeric) {
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+ }
+ }
}
diff --git a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/Simulator.java b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/Simulator.java
index 681eb16..3dacf23 100644
--- a/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/Simulator.java
+++ b/modules/helpers/metrics-simulator/src/main/java/org/rhq/metrics/simulator/Simulator.java
@@ -58,6 +58,7 @@ import org.rhq.metrics.simulator.plan.ScheduleGroup;
import org.rhq.metrics.simulator.plan.SimulationPlan;
import org.rhq.metrics.simulator.stats.Stats;
import org.rhq.server.metrics.DateTimeService;
+import org.rhq.server.metrics.MetricsDAO;
import org.rhq.server.metrics.MetricsServer;
/**
@@ -95,8 +96,10 @@ public class Simulator implements ShutdownManager {
session = createSession(nodes, compression);
}
+ MetricsDAO metricsDAO = new MetricsDAO(session, plan.getMetricsServerConfiguration());
MetricsServer metricsServer = new MetricsServer();
metricsServer.setSession(session);
+ metricsServer.setDAO(metricsDAO);
metricsServer.setConfiguration(plan.getMetricsServerConfiguration());
DateTimeService dateTimeService = new DateTimeService();
commit c7d8e5d6bb4d2921761a06ac1fe225e100c6bab5
Author: John Sanda <jsanda(a)redhat.com>
Date: Wed May 22 20:20:35 2013 -0400
changing transaction demarcation for basline and OOB calculations
The queries and methods in MeasurementBaselineManagerBean and in
MeasurementOOBManagerBean were too coarse-grained. In a test environment, I
was seeing regular timeouts while querying for schedules that do not yet have
baselines. And the method that computes OOBs was experiencing frequent
transaction timeouts. The transaction demarcation has been changed so that we
are doing a small, fixed size amount of working for each transaction when
calculating baselines and OOBs. Calls to Cassandra have been moved to
non-transactional methods.
Lastly, fixing a bug in MetricsBaselineCalculator to make sure we always return
a baseline for each schedule that it receives.
diff --git a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementBaselineManagerBean.java b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementBaselineManagerBean.java
index 8fb615c..37af9de 100644
--- a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementBaselineManagerBean.java
+++ b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementBaselineManagerBean.java
@@ -20,6 +20,8 @@ package org.rhq.enterprise.server.measurement;
import java.util.Arrays;
import java.util.Date;
+import java.util.Iterator;
+import java.util.LinkedList;
import java.util.List;
import java.util.Properties;
@@ -88,7 +90,8 @@ public class MeasurementBaselineManagerBean implements MeasurementBaselineManage
private SessionManagerBean sessionManager;
private final Log log = LogFactory.getLog(MeasurementBaselineManagerBean.class);
- private static final int BASELINE_PROCESSING_LIMIT = 50000;
+
+ private static final int BASELINE_PROCESSING_LIMIT = 100;
@TransactionAttribute(TransactionAttributeType.NEVER)
public void calculateAutoBaselines() {
@@ -185,11 +188,14 @@ public class MeasurementBaselineManagerBean implements MeasurementBaselineManage
* In any event, an appropriate chunking solution needs to be found, and that partitioning strategy
* needs to replace the limits in the query today.
*/
- int schedulesWithoutBaselines = measurementBaselineManager
- ._calculateAutoBaselinesINSERT(amountOfData);
- totalProcessed += schedulesWithoutBaselines;
-
- if (schedulesWithoutBaselines < BASELINE_PROCESSING_LIMIT) {
+// int schedulesWithoutBaselines = measurementBaselineManager
+// ._calculateAutoBaselinesINSERT(amountOfData);
+ List<MeasurementSchedule> schedulesWithoutBaselines =
+ measurementBaselineManager.getSchedulesWithoutBaselines();
+ measurementBaselineManager.calculateBaselines(schedulesWithoutBaselines, now, amountOfData);
+ totalProcessed += schedulesWithoutBaselines.size();
+
+ if (schedulesWithoutBaselines.size() < BASELINE_PROCESSING_LIMIT) {
break;
}
}
@@ -217,6 +223,62 @@ public class MeasurementBaselineManagerBean implements MeasurementBaselineManage
@SuppressWarnings("unchecked")
@TransactionAttribute(TransactionAttributeType.REQUIRES_NEW)
+ public List<MeasurementSchedule> getSchedulesWithoutBaselines() {
+ Query query = this.entityManager
+ .createNamedQuery(MeasurementBaseline.QUERY_FIND_MEASUREMENT_SCHEDULES_WITHOUT_AUTOBASELINES);
+ query.setMaxResults(BASELINE_PROCESSING_LIMIT);
+ List<MeasurementSchedule> scheduleIdsWithoutBaselines = query.getResultList();
+
+ return scheduleIdsWithoutBaselines;
+ }
+
+ @TransactionAttribute(TransactionAttributeType.NOT_SUPPORTED)
+ public void calculateBaselines(List<MeasurementSchedule> schedules, long olderThan, long amountOfData) {
+ long endTime = olderThan;
+ long startTime = endTime - amountOfData;
+
+ log.debug("Computing baselines for " + schedules.size() + " schedules");
+ MetricsBaselineCalculator baselineCalculator = new MetricsBaselineCalculator(sessionManager.getMetricsDAO());
+ long calcStartTime = System.currentTimeMillis();
+ List<MeasurementBaseline> results = baselineCalculator.calculateBaselines(schedules, startTime, endTime);
+ long calcEndTime = System.currentTimeMillis();
+ int count = results.size();
+
+ if (log.isDebugEnabled()) {
+ log.debug("Finished computing " + count + " new baselines in " + (calcEndTime - calcStartTime) + " ms");
+ }
+
+ log.debug("Persisting baselines calculations");
+ long saveStartTime = System.currentTimeMillis();
+ Iterator<MeasurementBaseline> iterator = results.iterator();
+ List<MeasurementBaseline> queue = new LinkedList<MeasurementBaseline>();
+ while (iterator.hasNext()) {
+ if (queue.size() == 10) {
+ measurementBaselineManager.saveNewBaselines(queue);
+ queue = new LinkedList<MeasurementBaseline>();
+ }
+ queue.add(iterator.next());
+ }
+ if (!queue.isEmpty()) {
+ measurementBaselineManager.saveNewBaselines(queue);
+ }
+
+ long saveEndTime = System.currentTimeMillis();
+ if (log.isDebugEnabled()) {
+ log.debug("Finished persisting " + count + " baselines in " + (saveEndTime - saveStartTime) + " ms");
+ }
+ }
+
+ @Override
+ @TransactionAttribute(TransactionAttributeType.REQUIRES_NEW)
+ public void saveNewBaselines(List<MeasurementBaseline> baselines) {
+ for (MeasurementBaseline baseline : baselines) {
+ entityManager.merge(baseline);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ @TransactionAttribute(TransactionAttributeType.REQUIRES_NEW)
public int _calculateAutoBaselinesINSERT(long amountOfData) throws Exception {
long endTime = System.currentTimeMillis();
long startTime = endTime - amountOfData;
diff --git a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementBaselineManagerLocal.java b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementBaselineManagerLocal.java
index 7f7655a..8781fc5 100644
--- a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementBaselineManagerLocal.java
+++ b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementBaselineManagerLocal.java
@@ -24,6 +24,7 @@ import javax.ejb.Local;
import org.rhq.core.domain.auth.Subject;
import org.rhq.core.domain.measurement.MeasurementBaseline;
+import org.rhq.core.domain.measurement.MeasurementSchedule;
import org.rhq.core.domain.resource.Resource;
/**
@@ -68,10 +69,47 @@ public interface MeasurementBaselineManagerLocal {
* Inserts baselines "as appropriate" for measurements that have at least amountOfData
* @param amountOfData will use amountOfData to compute new min/max/mean for baselines as appropriate
* @return number of rows inserted
+ * @deprecated This method is pending removal after peer review. Because 1hr data is
+ * now stored in Cassandra, transaction boundaries needed to change. This method has
+ * been replaced by {@link #getSchedulesWithoutBaselines()}, {@link #calculateBaselines(java.util.List, long)},
+ * and {@link #saveNewBaselines(java.util.List)}.
* @throws Exception
*/
int _calculateAutoBaselinesINSERT(long amountOfData) throws Exception;
+ /**
+ * <strong>Note</strong> This method exists only for transaction demarcation.
+ *
+ * @return A list of schedules that do not have baselines. This list is not assumed
+ * to be an exhaustive list of schedules that lack a baseline. As such, this method
+ * will be called repeatedly during baseline calculations to get all of the necessary
+ * schedules.
+ */
+ List<MeasurementSchedule> getSchedulesWithoutBaselines();
+
+ /**
+ * Given a list of schedules, this method calculates and stores baselines using the
+ * amount of 1 hr data specified and older than the time specified.
+ * <br/><br/>
+ * <strong>Note</strong> This method exists only for transaction demarcation.
+ *
+ * @param schedules The schedules that do not yet have baselines
+ * @param olderThan Use 1 hr data prior to this time
+ * @param amountOfData The amount of data to use for calculating baselines. This value
+ * is treated as a duration. For example, a value of 259200000
+ * would be treated as 3 days.
+ */
+ void calculateBaselines(List<MeasurementSchedule> schedules, long olderThan, long amountOfData);
+
+ /**
+ * Persists the newly calculated baselines.
+ * <br/><br/>
+ * <strong>Note</strong> This method exists only for transaction demarcation.
+ *
+ * @param baselines The baselines to persist.
+ */
+ void saveNewBaselines(List<MeasurementBaseline> baselines);
+
MeasurementBaseline getBaselineIfEqual(Subject subject, int groupId, int definitionId);
/**
diff --git a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementOOBManagerBean.java b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementOOBManagerBean.java
index eecbbca..f742c7f 100644
--- a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementOOBManagerBean.java
+++ b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementOOBManagerBean.java
@@ -82,6 +82,9 @@ public class MeasurementOOBManagerBean implements MeasurementOOBManagerLocal {
@EJB
AuthorizationManagerLocal authMangager;
+ @EJB
+ MeasurementOOBManagerLocal oobManager;
+
/**
* Compute oobs from the values in the 1h measurement table that just got added.
* For the total result, this is an incremental computation. The idea is that
@@ -191,56 +194,15 @@ public class MeasurementOOBManagerBean implements MeasurementOOBManagerLocal {
}
@SuppressWarnings("unchecked")
+ @TransactionAttribute(value = TransactionAttributeType.NOT_SUPPORTED)
public void computeOOBsForLastHour(Subject subject, Iterable<AggregateNumericMetric> metrics) {
+ log.info("Computing OOBs");
int count = 0;
long startTime = System.currentTimeMillis();
try {
for (AggregateNumericMetric metric : metrics) {
try {
- List<MeasurementBaseline> baselines = entityManager.createQuery(
- "select baseline from MeasurementBaseline baseline where baseline.schedule.id = :scheduleId")
- .setParameter("scheduleId", metric.getScheduleId())
- .getResultList();
- if (baselines.isEmpty()) {
- continue;
- }
- MeasurementBaseline baseline = baselines.get(0);
- Long upperDelta = null;
- Long lowerDelta = null;
-
- if (isPastUpperBound(baseline, metric)) {
- upperDelta =
- Math.round(((metric.getMax() - baseline.getMax()) / (baseline.getMax() - baseline.getMin())) * 100);
- }
-
- if (isPastLowerBound(baseline, metric)) {
- lowerDelta =
- Math.round(((baseline.getMin() - metric.getMin()) / (baseline.getMax() - baseline.getMin())) * 100);
- }
-
- Integer oobFactor;
- if (upperDelta != null && lowerDelta == null) {
- oobFactor = upperDelta.intValue();
- } else if (upperDelta == null && lowerDelta != null) {
- oobFactor = lowerDelta.intValue();
- } else if (upperDelta != null && lowerDelta != null) {
- if (upperDelta > lowerDelta) {
- oobFactor = upperDelta.intValue();
- } else {
- oobFactor = lowerDelta.intValue();
- }
- } else { // both are null
- oobFactor = null;
- }
-
- if (oobFactor != null) {
- MeasurementOOB oob = new MeasurementOOB();
- oob.setScheduleId(metric.getScheduleId());
- oob.setTimestamp(metric.getTimestamp());
- oob.setOobFactor(oobFactor);
- entityManager.merge(oob);
- ++count;
- }
+ count += oobManager.calculateOOB(metric);
} catch (Exception e) {
log.error("An error occurred while calculating OOBs for " + metric, e);
throw new RuntimeException(e);
@@ -248,10 +210,66 @@ public class MeasurementOOBManagerBean implements MeasurementOOBManagerLocal {
}
} finally {
long endTime = System.currentTimeMillis();
+ if (log.isInfoEnabled()) {
+ log.info("Finished calculating " + count + " OOBs in " + (endTime - startTime) + " ms");
+ }
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ @TransactionAttribute(value = TransactionAttributeType.REQUIRES_NEW)
+ public int calculateOOB(AggregateNumericMetric metric) {
+ List<MeasurementBaseline> baselines = entityManager.createQuery(
+ "select baseline from MeasurementBaseline baseline where baseline.schedule.id = :scheduleId")
+ .setParameter("scheduleId", metric.getScheduleId())
+ .getResultList();
+ if (baselines.isEmpty()) {
+ return 0;
+ }
+ MeasurementBaseline baseline = baselines.get(0);
+ Long upperDelta = null;
+ Long lowerDelta = null;
+
+ if (isPastUpperBound(baseline, metric)) {
+ upperDelta =
+ Math.round(((metric.getMax() - baseline.getMax()) / (baseline.getMax() - baseline.getMin())) * 100);
+ }
+
+ if (isPastLowerBound(baseline, metric)) {
+ lowerDelta =
+ Math.round(((baseline.getMin() - metric.getMin()) / (baseline.getMax() - baseline.getMin())) * 100);
+ }
+
+ Integer oobFactor;
+ if (upperDelta != null && lowerDelta == null) {
+ oobFactor = upperDelta.intValue();
+ } else if (upperDelta == null && lowerDelta != null) {
+ oobFactor = lowerDelta.intValue();
+ } else if (upperDelta != null && lowerDelta != null) {
+ if (upperDelta > lowerDelta) {
+ oobFactor = upperDelta.intValue();
+ } else {
+ oobFactor = lowerDelta.intValue();
+ }
+ } else { // both are null
+ oobFactor = null;
+ }
+
+ if (oobFactor != null) {
+ MeasurementOOB oob = new MeasurementOOB();
+ oob.setScheduleId(metric.getScheduleId());
+ oob.setTimestamp(metric.getTimestamp());
+ oob.setOobFactor(oobFactor);
+
if (log.isDebugEnabled()) {
- log.debug("Finished calculating " + count + " OOBs in " + (endTime - startTime) + " ms");
+ log.debug("Generated OOB " + oob + " for 1 hr metric " + metric + " with baseline " + baseline);
}
+
+ entityManager.merge(oob);
+ return 1;
}
+
+ return 0;
}
private boolean isPastUpperBound(MeasurementBaseline baseline, AggregateNumericMetric metric) {
diff --git a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementOOBManagerLocal.java b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementOOBManagerLocal.java
index b452a28..853c5af 100644
--- a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementOOBManagerLocal.java
+++ b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/measurement/MeasurementOOBManagerLocal.java
@@ -45,9 +45,30 @@ public interface MeasurementOOBManagerLocal {
*/
void computeOOBsFromHourBeginingAt(Subject subject, long begin);
+ /**
+ * Computes OOBs using the provided 1 hr data which should be the most recent 1 hr
+ * aggregates. These metrics are provided as an argument as opposed to querying for
+ * them because we already have the 1 hr aggregates load in memory when metrics
+ * aggregation runs prior to calculating OOBs.
+ *
+ * @param subject
+ * @param metrics The most recent 1 hr aggregates
+ */
void computeOOBsForLastHour(Subject subject, Iterable<AggregateNumericMetric> metrics);
/**
+ * Determines and calculates an OOB if necessary, If an OOB is generated, this method
+ * saves it to the database.
+ * <br/><br/>
+ * <strong>Note</strong> This method exists only for transaction demarcation.
+ *
+ * @param metric The 1 hr metric that is used to determine whether or not an OOB should
+ * be generated
+ * @return 1 if an OOB is generated, 0 otherwise
+ */
+ int calculateOOB(AggregateNumericMetric metric);
+
+ /**
* Return OOB Composites that contain all information about the OOBs in a given time as aggregates.
* @param subject The caller
* @param metricNameFilter
diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsBaselineCalculator.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsBaselineCalculator.java
index 235e157..ef8eb44 100644
--- a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsBaselineCalculator.java
+++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/MetricsBaselineCalculator.java
@@ -39,10 +39,6 @@ public class MetricsBaselineCalculator {
private MetricsDAO metricsDAO;
-// public MetricsBaselineCalculator(Session session) {
-// this.metricsDAO = new MetricsDAO(session);
-// }
-
public MetricsBaselineCalculator(MetricsDAO metricsDAO) {
this.metricsDAO = metricsDAO;
}
@@ -108,6 +104,12 @@ public class MetricsBaselineCalculator {
return baseline;
}
- return null;
+ MeasurementBaseline baseline = new MeasurementBaseline();
+ baseline.setMax(Double.NaN);
+ baseline.setMin(Double.NaN);
+ baseline.setMean(Double.NaN);
+ baseline.setSchedule(schedule);
+
+ return baseline;
}
}
10 years, 12 months
[rhq] 2 commits - modules/core modules/enterprise
by Jay Shaughnessy
modules/core/domain/src/main/java/org/rhq/core/domain/criteria/Criteria.java | 47 +++++++++-
modules/core/domain/src/main/java/org/rhq/core/domain/criteria/JPADriftChangeSetCriteria.java | 9 -
modules/core/domain/src/main/java/org/rhq/core/domain/criteria/JPADriftCriteria.java | 9 -
modules/core/domain/src/main/java/org/rhq/core/domain/criteria/MeasurementDataTraitCriteria.java | 12 +-
modules/core/domain/src/main/java/org/rhq/core/domain/util/PageControl.java | 34 ++++++-
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/core/jaas/DelegatingLoginModule.java | 10 +-
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/util/CriteriaQuery.java | 15 ---
modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/util/CriteriaQueryGenerator.java | 10 ++
modules/enterprise/server/jar/src/test/java/org/rhq/enterprise/server/util/CriteriaQueryTest.java | 22 ++++
9 files changed, 126 insertions(+), 42 deletions(-)
New commits:
commit 471b8a20f984f79d448acf58c01ec8c38cee8e67
Author: Jay Shaughnessy <jshaughn(a)redhat.com>
Date: Fri May 24 19:45:41 2013 -0400
Bug 966665
- Criteria queries with paging and sorting on non unique fields may return same item on different pages
All paged criteria queries now apply ID as the least significant
ordering field (see exceptions below). Since ID is unique, this ensures
that queries for pages always return the same ordering of rows.
ID will not be automatically appended as an ordering field if:
- the paging is unlimited (i.e. no pagesize, all rows returned)
- the Criteria class does not support an ID sort
- the caller has set the Criteria class instance to not support the ID sort
- ID is already specified as an ordering field
- the max number of ordering fields has already been set (3)
Note that CriteriaQuery has had its support for implicit ID sorting removed,
as the new mechanism, which moves support to the CriteriaQueryGenerator,
replaces it.
diff --git a/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/Criteria.java b/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/Criteria.java
index 2dde3a4..d8552de 100644
--- a/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/Criteria.java
+++ b/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/Criteria.java
@@ -109,6 +109,7 @@ public abstract class Criteria implements Serializable, BaseCriteria {
private List<Permission> requiredPermissions;
private boolean strict;
private Restriction restriction = null;
+ private boolean supportsAddSortId = true;
protected Map<String, String> filterOverrides;
protected Map<String, String> sortOverrides;
@@ -173,16 +174,54 @@ public abstract class Criteria implements Serializable, BaseCriteria {
}
public void addSortId(PageOrdering sortId) {
- addSortField("id");
- this.sortId = sortId;
+ if (isSupportsAddSortId()) {
+ addSortField("id");
+ this.sortId = sortId;
+
+ } else {
+ throw new UnsupportedOperationException("ID sort is not supported by " + this.getClass().getSimpleName());
+ }
}
public void addFilterId(Integer filterId) {
- this.filterId = filterId;
+ if (isSupportsAddFilterId()) {
+ this.filterId = filterId;
+
+ } else {
+ throw new UnsupportedOperationException("ID filter is not supported by " + this.getClass().getSimpleName());
+ }
}
public void addFilterIds(Integer... filterIds) {
- this.filterIds = CriteriaUtils.getListIgnoringNulls(filterIds);
+ if (isSupportsAddFilterIds()) {
+ this.filterIds = CriteriaUtils.getListIgnoringNulls(filterIds);
+
+ } else {
+ throw new UnsupportedOperationException("IDS filter is not supported by " + this.getClass().getSimpleName());
+ }
+ }
+
+ /**
+ * By default all Criteria support sort on ID. And this sort is applied implicitly to criteria
+ * queries involving paging, to ensure consistent ordering of query results. If for some unlikely reason
+ * the caller needs to disable the implicit ID sort then call this, setting the value to false.
+ *
+ * @param supportsAddSortId
+ */
+ public void setSupportsAddSortId(boolean supportsAddSortId) {
+ this.supportsAddSortId = supportsAddSortId;
+ }
+
+ public boolean isSupportsAddSortId() {
+ return supportsAddSortId;
+ }
+
+ public boolean isSupportsAddFilterId() {
+ return true;
+ }
+
+ public boolean isSupportsAddFilterIds() {
+ return true;
}
protected void addSortField(String fieldName) {
diff --git a/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/JPADriftChangeSetCriteria.java b/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/JPADriftChangeSetCriteria.java
index af888ea..2b03228 100644
--- a/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/JPADriftChangeSetCriteria.java
+++ b/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/JPADriftChangeSetCriteria.java
@@ -292,13 +292,12 @@ public class JPADriftChangeSetCriteria extends Criteria implements DriftChangeSe
}
@Override
- public void addSortId(PageOrdering pageOrdering) {
- throw new UnsupportedOperationException("The sort is not supported by DriftChangeSetCriteria.");
+ public boolean isSupportsAddSortId() {
+ return false;
}
@Override
- public void addFilterIds(Integer... filterIds) {
- throw new UnsupportedOperationException("The filter is not supported by DriftChangeSetCriteria.");
+ public boolean isSupportsAddFilterIds() {
+ return false;
}
-
}
diff --git a/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/JPADriftCriteria.java b/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/JPADriftCriteria.java
index 56f0ddf..b13ca90 100644
--- a/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/JPADriftCriteria.java
+++ b/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/JPADriftCriteria.java
@@ -245,13 +245,12 @@ public class JPADriftCriteria extends Criteria implements DriftCriteria {
}
@Override
- public void addFilterIds(Integer... filterIds) {
- throw new UnsupportedOperationException("The filter is not supported by DriftCriteria.");
+ public boolean isSupportsAddSortId() {
+ return false;
}
@Override
- public void addSortId(PageOrdering pageOrdering) {
- throw new UnsupportedOperationException("The sort is not supported by DriftCriteria.");
+ public boolean isSupportsAddFilterIds() {
+ return false;
}
-
}
diff --git a/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/MeasurementDataTraitCriteria.java b/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/MeasurementDataTraitCriteria.java
index a27e58d..3a2a2c1 100644
--- a/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/MeasurementDataTraitCriteria.java
+++ b/modules/core/domain/src/main/java/org/rhq/core/domain/criteria/MeasurementDataTraitCriteria.java
@@ -135,17 +135,17 @@ public class MeasurementDataTraitCriteria extends Criteria {
}
@Override
- public void addSortId(PageOrdering sortId) {
- throw new UnsupportedOperationException("The id column is not defined on this entity.");
+ public boolean isSupportsAddSortId() {
+ return false;
}
@Override
- public void addFilterId(Integer filterId) {
- throw new UnsupportedOperationException("The id column is not defined on this entity.");
+ public boolean isSupportsAddFilterId() {
+ return false;
}
@Override
- public void addFilterIds(Integer... filterIds) {
- throw new UnsupportedOperationException("The id column is not defined on this entity.");
+ public boolean isSupportsAddFilterIds() {
+ return false;
}
}
diff --git a/modules/core/domain/src/main/java/org/rhq/core/domain/util/PageControl.java b/modules/core/domain/src/main/java/org/rhq/core/domain/util/PageControl.java
index f87a814..888c821 100644
--- a/modules/core/domain/src/main/java/org/rhq/core/domain/util/PageControl.java
+++ b/modules/core/domain/src/main/java/org/rhq/core/domain/util/PageControl.java
@@ -82,14 +82,22 @@ public class PageControl implements Serializable, Cloneable {
return new PageControl(0, 1);
}
+ /**
+ * Equivalent to initDefaultOrderingField(defaultField, PageOrdering.ASC).
+ *
+ * @param defaultField
+ * @see #initDefaultOrderingField(String, PageOrdering)
+ */
public void initDefaultOrderingField(String defaultField) {
initDefaultOrderingField(defaultField, PageOrdering.ASC);
}
- public void addDefaultOrderingField(String defaultField) {
- addDefaultOrderingField(defaultField, PageOrdering.ASC);
- }
-
+ /**
+ * Sets initial sort. If sorting is already defined this call will have no effect.
+ *
+ * @param defaultField
+ * @param defaultPageOrdering
+ */
public void initDefaultOrderingField(String defaultField, PageOrdering defaultPageOrdering) {
if (orderingFields.size() > 0) {
return;
@@ -98,6 +106,24 @@ public class PageControl implements Serializable, Cloneable {
addDefaultOrderingField(defaultField, defaultPageOrdering);
}
+ /**
+ * Equivalent to addDefaultOrderingField(defaultField, PageOrdering.ASC).
+ *
+ * @param defaultField
+ * @see #addDefaultOrderingField(String, PageOrdering)
+ */
+ public void addDefaultOrderingField(String defaultField) {
+ addDefaultOrderingField(defaultField, PageOrdering.ASC);
+ }
+
+ /**
+ * Add a default ordering field. If the maximum number of sort fields (currently 3) are already
+ * defined this call will have no effect. If the field is already a sort field this call will have no
+ * effect. Otherwise, the ordering field will be appended to the existing ordering fields.
+ *
+ * @param defaultField
+ * @param defaultPageOrdering
+ */
public void addDefaultOrderingField(String defaultField, PageOrdering defaultPageOrdering) {
if (orderingFields.size() >= MAX_ORDERING_FIELD_COUNT) {
return; // only need to add defaults if there are less than 3 sort orders
diff --git a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/util/CriteriaQuery.java b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/util/CriteriaQuery.java
index 2d40388..a2bbfee 100644
--- a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/util/CriteriaQuery.java
+++ b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/util/CriteriaQuery.java
@@ -27,7 +27,6 @@ import java.util.NoSuchElementException;
import org.rhq.core.domain.criteria.BaseCriteria;
import org.rhq.core.domain.util.PageControl;
import org.rhq.core.domain.util.PageList;
-import org.rhq.core.domain.util.PageOrdering;
/** This class provides a way to make PageList results easily iterable with 'for each','while',etc. loops
* and importantly automatically handles iteration through all PageControl results. This
@@ -62,17 +61,6 @@ public class CriteriaQuery<T, C extends BaseCriteria> implements Iterable<T> {
public CriteriaQuery(C criteria, CriteriaQueryExecutor<T, C> queryExecutor) {
this.criteria = criteria;
this.queryExecutor = queryExecutor;
-
- // make sure we have at least a default sort, otherwise chunking doesn't work
- PageControl pageControlOverrides = this.criteria.getPageControlOverrides();
- if (null != pageControlOverrides) {
- if (pageControlOverrides.getOrderingFields().isEmpty()) {
- pageControlOverrides.addDefaultOrderingField("id");
- }
-
- } else if (this.criteria.getOrderingFieldNames().isEmpty()) {
- this.criteria.addSortId(PageOrdering.ASC);
- }
}
/** Returns iterator for a single page of results as defined by
@@ -110,7 +98,8 @@ public class CriteriaQuery<T, C extends BaseCriteria> implements Iterable<T> {
public QueryResultsIterator(PageList<T> firstPage) {
currentPage = firstPage;
iterator = currentPage.iterator();
- count = firstPage == null || firstPage.getPageControl() == null ? 0 : firstPage.getPageControl().getStartRow();
+ count = firstPage == null || firstPage.getPageControl() == null ? 0 : firstPage.getPageControl()
+ .getStartRow();
}
@Override
diff --git a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/util/CriteriaQueryGenerator.java b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/util/CriteriaQueryGenerator.java
index d1d2f73..574fe73 100644
--- a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/util/CriteriaQueryGenerator.java
+++ b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/util/CriteriaQueryGenerator.java
@@ -951,6 +951,7 @@ public final class CriteriaQueryGenerator {
} else {
pc = new PageControl(criteria.getPageNumber(), criteria.getPageSize());
}
+
for (String fieldName : criteria.getOrderingFieldNames()) {
for (Field sortField : getFields(criteria, Criteria.Type.SORT)) {
if (sortField.getName().equals(fieldName) == false) {
@@ -969,6 +970,15 @@ public final class CriteriaQueryGenerator {
}
}
}
+
+ // Unless paging is unlimited or it's not supported, add a sort on ID. This ensures that when paging
+ // we always have a consistent ordering. In other words, if the data set is unchanged between pages, there
+ // will no overlap/repetition of rows. Note that this applies even if other sort fields have been
+ // set, because they may still not have unique values. See https://bugzilla.redhat.com/show_bug.cgi?id=966665.
+ if (!pc.isUnlimited() && criteria.isSupportsAddSortId()) {
+ pc.addDefaultOrderingField("id");
+ }
+
return pc;
}
diff --git a/modules/enterprise/server/jar/src/test/java/org/rhq/enterprise/server/util/CriteriaQueryTest.java b/modules/enterprise/server/jar/src/test/java/org/rhq/enterprise/server/util/CriteriaQueryTest.java
index 2e1c897..b1f5272 100644
--- a/modules/enterprise/server/jar/src/test/java/org/rhq/enterprise/server/util/CriteriaQueryTest.java
+++ b/modules/enterprise/server/jar/src/test/java/org/rhq/enterprise/server/util/CriteriaQueryTest.java
@@ -24,11 +24,15 @@ import static org.testng.Assert.assertEquals;
import java.util.ArrayList;
import java.util.List;
+import org.testng.AssertJUnit;
import org.testng.annotations.Test;
import org.rhq.core.domain.criteria.Criteria;
+import org.rhq.core.domain.criteria.JPADriftCriteria;
+import org.rhq.core.domain.criteria.ResourceCriteria;
import org.rhq.core.domain.util.PageControl;
import org.rhq.core.domain.util.PageList;
+import org.rhq.core.domain.util.PageOrdering;
public class CriteriaQueryTest {
@@ -229,4 +233,22 @@ public class CriteriaQueryTest {
assert e.getMessage().contains("NonUniqueResultException");
}
}
+
+ @Test
+ public void testAddSort() {
+ try {
+ JPADriftCriteria c = new JPADriftCriteria();
+ c.addSortId(PageOrdering.ASC);
+ AssertJUnit.fail("Should have thrown exception");
+ } catch (UnsupportedOperationException e) {
+ //expected
+ }
+
+ ResourceCriteria c = new ResourceCriteria();
+ c.addSortName(PageOrdering.ASC);
+
+ assertEquals(2, CriteriaQueryGenerator.getPageControl(c).getOrderingFields().size());
+ assertEquals("name", CriteriaQueryGenerator.getPageControl(c).getOrderingFields().get(0).getField());
+ assertEquals("id", CriteriaQueryGenerator.getPageControl(c).getOrderingFields().get(1).getField());
+ }
}
commit 6613ba3db6f3c57c2d2ecac4823e368269a07152
Author: Jay Shaughnessy <jshaughn(a)redhat.com>
Date: Fri May 24 15:40:51 2013 -0400
Use only apache logger and not the jboss logger from the base class.
diff --git a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/core/jaas/DelegatingLoginModule.java b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/core/jaas/DelegatingLoginModule.java
index d89c04b..b83a037 100644
--- a/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/core/jaas/DelegatingLoginModule.java
+++ b/modules/enterprise/server/jar/src/main/java/org/rhq/enterprise/server/core/jaas/DelegatingLoginModule.java
@@ -84,7 +84,7 @@ public class DelegatingLoginModule extends UsernamePasswordLoginModule {
public void initialize(Subject subject, CallbackHandler callbackHandler, Map<String, ?> sharedState,
Map<String, ?> options) {
- debugEnabled = log.isDebugEnabled();
+ debugEnabled = LOG.isDebugEnabled();
super.initialize(subject, callbackHandler, sharedState, options);
@@ -101,14 +101,14 @@ public class DelegatingLoginModule extends UsernamePasswordLoginModule {
}
if (debugEnabled) {
- log.debug("Delegating to " + delegateTo + " with roles " + additionalRoles);
+ LOG.debug("Delegating to " + delegateTo + " with roles " + additionalRoles);
}
// Now create the context for later use
try {
loginContext = new LoginContext(delegateTo, new DelegateCallbackHandler());
} catch (LoginException e) {
- log.warn("Initialize failed : " + e.getMessage());
+ LOG.warn("Initialize failed : " + e.getMessage());
}
}
@@ -123,7 +123,7 @@ public class DelegatingLoginModule extends UsernamePasswordLoginModule {
@Override
public boolean login() throws LoginException {
try {
- // Get the username / password the user entred and save if for later use
+ // Get the username / password the user entered and save if for later use
usernamePassword = super.getUsernameAndPassword();
// Try to log in via the delegate
@@ -141,7 +141,7 @@ public class DelegatingLoginModule extends UsernamePasswordLoginModule {
loginOk = true;
if (debugEnabled) {
- log.debug("Login ok for " + usernamePassword[0]);
+ LOG.debug("Login ok for " + usernamePassword[0]);
}
return true;
10 years, 12 months