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@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); - } }