modules/core/domain/src/main/java/org/rhq/core/domain/measurement/MeasurementDataNumeric1D.java
| 4
modules/core/domain/src/main/java/org/rhq/core/domain/measurement/MeasurementDataNumeric1H.java
| 4
modules/core/domain/src/main/java/org/rhq/core/domain/measurement/MeasurementDataNumeric6H.java
| 4
modules/core/domain/src/main/java/org/rhq/core/domain/measurement/MeasurementDataNumericAggregateInterface.java
| 45 +++
modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/DataMigrator.java
| 150 +++++-----
5 files changed, 141 insertions(+), 66 deletions(-)
New commits:
commit 4c932c2c765aef27d333852f469286dbb50825aa
Author: Stefan Negrea <snegrea(a)redhat.com>
Date: Thu Dec 27 17:49:40 2012 -0600
Update data migrator to use interfaces to compact the code and also make the process
configurable.
diff --git
a/modules/core/domain/src/main/java/org/rhq/core/domain/measurement/MeasurementDataNumeric1D.java
b/modules/core/domain/src/main/java/org/rhq/core/domain/measurement/MeasurementDataNumeric1D.java
index 35c958b..e36f136 100644
---
a/modules/core/domain/src/main/java/org/rhq/core/domain/measurement/MeasurementDataNumeric1D.java
+++
b/modules/core/domain/src/main/java/org/rhq/core/domain/measurement/MeasurementDataNumeric1D.java
@@ -43,7 +43,9 @@ import javax.persistence.Table;
@NamedQuery(name = MeasurementDataNumeric1D.QUERY_FIND_ALL, query = "SELECT m
From MeasurementDataNumeric1D m"),
@NamedQuery(name = MeasurementDataNumeric1D.QUERY_DELETE_ALL, query = "DELETE
FROM MeasurementDataNumeric1D m ") })
@Table(name = "RHQ_MEASUREMENT_DATA_NUM_1D")
-public class MeasurementDataNumeric1D extends MeasurementData implements Serializable {
+public class MeasurementDataNumeric1D extends MeasurementData implements Serializable,
+ MeasurementDataNumericAggregateInterface {
+
private static final long serialVersionUID = 1L;
public static final String GET_NUM_AGGREGATE =
"MeasurementDataNumeric1D.getNumAggregate";
diff --git
a/modules/core/domain/src/main/java/org/rhq/core/domain/measurement/MeasurementDataNumeric1H.java
b/modules/core/domain/src/main/java/org/rhq/core/domain/measurement/MeasurementDataNumeric1H.java
index 6766293..ad43580 100644
---
a/modules/core/domain/src/main/java/org/rhq/core/domain/measurement/MeasurementDataNumeric1H.java
+++
b/modules/core/domain/src/main/java/org/rhq/core/domain/measurement/MeasurementDataNumeric1H.java
@@ -41,7 +41,9 @@ import javax.persistence.Table;
@NamedQuery(name = MeasurementDataNumeric1H.QUERY_FIND_ALL, query = "SELECT m
From MeasurementDataNumeric1H m"),
@NamedQuery(name = MeasurementDataNumeric1H.QUERY_DELETE_ALL, query = "DELETE
FROM MeasurementDataNumeric1D m ") })
@Table(name = "RHQ_MEASUREMENT_DATA_NUM_1H")
-public class MeasurementDataNumeric1H extends MeasurementData implements Serializable {
+public class MeasurementDataNumeric1H extends MeasurementData implements Serializable,
+ MeasurementDataNumericAggregateInterface {
+
private static final long serialVersionUID = 1L;
public static final String GET_MAX_TIMESTAMP =
"MeasurementDataNumeric1H.getMaxTimestamp";
diff --git
a/modules/core/domain/src/main/java/org/rhq/core/domain/measurement/MeasurementDataNumeric6H.java
b/modules/core/domain/src/main/java/org/rhq/core/domain/measurement/MeasurementDataNumeric6H.java
index 75f3aed..3cb718c 100644
---
a/modules/core/domain/src/main/java/org/rhq/core/domain/measurement/MeasurementDataNumeric6H.java
+++
b/modules/core/domain/src/main/java/org/rhq/core/domain/measurement/MeasurementDataNumeric6H.java
@@ -43,7 +43,9 @@ import javax.persistence.Table;
@NamedQuery(name = MeasurementDataNumeric6H.QUERY_FIND_ALL, query = "SELECT m
From MeasurementDataNumeric6H m"),
@NamedQuery(name = MeasurementDataNumeric6H.QUERY_DELETE_ALL, query = "DELETE
FROM MeasurementDataNumeric6H m ") })
@Table(name = "RHQ_MEASUREMENT_DATA_NUM_6H")
-public class MeasurementDataNumeric6H extends MeasurementData implements Serializable {
+public class MeasurementDataNumeric6H extends MeasurementData implements Serializable,
+ MeasurementDataNumericAggregateInterface {
+
private static final long serialVersionUID = 1L;
public static final String GET_NUM_AGGREGATE =
"MeasurementDataNumeric6H.getNumAggregate";
diff --git
a/modules/core/domain/src/main/java/org/rhq/core/domain/measurement/MeasurementDataNumericAggregateInterface.java
b/modules/core/domain/src/main/java/org/rhq/core/domain/measurement/MeasurementDataNumericAggregateInterface.java
new file mode 100644
index 0000000..d4a0484
--- /dev/null
+++
b/modules/core/domain/src/main/java/org/rhq/core/domain/measurement/MeasurementDataNumericAggregateInterface.java
@@ -0,0 +1,45 @@
+/*
+ *
+ * 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.core.domain.measurement;
+
+/**
+ * Transition interface for data migration between SQL and Cassandra
+ *
+ * @author Stefan Negrea
+ */
+@Deprecated
+public interface MeasurementDataNumericAggregateInterface {
+
+ public int getScheduleId();
+
+ public long getTimestamp();
+
+ public Object getValue();
+
+ public Double getMin();
+
+ public Double getMax();
+
+}
diff --git
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/DataMigrator.java
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/DataMigrator.java
index c783f93..b5c1f6c 100644
---
a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/DataMigrator.java
+++
b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/DataMigrator.java
@@ -34,6 +34,7 @@ import com.datastax.driver.core.exceptions.NoHostAvailableException;
import org.rhq.core.domain.measurement.MeasurementDataNumeric1D;
import org.rhq.core.domain.measurement.MeasurementDataNumeric1H;
import org.rhq.core.domain.measurement.MeasurementDataNumeric6H;
+import org.rhq.core.domain.measurement.MeasurementDataNumericAggregateInterface;
/**
* @author Stefan Negrea
@@ -41,76 +42,120 @@ import org.rhq.core.domain.measurement.MeasurementDataNumeric6H;
*/
public class DataMigrator {
+ private static final int MAX_RECORDS_TO_MIGRATE = 1000;
+
private final EntityManager entityManager;
+ private final Session session;
- private Session session;
+ private boolean deleteDataImmediatelyAfterMigration;
+ private boolean deleteAllDataAtTheEndOfMigration;
+ private boolean runRawDataMigration;
+ private boolean run1HAggregateDataMigration;
+ private boolean run6HAggregateDataMigration;
+ private boolean run1DAggregateDataMigration;
public DataMigrator(EntityManager entityManager, Session session) {
this.entityManager = entityManager;
this.session = session;
+
+ this.deleteDataImmediatelyAfterMigration = true;
+ this.deleteAllDataAtTheEndOfMigration = false;
+ this.runRawDataMigration = true;
+ this.run1HAggregateDataMigration = true;
+ this.run6HAggregateDataMigration = true;
+ this.run1DAggregateDataMigration = true;
}
- public void migrateData() {
+ public void run1HAggregateDataMigration(boolean value) {
+ this.run1HAggregateDataMigration = value;
+ }
- migrateRawData();
- migrateOneHourData();
- migrateSixHourData();
- migrateTwentyFourHourData();
+ public void run6HAggregateDataMigration(boolean value) {
+ this.run6HAggregateDataMigration = value;
+ }
- clearAllData();
+ public void run1DAggregateDataMigration(boolean value) {
+ this.run1DAggregateDataMigration = value;
}
- private void migrateRawData() {
+ public void deleteDataImmediatelyAfterMigration(boolean value) {
+ this.deleteDataImmediatelyAfterMigration = value;
+ this.deleteAllDataAtTheEndOfMigration = !value;
+ }
+ public void deleteAllDataAtTheEndOfMigration(boolean value) {
+ this.deleteAllDataAtTheEndOfMigration = value;
+ this.deleteDataImmediatelyAfterMigration = !value;
}
- @SuppressWarnings("unchecked")
- private void migrateOneHourData() {
- Query q =
this.entityManager.createNamedQuery(MeasurementDataNumeric1H.QUERY_FIND_ALL);
- List<MeasurementDataNumeric1H> existingData = q.getResultList();
+ public void migrateData() {
+ if (runRawDataMigration) {
+ migrateRawData();
+ }
- try {
- PreparedStatement statement =
createPreparedStatement(MetricsTable.ONE_HOUR);
+ if (run1HAggregateDataMigration) {
+ migrateAggregatedMetricsData(MeasurementDataNumeric1H.QUERY_FIND_ALL,
MetricsTable.ONE_HOUR);
+ }
- for (MeasurementDataNumeric1H measurement : existingData) {
- insertData(statement, measurement.getScheduleId(), measurement.getMin(),
measurement.getMax(),
- Double.parseDouble(measurement.getValue().toString()),
measurement.getTimestamp());
- }
- } catch (NoHostAvailableException e) {
- throw new CQLException(e);
+ if (run6HAggregateDataMigration) {
+ migrateAggregatedMetricsData(MeasurementDataNumeric6H.QUERY_FIND_ALL,
MetricsTable.SIX_HOUR);
+ }
+
+ if (run1DAggregateDataMigration) {
+ migrateAggregatedMetricsData(MeasurementDataNumeric1D.QUERY_FIND_ALL,
MetricsTable.TWENTY_FOUR_HOUR);
+ }
+
+ if (deleteAllDataAtTheEndOfMigration) {
+ this.clearAllData();
}
}
+ private void migrateRawData() {
+ //possibly need to add raw SQL code here because data is split among several
tables
+ }
+
@SuppressWarnings("unchecked")
- private void migrateSixHourData() {
- Query q =
this.entityManager.createNamedQuery(MeasurementDataNumeric6H.QUERY_FIND_ALL);
- List<MeasurementDataNumeric6H> existingData = q.getResultList();
+ private void migrateAggregatedMetricsData(String query, MetricsTable metricsTable) {
+ List<MeasurementDataNumericAggregateInterface> existingData = null;
- try {
- PreparedStatement statement =
createPreparedStatement(MetricsTable.SIX_HOUR);
+ while (true) {
+ Query q = this.entityManager.createNamedQuery(query);
+ q.setMaxResults(MAX_RECORDS_TO_MIGRATE);
+ existingData = (List<MeasurementDataNumericAggregateInterface>)
q.getResultList();
- for (MeasurementDataNumeric6H measurement : existingData) {
- insertData(statement, measurement.getScheduleId(), measurement.getMin(),
measurement.getMax(),
- Double.parseDouble(measurement.getValue().toString()),
measurement.getTimestamp());
+ if (existingData.size() == 0) {
+ break;
}
- } catch (NoHostAvailableException e) {
- throw new CQLException(e);
- }
- }
- private void migrateTwentyFourHourData() {
- Query q =
this.entityManager.createNamedQuery(MeasurementDataNumeric1D.QUERY_FIND_ALL);
- List<MeasurementDataNumeric1D> existingData = q.getResultList();
+ try {
+ String cql = "INSERT INTO " + metricsTable
+ + " (schedule_id, time, type, value) VALUES (?, ?, ?, ?) USING
TTL " + metricsTable.getTTL();
+ PreparedStatement statement = session.prepare(cql);
+
+ for (MeasurementDataNumericAggregateInterface measurement : existingData)
{
- try {
- PreparedStatement statement =
createPreparedStatement(MetricsTable.TWENTY_FOUR_HOUR);
+ BoundStatement boundStatement =
statement.bind(measurement.getScheduleId(),
+ new Date(measurement.getTimestamp()),
AggregateType.MIN.ordinal(), measurement.getMin());
+ session.execute(boundStatement);
- for (MeasurementDataNumeric1D measurement : existingData) {
- insertData(statement, measurement.getScheduleId(), measurement.getMin(),
measurement.getMax(),
- Double.parseDouble(measurement.getValue().toString()),
measurement.getTimestamp());
+ boundStatement = statement.bind(measurement.getScheduleId(), new
Date(measurement.getTimestamp()),
+ AggregateType.MAX.ordinal(), measurement.getMax());
+ session.execute(boundStatement);
+
+ boundStatement = statement.bind(measurement.getScheduleId(), new
Date(measurement.getTimestamp()),
+ AggregateType.AVG.ordinal(),
Double.parseDouble(measurement.getValue().toString()));
+ session.execute(boundStatement);
+ }
+ } catch (NoHostAvailableException e) {
+ throw new CQLException(e);
+ }
+
+ if (this.deleteDataImmediatelyAfterMigration) {
+ for (Object entity : existingData) {
+ this.entityManager.remove(entity);
+ }
+ this.entityManager.flush();
}
- } catch (NoHostAvailableException e) {
- throw new CQLException(e);
}
}
@@ -124,25 +169,4 @@ public class DataMigrator {
q =
this.entityManager.createNamedQuery(MeasurementDataNumeric1D.QUERY_DELETE_ALL);
q.executeUpdate();
}
-
- private PreparedStatement createPreparedStatement(MetricsTable metricsTable) throws
NoHostAvailableException {
- String cql = "INSERT INTO " + metricsTable + " (schedule_id, time,
type, value) VALUES (?, ?, ?, ?) USING TTL "
- + metricsTable.getTTL();
- PreparedStatement statement = session.prepare(cql);
- return statement;
- }
-
- private void insertData(PreparedStatement statement, int scheduleId, double min,
double max, double average,
- long timestamp)
- throws NoHostAvailableException {
- BoundStatement boundStatement = statement.bind(scheduleId, new Date(timestamp),
AggregateType.MIN.ordinal(),
- min);
- session.execute(boundStatement);
-
- boundStatement = statement.bind(scheduleId, new Date(timestamp),
AggregateType.MAX.ordinal(), max);
- session.execute(boundStatement);
-
- boundStatement = statement.bind(scheduleId, new Date(timestamp),
AggregateType.AVG.ordinal(), average);
- session.execute(boundStatement);
- }
}