[rhq] Branch 'feature/cassandra-backend' - 5 commits - .classpath modules/enterprise modules/plugins

.classpath | 1 modules/enterprise/server/server-metrics/pom.xml | 4 modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/DataMigrator.java | 705 --------- modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/DataMigratorRunner.java | 644 -------- modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/DataMigrator.java | 730 ++++++++++ modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/DataMigratorRunner.java | 668 +++++++++ modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/ExistingDataBulkExportSource.java | 124 + modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/ExistingDataJPABulkExportSource.java | 58 modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/ExistingDataJPASource.java | 64 modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/ExistingDataSource.java | 34 modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/ExistingPostgresDataBulkExportSource.java | 68 modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/DataSourceTest.java | 169 ++ modules/plugins/cassandra/pom.xml | 56 modules/plugins/cassandra/src/main/java/org/rhq/plugins/cassandra/CassandraNodeComponent.java | 74 - modules/plugins/cassandra/src/main/java/org/rhq/plugins/cassandra/CassandraNodeDiscoveryComponent.java | 86 + modules/plugins/cassandra/src/main/resources/META-INF/rhq-plugin.xml | 11 16 files changed, 2047 insertions(+), 1449 deletions(-) New commits: commit 4703c030ca9d63510824bf0544aa855361049633 Author: Stefan Negrea <snegrea(a)redhat.com&gt; Date: Tue Apr 30 17:15:30 2013 -0500 Add configurable authenticator property to the plugin. The authenticator is detected on discovery and then used for CQL connections. diff --git a/modules/plugins/cassandra/src/main/java/org/rhq/plugins/cassandra/CassandraNodeComponent.java b/modules/plugins/cassandra/src/main/java/org/rhq/plugins/cassandra/CassandraNodeComponent.java index e16fd88..0dd282e 100644 --- a/modules/plugins/cassandra/src/main/java/org/rhq/plugins/cassandra/CassandraNodeComponent.java +++ b/modules/plugins/cassandra/src/main/java/org/rhq/plugins/cassandra/CassandraNodeComponent.java @@ -30,7 +30,7 @@ import static org.rhq.core.system.OperatingSystemType.WINDOWS; import java.io.File; import com.datastax.driver.core.Cluster; -import com.datastax.driver.core.ProtocolOptions.Compression; +import com.datastax.driver.core.Cluster.Builder; import com.datastax.driver.core.Session; import com.datastax.driver.core.SimpleAuthInfoProvider; @@ -73,7 +73,9 @@ public class CassandraNodeComponent extends JMXServerComponent<ResourceComponent String host = context.getPluginConfiguration().getSimpleValue("host", "localhost"); String clusterName = context.getPluginConfiguration().getSimpleValue("clusterName", "unknown"); String username = context.getPluginConfiguration().getSimpleValue("username", "cassandra"); - String password = context.getPluginConfiguration().getSimpleValue("password", "password"); + String password = context.getPluginConfiguration().getSimpleValue("password", "cassandra"); + String authenticatorClassName = context.getPluginConfiguration().getSimpleValue("authenticator", + "org.apache.cassandra.auth.AllowAllAuthenticator"); Integer nativePort = 9042; try { @@ -83,16 +85,20 @@ public class CassandraNodeComponent extends JMXServerComponent<ResourceComponent log.debug("Native transport port parsing failed...", e); } + try { - Cluster cluster = Cluster + Builder clusterBuilder = Cluster .builder() .addContactPoints(new String[] { host }) .withoutMetrics() - .withPort(nativePort) - .withCompression(Compression.NONE) - .withAuthInfoProvider(new SimpleAuthInfoProvider().add("username", username).add("password", password)) - .build(); - this.cassandraSession = cluster.connect(clusterName); + .withPort(nativePort); + + if (authenticatorClassName.endsWith("PasswordAuthenticator")) { + clusterBuilder = clusterBuilder.withAuthInfoProvider(new SimpleAuthInfoProvider().add("username", + username).add("password", password)); + } + + this.cassandraSession = clusterBuilder.build().connect(clusterName); } catch (Exception e) { log.error("Connect to Cassandra " + host + ":" + nativePort, e); throw e; diff --git a/modules/plugins/cassandra/src/main/java/org/rhq/plugins/cassandra/CassandraNodeDiscoveryComponent.java b/modules/plugins/cassandra/src/main/java/org/rhq/plugins/cassandra/CassandraNodeDiscoveryComponent.java index febddf1..afb5cd9 100644 --- a/modules/plugins/cassandra/src/main/java/org/rhq/plugins/cassandra/CassandraNodeDiscoveryComponent.java +++ b/modules/plugins/cassandra/src/main/java/org/rhq/plugins/cassandra/CassandraNodeDiscoveryComponent.java @@ -120,6 +120,10 @@ public class CassandraNodeDiscoveryComponent extends JMXDiscoveryComponent { pluginConfig.put(new PropertySimple("nativeTransportPort", parsedProperties .get("native_transport_port"))); } + + if (parsedProperties.get("authenticator") != null) { + pluginConfig.put(new PropertySimple("authenticator", parsedProperties.get("authenticator"))); + } } catch (Exception e) { log.error("YAML Configuration load exception ", e); } finally { diff --git a/modules/plugins/cassandra/src/main/resources/META-INF/rhq-plugin.xml b/modules/plugins/cassandra/src/main/resources/META-INF/rhq-plugin.xml index 905d4b8..66f867d 100644 --- a/modules/plugins/cassandra/src/main/resources/META-INF/rhq-plugin.xml +++ b/modules/plugins/cassandra/src/main/resources/META-INF/rhq-plugin.xml @@ -47,6 +47,12 @@ <c:simple-property name="nativeTransportPort" description="The port on which Cassandra listens for CQL client connections." default="9042" type="integer"/> <c:simple-property name="host" description="The host on which cassandra listens to CQL client connections" default="localhost"/> <c:simple-property name="clusterName" description="Cluster name" default="localhost"/> + <c:simple-property name="authenticator" required="true" default="org.apache.cassandra.auth.AllowAllAuthenticator"> + <c:property-options> + <c:option name="org.apache.cassandra.auth.AllowAllAuthenticator" value="org.apache.cassandra.auth.AllowAllAuthenticator"/> + <c:option name="org.apache.cassandra.auth.PasswordAuthenticator" value="org.apache.cassandra.auth.PasswordAuthenticator"/> + </c:property-options> + </c:simple-property> </plugin-configuration> <process-scan name="CassandraDaemon" query="process|basename|match=^java.*,arg|org.apache.cassandra.service.CassandraDaemon|match=.*"/> commit befa92e41115183f94bda35f1a50c3d2a9e67042 Author: Stefan Negrea <snegrea(a)redhat.com&gt; Date: Tue Apr 30 15:50:00 2013 -0500 Integrate the experimental bulk export for Postgres into the data migrator code. Also, move all the migrator related code to a sub-package diff --git a/modules/enterprise/server/server-metrics/pom.xml b/modules/enterprise/server/server-metrics/pom.xml index 5220281..42f6216 100644 --- a/modules/enterprise/server/server-metrics/pom.xml +++ b/modules/enterprise/server/server-metrics/pom.xml @@ -188,7 +188,7 @@ </descriptorRefs> <archive> <manifest> - <mainClass>org.rhq.server.metrics.DataMigratorRunner</mainClass> + <mainClass>org.rhq.server.metrics.migrator.DataMigratorRunner</mainClass> <addClasspath>true</addClasspath> </manifest> </archive> @@ -293,7 +293,7 @@ </descriptorRefs> <archive> <manifest> - <mainClass>org.rhq.server.metrics.DataMigratorRunner</mainClass> + <mainClass>org.rhq.server.metrics.migrator.DataMigratorRunner</mainClass> <addClasspath>true</addClasspath> </manifest> </archive> 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 deleted file mode 100644 index 523e4d4..0000000 --- a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/DataMigrator.java +++ /dev/null @@ -1,705 +0,0 @@ -/* - * RHQ Management Platform - * Copyright 2011, Red Hat Middleware LLC, and individual contributors - * as indicated by the @author tags. See the copyright.txt file in the - * distribution for a full listing of individual contributors. - * - * This program is free software; you can redistribute it and/or modify - * it under the terms of the GNU General Public License as published by - * the Free Software Foundation version 2 of the License. - * - * 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 for more details. - * - * You should have received a copy of the GNU General Public License - * along with this program; if not, write to the Free Software - * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA. - */ - -package org.rhq.server.metrics; - -import static com.datastax.driver.core.querybuilder.QueryBuilder.ttl; - -import java.util.ArrayList; -import java.util.Arrays; -import java.util.Date; -import java.util.LinkedList; -import java.util.List; -import java.util.Queue; - -import javax.persistence.EntityManager; -import javax.persistence.Query; - -import com.datastax.driver.core.ResultSetFuture; -import com.datastax.driver.core.Session; -import com.datastax.driver.core.querybuilder.Batch; -import com.datastax.driver.core.querybuilder.QueryBuilder; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; - -import org.rhq.server.metrics.domain.AggregateType; -import org.rhq.server.metrics.domain.MetricsTable; - -/** - * @author Stefan Negrea - * - */ -public class DataMigrator { - - private final Log log = LogFactory.getLog(DataMigrator.class); - - - private static final int MAX_RECORDS_TO_LOAD_FROM_SQL = 30000; - private static final int MAX_RAW_BATCH_TO_CASSANDRA = 100; - private static final int MAX_AGGREGATE_BATCH_TO_CASSANDRA = 50; - private static final int MAX_NUMBER_OF_FAILURES = 5; - private static final long NUMBER_OF_BATCHES_FOR_ESTIMATION = 4; - private static final double UNDER_ESTIMATION_FACTOR = .10; - - - private enum MigrationQuery { - SELECT_1H_DATA("SELECT schedule_id, time_stamp, value, minvalue, maxvalue FROM RHQ_MEASUREMENT_DATA_NUM_1H"), - SELECT_6H_DATA("SELECT schedule_id, time_stamp, value, minvalue, maxvalue FROM RHQ_MEASUREMENT_DATA_NUM_6H"), - SELECT_1D_DATA("SELECT schedule_id, time_stamp, value, minvalue, maxvalue FROM RHQ_MEASUREMENT_DATA_NUM_1D"), - - DELETE_1H_DATA("DELETE FROM RHQ_MEASUREMENT_DATA_NUM_1H"), - DELETE_6H_DATA("DELETE FROM RHQ_MEASUREMENT_DATA_NUM_6H"), - DELETE_1D_DATA("DELETE FROM RHQ_MEASUREMENT_DATA_NUM_1D"), - - - COUNT_1H_DATA("SELECT COUNT(*) FROM RHQ_MEASUREMENT_DATA_NUM_1H"), - COUNT_6H_DATA("SELECT COUNT(*) FROM RHQ_MEASUREMENT_DATA_NUM_6H"), - COUNT_1D_DATA("SELECT COUNT(*) FROM RHQ_MEASUREMENT_DATA_NUM_1D"), - - COUNT_RAW("SELECT COUNT(*) FROM %s"), - SELECT_RAW_DATA("SELECT schedule_id, time_stamp, value FROM %s"), - DELETE_RAW_ALL_DATA("DELETE FROM %s"), - DELETE_RAW_ENTRY("DELETE FROM %s WHERE schedule_id = ?"); - - public static final int SCHEDULE_INDEX = 0; - public static final int TIMESTAMP_INDEX = 1; - public static final int VALUE_INDEX = 2; - public static final int MIN_VALUE_INDEX = 3; - public static final int MAX_VALUE_INDEX = 4; - - private String query; - - private MigrationQuery(String query){ - this.query = query; - } - - /** - * @return the query - */ - public String getQuery() { - return query; - } - - @Override - public String toString() { - return query; - } - } - - private final EntityManager entityManager; - - private final Session session; - - private boolean deleteDataImmediatelyAfterMigration; - private boolean deleteAllDataAtEndOfMigration; - - private boolean runRawDataMigration; - private boolean run1HAggregateDataMigration; - private boolean run6HAggregateDataMigration; - private boolean run1DAggregateDataMigration; - - private long estimation; - - public DataMigrator(EntityManager entityManager, Session session) { - this.entityManager = entityManager; - this.session = session; - - this.deleteDataImmediatelyAfterMigration = false; - this.deleteAllDataAtEndOfMigration = false; - this.runRawDataMigration = true; - this.run1HAggregateDataMigration = true; - this.run6HAggregateDataMigration = true; - this.run1DAggregateDataMigration = true; - } - - public void runRawDataMigration(boolean value) { - this.runRawDataMigration = value; - } - - public void run1HAggregateDataMigration(boolean value) { - this.run1HAggregateDataMigration = value; - } - - public void run6HAggregateDataMigration(boolean value) { - this.run6HAggregateDataMigration = value; - } - - public void run1DAggregateDataMigration(boolean value) { - this.run1DAggregateDataMigration = value; - } - - - public void deleteDataImmediatelyAfterMigration() { - this.deleteDataImmediatelyAfterMigration = true; - this.deleteAllDataAtEndOfMigration = false; - } - - public void deleteAllDataAtEndOfMigration() { - this.deleteAllDataAtEndOfMigration = true; - this.deleteDataImmediatelyAfterMigration = false; - } - - public void preserveData() { - this.deleteAllDataAtEndOfMigration = false; - this.deleteDataImmediatelyAfterMigration = false; - } - - public long estimate() throws Exception { - this.estimation = 0; - if (runRawDataMigration) { - retryOnFailure(new RawDataMigrator(), Task.Estimate); - } - - if (run1HAggregateDataMigration) { - retryOnFailure(new AggregateDataMigrator(MetricsTable.ONE_HOUR), Task.Estimate); - } - - if (run6HAggregateDataMigration) { - retryOnFailure(new AggregateDataMigrator(MetricsTable.SIX_HOUR), Task.Estimate); - } - - if (run1DAggregateDataMigration) { - retryOnFailure(new AggregateDataMigrator(MetricsTable.TWENTY_FOUR_HOUR), Task.Estimate); - } - - if (deleteAllDataAtEndOfMigration) { - retryOnFailure(new DeleteAllData(), Task.Estimate); - } - - estimation = (long) (estimation + estimation * UNDER_ESTIMATION_FACTOR); - - return estimation; - } - - public void migrateData() throws Exception { - if (runRawDataMigration) { - retryOnFailure(new RawDataMigrator(), Task.Migrate); - } - - if (run1HAggregateDataMigration) { - retryOnFailure(new AggregateDataMigrator(MetricsTable.ONE_HOUR), Task.Migrate); - } - - if (run6HAggregateDataMigration) { - retryOnFailure(new AggregateDataMigrator(MetricsTable.SIX_HOUR), Task.Migrate); - } - - if (run1DAggregateDataMigration) { - retryOnFailure(new AggregateDataMigrator(MetricsTable.TWENTY_FOUR_HOUR), Task.Migrate); - } - - if (deleteAllDataAtEndOfMigration) { - retryOnFailure(new DeleteAllData(), Task.Migrate); - } - } - - public void deleteOldData() throws Exception { - if (deleteAllDataAtEndOfMigration) { - retryOnFailure(new DeleteAllData(), Task.Migrate); - } - } - - /** - * Retries the migration {@link #MAX_NUMBER_OF_FAILURES} times before - * failing the migration operation. - * - * @param migrator - * @throws Exception - */ - private Thread retryOnFailure(final CallableMigrationWorker migrator, final Task task) - throws Exception { - - RunnableWithException runnable = new RunnableWithException() { - private Exception exception; - - @Override - public void run() { - int numberOfFailures = 0; - Exception caughtException = null; - - log.info(migrator.getClass()); - - while (numberOfFailures < MAX_NUMBER_OF_FAILURES) { - try { - if (task == Task.Estimate) { - estimation += migrator.estimate(); - } else { - migrator.migrate(); - } - return; - } catch (Exception e) { - log.error("Migrator " + migrator.getClass() + " failed. Retrying!", e); - - caughtException = e; - numberOfFailures++; - } - } - - this.exception = caughtException; - } - - @Override - public Exception getException() { - return this.exception; - } - }; - - Thread localThread = new Thread(runnable); - localThread.start(); - localThread.join(); - - if (runnable.getException() != null) { - throw runnable.getException(); - } - - return localThread; - } - - /** - * Returns a list of all the raw SQL metric tables. - * There is no equivalent in Cassandra, all raw data is stored in a single column family. - * - * @return SQL raw metric tables - */ - private String[] getRawDataTables() { - int tableCount = 15; - String tablePrefix = "RHQ_MEAS_DATA_NUM_R"; - - String[] tables = new String[tableCount]; - for (int i = 0; i < tableCount; i++) { - if (i < 10) { - tables[i] = tablePrefix + "0" + i; - } else { - tables[i] = tablePrefix + i; - } - } - - return tables; - } - - enum Task { - Migrate, Estimate - } - - private interface CallableMigrationWorker { - - - long estimate() throws Exception; - - void migrate() throws Exception; - } - - private interface RunnableWithException extends Runnable { - Exception getException(); - } - - private class AggregateDataMigrator implements CallableMigrationWorker { - - private final String selectQuery; - private final String deleteQuery; - private final String countQuery; - private final MetricsTable metricsTable; - - /** - * @param query - * @param metricsTable - */ - public AggregateDataMigrator(MetricsTable metricsTable) throws Exception { - this.metricsTable = metricsTable; - - if (MetricsTable.ONE_HOUR.equals(this.metricsTable)) { - this.selectQuery = MigrationQuery.SELECT_1H_DATA.toString(); - this.deleteQuery = MigrationQuery.DELETE_1H_DATA.toString(); - this.countQuery = MigrationQuery.COUNT_1H_DATA.toString(); - } else if (MetricsTable.SIX_HOUR.equals(this.metricsTable)) { - this.selectQuery = MigrationQuery.SELECT_6H_DATA.toString(); - this.deleteQuery = MigrationQuery.DELETE_6H_DATA.toString(); - this.countQuery = MigrationQuery.COUNT_6H_DATA.toString(); - } else if (MetricsTable.TWENTY_FOUR_HOUR.equals(this.metricsTable)) { - this.selectQuery = MigrationQuery.SELECT_1D_DATA.toString(); - this.deleteQuery = MigrationQuery.DELETE_1D_DATA.toString(); - this.countQuery = MigrationQuery.COUNT_1D_DATA.toString(); - } else { - throw new Exception("MetricsTable " + metricsTable.toString() + " not supported by this migrator."); - } - } - - @Override - public long estimate() throws Exception { - Query nativeQuery = entityManager.createNativeQuery(this.countQuery); - long recordCount = Long.parseLong(nativeQuery.getSingleResult().toString()); - long estimatedTimeToMigrate = this.performMigration(Task.Estimate); - - long estimation = (recordCount / (long) MAX_RECORDS_TO_LOAD_FROM_SQL / NUMBER_OF_BATCHES_FOR_ESTIMATION) - * estimatedTimeToMigrate; - return estimation; - } - - public void migrate() throws Exception { - performMigration(Task.Migrate); - if (deleteDataImmediatelyAfterMigration) { - deleteTableData(); - } - } - - private void deleteTableData() throws Exception { - int failureCount = 0; - while (failureCount < MAX_NUMBER_OF_FAILURES) { - try { - entityManager.getTransaction().begin(); - Query nativeQuery = entityManager.createNativeQuery(this.deleteQuery); - nativeQuery.executeUpdate(); - entityManager.getTransaction().commit(); - log.info("- " + metricsTable.toString() + " - Cleaned -"); - } catch (Exception e) { - log.error("Failed to delete " + metricsTable.toString() - + " data. Attempting to delete data one more time..."); - - failureCount++; - if (failureCount == MAX_NUMBER_OF_FAILURES) { - throw e; - } - } - } - } - - @SuppressWarnings("unchecked") - private long performMigration(Task task) throws Exception { - long migrationStartTime = System.currentTimeMillis(); - long numberOfBatchesMigrated = 0; - - List<Object[]> existingData; - int failureCount; - Query nativeQuery; - - int lastMigratedRecord = 0; - - while (true) { - nativeQuery = entityManager.createNativeQuery(this.selectQuery); - nativeQuery.setFirstResult(lastMigratedRecord); - nativeQuery.setMaxResults(MAX_RECORDS_TO_LOAD_FROM_SQL); - - existingData = nativeQuery.getResultList(); - - if (existingData.size() == 0) { - break; - } - - lastMigratedRecord += existingData.size(); - - failureCount = 0; - while (failureCount < MAX_NUMBER_OF_FAILURES) { - try { - insertDataToCassandra(existingData); - break; - } catch (Exception e) { - log.error("Failed to insert " + metricsTable.toString() - + " data. Attempting to insert the current batch of data one more time"); - log.error(e); - - failureCount++; - if (failureCount == MAX_NUMBER_OF_FAILURES) { - throw e; - } - } - } - - log.info("- " + metricsTable + " - " + lastMigratedRecord + " -"); - - numberOfBatchesMigrated++; - if (Task.Estimate.equals(task) && numberOfBatchesMigrated >= NUMBER_OF_BATCHES_FOR_ESTIMATION) { - break; - } - } - - return System.currentTimeMillis() - migrationStartTime; - } - - private void insertDataToCassandra(List<Object[]> existingData) - throws Exception { - List<ResultSetFuture> resultSetFutures = new ArrayList<ResultSetFuture>(); - Batch batch = QueryBuilder.batch(); - int batchSize = 0; - - //only need approximate TTL to speed up processing - //given that each batch is processed within seconds, getting the - //system time once per batch has minimal impact on the record retention - long creationTimeMillis; - long itemTTLSeconds; - long currentTimeMillis = System.currentTimeMillis(); - long expectedTTLMillis = metricsTable.getTTLinMilliseconds() * 10; - - - for (Object[] rawMeasurement : existingData) { - creationTimeMillis = Long.parseLong(rawMeasurement[MigrationQuery.TIMESTAMP_INDEX].toString()); - itemTTLSeconds = (expectedTTLMillis - currentTimeMillis + creationTimeMillis) / 1000l; - - if(itemTTLSeconds > 0 ){ - batch.add(QueryBuilder.insertInto(metricsTable.toString()) - .value("schedule_id",Integer.parseInt(rawMeasurement[MigrationQuery.SCHEDULE_INDEX].toString())) - .value("time", new Date(creationTimeMillis)) - .value("type", AggregateType.AVG.ordinal()) - .value("value", Double.parseDouble(rawMeasurement[MigrationQuery.VALUE_INDEX].toString())) - .using(ttl((int) itemTTLSeconds))); - - batch.add(QueryBuilder.insertInto(metricsTable.toString()) - .value("schedule_id", Integer.parseInt(rawMeasurement[MigrationQuery.SCHEDULE_INDEX].toString())) - .value("time", new Date(creationTimeMillis)) - .value("type", AggregateType.MIN.ordinal()) - .value("value", Double.parseDouble(rawMeasurement[MigrationQuery.MIN_VALUE_INDEX].toString())) - .using(ttl((int) itemTTLSeconds))); - - batch.add(QueryBuilder.insertInto(metricsTable.toString()) - .value("schedule_id", Integer.parseInt(rawMeasurement[MigrationQuery.SCHEDULE_INDEX].toString())) - .value("time", new Date(creationTimeMillis)) - .value("type", AggregateType.MAX.ordinal()) - .value("value", Double.parseDouble(rawMeasurement[MigrationQuery.MAX_VALUE_INDEX].toString())) - .using(ttl((int) itemTTLSeconds))); - - batchSize += 3; - } - - if (batchSize >= MAX_AGGREGATE_BATCH_TO_CASSANDRA) { - resultSetFutures.add(session.executeAsync(batch)); - batch = QueryBuilder.batch(); - batchSize = 0; - } - } - - if (batchSize != 0) { - resultSetFutures.add(session.executeAsync(batch)); - } - - for (ResultSetFuture future : resultSetFutures) { - future.get(); - } - } - } - - - private class RawDataMigrator implements CallableMigrationWorker { - - Queue<String> tablesNotProcessed = new LinkedList<String>(Arrays.asList(getRawDataTables())); - - public long estimate() throws Exception { - long recordCount = 0; - for (String table : getRawDataTables()) { - String countQuery = String.format(MigrationQuery.COUNT_RAW.toString(), table); - Query nativeQuery = entityManager.createNativeQuery(countQuery); - recordCount += Long.parseLong(nativeQuery.getSingleResult().toString()); - } - - long estimatedTimeToMigrate = this.performMigration(Task.Estimate); - long estimation = (recordCount / (long) MAX_RECORDS_TO_LOAD_FROM_SQL / NUMBER_OF_BATCHES_FOR_ESTIMATION) - * estimatedTimeToMigrate; - return estimation; - } - - public void migrate() throws Exception { - performMigration(Task.Migrate); - } - - @SuppressWarnings("unchecked") - private long performMigration(Task task) throws Exception { - long migrationStartTime = System.currentTimeMillis(); - long numberOfBatchesMigrated = 0; - - List<Object[]> existingData; - int failureCount; - - while (!tablesNotProcessed.isEmpty()) { - String table = tablesNotProcessed.peek(); - - String selectQuery = String.format(MigrationQuery.SELECT_RAW_DATA.toString(), table); - Query nativeSelectQuery = entityManager.createNativeQuery(selectQuery); - - log.info("Start migrating raw table: " + table); - - int lastMigratedRecord = 0; - while (true) { - nativeSelectQuery.setFirstResult(lastMigratedRecord); - nativeSelectQuery.setMaxResults(MAX_RECORDS_TO_LOAD_FROM_SQL); - - existingData = nativeSelectQuery.getResultList(); - - if (existingData.size() == 0) { - break; - } - - lastMigratedRecord += existingData.size(); - - failureCount = 0; - while (failureCount < MAX_NUMBER_OF_FAILURES) { - try { - insertDataToCassandra(existingData); - break; - } catch (Exception e) { - log.error("Failed to insert " + MetricsTable.RAW.toString() - + " data. Attempting to insert the current batch of data one more time"); - log.error(e); - - - failureCount++; - if (failureCount == MAX_AGGREGATE_BATCH_TO_CASSANDRA) { - throw e; - } - } - } - - log.info("- " + table + " - " + lastMigratedRecord + " -"); - - numberOfBatchesMigrated++; - if (Task.Estimate.equals(task) && numberOfBatchesMigrated >= NUMBER_OF_BATCHES_FOR_ESTIMATION) { - break; - } - } - - if (Task.Migrate.equals(task)) { - log.info("Done migrating raw table" + table + "---------------------"); - - if (deleteDataImmediatelyAfterMigration) { - deleteTableData(table); - } - } else if (numberOfBatchesMigrated >= NUMBER_OF_BATCHES_FOR_ESTIMATION) { - break; - } - - tablesNotProcessed.poll(); - } - - return System.currentTimeMillis() - migrationStartTime; - } - - private void deleteTableData(String table) throws Exception { - String deleteQuery = String.format(MigrationQuery.DELETE_RAW_ENTRY.toString(), table); - int failureCount = 0; - while (failureCount < MAX_NUMBER_OF_FAILURES) { - try { - entityManager.getTransaction().begin(); - Query nativeQuery = entityManager.createNativeQuery(deleteQuery); - nativeQuery.executeUpdate(); - entityManager.getTransaction().commit(); - log.info("- " + table + " - Cleaned -"); - } catch (Exception e) { - log.error("Failed to delete " + table + " data. Attempting to delete data one more time..."); - - failureCount++; - if (failureCount == MAX_NUMBER_OF_FAILURES) { - throw e; - } - } - } - } - - private void insertDataToCassandra(List<Object[]> existingData) throws Exception { - List<ResultSetFuture> resultSetFutures = new ArrayList<ResultSetFuture>(); - Batch batch = QueryBuilder.batch(); - int batchSize = 0; - - //only need approximate TTL to speed up processing - //given that each batch is processed within seconds, getting the - //system time once per batch has minimal impact on the record retention - long creationTimeMillis; - long itemTTLSeconds; - long currentTimeMillis = System.currentTimeMillis(); - long expectedTTLMillis = MetricsTable.RAW.getTTLinMilliseconds() * 10; - - - for (Object[] rawDataPoint : existingData) { - creationTimeMillis = Long.parseLong(rawDataPoint[MigrationQuery.TIMESTAMP_INDEX].toString()); - itemTTLSeconds = (expectedTTLMillis - currentTimeMillis + creationTimeMillis) / 1000l; - - if (itemTTLSeconds > 0) { - batch.add(QueryBuilder.insertInto(MetricsTable.RAW.toString()) - .value("schedule_id", Integer.parseInt(rawDataPoint[MigrationQuery.SCHEDULE_INDEX].toString())) - .value("time", new Date(creationTimeMillis)) - .value("value", Double.parseDouble(rawDataPoint[MigrationQuery.VALUE_INDEX].toString())) - .using(ttl((int) itemTTLSeconds))); - batchSize++; - } - - if (batchSize >= MAX_RAW_BATCH_TO_CASSANDRA) { - resultSetFutures.add(session.executeAsync(batch)); - batch = QueryBuilder.batch(); - batchSize = 0; - } - } - - if (batchSize != 0) { - resultSetFutures.add(session.executeAsync(batch)); - } - - for (ResultSetFuture future : resultSetFutures) { - future.get(); - } - } - } - - - private class DeleteAllData implements CallableMigrationWorker { - - public void migrate() { - Query nativeQuery; - - if (run1HAggregateDataMigration) { - entityManager.getTransaction().begin(); - nativeQuery = entityManager.createNativeQuery(MigrationQuery.DELETE_1H_DATA.toString()); - nativeQuery.executeUpdate(); - entityManager.getTransaction().commit(); - log.info("- RHQ_MEASUREMENT_DATA_NUM_1H - Cleaned -"); - } - - if (run6HAggregateDataMigration) { - entityManager.getTransaction().begin(); - nativeQuery = entityManager.createNativeQuery(MigrationQuery.DELETE_6H_DATA.toString()); - nativeQuery.executeUpdate(); - entityManager.getTransaction().commit(); - log.info("- RHQ_MEASUREMENT_DATA_NUM_6H - Cleaned -"); - } - - if (run1DAggregateDataMigration) { - entityManager.getTransaction().begin(); - nativeQuery = entityManager.createNativeQuery(MigrationQuery.DELETE_1D_DATA.toString()); - nativeQuery.executeUpdate(); - entityManager.getTransaction().commit(); - log.info("- RHQ_MEASUREMENT_DATA_NUM_1D - Cleaned -"); - } - - if (runRawDataMigration) { - for (String table : getRawDataTables()) { - entityManager.getTransaction().begin(); - String deleteAllData = String.format(MigrationQuery.DELETE_RAW_ALL_DATA.toString(), table); - nativeQuery = entityManager.createNativeQuery(deleteAllData); - nativeQuery.executeUpdate(); - entityManager.getTransaction().commit(); - log.info("- " + table + " - Cleaned -"); - } - } - } - - @Override - public long estimate() throws Exception { - return 300000; // return return 5 minutes for now without any database side checks. - } - } -} - - diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/DataMigratorRunner.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/DataMigratorRunner.java deleted file mode 100644 index dc3aaf4..0000000 --- a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/DataMigratorRunner.java +++ /dev/null @@ -1,644 +0,0 @@ -/* - * RHQ Management Platform - * Copyright 2011, Red Hat Middleware LLC, and individual contributors - * as indicated by the @author tags. See the copyright.txt file in the - * distribution for a full listing of individual contributors. - * - * This program is free software; you can redistribute it and/or modify - * it under the terms of the GNU General Public License as published by - * the Free Software Foundation version 2 of the License. - * - * 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 for more details. - * - * You should have received a copy of the GNU General Public License - * along with this program; if not, write to the Free Software - * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA. - */ - -package org.rhq.server.metrics; - -import java.io.File; -import java.io.FileInputStream; -import java.io.FileNotFoundException; -import java.util.HashMap; -import java.util.Map; -import java.util.Properties; -import java.util.concurrent.TimeUnit; - -import javax.persistence.EntityManager; -import javax.persistence.EntityManagerFactory; - -import com.datastax.driver.core.Cluster; -import com.datastax.driver.core.ProtocolOptions.Compression; -import com.datastax.driver.core.Session; -import com.datastax.driver.core.SimpleAuthInfoProvider; -import com.datastax.driver.core.exceptions.NoHostAvailableException; - -import org.apache.commons.cli.CommandLine; -import org.apache.commons.cli.CommandLineParser; -import org.apache.commons.cli.HelpFormatter; -import org.apache.commons.cli.Option; -import org.apache.commons.cli.OptionBuilder; -import org.apache.commons.cli.Options; -import org.apache.commons.cli.ParseException; -import org.apache.commons.cli.PosixParser; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.apache.log4j.ConsoleAppender; -import org.apache.log4j.Level; -import org.apache.log4j.Logger; -import org.apache.log4j.PatternLayout; -import org.hibernate.ejb.Ejb3Configuration; - - -/** - * @author Stefan Negrea - * - * Only postgres is supported by the runner, however the data migrator itself can run - * with any database. - * - * Maven command to run this from the command line: - * - * mvn install -DskipTests exec:java -Dexec.mainClass="org.rhq.server.metrics.DataMigratorRunner" - * - * - */ -@SuppressWarnings({ "static-access", "deprecation" }) -public class DataMigratorRunner { - - private final Log log = LogFactory.getLog(DataMigratorRunner.class); - - //Cassandra - private Option cassandraUserOption = OptionBuilder.withLongOpt("cassandra-user").hasArg().withType(String.class) - .withDescription("Cassandra user (default: rhqadmin)").create(); - private Option cassandraPasswordOption = OptionBuilder.withLongOpt("cassandra-password").hasArg() - .withDescription("Cassandra password (default: rhqadmin)").withType(String.class).create(); - private Option cassandraHostsOption = OptionBuilder.withLongOpt("cassandra-hosts").hasArg().withType(String.class) - .withDescription("Cassandra hosts, format host_ip_1,host_ip_2,... (default: 127.0.0.1") - .create(); - private Option cassandraPortOption = OptionBuilder.withLongOpt("cassandra-port").hasArg().withType(Integer.class) - .withDescription("Cassandra native binary protocol port (default: 9142)").create(); - private Option cassandraCompressionOption = OptionBuilder.withLongOpt("cassandra-compression").hasOptionalArg() - .withType(String.class).withDescription("Enable compression for communication with Cassandra (default: true)") - .create(); - - //SQL - private Option sqlUserOption = OptionBuilder.withLongOpt("sql-user").hasArg().withType(String.class) - .withDescription("SQL server user (default: rhqadmin)").create(); - private Option sqlPasswordOption = OptionBuilder.withLongOpt("sql-password").hasArg().withType(String.class) - .withDescription("SQL server password (default: rhqadmin)").create(); - private Option sqlHostOption = OptionBuilder.withLongOpt("sql-host").hasArg().withType(String.class) - .withDescription("SQL server host address (default: localhost)").create(); - private Option sqlPortOption = OptionBuilder.withLongOpt("sql-port").hasArg().withType(String.class) - .withDescription("SQL server port (default: 5432)").create(); - private Option sqlDBOption = OptionBuilder.withLongOpt("sql-db").hasArg().withType(String.class) - .withDescription("SQL database (default: rhq)").create(); - - private Option sqlServerType = OptionBuilder.withLongOpt("sql-server-type").hasArg().withType(String.class) - .withDescription("SQL server type, only postgres and oracle are supported (default: postgres)").create(); - private Option sqlPostgresServer = OptionBuilder.withLongOpt("sql-server-postgres").hasOptionalArg() - .withType(Boolean.class).withDescription("Postgres SQL server.").create(); - private Option sqlOracleServer = OptionBuilder.withLongOpt("sql-server-oracle").hasOptionalArg() - .withType(Boolean.class) - .withDescription("Oracle SQL server.").create(); - - //Migration - private Option disableRawOption = OptionBuilder.withLongOpt("disable-raw-migration").hasOptionalArg().withType(Boolean.class) - .withDescription("Disable raw table migration (default: false)").create(); - private Option disable1HOption = OptionBuilder.withLongOpt("disable-1h-migration").hasOptionalArg().withType(Boolean.class) - .withDescription("Disable 1 hour aggregates table migration (default: false)").create(); - private Option disable6HOption = OptionBuilder.withLongOpt("disable-6h-migration").hasOptionalArg().withType(Boolean.class) - .withDescription("Disable 6 hours aggregates table migration (default: false)").create(); - private Option disable1DOption = OptionBuilder.withLongOpt("disable-1d-migration").hasOptionalArg().withType(Boolean.class) - .withDescription("Disable 24 hours aggregates table migration (default: false)").create(); - private Option preserveDataOption = OptionBuilder.withLongOpt("preserve-data").hasOptionalArg().withType(Boolean.class) - .withDescription("Preserve SQL data post migration (default: true)").create(); - private Option deleteDataOption = OptionBuilder.withLongOpt("delete-data").hasOptionalArg().withType(Boolean.class) - .withDescription("Delete SQL data at the end of migration (default: false)").create(); - private Option estimateOnlyOption = OptionBuilder.withLongOpt("estimate-only").hasOptionalArg().withType(Boolean.class) - .withDescription("Only estimate how long the migration will take (default: false)").create(); - private Option deleteOnlyOption = OptionBuilder.withLongOpt("delete-only").hasOptionalArg().withType(Boolean.class) - .withDescription("Only delete data from the old SQL server, no migration will be performed (default: false)") - .create(); - - //Runner - private Option helpOption = OptionBuilder.withLongOpt("help").create("h"); - private Option debugLogOption = OptionBuilder.withLongOpt("debugLog") - .withDescription("Enable debug level logs for the communication with Cassandra and SQL Server (default: false)") - .create("X"); - private Option configFileOption = OptionBuilder.withLongOpt("config-file").hasArg() - .withDescription("Configuration file. All the command line options can be set in a typical properties file. " + - "Command line arguments take precedence over default and configuration file options.") - .create(); - - private Map<Object, Object> configuration = new HashMap<Object, Object>(); - private Options options; - - /** - * @param args - * @throws ParseException - */ - public static void main(String[] args) throws Exception { - initLogging(); - try{ - DataMigratorRunner runner = new DataMigratorRunner(); - runner.configure(args); - runner.run(); - } catch (HelpRequestedException h) { - //do nothing - } catch (Exception e) { - e.printStackTrace(); - System.exit(1); - } - - System.exit(0); - } - - private static void initLogging() { - Logger root = Logger.getRootLogger(); - if (!root.getAllAppenders().hasMoreElements()) { - root.addAppender(new ConsoleAppender(new PatternLayout(PatternLayout.TTCC_CONVERSION_PATTERN))); - setLogLevel(Level.ERROR); - } - } - - private static void setLogLevel(Level level) { - Logger root = Logger.getRootLogger(); - root.setLevel(level); - - Logger cassandraLogging = root.getLoggerRepository().getLogger("log4j.logger.org.apache.cassandra.cql.jdbc"); - cassandraLogging.setLevel(level); - - Logger cassandraDriverLogging = root.getLoggerRepository().getLogger("com.datastax.driver"); - cassandraDriverLogging.setLevel(level); - - Logger hibernateLogging = root.getLoggerRepository().getLogger("org.hibernate"); - hibernateLogging.setLevel(Level.ERROR); - - Logger migratorLogging = root.getLoggerRepository().getLogger("org.rhq"); - if (Level.DEBUG.equals(level)) { - migratorLogging.setLevel(Level.ALL); - } else { - migratorLogging.setLevel(level); - } - } - - private void configure(String args[]) throws Exception { - options = new Options(); - - options.addOption(cassandraUserOption); - options.addOption(cassandraPasswordOption); - options.addOption(cassandraHostsOption); - options.addOption(cassandraPortOption); - options.addOption(cassandraCompressionOption); - - options.addOption(sqlUserOption); - options.addOption(sqlPasswordOption); - options.addOption(sqlHostOption); - options.addOption(sqlPortOption); - options.addOption(sqlDBOption); - options.addOption(sqlServerType); - options.addOption(sqlPostgresServer); - options.addOption(sqlOracleServer); - - options.addOption(disableRawOption); - options.addOption(disable1HOption); - options.addOption(disable6HOption); - options.addOption(disable1DOption); - options.addOption(preserveDataOption); - options.addOption(deleteDataOption); - options.addOption(estimateOnlyOption); - options.addOption(deleteOnlyOption); - - options.addOption(helpOption); - options.addOption(debugLogOption); - options.addOption(configFileOption); - - CommandLine commandLine; - try { - CommandLineParser parser = new PosixParser(); - commandLine = parser.parse(options, args); - } catch (Exception e) { - HelpFormatter formatter = new HelpFormatter(); - formatter.setWidth(120); - formatter.printHelp("DataMigrationRunner", options); - throw new Exception("Error parsing command line arguments"); - } - - if (commandLine.hasOption(helpOption.getLongOpt()) || commandLine.hasOption(helpOption.getOpt())) { - HelpFormatter formatter = new HelpFormatter(); - formatter.printHelp("DataMigrationRunner", options); - throw new HelpRequestedException(); - } - - if (commandLine.hasOption(debugLogOption.getLongOpt()) || commandLine.hasOption(debugLogOption.getOpt())) { - DataMigratorRunner.setLogLevel(Level.DEBUG); - } - - loadDefaultConfiguration(); - if (commandLine.hasOption(configFileOption.getLongOpt())) { - loadConfigFile(commandLine.getOptionValue(configFileOption.getLongOpt())); - } - - parseCassandraOptions(commandLine); - parseSQLOptions(commandLine); - parseMigrationOptions(commandLine); - } - - /** - * Add default configuration options to the configuration store. - */ - private void loadDefaultConfiguration() { - //default Cassandra configuration - configuration.put(cassandraUserOption, "rhqadmin"); - configuration.put(cassandraPasswordOption, "rhqadmin"); - configuration.put(cassandraHostsOption, new String[] { "127.0.0.1" }); - configuration.put(cassandraPortOption, 9042); - configuration.put(cassandraCompressionOption, true); - - //default SQL configuration - configuration.put(sqlUserOption, "rhqadmin"); - configuration.put(sqlPasswordOption, "rhqadmin"); - configuration.put(sqlHostOption, "localhost"); - configuration.put(sqlPortOption, "5432"); - configuration.put(sqlDBOption, "rhq"); - configuration.put(sqlServerType, "postgres"); - - //default runner options - configuration.put(disableRawOption, false); - configuration.put(disable1HOption, false); - configuration.put(disable6HOption, false); - configuration.put(disable1DOption, false); - configuration.put(preserveDataOption, true); - configuration.put(estimateOnlyOption, false); - configuration.put(deleteOnlyOption, false); - } - - /** - * Load the configuration options from file and overlay them on top of the default - * options. - * - * @param file config file - */ - private void loadConfigFile(String file) { - try { - File configFile = new File(file); - if (!configFile.exists()) { - throw new FileNotFoundException("Configuration file not found!"); - } - - Properties configProperties = new Properties(); - FileInputStream stream = new FileInputStream(configFile); - configProperties.load(stream); - stream.close(); - - for (Object optionObject : options.getOptions()) { - Option option = (Option) optionObject; - Object optionValue; - - if ((optionValue = configProperties.get(option.getLongOpt())) != null) { - log.debug("Configuration option loaded: " + option.getLongOpt() + " (" + option.getType() + ") -> " - + optionValue); - - if (option.equals(cassandraHostsOption)) { - String[] cassandraHosts = parseCassandraHosts(optionValue.toString()); - configuration.put(option, cassandraHosts); - } else if (option.equals(sqlServerType)) { - if ("oracle".equals(optionValue)) { - configuration.put(option, "oracle"); - } else { - configuration.put(option, "postgres"); - } - } else if (option.equals(sqlPostgresServer)) { - boolean value = tryParseBoolean(optionValue.toString(), true); - if (value == true) { - configuration.put(sqlServerType, "postgres"); - } - } else if (option.equals(sqlOracleServer)) { - boolean value = tryParseBoolean(optionValue.toString(), true); - if (value == true) { - configuration.put(sqlServerType, "oracle"); - } - } else if (option.getType().equals(Boolean.class)) { - configuration.put(option, tryParseBoolean(optionValue.toString(), true)); - } else if (option.getType().equals(Integer.class)) { - configuration.put(option, tryParseInteger(optionValue.toString(), 0)); - } else { - configuration.put(option, optionValue.toString()); - } - } - } - } catch (Exception e) { - log.error("Unable to load or process the configuration file.", e); - System.exit(1); - } - - log.debug(configuration.toString()); - } - - /** - * Parse command line options for Cassandra. - * - * @param commandLine command line - * @throws NoHostAvailableException - */ - private void parseCassandraOptions(CommandLine commandLine) throws Exception { - if (commandLine.hasOption(cassandraUserOption.getLongOpt())) { - configuration.put(cassandraUserOption, commandLine.getOptionValue(cassandraUserOption.getLongOpt())); - } - - if (commandLine.hasOption(cassandraPasswordOption.getLongOpt())) { - configuration - .put(cassandraPasswordOption, commandLine.getOptionValue(cassandraPasswordOption.getLongOpt())); - } - - if (commandLine.hasOption(cassandraHostsOption.getLongOpt())) { - String[] cassandraHosts = parseCassandraHosts(commandLine.getOptionValue(cassandraHostsOption.getLongOpt())); - configuration.put(cassandraHostsOption, cassandraHosts); - } - - if (commandLine.hasOption(cassandraPortOption.getLongOpt())) { - Integer cassandraPort = tryParseInteger(commandLine.getOptionValue(cassandraPortOption.getLongOpt()), 9142); - configuration.put(cassandraPortOption, cassandraPort); - } - - if (commandLine.hasOption(cassandraCompressionOption.getLongOpt())) { - boolean value = tryParseBoolean(commandLine.getOptionValue(disableRawOption.getLongOpt()), true); - configuration.put(cassandraCompressionOption, value); - } - } - - /** - * Parse command line options for SQL. - * - * @param commandLine command line - * @throws NoHostAvailableException - */ - private void parseSQLOptions(CommandLine commandLine) throws NoHostAvailableException { - if (commandLine.hasOption(sqlUserOption.getLongOpt())) { - configuration.put(sqlUserOption, commandLine.getOptionValue(sqlUserOption.getLongOpt())); - } - - if (commandLine.hasOption(sqlPasswordOption.getLongOpt())) { - configuration.put(sqlPasswordOption, commandLine.getOptionValue(sqlPasswordOption.getLongOpt())); - } - - if (commandLine.hasOption(sqlHostOption.getLongOpt())) { - configuration.put(sqlHostOption, commandLine.getOptionValue(sqlHostOption.getLongOpt())); - } - - if (commandLine.hasOption(sqlPortOption.getLongOpt())) { - configuration.put(sqlPortOption, commandLine.getOptionValue(sqlPortOption.getLongOpt())); - } - - if (commandLine.hasOption(sqlDBOption.getLongOpt())) { - configuration.put(sqlDBOption, commandLine.getOptionValue(sqlDBOption.getLongOpt())); - } - - if (commandLine.hasOption(sqlServerType.getLongOpt())) { - if ("oracle".equals(commandLine.getOptionValue(sqlServerType.getLongOpt()))) { - configuration.put(sqlServerType, "oracle"); - } else { - configuration.put(sqlServerType, "postgres"); - } - } else if (commandLine.hasOption(sqlPostgresServer.getLongOpt())) { - configuration.put(sqlServerType, "postgres"); - } else if (commandLine.hasOption(sqlOracleServer.getLongOpt())) { - configuration.put(sqlServerType, "oracle"); - } - } - - /** - * Parse command line options for the actual migration progress. - * - * @param commandLine - */ - private void parseMigrationOptions(CommandLine commandLine) { - boolean value; - - if (commandLine.hasOption(disableRawOption.getLongOpt())) { - value = tryParseBoolean(commandLine.getOptionValue(disableRawOption.getLongOpt()), true); - configuration.put(disableRawOption, value); - } - - if (commandLine.hasOption(disable1HOption.getLongOpt())) { - value = tryParseBoolean(commandLine.getOptionValue(disable1HOption.getLongOpt()), true); - configuration.put(disable1HOption, value); - } - - if (commandLine.hasOption(disable6HOption.getLongOpt())) { - value = tryParseBoolean(commandLine.getOptionValue(disable6HOption.getLongOpt()), true); - configuration.put(disable6HOption, value); - } - - if (commandLine.hasOption(disable1DOption.getLongOpt())) { - value = tryParseBoolean(commandLine.getOptionValue(disable1DOption.getLongOpt()), true); - configuration.put(disable1DOption, value); - } - - if (commandLine.hasOption(preserveDataOption.getLongOpt())) { - value = tryParseBoolean(commandLine.getOptionValue(preserveDataOption.getLongOpt()), true); - configuration.put(preserveDataOption, value); - } else if (commandLine.hasOption(deleteDataOption.getLongOpt())) { - value = tryParseBoolean(commandLine.getOptionValue(deleteDataOption.getLongOpt()), true); - configuration.put(preserveDataOption, value); - } - - if (commandLine.hasOption(estimateOnlyOption.getLongOpt())) { - value = tryParseBoolean(commandLine.getOptionValue(estimateOnlyOption.getLongOpt()), true); - configuration.put(estimateOnlyOption, true); - } - } - - private void run() throws Exception { - log.debug("Creating Entity Manager"); - EntityManager entityManager = this.createEntityManager(); - log.debug("Done creating Entity Manager"); - - log.debug("Creating Cassandra session"); - Session session = this.createCassandraSession(); - log.debug("Done creating Cassandra session"); - - DataMigrator migrator = new DataMigrator(entityManager, session); - - if (!(Boolean) configuration.get(deleteOnlyOption)) { - if ((Boolean) configuration.get(preserveDataOption)) { - migrator.preserveData(); - } else { - migrator.deleteAllDataAtEndOfMigration(); - } - - migrator.runRawDataMigration(!(Boolean) configuration.get(disableRawOption)); - migrator.run1HAggregateDataMigration(!(Boolean) configuration.get(disable1HOption)); - migrator.run6HAggregateDataMigration(!(Boolean) configuration.get(disable6HOption)); - migrator.run1DAggregateDataMigration(!(Boolean) configuration.get(disable1DOption)); - - System.out.println("Estimation process - starting\n"); - long estimate = migrator.estimate(); - System.out.println("The migration process will take approximately: " - + TimeUnit.MILLISECONDS.toMinutes(estimate) + " minutes (or " + estimate + " milliseconds)\n"); - System.out.println("Estimation process - ended\n\n"); - - if (!(Boolean) configuration.get(estimateOnlyOption)) { - System.out.println("Migration process - starting\n"); - long startTime = System.currentTimeMillis(); - migrator.migrateData(); - long duration = System.currentTimeMillis() - startTime; - System.out.println("The migration process took: " + TimeUnit.MILLISECONDS.toMinutes(duration) - + " minutes (or " + duration + " milliseconds)\n"); - System.out.println("Migration process - ended\n"); - } - } else { - migrator.deleteAllDataAtEndOfMigration(); - migrator.runRawDataMigration(true); - migrator.run1HAggregateDataMigration(true); - migrator.run6HAggregateDataMigration(true); - migrator.run1DAggregateDataMigration(true); - - System.out.println("Estimation process - starting\n"); - long estimate = migrator.estimate(); - System.out.println("The deletion of old data will take approximately: " - + TimeUnit.MILLISECONDS.toMinutes(estimate) + " minutes (or " + estimate + " milliseconds)\n"); - System.out.println("Estimation process - ended\n\n"); - - if (!(Boolean) configuration.get(estimateOnlyOption)) { - System.out.println("Old data deletion process - starting\n"); - long startTime = System.currentTimeMillis(); - migrator.deleteOldData(); - long duration = System.currentTimeMillis() - startTime; - System.out.println("The deletion process took: " + TimeUnit.MILLISECONDS.toMinutes(duration) - + " minutes (or " + duration + " milliseconds)\n"); - System.out.println("Old data deletion process - ended\n"); - } - } - } - - /** - * Create a Cassandra session based on configuration options. - * - * @return Cassandra session - * @throws Exception - */ - private Session createCassandraSession() throws Exception { - Compression selectedCompression = Compression.NONE; - if ((Boolean) configuration.get(cassandraCompressionOption)) { - selectedCompression = Compression.SNAPPY; - } - - Cluster cluster = Cluster - .builder() - .addContactPoints((String[]) configuration.get(cassandraHostsOption)) - .withPort((Integer) configuration.get(cassandraPortOption)) - .withCompression(selectedCompression) - .withoutMetrics() - .withAuthInfoProvider( - new SimpleAuthInfoProvider().add("username", (String) configuration.get(cassandraUserOption)).add( - "password", (String) configuration.get(cassandraPasswordOption))).build(); - - return cluster.connect("rhq"); - } - - /** - * Create a hibernate session to the SQL server. - * - * @return - * @throws Exception - */ - private EntityManager createEntityManager() throws Exception { - Properties properties = new Properties(); - properties.put("javax.persistence.provider", "org.hibernate.ejb.HibernatePersistence"); - properties.put("hibernate.connection.username", (String) configuration.get(sqlUserOption)); - properties.put("hibernate.connection.password", (String) configuration.get(sqlPasswordOption)); - - if ("oracle".equals(configuration.get(sqlServerType))) { - String driverClassName = "oracle.jdbc.driver.OracleDriver"; - - try { - //Required to preload the driver manually. - //Without this the driver load will fail due to the packaging. - Class.forName(driverClassName); - } catch (ClassNotFoundException e) { - log.debug(e); - throw new Exception("Oracle SQL Driver class could not be loaded. Missing class: " + driverClassName); - } - - properties.put("hibernate.dialect", "org.hibernate.dialect.Oracle10gDialect"); - properties.put("hibernate.driver_class", driverClassName); - properties.put("hibernate.connection.url", "jdbc:oracle:thin:@" + (String) configuration.get(sqlHostOption) - + ":" + (String) configuration.get(sqlPortOption) + ":" + (String) configuration.get(sqlDBOption)); - properties.put("hibernate.default_schema", (String) configuration.get(sqlDBOption)); - } else { - String driverClassName = "org.postgresql.Driver"; - - try { - //Required to preload the driver manually. - //Without this the driver load will fail due to the packaging. - Class.forName(driverClassName); - } catch (ClassNotFoundException e) { - log.debug(e); - throw new Exception("Postgres SQL Driver class could not be loaded. Missing class: " + driverClassName); - } - - properties.put("hibernate.dialect", "org.hibernate.dialect.PostgreSQLDialect"); - properties.put("hibernate.driver_class", driverClassName); - properties.put("hibernate.connection.url", "jdbc:postgresql://" + (String) configuration.get(sqlHostOption) - + ":" + (String) configuration.get(sqlPortOption) + "/" + (String) configuration.get(sqlDBOption)); - } - - log.debug("Creating entity manager with the following configuration:"); - log.debug(properties); - - Ejb3Configuration configuration = new Ejb3Configuration(); - configuration.setProperties(properties); - EntityManagerFactory factory = configuration.buildEntityManagerFactory(); - return factory.createEntityManager(); - } - - /** - * Parse Cassandra host information submitted in the form: - * host_addres,thrift_port,native_port|host_address_2,thrift_port,native_port - * - * @param stringValue - * @return - */ - private String[] parseCassandraHosts(String stringValue) { - String[] seeds = stringValue.split(","); - return seeds; - } - - /** - * @param value object value to parse - * @param defaultValue default value - * @return - */ - private boolean tryParseBoolean(Object value, boolean defaultValue) { - try { - return Boolean.parseBoolean(value.toString()); - } catch (Exception e) { - return defaultValue; - } - } - - /** - * @param value object value to parse - * @param defaultValue default value - * @return - */ - private Integer tryParseInteger(Object value, int defaultValue) { - try { - return Integer.parseInt(value.toString()); - } catch (Exception e) { - return defaultValue; - } - } - - @SuppressWarnings("serial") - private class HelpRequestedException extends Exception { - public HelpRequestedException() { - super("Help Requested"); - } - } -} \ No newline at end of file diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataBulkExportSource.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataBulkExportSource.java deleted file mode 100644 index d817002..0000000 --- a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataBulkExportSource.java +++ /dev/null @@ -1,129 +0,0 @@ -/* - * RHQ Management Platform - * Copyright (C) 2005-2013 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 as published by - * the Free Software Foundation version 2 of the License. - * - * 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 for more details. - * - * You should have received a copy of the GNU 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 java.io.BufferedReader; -import java.io.File; -import java.io.FileReader; -import java.sql.Connection; -import java.sql.SQLException; -import java.util.LinkedList; -import java.util.List; -import java.util.StringTokenizer; - -import javax.persistence.EntityManager; - -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.hibernate.Session; -import org.hibernate.engine.spi.SessionFactoryImplementor; -import org.hibernate.service.jdbc.connections.spi.ConnectionProvider; - -import org.rhq.core.util.stream.StreamUtil; - -/** - * @author Thomas Segismont - */ -public abstract class ExistingDataBulkExportSource implements ExistingDataSource { - - private static final Log LOG = LogFactory.getLog(ExistingDataBulkExportSource.class); - - protected static final int IO_BUFFER_SIZE = 1024 * 64; - - protected static final String DELIMITER = "|"; - - private EntityManager entityManager; - - private String selectNativeQuery; - - private File workDirectory; - - private String fileName; - - private File existingDataFile; - - private BufferedReader existingDataFileReader; - - private int currentIndex; - - public ExistingDataBulkExportSource(EntityManager entityManager, String selectNativeQuery, File workDirectory, - String fileName) { - this.entityManager = entityManager; - this.selectNativeQuery = selectNativeQuery; - this.workDirectory = workDirectory; - this.fileName = fileName; - existingDataFile = new File(workDirectory, fileName); - } - - protected String getSelectNativeQuery() { - return selectNativeQuery; - } - - protected File getExistingDataFile() { - return existingDataFile; - } - - protected Connection getConnection() throws SQLException { - Session session = (Session) entityManager.getDelegate(); - SessionFactoryImplementor sfi = (SessionFactoryImplementor) session.getSessionFactory(); - ConnectionProvider cp = sfi.getConnectionProvider(); - return cp.getConnection(); - } - - public abstract void exportExistingData() throws Exception; - - public void startReading() throws Exception { - if (!existingDataFile.exists() && !existingDataFile.isFile() && !existingDataFile.canRead()) { - throw new IllegalStateException(); - } - existingDataFileReader = new BufferedReader(new FileReader(existingDataFile)); - currentIndex = 0; - } - - public void stopReading() { - StreamUtil.safeClose(existingDataFileReader); - } - - @Override - public List<Object[]> getExistingData(int fromIndex, int maxResults) throws Exception { - if (LOG.isDebugEnabled()) { - LOG.debug("Reading lines " + fromIndex + "-" + (fromIndex + maxResults)); - } - if (fromIndex != currentIndex) { - throw new IllegalStateException(); - } - List<Object[]> results = new LinkedList<Object[]>(); - for (int i = 0; i < maxResults; i++) { - String nextLine = existingDataFileReader.readLine(); - if (nextLine == null) { - break; - } - currentIndex++; - StringTokenizer stringTokenizer = new StringTokenizer(nextLine, DELIMITER); - Object[] row = new Object[stringTokenizer.countTokens()]; - for (int j = 0; j < row.length; j++) { - row[j] = stringTokenizer.nextToken(); - } - results.add(row); - } - return results; - } - -} diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataJPABulkExportSource.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataJPABulkExportSource.java deleted file mode 100644 index d61fdf1..0000000 --- a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataJPABulkExportSource.java +++ /dev/null @@ -1,124 +0,0 @@ -package org.rhq.server.metrics; - -import java.io.BufferedWriter; -import java.io.File; -import java.io.FileWriter; -import java.sql.Connection; -import java.sql.PreparedStatement; -import java.sql.ResultSet; -import java.util.List; -import java.util.Properties; - -import javax.persistence.EntityManager; -import javax.persistence.EntityManagerFactory; - -import org.apache.commons.lang.time.StopWatch; -import org.apache.log4j.BasicConfigurator; -import org.apache.log4j.Level; -import org.apache.log4j.Logger; -import org.hibernate.ejb.Ejb3Configuration; - -import org.rhq.core.util.jdbc.JDBCUtil; -import org.rhq.core.util.stream.StreamUtil; - -/** - * @author Thomas Segismont - */ -public class ExistingDataJPABulkExportSource extends ExistingDataBulkExportSource { - - public ExistingDataJPABulkExportSource(EntityManager entityManager, String selectNativeQuery, File workDirectory, - String fileName) { - super(entityManager, selectNativeQuery, workDirectory, fileName); - } - - public void exportExistingData() throws Exception { - BufferedWriter fileWriter = null; - Connection connection = null; - PreparedStatement statement = null; - ResultSet resultSet = null; - try { - fileWriter = new BufferedWriter(new FileWriter(getExistingDataFile()), IO_BUFFER_SIZE); - connection = getConnection(); - statement = connection.prepareStatement(getSelectNativeQuery()); - int columnCount = resultSet.getMetaData().getColumnCount(); - resultSet = statement.executeQuery(); - while (resultSet.next()) { - for (int i = 1; i < columnCount + 1; i++) { - if (i > 1) { - fileWriter.write(DELIMITER); - } - fileWriter.write(resultSet.getString(i)); - } - fileWriter.write("\n"); - } - } finally { - StreamUtil.safeClose(fileWriter); - JDBCUtil.safeClose(connection, statement, resultSet); - } - } - - public static void main(String[] args) throws Exception { - BasicConfigurator.configure(); - Logger.getRootLogger().setLevel(Level.INFO); - Logger.getLogger("org.rhq").setLevel(Level.DEBUG); - EntityManagerFactory entityManagerFactory = null; - EntityManager entityManager = null; - ExistingDataBulkExportSource source = null; - try { - entityManagerFactory = createEntityManager(); - entityManager = entityManagerFactory.createEntityManager(); - source = new ExistingPostgresDataBulkExportSource(entityManager, - "SELECT schedule_id, time_stamp, value, minvalue, maxvalue FROM RHQ_MEASUREMENT_DATA_NUM_1D", - // "SELECT count(*) FROM RHQ_MEASUREMENT_DATA_NUM_1D limit 1" - new File(System.getProperty("java.io.tmpdir")), "poupoupidou.txt"); - StopWatch stopWatch = new StopWatch(); - stopWatch.start(); - source.exportExistingData(); - source.startReading(); - int rowIndex = 0; - int maxResults = 30000; - for (;;) { - List<Object[]> existingData = source.getExistingData(rowIndex, maxResults); - if (existingData.size() < maxResults) { - break; - } else { - rowIndex += maxResults; - } - } - stopWatch.stop(); - System.out.println("Execution: " + stopWatch); - } finally { - if (source != null) { - source.stopReading(); - } - if (entityManager != null) { - entityManager.close(); - } - if (entityManagerFactory != null) { - entityManagerFactory.close(); - } - } - } - - private static EntityManagerFactory createEntityManager() throws Exception { - Properties properties = new Properties(); - properties.put("javax.persistence.provider", "org.hibernate.ejb.HibernatePersistence"); - properties.put("hibernate.connection.username", "rhqadmin"); - properties.put("hibernate.connection.password", "rhqadmin"); - properties.put("hibernate.dialect", "org.hibernate.dialect.PostgreSQLDialect"); - String driverClassName = "org.postgresql.Driver"; - try { - //Required to preload the driver manually. - //Without this the driver load will fail due to the packaging. - Class.forName(driverClassName); - } catch (ClassNotFoundException e) { - throw new Exception("Postgres SQL Driver class could not be loaded. Missing class: " + driverClassName); - } - properties.put("hibernate.driver_class", driverClassName); - properties.put("hibernate.connection.url", "jdbc:postgresql://localhost:5432/rhqdev"); - Ejb3Configuration configuration = new Ejb3Configuration(); - configuration.setProperties(properties); - return configuration.buildEntityManagerFactory(); - } - -} diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataJPASource.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataJPASource.java deleted file mode 100644 index b072f58..0000000 --- a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataJPASource.java +++ /dev/null @@ -1,119 +0,0 @@ -/* - * RHQ Management Platform - * Copyright (C) 2005-2013 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 as published by - * the Free Software Foundation version 2 of the License. - * - * 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 for more details. - * - * You should have received a copy of the GNU 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 java.util.List; -import java.util.Properties; - -import javax.persistence.EntityManager; -import javax.persistence.EntityManagerFactory; - -import org.apache.commons.lang.time.StopWatch; -import org.apache.commons.logging.Log; -import org.apache.commons.logging.LogFactory; -import org.apache.log4j.BasicConfigurator; -import org.apache.log4j.Level; -import org.apache.log4j.Logger; -import org.hibernate.ejb.Ejb3Configuration; - -/** - * @author Thomas Segismont - */ -public class ExistingDataJPASource implements ExistingDataSource { - - private static final Log LOG = LogFactory.getLog(ExistingDataJPASource.class); - - private EntityManager entityManager; - - private String selectNativeQuery; - - public ExistingDataJPASource(EntityManager entityManager, String selectNativeQuery) { - this.entityManager = entityManager; - this.selectNativeQuery = selectNativeQuery; - } - - @Override - public List<Object[]> getExistingData(int fromIndex, int maxResults) throws Exception { - if (LOG.isDebugEnabled()) { - LOG.debug("Reading lines " + fromIndex + "-" + (fromIndex + maxResults)); - } - return entityManager.createNativeQuery(selectNativeQuery).setFirstResult(fromIndex).setMaxResults(maxResults) - .getResultList(); - } - - public static void main(String[] args) throws Exception { - BasicConfigurator.configure(); - Logger.getRootLogger().setLevel(Level.INFO); - Logger.getLogger("org.rhq").setLevel(Level.DEBUG); - EntityManagerFactory entityManagerFactory = null; - EntityManager entityManager = null; - ExistingDataJPASource source = null; - try { - entityManagerFactory = createEntityManager(); - entityManager = entityManagerFactory.createEntityManager(); - source = new ExistingDataJPASource( - entityManager, - "SELECT schedule_id, time_stamp, value, minvalue, maxvalue FROM RHQ_MEASUREMENT_DATA_NUM_1D"); - StopWatch stopWatch = new StopWatch(); - stopWatch.start(); - int rowIndex = 0; - int maxResults = 30000; - for (; ; ) { - List<Object[]> existingData = source.getExistingData(rowIndex, maxResults); - if (existingData.size() < maxResults) { - break; - } else { - rowIndex += maxResults; - } - } - stopWatch.stop(); - System.out.println("Execution: " + stopWatch); - } finally { - if (entityManager != null) { - entityManager.close(); - } - if (entityManagerFactory != null) { - entityManagerFactory.close(); - } - } - } - - private static EntityManagerFactory createEntityManager() throws Exception { - Properties properties = new Properties(); - properties.put("javax.persistence.provider", "org.hibernate.ejb.HibernatePersistence"); - properties.put("hibernate.connection.username", "rhqadmin"); - properties.put("hibernate.connection.password", "rhqadmin"); - properties.put("hibernate.dialect", "org.hibernate.dialect.PostgreSQLDialect"); - String driverClassName = "org.postgresql.Driver"; - try { - //Required to preload the driver manually. - //Without this the driver load will fail due to the packaging. - Class.forName(driverClassName); - } catch (ClassNotFoundException e) { - throw new Exception("Postgres SQL Driver class could not be loaded. Missing class: " + driverClassName); - } - properties.put("hibernate.driver_class", driverClassName); - properties.put("hibernate.connection.url", "jdbc:postgresql://localhost:5432/rhq"); - Ejb3Configuration configuration = new Ejb3Configuration(); - configuration.setProperties(properties); - return configuration.buildEntityManagerFactory(); - } - -} diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataSource.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataSource.java deleted file mode 100644 index d6e0203..0000000 --- a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataSource.java +++ /dev/null @@ -1,31 +0,0 @@ -/* - * RHQ Management Platform - * Copyright (C) 2005-2013 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 as published by - * the Free Software Foundation version 2 of the License. - * - * 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 for more details. - * - * You should have received a copy of the GNU 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 java.util.List; - -/** - * @author Thomas Segismont - */ -public interface ExistingDataSource { - - List<Object[]> getExistingData(int fromIndex, int maxResults) throws Exception; - -} diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingPostgresDataBulkExportSource.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingPostgresDataBulkExportSource.java deleted file mode 100644 index 8d049ed..0000000 --- a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingPostgresDataBulkExportSource.java +++ /dev/null @@ -1,133 +0,0 @@ -/* - * RHQ Management Platform - * Copyright (C) 2005-2013 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 as published by - * the Free Software Foundation version 2 of the License. - * - * 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 for more details. - * - * You should have received a copy of the GNU 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 java.io.BufferedWriter; -import java.io.File; -import java.io.FileWriter; -import java.sql.Connection; -import java.util.List; -import java.util.Properties; - -import javax.persistence.EntityManager; -import javax.persistence.EntityManagerFactory; - -import org.apache.commons.lang.time.StopWatch; -import org.apache.log4j.BasicConfigurator; -import org.apache.log4j.Level; -import org.apache.log4j.Logger; -import org.hibernate.ejb.Ejb3Configuration; -import org.postgresql.copy.CopyManager; -import org.postgresql.core.BaseConnection; - -import org.rhq.core.util.jdbc.JDBCUtil; -import org.rhq.core.util.stream.StreamUtil; - -/** - * @author Thomas Segismont - */ -public class ExistingPostgresDataBulkExportSource extends ExistingDataBulkExportSource { - - public ExistingPostgresDataBulkExportSource(EntityManager entityManager, String selectNativeQuery, - File workDirectory, String fileName) { - super(entityManager, selectNativeQuery, workDirectory, fileName); - } - - public void exportExistingData() throws Exception { - BufferedWriter fileWriter = null; - Connection connection = null; - try { - fileWriter = new BufferedWriter(new FileWriter(getExistingDataFile()), IO_BUFFER_SIZE); - connection = getConnection(); - CopyManager copyManager = new CopyManager((BaseConnection) connection); - copyManager.copyOut("COPY (" + getSelectNativeQuery() + ") TO STDOUT WITH DELIMITER '" + DELIMITER + "'", - fileWriter); - } finally { - StreamUtil.safeClose(fileWriter); - JDBCUtil.safeClose(connection); - } - } - - public static void main(String[] args) throws Exception { - BasicConfigurator.configure(); - Logger.getRootLogger().setLevel(Level.INFO); - Logger.getLogger("org.rhq").setLevel(Level.DEBUG); - EntityManagerFactory entityManagerFactory = null; - EntityManager entityManager = null; - ExistingDataBulkExportSource source = null; - try { - entityManagerFactory = createEntityManager(); - entityManager = entityManagerFactory.createEntityManager(); - source = new ExistingPostgresDataBulkExportSource( - entityManager, - "SELECT schedule_id, time_stamp, value, minvalue, maxvalue FROM RHQ_MEASUREMENT_DATA_NUM_1D", - // "SELECT count(*) FROM RHQ_MEASUREMENT_DATA_NUM_1D limit 1" - new File(System.getProperty("java.io.tmpdir")), "poupoupidou.txt"); - StopWatch stopWatch = new StopWatch(); - stopWatch.start(); - source.exportExistingData(); - source.startReading(); - int rowIndex = 0; - int maxResults = 30000; - for (;;) { - List<Object[]> existingData = source.getExistingData(rowIndex, maxResults); - if (existingData.size() < maxResults) { - break; - } else { - rowIndex += maxResults; - } - } - stopWatch.stop(); - System.out.println("Execution: " + stopWatch); - } finally { - if (source != null) { - source.stopReading(); - } - if (entityManager != null) { - entityManager.close(); - } - if (entityManagerFactory != null) { - entityManagerFactory.close(); - } - } - } - - private static EntityManagerFactory createEntityManager() throws Exception { - Properties properties = new Properties(); - properties.put("javax.persistence.provider", "org.hibernate.ejb.HibernatePersistence"); - properties.put("hibernate.connection.username", "rhqadmin"); - properties.put("hibernate.connection.password", "rhqadmin"); - properties.put("hibernate.dialect", "org.hibernate.dialect.PostgreSQLDialect"); - String driverClassName = "org.postgresql.Driver"; - try { - //Required to preload the driver manually. - //Without this the driver load will fail due to the packaging. - Class.forName(driverClassName); - } catch (ClassNotFoundException e) { - throw new Exception("Postgres SQL Driver class could not be loaded. Missing class: " + driverClassName); - } - properties.put("hibernate.driver_class", driverClassName); - properties.put("hibernate.connection.url", "jdbc:postgresql://localhost:5432/rhq"); - Ejb3Configuration configuration = new Ejb3Configuration(); - configuration.setProperties(properties); - return configuration.buildEntityManagerFactory(); - } - -} diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/DataMigrator.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/DataMigrator.java new file mode 100644 index 0000000..bcddbd8 --- /dev/null +++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/DataMigrator.java @@ -0,0 +1,730 @@ +/* + * RHQ Management Platform + * Copyright 2011, Red Hat Middleware LLC, and individual contributors + * as indicated by the @author tags. See the copyright.txt file in the + * distribution for a full listing of individual contributors. + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation version 2 of the License. + * + * 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 for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA. + */ + +package org.rhq.server.metrics.migrator; + +import static com.datastax.driver.core.querybuilder.QueryBuilder.ttl; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Date; +import java.util.LinkedList; +import java.util.List; +import java.util.Queue; + +import javax.persistence.EntityManager; +import javax.persistence.Query; + +import com.datastax.driver.core.ResultSetFuture; +import com.datastax.driver.core.Session; +import com.datastax.driver.core.querybuilder.Batch; +import com.datastax.driver.core.querybuilder.QueryBuilder; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.rhq.server.metrics.domain.AggregateType; +import org.rhq.server.metrics.domain.MetricsTable; + + +/** + * @author Stefan Negrea + * + */ +public class DataMigrator { + + public enum DatabaseType { + Postgres, Oracle + } + + private final Log log = LogFactory.getLog(DataMigrator.class); + + + private static final int MAX_RECORDS_TO_LOAD_FROM_SQL = 30000; + private static final int MAX_RAW_BATCH_TO_CASSANDRA = 100; + private static final int MAX_AGGREGATE_BATCH_TO_CASSANDRA = 50; + private static final int MAX_NUMBER_OF_FAILURES = 5; + private static final long NUMBER_OF_BATCHES_FOR_ESTIMATION = 4; + private static final double UNDER_ESTIMATION_FACTOR = .10; + + + private enum MigrationQuery { + SELECT_1H_DATA("SELECT schedule_id, time_stamp, value, minvalue, maxvalue FROM RHQ_MEASUREMENT_DATA_NUM_1H"), + SELECT_6H_DATA("SELECT schedule_id, time_stamp, value, minvalue, maxvalue FROM RHQ_MEASUREMENT_DATA_NUM_6H"), + SELECT_1D_DATA("SELECT schedule_id, time_stamp, value, minvalue, maxvalue FROM RHQ_MEASUREMENT_DATA_NUM_1D"), + + DELETE_1H_DATA("DELETE FROM RHQ_MEASUREMENT_DATA_NUM_1H"), + DELETE_6H_DATA("DELETE FROM RHQ_MEASUREMENT_DATA_NUM_6H"), + DELETE_1D_DATA("DELETE FROM RHQ_MEASUREMENT_DATA_NUM_1D"), + + + COUNT_1H_DATA("SELECT COUNT(*) FROM RHQ_MEASUREMENT_DATA_NUM_1H"), + COUNT_6H_DATA("SELECT COUNT(*) FROM RHQ_MEASUREMENT_DATA_NUM_6H"), + COUNT_1D_DATA("SELECT COUNT(*) FROM RHQ_MEASUREMENT_DATA_NUM_1D"), + + COUNT_RAW("SELECT COUNT(*) FROM %s"), + SELECT_RAW_DATA("SELECT schedule_id, time_stamp, value FROM %s"), + DELETE_RAW_ALL_DATA("DELETE FROM %s"), + DELETE_RAW_ENTRY("DELETE FROM %s WHERE schedule_id = ?"); + + public static final int SCHEDULE_INDEX = 0; + public static final int TIMESTAMP_INDEX = 1; + public static final int VALUE_INDEX = 2; + public static final int MIN_VALUE_INDEX = 3; + public static final int MAX_VALUE_INDEX = 4; + + private String query; + + private MigrationQuery(String query){ + this.query = query; + } + + /** + * @return the query + */ + public String getQuery() { + return query; + } + + @Override + public String toString() { + return query; + } + } + + private final EntityManager entityManager; + private final Session session; + private final DatabaseType databaseType; + private final boolean experimentalDataSource; + + private boolean deleteDataImmediatelyAfterMigration; + private boolean deleteAllDataAtEndOfMigration; + + private boolean runRawDataMigration; + private boolean run1HAggregateDataMigration; + private boolean run6HAggregateDataMigration; + private boolean run1DAggregateDataMigration; + + private long estimation; + + public DataMigrator(EntityManager entityManager, Session session, DatabaseType databaseType) { + this(entityManager, session, databaseType, false); + } + + public DataMigrator(EntityManager entityManager, Session session, DatabaseType databaseType, + boolean experimentalDataSource) { + this.entityManager = entityManager; + this.session = session; + this.databaseType = databaseType; + + this.experimentalDataSource = experimentalDataSource; + + this.deleteDataImmediatelyAfterMigration = false; + this.deleteAllDataAtEndOfMigration = false; + this.runRawDataMigration = true; + this.run1HAggregateDataMigration = true; + this.run6HAggregateDataMigration = true; + this.run1DAggregateDataMigration = true; + } + + public void runRawDataMigration(boolean value) { + this.runRawDataMigration = value; + } + + public void run1HAggregateDataMigration(boolean value) { + this.run1HAggregateDataMigration = value; + } + + public void run6HAggregateDataMigration(boolean value) { + this.run6HAggregateDataMigration = value; + } + + public void run1DAggregateDataMigration(boolean value) { + this.run1DAggregateDataMigration = value; + } + + + public void deleteDataImmediatelyAfterMigration() { + this.deleteDataImmediatelyAfterMigration = true; + this.deleteAllDataAtEndOfMigration = false; + } + + public void deleteAllDataAtEndOfMigration() { + this.deleteAllDataAtEndOfMigration = true; + this.deleteDataImmediatelyAfterMigration = false; + } + + public void preserveData() { + this.deleteAllDataAtEndOfMigration = false; + this.deleteDataImmediatelyAfterMigration = false; + } + + public long estimate() throws Exception { + this.estimation = 0; + if (runRawDataMigration) { + retryOnFailure(new RawDataMigrator(), Task.Estimate); + } + + if (run1HAggregateDataMigration) { + retryOnFailure(new AggregateDataMigrator(MetricsTable.ONE_HOUR), Task.Estimate); + } + + if (run6HAggregateDataMigration) { + retryOnFailure(new AggregateDataMigrator(MetricsTable.SIX_HOUR), Task.Estimate); + } + + if (run1DAggregateDataMigration) { + retryOnFailure(new AggregateDataMigrator(MetricsTable.TWENTY_FOUR_HOUR), Task.Estimate); + } + + if (deleteAllDataAtEndOfMigration) { + retryOnFailure(new DeleteAllData(), Task.Estimate); + } + + estimation = (long) (estimation + estimation * UNDER_ESTIMATION_FACTOR); + + return estimation; + } + + public void migrateData() throws Exception { + if (runRawDataMigration) { + retryOnFailure(new RawDataMigrator(), Task.Migrate); + } + + if (run1HAggregateDataMigration) { + retryOnFailure(new AggregateDataMigrator(MetricsTable.ONE_HOUR), Task.Migrate); + } + + if (run6HAggregateDataMigration) { + retryOnFailure(new AggregateDataMigrator(MetricsTable.SIX_HOUR), Task.Migrate); + } + + if (run1DAggregateDataMigration) { + retryOnFailure(new AggregateDataMigrator(MetricsTable.TWENTY_FOUR_HOUR), Task.Migrate); + } + + if (deleteAllDataAtEndOfMigration) { + retryOnFailure(new DeleteAllData(), Task.Migrate); + } + } + + public void deleteOldData() throws Exception { + if (deleteAllDataAtEndOfMigration) { + retryOnFailure(new DeleteAllData(), Task.Migrate); + } + } + + /** + * Retries the migration {@link #MAX_NUMBER_OF_FAILURES} times before + * failing the migration operation. + * + * @param migrator + * @throws Exception + */ + private Thread retryOnFailure(final CallableMigrationWorker migrator, final Task task) + throws Exception { + + RunnableWithException runnable = new RunnableWithException() { + private Exception exception; + + @Override + public void run() { + int numberOfFailures = 0; + Exception caughtException = null; + + log.info(migrator.getClass()); + + while (numberOfFailures < MAX_NUMBER_OF_FAILURES) { + try { + if (task == Task.Estimate) { + estimation += migrator.estimate(); + } else { + migrator.migrate(); + } + return; + } catch (Exception e) { + log.error("Migrator " + migrator.getClass() + " failed. Retrying!", e); + + caughtException = e; + numberOfFailures++; + } + } + + this.exception = caughtException; + } + + @Override + public Exception getException() { + return this.exception; + } + }; + + Thread localThread = new Thread(runnable); + localThread.start(); + localThread.join(); + + if (runnable.getException() != null) { + throw runnable.getException(); + } + + return localThread; + } + + /** + * Returns a list of all the raw SQL metric tables. + * There is no equivalent in Cassandra, all raw data is stored in a single column family. + * + * @return SQL raw metric tables + */ + private String[] getRawDataTables() { + int tableCount = 15; + String tablePrefix = "RHQ_MEAS_DATA_NUM_R"; + + String[] tables = new String[tableCount]; + for (int i = 0; i < tableCount; i++) { + if (i < 10) { + tables[i] = tablePrefix + "0" + i; + } else { + tables[i] = tablePrefix + i; + } + } + + return tables; + } + + private ExistingDataSource getExistingDataSource(EntityManager entityManager, String query) { + if (this.databaseType == DatabaseType.Oracle) { + return new ExistingDataJPASource(entityManager, query); + } else { + if (!experimentalDataSource) { + return new ExistingDataJPASource(entityManager, query); + } else { + return new ExistingPostgresDataBulkExportSource(entityManager, query); + } + } + + //return new ExistingDataJPASource(entityManager, query); + } + + private enum Task { + Migrate, Estimate + } + + private interface CallableMigrationWorker { + + + long estimate() throws Exception; + + void migrate() throws Exception; + } + + private interface RunnableWithException extends Runnable { + Exception getException(); + } + + private class AggregateDataMigrator implements CallableMigrationWorker { + + private final String selectQuery; + private final String deleteQuery; + private final String countQuery; + private final MetricsTable metricsTable; + + /** + * @param query + * @param metricsTable + */ + public AggregateDataMigrator(MetricsTable metricsTable) throws Exception { + this.metricsTable = metricsTable; + + if (MetricsTable.ONE_HOUR.equals(this.metricsTable)) { + this.selectQuery = MigrationQuery.SELECT_1H_DATA.toString(); + this.deleteQuery = MigrationQuery.DELETE_1H_DATA.toString(); + this.countQuery = MigrationQuery.COUNT_1H_DATA.toString(); + } else if (MetricsTable.SIX_HOUR.equals(this.metricsTable)) { + this.selectQuery = MigrationQuery.SELECT_6H_DATA.toString(); + this.deleteQuery = MigrationQuery.DELETE_6H_DATA.toString(); + this.countQuery = MigrationQuery.COUNT_6H_DATA.toString(); + } else if (MetricsTable.TWENTY_FOUR_HOUR.equals(this.metricsTable)) { + this.selectQuery = MigrationQuery.SELECT_1D_DATA.toString(); + this.deleteQuery = MigrationQuery.DELETE_1D_DATA.toString(); + this.countQuery = MigrationQuery.COUNT_1D_DATA.toString(); + } else { + throw new Exception("MetricsTable " + metricsTable.toString() + " not supported by this migrator."); + } + } + + @Override + public long estimate() throws Exception { + Query nativeQuery = entityManager.createNativeQuery(this.countQuery); + long recordCount = Long.parseLong(nativeQuery.getSingleResult().toString()); + long estimatedTimeToMigrate = this.performMigration(Task.Estimate); + + long estimation = (recordCount / (long) MAX_RECORDS_TO_LOAD_FROM_SQL / NUMBER_OF_BATCHES_FOR_ESTIMATION) + * estimatedTimeToMigrate; + return estimation; + } + + public void migrate() throws Exception { + performMigration(Task.Migrate); + if (deleteDataImmediatelyAfterMigration) { + deleteTableData(); + } + } + + private void deleteTableData() throws Exception { + int failureCount = 0; + while (failureCount < MAX_NUMBER_OF_FAILURES) { + try { + entityManager.getTransaction().begin(); + Query nativeQuery = entityManager.createNativeQuery(this.deleteQuery); + nativeQuery.executeUpdate(); + entityManager.getTransaction().commit(); + log.info("- " + metricsTable.toString() + " - Cleaned -"); + } catch (Exception e) { + log.error("Failed to delete " + metricsTable.toString() + + " data. Attempting to delete data one more time..."); + + failureCount++; + if (failureCount == MAX_NUMBER_OF_FAILURES) { + throw e; + } + } + } + } + + private long performMigration(Task task) throws Exception { + long migrationStartTime = System.currentTimeMillis(); + long numberOfBatchesMigrated = 0; + + List<Object[]> existingData; + int failureCount; + + int lastMigratedRecord = 0; + ExistingDataSource dataSource = getExistingDataSource(entityManager, selectQuery); + dataSource.initialize(); + + while (true) { + existingData = dataSource.getData(lastMigratedRecord, MAX_RECORDS_TO_LOAD_FROM_SQL); + + if (existingData.size() == 0) { + break; + } + + lastMigratedRecord += existingData.size(); + + failureCount = 0; + while (failureCount < MAX_NUMBER_OF_FAILURES) { + try { + insertDataToCassandra(existingData); + break; + } catch (Exception e) { + log.error("Failed to insert " + metricsTable.toString() + + " data. Attempting to insert the current batch of data one more time"); + log.error(e); + + failureCount++; + if (failureCount == MAX_NUMBER_OF_FAILURES) { + throw e; + } + } + } + + log.info("- " + metricsTable + " - " + lastMigratedRecord + " -"); + + numberOfBatchesMigrated++; + if (Task.Estimate.equals(task) && numberOfBatchesMigrated >= NUMBER_OF_BATCHES_FOR_ESTIMATION) { + break; + } + } + + dataSource.close(); + + return System.currentTimeMillis() - migrationStartTime; + } + + private void insertDataToCassandra(List<Object[]> existingData) + throws Exception { + List<ResultSetFuture> resultSetFutures = new ArrayList<ResultSetFuture>(); + Batch batch = QueryBuilder.batch(); + int batchSize = 0; + + //only need approximate TTL to speed up processing + //given that each batch is processed within seconds, getting the + //system time once per batch has minimal impact on the record retention + long creationTimeMillis; + long itemTTLSeconds; + long currentTimeMillis = System.currentTimeMillis(); + long expectedTTLMillis = metricsTable.getTTLinMilliseconds() * 10; + + + for (Object[] rawMeasurement : existingData) { + creationTimeMillis = Long.parseLong(rawMeasurement[MigrationQuery.TIMESTAMP_INDEX].toString()); + itemTTLSeconds = (expectedTTLMillis - currentTimeMillis + creationTimeMillis) / 1000l; + + if(itemTTLSeconds > 0 ){ + batch.add(QueryBuilder.insertInto(metricsTable.toString()) + .value("schedule_id",Integer.parseInt(rawMeasurement[MigrationQuery.SCHEDULE_INDEX].toString())) + .value("time", new Date(creationTimeMillis)) + .value("type", AggregateType.AVG.ordinal()) + .value("value", Double.parseDouble(rawMeasurement[MigrationQuery.VALUE_INDEX].toString())) + .using(ttl((int) itemTTLSeconds))); + + batch.add(QueryBuilder.insertInto(metricsTable.toString()) + .value("schedule_id", Integer.parseInt(rawMeasurement[MigrationQuery.SCHEDULE_INDEX].toString())) + .value("time", new Date(creationTimeMillis)) + .value("type", AggregateType.MIN.ordinal()) + .value("value", Double.parseDouble(rawMeasurement[MigrationQuery.MIN_VALUE_INDEX].toString())) + .using(ttl((int) itemTTLSeconds))); + + batch.add(QueryBuilder.insertInto(metricsTable.toString()) + .value("schedule_id", Integer.parseInt(rawMeasurement[MigrationQuery.SCHEDULE_INDEX].toString())) + .value("time", new Date(creationTimeMillis)) + .value("type", AggregateType.MAX.ordinal()) + .value("value", Double.parseDouble(rawMeasurement[MigrationQuery.MAX_VALUE_INDEX].toString())) + .using(ttl((int) itemTTLSeconds))); + + batchSize += 3; + } + + if (batchSize >= MAX_AGGREGATE_BATCH_TO_CASSANDRA) { + resultSetFutures.add(session.executeAsync(batch)); + batch = QueryBuilder.batch(); + batchSize = 0; + } + } + + if (batchSize != 0) { + resultSetFutures.add(session.executeAsync(batch)); + } + + for (ResultSetFuture future : resultSetFutures) { + future.get(); + } + } + } + + + private class RawDataMigrator implements CallableMigrationWorker { + + Queue<String> tablesNotProcessed = new LinkedList<String>(Arrays.asList(getRawDataTables())); + + public long estimate() throws Exception { + long recordCount = 0; + for (String table : getRawDataTables()) { + String countQuery = String.format(MigrationQuery.COUNT_RAW.toString(), table); + Query nativeQuery = entityManager.createNativeQuery(countQuery); + recordCount += Long.parseLong(nativeQuery.getSingleResult().toString()); + } + + long estimatedTimeToMigrate = this.performMigration(Task.Estimate); + long estimation = (recordCount / (long) MAX_RECORDS_TO_LOAD_FROM_SQL / NUMBER_OF_BATCHES_FOR_ESTIMATION) + * estimatedTimeToMigrate; + return estimation; + } + + public void migrate() throws Exception { + performMigration(Task.Migrate); + } + + private long performMigration(Task task) throws Exception { + long migrationStartTime = System.currentTimeMillis(); + long numberOfBatchesMigrated = 0; + + List<Object[]> existingData; + int failureCount; + + while (!tablesNotProcessed.isEmpty()) { + String table = tablesNotProcessed.peek(); + + String selectQuery = String.format(MigrationQuery.SELECT_RAW_DATA.toString(), table); + + ExistingDataSource dataSource = getExistingDataSource(entityManager, selectQuery); + dataSource.initialize(); + + log.info("Start migrating raw table: " + table); + + int lastMigratedRecord = 0; + while (true) { + existingData = dataSource.getData(lastMigratedRecord, MAX_RECORDS_TO_LOAD_FROM_SQL); + + if (existingData.size() == 0) { + break; + } + + lastMigratedRecord += existingData.size(); + + failureCount = 0; + while (failureCount < MAX_NUMBER_OF_FAILURES) { + try { + insertDataToCassandra(existingData); + break; + } catch (Exception e) { + log.error("Failed to insert " + MetricsTable.RAW.toString() + + " data. Attempting to insert the current batch of data one more time"); + log.error(e); + + + failureCount++; + if (failureCount == MAX_AGGREGATE_BATCH_TO_CASSANDRA) { + throw e; + } + } + } + + log.info("- " + table + " - " + lastMigratedRecord + " -"); + + numberOfBatchesMigrated++; + if (Task.Estimate.equals(task) && numberOfBatchesMigrated >= NUMBER_OF_BATCHES_FOR_ESTIMATION) { + break; + } + } + + if (Task.Migrate.equals(task)) { + log.info("Done migrating raw table" + table + "---------------------"); + + if (deleteDataImmediatelyAfterMigration) { + deleteTableData(table); + } + } else if (numberOfBatchesMigrated >= NUMBER_OF_BATCHES_FOR_ESTIMATION) { + break; + } + + dataSource.close(); + tablesNotProcessed.poll(); + } + + return System.currentTimeMillis() - migrationStartTime; + } + + private void deleteTableData(String table) throws Exception { + String deleteQuery = String.format(MigrationQuery.DELETE_RAW_ENTRY.toString(), table); + int failureCount = 0; + while (failureCount < MAX_NUMBER_OF_FAILURES) { + try { + entityManager.getTransaction().begin(); + Query nativeQuery = entityManager.createNativeQuery(deleteQuery); + nativeQuery.executeUpdate(); + entityManager.getTransaction().commit(); + log.info("- " + table + " - Cleaned -"); + } catch (Exception e) { + log.error("Failed to delete " + table + " data. Attempting to delete data one more time..."); + + failureCount++; + if (failureCount == MAX_NUMBER_OF_FAILURES) { + throw e; + } + } + } + } + + private void insertDataToCassandra(List<Object[]> existingData) throws Exception { + List<ResultSetFuture> resultSetFutures = new ArrayList<ResultSetFuture>(); + Batch batch = QueryBuilder.batch(); + int batchSize = 0; + + //only need approximate TTL to speed up processing + //given that each batch is processed within seconds, getting the + //system time once per batch has minimal impact on the record retention + long creationTimeMillis; + long itemTTLSeconds; + long currentTimeMillis = System.currentTimeMillis(); + long expectedTTLMillis = MetricsTable.RAW.getTTLinMilliseconds() * 10; + + + for (Object[] rawDataPoint : existingData) { + creationTimeMillis = Long.parseLong(rawDataPoint[MigrationQuery.TIMESTAMP_INDEX].toString()); + itemTTLSeconds = (expectedTTLMillis - currentTimeMillis + creationTimeMillis) / 1000l; + + if (itemTTLSeconds > 0) { + batch.add(QueryBuilder.insertInto(MetricsTable.RAW.toString()) + .value("schedule_id", Integer.parseInt(rawDataPoint[MigrationQuery.SCHEDULE_INDEX].toString())) + .value("time", new Date(creationTimeMillis)) + .value("value", Double.parseDouble(rawDataPoint[MigrationQuery.VALUE_INDEX].toString())) + .using(ttl((int) itemTTLSeconds))); + batchSize++; + } + + if (batchSize >= MAX_RAW_BATCH_TO_CASSANDRA) { + resultSetFutures.add(session.executeAsync(batch)); + batch = QueryBuilder.batch(); + batchSize = 0; + } + } + + if (batchSize != 0) { + resultSetFutures.add(session.executeAsync(batch)); + } + + for (ResultSetFuture future : resultSetFutures) { + future.get(); + } + } + } + + + private class DeleteAllData implements CallableMigrationWorker { + + public void migrate() { + Query nativeQuery; + + if (run1HAggregateDataMigration) { + entityManager.getTransaction().begin(); + nativeQuery = entityManager.createNativeQuery(MigrationQuery.DELETE_1H_DATA.toString()); + nativeQuery.executeUpdate(); + entityManager.getTransaction().commit(); + log.info("- RHQ_MEASUREMENT_DATA_NUM_1H - Cleaned -"); + } + + if (run6HAggregateDataMigration) { + entityManager.getTransaction().begin(); + nativeQuery = entityManager.createNativeQuery(MigrationQuery.DELETE_6H_DATA.toString()); + nativeQuery.executeUpdate(); + entityManager.getTransaction().commit(); + log.info("- RHQ_MEASUREMENT_DATA_NUM_6H - Cleaned -"); + } + + if (run1DAggregateDataMigration) { + entityManager.getTransaction().begin(); + nativeQuery = entityManager.createNativeQuery(MigrationQuery.DELETE_1D_DATA.toString()); + nativeQuery.executeUpdate(); + entityManager.getTransaction().commit(); + log.info("- RHQ_MEASUREMENT_DATA_NUM_1D - Cleaned -"); + } + + if (runRawDataMigration) { + for (String table : getRawDataTables()) { + entityManager.getTransaction().begin(); + String deleteAllData = String.format(MigrationQuery.DELETE_RAW_ALL_DATA.toString(), table); + nativeQuery = entityManager.createNativeQuery(deleteAllData); + nativeQuery.executeUpdate(); + entityManager.getTransaction().commit(); + log.info("- " + table + " - Cleaned -"); + } + } + } + + @Override + public long estimate() throws Exception { + return 300000; // return return 5 minutes for now without any database side checks. + } + } +} + + diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/DataMigratorRunner.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/DataMigratorRunner.java new file mode 100644 index 0000000..adcd1c0 --- /dev/null +++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/DataMigratorRunner.java @@ -0,0 +1,668 @@ +/* + * RHQ Management Platform + * Copyright 2011, Red Hat Middleware LLC, and individual contributors + * as indicated by the @author tags. See the copyright.txt file in the + * distribution for a full listing of individual contributors. + * + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License as published by + * the Free Software Foundation version 2 of the License. + * + * 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 for more details. + * + * You should have received a copy of the GNU General Public License + * along with this program; if not, write to the Free Software + * Foundation, Inc., 675 Mass Ave, Cambridge, MA 02139, USA. + */ + +package org.rhq.server.metrics.migrator; + +import java.io.File; +import java.io.FileInputStream; +import java.io.FileNotFoundException; +import java.util.HashMap; +import java.util.Map; +import java.util.Properties; +import java.util.concurrent.TimeUnit; + +import javax.persistence.EntityManager; +import javax.persistence.EntityManagerFactory; + +import com.datastax.driver.core.Cluster; +import com.datastax.driver.core.ProtocolOptions.Compression; +import com.datastax.driver.core.Session; +import com.datastax.driver.core.SimpleAuthInfoProvider; +import com.datastax.driver.core.exceptions.NoHostAvailableException; + +import org.apache.commons.cli.CommandLine; +import org.apache.commons.cli.CommandLineParser; +import org.apache.commons.cli.HelpFormatter; +import org.apache.commons.cli.Option; +import org.apache.commons.cli.OptionBuilder; +import org.apache.commons.cli.Options; +import org.apache.commons.cli.ParseException; +import org.apache.commons.cli.PosixParser; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.log4j.ConsoleAppender; +import org.apache.log4j.Level; +import org.apache.log4j.Logger; +import org.apache.log4j.PatternLayout; +import org.hibernate.ejb.Ejb3Configuration; + +import org.rhq.server.metrics.migrator.DataMigrator.DatabaseType; + + +/** + * @author Stefan Negrea + * + * Only postgres is supported by the runner, however the data migrator itself can run + * with any database. + * + * Maven command to run this from the command line: + * + * mvn install -DskipTests exec:java -Dexec.mainClass="org.rhq.server.metrics.DataMigratorRunner" + * + * + */ +@SuppressWarnings({ "static-access", "deprecation" }) +public class DataMigratorRunner { + + private final Log log = LogFactory.getLog(DataMigratorRunner.class); + + //Cassandra + private Option cassandraUserOption = OptionBuilder.withLongOpt("cassandra-user").hasArg().withType(String.class) + .withDescription("Cassandra user (default: rhqadmin)").create(); + private Option cassandraPasswordOption = OptionBuilder.withLongOpt("cassandra-password").hasArg() + .withDescription("Cassandra password (default: rhqadmin)").withType(String.class).create(); + private Option cassandraHostsOption = OptionBuilder.withLongOpt("cassandra-hosts").hasArg().withType(String.class) + .withDescription("Cassandra hosts, format host_ip_1,host_ip_2,... (default: 127.0.0.1") + .create(); + private Option cassandraPortOption = OptionBuilder.withLongOpt("cassandra-port").hasArg().withType(Integer.class) + .withDescription("Cassandra native binary protocol port (default: 9142)").create(); + private Option cassandraCompressionOption = OptionBuilder.withLongOpt("cassandra-compression").hasOptionalArg() + .withType(String.class).withDescription("Enable compression for communication with Cassandra (default: true)") + .create(); + + //SQL + private Option sqlUserOption = OptionBuilder.withLongOpt("sql-user").hasArg().withType(String.class) + .withDescription("SQL server user (default: rhqadmin)").create(); + private Option sqlPasswordOption = OptionBuilder.withLongOpt("sql-password").hasArg().withType(String.class) + .withDescription("SQL server password (default: rhqadmin)").create(); + private Option sqlHostOption = OptionBuilder.withLongOpt("sql-host").hasArg().withType(String.class) + .withDescription("SQL server host address (default: localhost)").create(); + private Option sqlPortOption = OptionBuilder.withLongOpt("sql-port").hasArg().withType(String.class) + .withDescription("SQL server port (default: 5432)").create(); + private Option sqlDBOption = OptionBuilder.withLongOpt("sql-db").hasArg().withType(String.class) + .withDescription("SQL database (default: rhq)").create(); + + private Option sqlServerType = OptionBuilder.withLongOpt("sql-server-type").hasArg().withType(String.class) + .withDescription("SQL server type, only postgres and oracle are supported (default: postgres)").create(); + private Option sqlPostgresServer = OptionBuilder.withLongOpt("sql-server-postgres").hasOptionalArg() + .withType(Boolean.class).withDescription("Postgres SQL server.").create(); + private Option sqlOracleServer = OptionBuilder.withLongOpt("sql-server-oracle").hasOptionalArg() + .withType(Boolean.class) + .withDescription("Oracle SQL server.").create(); + + //Migration + private Option disableRawOption = OptionBuilder.withLongOpt("disable-raw-migration").hasOptionalArg().withType(Boolean.class) + .withDescription("Disable raw table migration (default: false)").create(); + private Option disable1HOption = OptionBuilder.withLongOpt("disable-1h-migration").hasOptionalArg().withType(Boolean.class) + .withDescription("Disable 1 hour aggregates table migration (default: false)").create(); + private Option disable6HOption = OptionBuilder.withLongOpt("disable-6h-migration").hasOptionalArg().withType(Boolean.class) + .withDescription("Disable 6 hours aggregates table migration (default: false)").create(); + private Option disable1DOption = OptionBuilder.withLongOpt("disable-1d-migration").hasOptionalArg().withType(Boolean.class) + .withDescription("Disable 24 hours aggregates table migration (default: false)").create(); + private Option preserveDataOption = OptionBuilder.withLongOpt("preserve-data").hasOptionalArg().withType(Boolean.class) + .withDescription("Preserve SQL data post migration (default: true)").create(); + private Option deleteDataOption = OptionBuilder.withLongOpt("delete-data").hasOptionalArg().withType(Boolean.class) + .withDescription("Delete SQL data at the end of migration (default: false)").create(); + private Option estimateOnlyOption = OptionBuilder.withLongOpt("estimate-only").hasOptionalArg().withType(Boolean.class) + .withDescription("Only estimate how long the migration will take (default: false)").create(); + private Option deleteOnlyOption = OptionBuilder.withLongOpt("delete-only").hasOptionalArg().withType(Boolean.class) + .withDescription("Only delete data from the old SQL server, no migration will be performed (default: false)") + .create(); + private Option experimentalExportOption = OptionBuilder.withLongOpt("experimental-export").hasOptionalArg().withType(Boolean.class) + .withDescription("Enable experimental bulk export for Postgres, option ignored for Oracle migration (default: false)") + .create(); + + //Runner + private Option helpOption = OptionBuilder.withLongOpt("help").create("h"); + private Option debugLogOption = OptionBuilder.withLongOpt("debugLog") + .withDescription("Enable debug level logs for the communication with Cassandra and SQL Server (default: false)") + .create("X"); + private Option configFileOption = OptionBuilder.withLongOpt("config-file").hasArg() + .withDescription("Configuration file. All the command line options can be set in a typical properties file. " + + "Command line arguments take precedence over default and configuration file options.") + .create(); + + private Map<Object, Object> configuration = new HashMap<Object, Object>(); + private Options options; + + /** + * @param args + * @throws ParseException + */ + public static void main(String[] args) throws Exception { + initLogging(); + try{ + DataMigratorRunner runner = new DataMigratorRunner(); + runner.configure(args); + runner.run(); + } catch (HelpRequestedException h) { + //do nothing + } catch (Exception e) { + e.printStackTrace(); + System.exit(1); + } + + System.exit(0); + } + + private static void initLogging() { + Logger root = Logger.getRootLogger(); + if (!root.getAllAppenders().hasMoreElements()) { + root.addAppender(new ConsoleAppender(new PatternLayout(PatternLayout.TTCC_CONVERSION_PATTERN))); + setLogLevel(Level.ERROR); + } + } + + private static void setLogLevel(Level level) { + Logger root = Logger.getRootLogger(); + root.setLevel(level); + + Logger cassandraLogging = root.getLoggerRepository().getLogger("log4j.logger.org.apache.cassandra.cql.jdbc"); + cassandraLogging.setLevel(level); + + Logger cassandraDriverLogging = root.getLoggerRepository().getLogger("com.datastax.driver"); + cassandraDriverLogging.setLevel(level); + + Logger hibernateLogging = root.getLoggerRepository().getLogger("org.hibernate"); + hibernateLogging.setLevel(Level.ERROR); + + Logger migratorLogging = root.getLoggerRepository().getLogger("org.rhq"); + if (Level.DEBUG.equals(level)) { + migratorLogging.setLevel(Level.ALL); + } else { + migratorLogging.setLevel(level); + } + } + + private void configure(String args[]) throws Exception { + options = new Options(); + + options.addOption(cassandraUserOption); + options.addOption(cassandraPasswordOption); + options.addOption(cassandraHostsOption); + options.addOption(cassandraPortOption); + options.addOption(cassandraCompressionOption); + + options.addOption(sqlUserOption); + options.addOption(sqlPasswordOption); + options.addOption(sqlHostOption); + options.addOption(sqlPortOption); + options.addOption(sqlDBOption); + options.addOption(sqlServerType); + options.addOption(sqlPostgresServer); + options.addOption(sqlOracleServer); + + options.addOption(disableRawOption); + options.addOption(disable1HOption); + options.addOption(disable6HOption); + options.addOption(disable1DOption); + options.addOption(preserveDataOption); + options.addOption(deleteDataOption); + options.addOption(estimateOnlyOption); + options.addOption(deleteOnlyOption); + options.addOption(experimentalExportOption); + + options.addOption(helpOption); + options.addOption(debugLogOption); + options.addOption(configFileOption); + + CommandLine commandLine; + try { + CommandLineParser parser = new PosixParser(); + commandLine = parser.parse(options, args); + } catch (Exception e) { + HelpFormatter formatter = new HelpFormatter(); + formatter.setWidth(120); + formatter.printHelp("DataMigrationRunner", options); + throw new Exception("Error parsing command line arguments"); + } + + if (commandLine.hasOption(helpOption.getLongOpt()) || commandLine.hasOption(helpOption.getOpt())) { + HelpFormatter formatter = new HelpFormatter(); + formatter.printHelp("DataMigrationRunner", options); + throw new HelpRequestedException(); + } + + if (commandLine.hasOption(debugLogOption.getLongOpt()) || commandLine.hasOption(debugLogOption.getOpt())) { + DataMigratorRunner.setLogLevel(Level.DEBUG); + } + + loadDefaultConfiguration(); + if (commandLine.hasOption(configFileOption.getLongOpt())) { + loadConfigFile(commandLine.getOptionValue(configFileOption.getLongOpt())); + } + + parseCassandraOptions(commandLine); + parseSQLOptions(commandLine); + parseMigrationOptions(commandLine); + } + + /** + * Add default configuration options to the configuration store. + */ + private void loadDefaultConfiguration() { + //default Cassandra configuration + configuration.put(cassandraUserOption, "rhqadmin"); + configuration.put(cassandraPasswordOption, "rhqadmin"); + configuration.put(cassandraHostsOption, new String[] { "127.0.0.1" }); + configuration.put(cassandraPortOption, 9042); + configuration.put(cassandraCompressionOption, true); + + //default SQL configuration + configuration.put(sqlUserOption, "rhqadmin"); + configuration.put(sqlPasswordOption, "rhqadmin"); + configuration.put(sqlHostOption, "localhost"); + configuration.put(sqlPortOption, "5432"); + configuration.put(sqlDBOption, "rhq"); + configuration.put(sqlServerType, "postgres"); + + //default runner options + configuration.put(disableRawOption, false); + configuration.put(disable1HOption, false); + configuration.put(disable6HOption, false); + configuration.put(disable1DOption, false); + configuration.put(preserveDataOption, true); + configuration.put(estimateOnlyOption, false); + configuration.put(deleteOnlyOption, false); + configuration.put(experimentalExportOption, false); + } + + /** + * Load the configuration options from file and overlay them on top of the default + * options. + * + * @param file config file + */ + private void loadConfigFile(String file) { + try { + File configFile = new File(file); + if (!configFile.exists()) { + throw new FileNotFoundException("Configuration file not found!"); + } + + Properties configProperties = new Properties(); + FileInputStream stream = new FileInputStream(configFile); + configProperties.load(stream); + stream.close(); + + for (Object optionObject : options.getOptions()) { + Option option = (Option) optionObject; + Object optionValue; + + if ((optionValue = configProperties.get(option.getLongOpt())) != null) { + log.debug("Configuration option loaded: " + option.getLongOpt() + " (" + option.getType() + ") -> " + + optionValue); + + if (option.equals(cassandraHostsOption)) { + String[] cassandraHosts = parseCassandraHosts(optionValue.toString()); + configuration.put(option, cassandraHosts); + } else if (option.equals(sqlServerType)) { + if ("oracle".equals(optionValue)) { + configuration.put(option, "oracle"); + } else { + configuration.put(option, "postgres"); + } + } else if (option.equals(sqlPostgresServer)) { + boolean value = tryParseBoolean(optionValue.toString(), true); + if (value == true) { + configuration.put(sqlServerType, "postgres"); + } + } else if (option.equals(sqlOracleServer)) { + boolean value = tryParseBoolean(optionValue.toString(), true); + if (value == true) { + configuration.put(sqlServerType, "oracle"); + } + } else if (option.getType().equals(Boolean.class)) { + configuration.put(option, tryParseBoolean(optionValue.toString(), true)); + } else if (option.getType().equals(Integer.class)) { + configuration.put(option, tryParseInteger(optionValue.toString(), 0)); + } else { + configuration.put(option, optionValue.toString()); + } + } + } + } catch (Exception e) { + log.error("Unable to load or process the configuration file.", e); + System.exit(1); + } + + log.debug(configuration.toString()); + } + + /** + * Parse command line options for Cassandra. + * + * @param commandLine command line + * @throws NoHostAvailableException + */ + private void parseCassandraOptions(CommandLine commandLine) throws Exception { + if (commandLine.hasOption(cassandraUserOption.getLongOpt())) { + configuration.put(cassandraUserOption, commandLine.getOptionValue(cassandraUserOption.getLongOpt())); + } + + if (commandLine.hasOption(cassandraPasswordOption.getLongOpt())) { + configuration + .put(cassandraPasswordOption, commandLine.getOptionValue(cassandraPasswordOption.getLongOpt())); + } + + if (commandLine.hasOption(cassandraHostsOption.getLongOpt())) { + String[] cassandraHosts = parseCassandraHosts(commandLine.getOptionValue(cassandraHostsOption.getLongOpt())); + configuration.put(cassandraHostsOption, cassandraHosts); + } + + if (commandLine.hasOption(cassandraPortOption.getLongOpt())) { + Integer cassandraPort = tryParseInteger(commandLine.getOptionValue(cassandraPortOption.getLongOpt()), 9142); + configuration.put(cassandraPortOption, cassandraPort); + } + + if (commandLine.hasOption(cassandraCompressionOption.getLongOpt())) { + boolean value = tryParseBoolean(commandLine.getOptionValue(disableRawOption.getLongOpt()), true); + configuration.put(cassandraCompressionOption, value); + } + } + + /** + * Parse command line options for SQL. + * + * @param commandLine command line + * @throws NoHostAvailableException + */ + private void parseSQLOptions(CommandLine commandLine) throws NoHostAvailableException { + if (commandLine.hasOption(sqlUserOption.getLongOpt())) { + configuration.put(sqlUserOption, commandLine.getOptionValue(sqlUserOption.getLongOpt())); + } + + if (commandLine.hasOption(sqlPasswordOption.getLongOpt())) { + configuration.put(sqlPasswordOption, commandLine.getOptionValue(sqlPasswordOption.getLongOpt())); + } + + if (commandLine.hasOption(sqlHostOption.getLongOpt())) { + configuration.put(sqlHostOption, commandLine.getOptionValue(sqlHostOption.getLongOpt())); + } + + if (commandLine.hasOption(sqlPortOption.getLongOpt())) { + configuration.put(sqlPortOption, commandLine.getOptionValue(sqlPortOption.getLongOpt())); + } + + if (commandLine.hasOption(sqlDBOption.getLongOpt())) { + configuration.put(sqlDBOption, commandLine.getOptionValue(sqlDBOption.getLongOpt())); + } + + if (commandLine.hasOption(sqlServerType.getLongOpt())) { + if ("oracle".equals(commandLine.getOptionValue(sqlServerType.getLongOpt()))) { + configuration.put(sqlServerType, "oracle"); + } else { + configuration.put(sqlServerType, "postgres"); + } + } else if (commandLine.hasOption(sqlPostgresServer.getLongOpt())) { + configuration.put(sqlServerType, "postgres"); + } else if (commandLine.hasOption(sqlOracleServer.getLongOpt())) { + configuration.put(sqlServerType, "oracle"); + } + } + + /** + * Parse command line options for the actual migration progress. + * + * @param commandLine + */ + private void parseMigrationOptions(CommandLine commandLine) { + boolean value; + + if (commandLine.hasOption(disableRawOption.getLongOpt())) { + value = tryParseBoolean(commandLine.getOptionValue(disableRawOption.getLongOpt()), true); + configuration.put(disableRawOption, value); + } + + if (commandLine.hasOption(disable1HOption.getLongOpt())) { + value = tryParseBoolean(commandLine.getOptionValue(disable1HOption.getLongOpt()), true); + configuration.put(disable1HOption, value); + } + + if (commandLine.hasOption(disable6HOption.getLongOpt())) { + value = tryParseBoolean(commandLine.getOptionValue(disable6HOption.getLongOpt()), true); + configuration.put(disable6HOption, value); + } + + if (commandLine.hasOption(disable1DOption.getLongOpt())) { + value = tryParseBoolean(commandLine.getOptionValue(disable1DOption.getLongOpt()), true); + configuration.put(disable1DOption, value); + } + + if (commandLine.hasOption(preserveDataOption.getLongOpt())) { + value = tryParseBoolean(commandLine.getOptionValue(preserveDataOption.getLongOpt()), true); + configuration.put(preserveDataOption, value); + } else if (commandLine.hasOption(deleteDataOption.getLongOpt())) { + value = tryParseBoolean(commandLine.getOptionValue(deleteDataOption.getLongOpt()), true); + configuration.put(preserveDataOption, value); + } + + if (commandLine.hasOption(estimateOnlyOption.getLongOpt())) { + value = tryParseBoolean(commandLine.getOptionValue(estimateOnlyOption.getLongOpt()), true); + configuration.put(estimateOnlyOption, value); + } + + if (commandLine.hasOption(experimentalExportOption.getLongOpt())) { + value = tryParseBoolean(commandLine.getOptionValue(experimentalExportOption.getLongOpt()), true); + configuration.put(experimentalExportOption, value); + } + } + + private void run() throws Exception { + log.debug("Creating Entity Manager"); + EntityManager entityManager = this.createEntityManager(); + log.debug("Done creating Entity Manager"); + + log.debug("Creating Cassandra session"); + Session session = this.createCassandraSession(); + log.debug("Done creating Cassandra session"); + + DatabaseType databaseType = DatabaseType.Postgres; + if ("oracle".equals(configuration.get(sqlServerType))) { + databaseType = databaseType.Oracle; + } + DataMigrator migrator = new DataMigrator(entityManager, session, databaseType, tryParseBoolean( + configuration.get(experimentalExportOption), false)); + + if (!(Boolean) configuration.get(deleteOnlyOption)) { + if ((Boolean) configuration.get(preserveDataOption)) { + migrator.preserveData(); + } else { + migrator.deleteAllDataAtEndOfMigration(); + } + + migrator.runRawDataMigration(!(Boolean) configuration.get(disableRawOption)); + migrator.run1HAggregateDataMigration(!(Boolean) configuration.get(disable1HOption)); + migrator.run6HAggregateDataMigration(!(Boolean) configuration.get(disable6HOption)); + migrator.run1DAggregateDataMigration(!(Boolean) configuration.get(disable1DOption)); + + System.out.println("Estimation process - starting\n"); + long estimate = migrator.estimate(); + System.out.println("The migration process will take approximately: " + + TimeUnit.MILLISECONDS.toMinutes(estimate) + " minutes (or " + estimate + " milliseconds)\n"); + System.out.println("Estimation process - ended\n\n"); + + if (!(Boolean) configuration.get(estimateOnlyOption)) { + System.out.println("Migration process - starting\n"); + long startTime = System.currentTimeMillis(); + migrator.migrateData(); + long duration = System.currentTimeMillis() - startTime; + System.out.println("The migration process took: " + TimeUnit.MILLISECONDS.toMinutes(duration) + + " minutes (or " + duration + " milliseconds)\n"); + System.out.println("Migration process - ended\n"); + } + } else { + migrator.deleteAllDataAtEndOfMigration(); + migrator.runRawDataMigration(true); + migrator.run1HAggregateDataMigration(true); + migrator.run6HAggregateDataMigration(true); + migrator.run1DAggregateDataMigration(true); + + System.out.println("Estimation process - starting\n"); + long estimate = migrator.estimate(); + System.out.println("The deletion of old data will take approximately: " + + TimeUnit.MILLISECONDS.toMinutes(estimate) + " minutes (or " + estimate + " milliseconds)\n"); + System.out.println("Estimation process - ended\n\n"); + + if (!(Boolean) configuration.get(estimateOnlyOption)) { + System.out.println("Old data deletion process - starting\n"); + long startTime = System.currentTimeMillis(); + migrator.deleteOldData(); + long duration = System.currentTimeMillis() - startTime; + System.out.println("The deletion process took: " + TimeUnit.MILLISECONDS.toMinutes(duration) + + " minutes (or " + duration + " milliseconds)\n"); + System.out.println("Old data deletion process - ended\n"); + } + } + + /*if (entityManager != null) { + entityManager.close(); + } + if (entityManagerFactory != null) { + entityManagerFactory.close(); + }*/ + } + + /** + * Create a Cassandra session based on configuration options. + * + * @return Cassandra session + * @throws Exception + */ + private Session createCassandraSession() throws Exception { + Compression selectedCompression = Compression.NONE; + if ((Boolean) configuration.get(cassandraCompressionOption)) { + selectedCompression = Compression.SNAPPY; + } + + Cluster cluster = Cluster + .builder() + .addContactPoints((String[]) configuration.get(cassandraHostsOption)) + .withPort((Integer) configuration.get(cassandraPortOption)) + .withCompression(selectedCompression) + .withoutMetrics() + .withAuthInfoProvider( + new SimpleAuthInfoProvider().add("username", (String) configuration.get(cassandraUserOption)).add( + "password", (String) configuration.get(cassandraPasswordOption))).build(); + + return cluster.connect("rhq"); + } + + /** + * Create a hibernate session to the SQL server. + * + * @return + * @throws Exception + */ + private EntityManager createEntityManager() throws Exception { + Properties properties = new Properties(); + properties.put("javax.persistence.provider", "org.hibernate.ejb.HibernatePersistence"); + properties.put("hibernate.connection.username", (String) configuration.get(sqlUserOption)); + properties.put("hibernate.connection.password", (String) configuration.get(sqlPasswordOption)); + + if ("oracle".equals(configuration.get(sqlServerType))) { + String driverClassName = "oracle.jdbc.driver.OracleDriver"; + + try { + //Required to preload the driver manually. + //Without this the driver load will fail due to the packaging. + Class.forName(driverClassName); + } catch (ClassNotFoundException e) { + log.debug(e); + throw new Exception("Oracle SQL Driver class could not be loaded. Missing class: " + driverClassName); + } + + properties.put("hibernate.dialect", "org.hibernate.dialect.Oracle10gDialect"); + properties.put("hibernate.driver_class", driverClassName); + properties.put("hibernate.connection.url", "jdbc:oracle:thin:@" + (String) configuration.get(sqlHostOption) + + ":" + (String) configuration.get(sqlPortOption) + ":" + (String) configuration.get(sqlDBOption)); + properties.put("hibernate.default_schema", (String) configuration.get(sqlDBOption)); + } else { + String driverClassName = "org.postgresql.Driver"; + + try { + //Required to preload the driver manually. + //Without this the driver load will fail due to the packaging. + Class.forName(driverClassName); + } catch (ClassNotFoundException e) { + log.debug(e); + throw new Exception("Postgres SQL Driver class could not be loaded. Missing class: " + driverClassName); + } + + properties.put("hibernate.dialect", "org.hibernate.dialect.PostgreSQLDialect"); + properties.put("hibernate.driver_class", driverClassName); + properties.put("hibernate.connection.url", "jdbc:postgresql://" + (String) configuration.get(sqlHostOption) + + ":" + (String) configuration.get(sqlPortOption) + "/" + (String) configuration.get(sqlDBOption)); + } + + log.debug("Creating entity manager with the following configuration:"); + log.debug(properties); + + Ejb3Configuration configuration = new Ejb3Configuration(); + configuration.setProperties(properties); + EntityManagerFactory factory = configuration.buildEntityManagerFactory(); + return factory.createEntityManager(); + } + + /** + * Parse Cassandra host information submitted in the form: + * host_addres,thrift_port,native_port|host_address_2,thrift_port,native_port + * + * @param stringValue + * @return + */ + private String[] parseCassandraHosts(String stringValue) { + String[] seeds = stringValue.split(","); + return seeds; + } + + /** + * @param value object value to parse + * @param defaultValue default value + * @return + */ + private boolean tryParseBoolean(Object value, boolean defaultValue) { + try { + return Boolean.parseBoolean(value.toString()); + } catch (Exception e) { + return defaultValue; + } + } + + /** + * @param value object value to parse + * @param defaultValue default value + * @return + */ + private Integer tryParseInteger(Object value, int defaultValue) { + try { + return Integer.parseInt(value.toString()); + } catch (Exception e) { + return defaultValue; + } + } + + @SuppressWarnings("serial") + private class HelpRequestedException extends Exception { + public HelpRequestedException() { + super("Help Requested"); + } + } +} \ No newline at end of file diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/ExistingDataBulkExportSource.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/ExistingDataBulkExportSource.java new file mode 100644 index 0000000..756d1b1 --- /dev/null +++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/ExistingDataBulkExportSource.java @@ -0,0 +1,124 @@ +/* + * RHQ Management Platform + * Copyright (C) 2005-2013 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 as published by + * the Free Software Foundation version 2 of the License. + * + * 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 for more details. + * + * You should have received a copy of the GNU 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.migrator; + +import java.io.BufferedReader; +import java.io.File; +import java.io.FileReader; +import java.util.LinkedList; +import java.util.List; +import java.util.Random; +import java.util.StringTokenizer; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +import org.rhq.core.util.stream.StreamUtil; + +/** + * @author Thomas Segismont + */ +public abstract class ExistingDataBulkExportSource implements ExistingDataSource { + + private static final Log log = LogFactory.getLog(ExistingDataBulkExportSource.class); + + protected static final int IO_BUFFER_SIZE = 1024 * 64; + + protected static final String DELIMITER = "|"; + + private File existingDataFile; + + private BufferedReader existingDataFileReader; + + private int currentIndex; + + public ExistingDataBulkExportSource() { + Random random = new Random(); + String exportFileName = System.currentTimeMillis() + "." + random.nextInt() + ".export"; + this.existingDataFile = new File(System.getProperty("java.io.tmpdir"), exportFileName); + } + + @Override + public void initialize() throws Exception { + this.exportExistingData(); + this.startReading(); + } + + @Override + public void close() throws Exception { + this.stopReading(); + this.removeTempFile(); + } + + protected abstract void exportExistingData() throws Exception; + + protected File getExistingDataFile() { + return existingDataFile; + } + + protected void startReading() throws Exception { + if (!existingDataFile.exists() && !existingDataFile.isFile() && !existingDataFile.canRead()) { + throw new IllegalStateException(); + } + existingDataFileReader = new BufferedReader(new FileReader(existingDataFile)); + currentIndex = 0; + } + + protected void stopReading() { + StreamUtil.safeClose(existingDataFileReader); + } + + protected void removeTempFile(){ + if(this.existingDataFile.exists()){ + try{ + this.existingDataFile.delete(); + }catch(Exception e){ + log.debug("Unable to clean temporary file " + this.existingDataFile, e); + } + } + } + + @Override + public List<Object[]> getData(int fromIndex, int maxResults) throws Exception { + if (log.isDebugEnabled()) { + log.debug("Reading lines " + fromIndex + " to " + (fromIndex + maxResults)); + } + + if (fromIndex != currentIndex) { + throw new IllegalStateException(); + } + + List<Object[]> results = new LinkedList<Object[]>(); + for (int i = 0; i < maxResults; i++) { + String nextLine = existingDataFileReader.readLine(); + if (nextLine == null) { + break; + } + currentIndex++; + StringTokenizer stringTokenizer = new StringTokenizer(nextLine, DELIMITER); + Object[] row = new Object[stringTokenizer.countTokens()]; + for (int j = 0; j < row.length; j++) { + row[j] = stringTokenizer.nextToken(); + } + results.add(row); + } + return results; + } +} diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/ExistingDataJPABulkExportSource.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/ExistingDataJPABulkExportSource.java new file mode 100644 index 0000000..fde9096 --- /dev/null +++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/ExistingDataJPABulkExportSource.java @@ -0,0 +1,58 @@ +package org.rhq.server.metrics.migrator; + +import java.io.BufferedWriter; +import java.io.FileWriter; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; + +import javax.persistence.EntityManager; + +import org.hibernate.Session; +import org.hibernate.engine.spi.SessionFactoryImplementor; +import org.hibernate.service.jdbc.connections.spi.ConnectionProvider; + +import org.rhq.core.util.jdbc.JDBCUtil; +import org.rhq.core.util.stream.StreamUtil; + +/** + * @author Thomas Segismont + */ +public class ExistingDataJPABulkExportSource extends ExistingDataBulkExportSource { + + private EntityManager entityManager; + private String selectNativeQuery; + + public ExistingDataJPABulkExportSource(EntityManager entityManager, String selectNativeQuery) { + super(); + } + + protected void exportExistingData() throws Exception { + BufferedWriter fileWriter = null; + Connection connection = null; + PreparedStatement statement = null; + ResultSet resultSet = null; + try { + fileWriter = new BufferedWriter(new FileWriter(getExistingDataFile()), IO_BUFFER_SIZE); + Session session = (Session) this.entityManager.getDelegate(); + SessionFactoryImplementor sfi = (SessionFactoryImplementor) session.getSessionFactory(); + ConnectionProvider cp = sfi.getConnectionProvider(); + connection = cp.getConnection(); + statement = connection.prepareStatement(this.selectNativeQuery); + resultSet = statement.executeQuery(); + int columnCount = resultSet.getMetaData().getColumnCount(); + while (resultSet.next()) { + for (int i = 1; i < columnCount + 1; i++) { + if (i > 1) { + fileWriter.write(DELIMITER); + } + fileWriter.write(resultSet.getString(i)); + } + fileWriter.write("\n"); + } + } finally { + StreamUtil.safeClose(fileWriter); + JDBCUtil.safeClose(connection, statement, resultSet); + } + } +} diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/ExistingDataJPASource.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/ExistingDataJPASource.java new file mode 100644 index 0000000..9aca36a --- /dev/null +++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/ExistingDataJPASource.java @@ -0,0 +1,64 @@ +/* + * RHQ Management Platform + * Copyright (C) 2005-2013 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 as published by + * the Free Software Foundation version 2 of the License. + * + * 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 for more details. + * + * You should have received a copy of the GNU 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.migrator; + +import java.util.List; + +import javax.persistence.EntityManager; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; + +/** + * @author Thomas Segismont + */ +public class ExistingDataJPASource implements ExistingDataSource { + + private static final Log log = LogFactory.getLog(ExistingDataJPASource.class); + + private EntityManager entityManager; + private String selectNativeQuery; + + public ExistingDataJPASource(EntityManager entityManager, String selectNativeQuery) { + this.entityManager = entityManager; + this.selectNativeQuery = selectNativeQuery; + } + + @SuppressWarnings("unchecked") + @Override + public List<Object[]> getData(int fromIndex, int maxResults) throws Exception { + if (log.isDebugEnabled()) { + log.debug("Reading lines " + fromIndex + " to " + (fromIndex + maxResults)); + } + + return (List<Object[]>) entityManager.createNativeQuery(selectNativeQuery).setFirstResult(fromIndex) + .setMaxResults(maxResults).getResultList(); + } + + @Override + public void initialize() { + //nothing to do since it just implements a simple query with limits + } + + @Override + public void close() { + //nothing to do since it just implements a simple query with limits + } +} diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/ExistingDataSource.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/ExistingDataSource.java new file mode 100644 index 0000000..0096538 --- /dev/null +++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/ExistingDataSource.java @@ -0,0 +1,34 @@ +/* + * RHQ Management Platform + * Copyright (C) 2005-2013 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 as published by + * the Free Software Foundation version 2 of the License. + * + * 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 for more details. + * + * You should have received a copy of the GNU 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.migrator; + +import java.util.List; + +/** + * @author Thomas Segismont + */ +public interface ExistingDataSource { + + List<Object[]> getData(int fromIndex, int maxResults) throws Exception; + + void initialize() throws Exception; + + void close() throws Exception; +} diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/ExistingPostgresDataBulkExportSource.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/ExistingPostgresDataBulkExportSource.java new file mode 100644 index 0000000..563e332 --- /dev/null +++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/migrator/ExistingPostgresDataBulkExportSource.java @@ -0,0 +1,68 @@ +/* + * RHQ Management Platform + * Copyright (C) 2005-2013 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 as published by + * the Free Software Foundation version 2 of the License. + * + * 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 for more details. + * + * You should have received a copy of the GNU 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.migrator; + +import java.io.BufferedWriter; +import java.io.FileWriter; +import java.sql.Connection; + +import javax.persistence.EntityManager; + +import org.hibernate.Session; +import org.hibernate.engine.spi.SessionFactoryImplementor; +import org.hibernate.service.jdbc.connections.spi.ConnectionProvider; +import org.postgresql.copy.CopyManager; +import org.postgresql.core.BaseConnection; + +import org.rhq.core.util.jdbc.JDBCUtil; +import org.rhq.core.util.stream.StreamUtil; + +/** + * @author Thomas Segismont + */ +public class ExistingPostgresDataBulkExportSource extends ExistingDataBulkExportSource { + + private EntityManager entityManager; + private String selectNativeQuery; + + public ExistingPostgresDataBulkExportSource(EntityManager entityManager, String selectNativeQuery) { + super(); + this.entityManager = entityManager; + this.selectNativeQuery = selectNativeQuery; + } + + protected void exportExistingData() throws Exception { + BufferedWriter fileWriter = null; + Connection connection = null; + try { + fileWriter = new BufferedWriter(new FileWriter(getExistingDataFile()), IO_BUFFER_SIZE); + Session session = (Session) entityManager.getDelegate(); + SessionFactoryImplementor sfi = (SessionFactoryImplementor) session.getSessionFactory(); + ConnectionProvider cp = sfi.getConnectionProvider(); + connection = cp.getConnection(); + CopyManager copyManager = new CopyManager((BaseConnection) connection); + copyManager.copyOut("COPY (" + this.selectNativeQuery + ") TO STDOUT WITH DELIMITER '" + DELIMITER + "'", + fileWriter); + } finally { + StreamUtil.safeClose(fileWriter); + JDBCUtil.safeClose(connection); + } + } +} diff --git a/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/DataSourceTest.java b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/DataSourceTest.java new file mode 100644 index 0000000..cda04ca --- /dev/null +++ b/modules/enterprise/server/server-metrics/src/test/java/org/rhq/server/metrics/DataSourceTest.java @@ -0,0 +1,169 @@ +/* + * RHQ Management Platform + * Copyright (C) 2005-2013 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 as published by + * the Free Software Foundation version 2 of the License. + * + * 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 for more details. + * + * You should have received a copy of the GNU 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 java.util.List; +import java.util.Properties; + +import javax.persistence.EntityManager; +import javax.persistence.EntityManagerFactory; + +import org.apache.commons.lang.time.StopWatch; +import org.apache.log4j.BasicConfigurator; +import org.apache.log4j.Level; +import org.apache.log4j.Logger; +import org.hibernate.ejb.Ejb3Configuration; + +import org.rhq.server.metrics.migrator.ExistingDataBulkExportSource; +import org.rhq.server.metrics.migrator.ExistingPostgresDataBulkExportSource; + +/** + * @author Thomas Segismont + */ + + +public class DataSourceTest { + + //ExistingPostgresDataBulkExport + public static void main(String[] args) throws Exception { + BasicConfigurator.configure(); + Logger.getRootLogger().setLevel(Level.INFO); + Logger.getLogger("org.rhq").setLevel(Level.DEBUG); + EntityManagerFactory entityManagerFactory = null; + EntityManager entityManager = null; + ExistingDataBulkExportSource source = null; + try { + entityManagerFactory = createEntityManager(); + entityManager = entityManagerFactory.createEntityManager(); + source = new ExistingPostgresDataBulkExportSource(entityManager, + "SELECT schedule_id, time_stamp, value, minvalue, maxvalue FROM RHQ_MEASUREMENT_DATA_NUM_1D"); + StopWatch stopWatch = new StopWatch(); + stopWatch.start(); + source.initialize(); + int rowIndex = 0; + int maxResults = 30000; + for (;;) { + List<Object[]> existingData = source.getData(rowIndex, maxResults); + if (existingData.size() < maxResults) { + break; + } else { + rowIndex += maxResults; + } + } + stopWatch.stop(); + System.out.println("Execution: " + stopWatch); + } finally { + if (source != null) { + source.close(); + } + if (entityManager != null) { + entityManager.close(); + } + if (entityManagerFactory != null) { + entityManagerFactory.close(); + } + } + } + + private static EntityManagerFactory createEntityManager() throws Exception { + Properties properties = new Properties(); + properties.put("javax.persistence.provider", "org.hibernate.ejb.HibernatePersistence"); + properties.put("hibernate.connection.username", "rhqadmin"); + properties.put("hibernate.connection.password", "rhqadmin"); + properties.put("hibernate.dialect", "org.hibernate.dialect.PostgreSQLDialect"); + String driverClassName = "org.postgresql.Driver"; + try { + //Required to preload the driver manually. + //Without this the driver load will fail due to the packaging. + Class.forName(driverClassName); + } catch (ClassNotFoundException e) { + throw new Exception("Postgres SQL Driver class could not be loaded. Missing class: " + driverClassName); + } + properties.put("hibernate.driver_class", driverClassName); + properties.put("hibernate.connection.url", "jdbc:postgresql://localhost:5432/rhq"); + Ejb3Configuration configuration = new Ejb3Configuration(); + configuration.setProperties(properties); + return configuration.buildEntityManagerFactory(); + } + + //ExistingDataJPABulkExportSource + + public static void main2(String[] args) throws Exception { + BasicConfigurator.configure(); + Logger.getRootLogger().setLevel(Level.INFO); + Logger.getLogger("org.rhq").setLevel(Level.DEBUG); + EntityManagerFactory entityManagerFactory = null; + EntityManager entityManager = null; + ExistingDataBulkExportSource source = null; + try { + entityManagerFactory = createEntityManager(); + entityManager = entityManagerFactory.createEntityManager(); + source = new ExistingPostgresDataBulkExportSource(entityManager, + "SELECT schedule_id, time_stamp, value, minvalue, maxvalue FROM RHQ_MEASUREMENT_DATA_NUM_1D"); + StopWatch stopWatch = new StopWatch(); + stopWatch.start(); + source.initialize(); + int rowIndex = 0; + int maxResults = 30000; + for (;;) { + List<Object[]> existingData = source.getData(rowIndex, maxResults); + if (existingData.size() < maxResults) { + break; + } else { + rowIndex += maxResults; + } + } + stopWatch.stop(); + System.out.println("Execution: " + stopWatch); + } finally { + if (source != null) { + source.close(); + } + if (entityManager != null) { + entityManager.close(); + } + if (entityManagerFactory != null) { + entityManagerFactory.close(); + } + } + } + + private static EntityManagerFactory createEntityManager2() throws Exception { + Properties properties = new Properties(); + properties.put("javax.persistence.provider", "org.hibernate.ejb.HibernatePersistence"); + properties.put("hibernate.connection.username", "rhqadmin"); + properties.put("hibernate.connection.password", "rhqadmin"); + properties.put("hibernate.dialect", "org.hibernate.dialect.PostgreSQLDialect"); + String driverClassName = "org.postgresql.Driver"; + try { + //Required to preload the driver manually. + //Without this the driver load will fail due to the packaging. + Class.forName(driverClassName); + } catch (ClassNotFoundException e) { + throw new Exception("Postgres SQL Driver class could not be loaded. Missing class: " + driverClassName); + } + properties.put("hibernate.driver_class", driverClassName); + properties.put("hibernate.connection.url", "jdbc:postgresql://localhost:5432/rhqdev"); + Ejb3Configuration configuration = new Ejb3Configuration(); + configuration.setProperties(properties); + return configuration.buildEntityManagerFactory(); + } + +} \ No newline at end of file commit d3a4a1539804a69de22bfbf909ab58156d6e4204 Author: Stefan Negrea <snegrea(a)redhat.com&gt; Date: Tue Apr 30 08:17:10 2013 -0500 Revamp the way to get information about the current node. The address, native port, and cluster name are now retrieved from the actual cassandra yaml file on discovery. diff --git a/.classpath b/.classpath index e44a17b..62c7e54 100644 --- a/.classpath +++ b/.classpath @@ -365,6 +365,7 @@ <classpathentry exported="true" kind="var" path="M2_REPO/org/picketbox/picketbox/4.0.7.Final/picketbox-4.0.7.Final.jar" sourcepath="/M2_REPO/org/picketbox/picketbox/4.0.7.Final/picketbox-4.0.7.Final-sources.jar"/> <classpathentry exported="true" kind="var" path="M2_REPO/org/python/jython-standalone/2.5.2/jython-standalone-2.5.2.jar"/> <classpathentry exported="true" kind="var" path="M2_REPO/com/wordnik/swagger-annotations_2.9.1/1.1.1-SNAPSHOT/swagger-annotations_2.9.1-1.1.1-20121031.024335-6.jar"/> + <classpathentry exported="true" kind="var" path="M2_REPO/org/yaml/snakeyaml/1.12/snakeyaml-1.12.jar"/> <classpathentry exported="true" kind="var" path="M2_REPO/joda-time/joda-time/2.1/joda-time-2.1.jar"/> <classpathentry exported="true" kind="var" path="M2_REPO/org/jboss/byteman/byteman-install/1.5.2/byteman-install-1.5.2.jar" sourcepath="M2_REPO/org/jboss/byteman/byteman-install/1.5.2/byteman-install-1.5.2-sources.jar"/> <classpathentry exported="true" kind="var" path="M2_REPO/org/jboss/byteman/byteman-submit/1.5.2/byteman-submit-1.5.2.jar" sourcepath="M2_REPO/org/jboss/byteman/byteman-submit/1.5.2/byteman-submit-1.5.2-sources.jar"/> diff --git a/modules/plugins/cassandra/pom.xml b/modules/plugins/cassandra/pom.xml index d13cbf4..a88d6e3 100644 --- a/modules/plugins/cassandra/pom.xml +++ b/modules/plugins/cassandra/pom.xml @@ -20,9 +20,14 @@ <version>${cassandra.driver.version}</version> </dependency> <dependency> - <groupId>commons-logging</groupId> - <artifactId>commons-logging</artifactId> - <scope>provided</scope> + <groupId>org.yaml</groupId> + <artifactId>snakeyaml</artifactId> + <version>1.12</version> + </dependency> + <dependency> + <groupId>org.xerial.snappy</groupId> + <artifactId>snappy-java</artifactId> + <version>1.0.5-M3-p1</version> </dependency> <dependency> <groupId>${project.groupId}</groupId> @@ -83,39 +88,8 @@ <artifactId>cassandra-driver-core</artifactId> </artifactItem> <artifactItem> - <groupId>io.netty</groupId> - <artifactId>netty</artifactId> - <version>${cassandra.driver.netty.version}</version> - </artifactItem> - <artifactItem> - <groupId>org.apache.cassandra</groupId> - <artifactId>cassandra-all</artifactId> - <version>${cassandra.version}</version> - </artifactItem> - <artifactItem> - <groupId>org.apache.cassandra</groupId> - <artifactId>cassandra-thrift</artifactId> - <version>${cassandra.version}</version> - </artifactItem> - <artifactItem> - <groupId>org.apache.thrift</groupId> - <artifactId>libthrift</artifactId> - <version>${cassandra.thrift.version}</version> - </artifactItem> - <artifactItem> - <groupId>com.google.guava</groupId> - <artifactId>guava</artifactId> - <version>14.0.1</version> - </artifactItem> - <artifactItem> - <groupId>com.yammer.metrics</groupId> - <artifactId>metrics-core</artifactId> - <version>2.2.0</version> - </artifactItem> - <artifactItem> - <groupId>org.slf4j</groupId> - <artifactId>slf4j-api</artifactId> - <version>1.7.2</version> + <groupId>org.yaml</groupId> + <artifactId>snakeyaml</artifactId> </artifactItem> <artifactItem> <groupId>org.slf4j</groupId> @@ -127,16 +101,6 @@ <artifactId>snappy-java</artifactId> <version>1.0.5-M3-p1</version> </artifactItem> - <artifactItem> - <groupId>org.codehaus.jackson</groupId> - <artifactId>jackson-core-asl</artifactId> - <version>1.9.2</version> - </artifactItem> - <artifactItem> - <groupId>org.codehaus.jackson</groupId> - <artifactId>jackson-mapper-asl</artifactId> - <version>1.9.2</version> - </artifactItem> </artifactItems> </configuration> </execution> diff --git a/modules/plugins/cassandra/src/main/java/org/rhq/plugins/cassandra/CassandraNodeComponent.java b/modules/plugins/cassandra/src/main/java/org/rhq/plugins/cassandra/CassandraNodeComponent.java index db28117..e16fd88 100644 --- a/modules/plugins/cassandra/src/main/java/org/rhq/plugins/cassandra/CassandraNodeComponent.java +++ b/modules/plugins/cassandra/src/main/java/org/rhq/plugins/cassandra/CassandraNodeComponent.java @@ -28,14 +28,11 @@ import static org.rhq.core.domain.measurement.AvailabilityType.UP; import static org.rhq.core.system.OperatingSystemType.WINDOWS; import java.io.File; -import java.util.Set; import com.datastax.driver.core.Cluster; -import com.datastax.driver.core.Query; -import com.datastax.driver.core.ResultSet; +import com.datastax.driver.core.ProtocolOptions.Compression; import com.datastax.driver.core.Session; import com.datastax.driver.core.SimpleAuthInfoProvider; -import com.datastax.driver.core.querybuilder.QueryBuilder; import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; @@ -45,14 +42,9 @@ import org.mc4j.ems.connection.bean.EmsBean; import org.mc4j.ems.connection.bean.operation.EmsOperation; import org.rhq.core.domain.configuration.Configuration; -import org.rhq.core.domain.configuration.PropertySimple; import org.rhq.core.domain.measurement.AvailabilityType; -import org.rhq.core.domain.measurement.MeasurementDataTrait; -import org.rhq.core.domain.measurement.MeasurementReport; -import org.rhq.core.domain.measurement.MeasurementScheduleRequest; import org.rhq.core.pluginapi.inventory.ResourceComponent; import org.rhq.core.pluginapi.inventory.ResourceContext; -import org.rhq.core.pluginapi.measurement.MeasurementFacet; import org.rhq.core.pluginapi.operation.OperationFacet; import org.rhq.core.pluginapi.operation.OperationResult; import org.rhq.core.pluginapi.util.ProcessExecutionUtility; @@ -67,8 +59,7 @@ import org.rhq.plugins.jmx.JMXServerComponent; /** * @author John Sanda */ -public class CassandraNodeComponent extends JMXServerComponent<ResourceComponent<?>> implements MeasurementFacet, - OperationFacet { +public class CassandraNodeComponent extends JMXServerComponent<ResourceComponent<?>> implements OperationFacet { private Log log = LogFactory.getLog(CassandraNodeComponent.class); @@ -78,20 +69,35 @@ public class CassandraNodeComponent extends JMXServerComponent<ResourceComponent @Override public void start(ResourceContext context) throws Exception { super.start(context); - Cluster cluster = Cluster.builder().addContactPoints(new String[] { "localhost" }).withoutMetrics() - .withPort(getCQLPort(context.getPluginConfiguration())) - .withAuthInfoProvider( - new SimpleAuthInfoProvider().add("username", - this.getResourceContext().getPluginConfiguration().getSimpleValue("username")).add("password", - this.getResourceContext().getPluginConfiguration().getSimpleValue("password"))) - .build(); - this.cassandraSession = cluster.connect("rhq"); - }; - private int getCQLPort(Configuration pluginConfig) { - String property = pluginConfig.getSimpleValue("nativeTransportPort", "9042"); - return Integer.parseInt(property); - } + String host = context.getPluginConfiguration().getSimpleValue("host", "localhost"); + String clusterName = context.getPluginConfiguration().getSimpleValue("clusterName", "unknown"); + String username = context.getPluginConfiguration().getSimpleValue("username", "cassandra"); + String password = context.getPluginConfiguration().getSimpleValue("password", "password"); + + Integer nativePort = 9042; + try { + nativePort = Integer.parseInt(context.getPluginConfiguration() + .getSimpleValue("nativeTransportPort", "9042")); + } catch (Exception e) { + log.debug("Native transport port parsing failed...", e); + } + + try { + Cluster cluster = Cluster + .builder() + .addContactPoints(new String[] { host }) + .withoutMetrics() + .withPort(nativePort) + .withCompression(Compression.NONE) + .withAuthInfoProvider(new SimpleAuthInfoProvider().add("username", username).add("password", password)) + .build(); + this.cassandraSession = cluster.connect(clusterName); + } catch (Exception e) { + log.error("Connect to Cassandra " + host + ":" + nativePort, e); + throw e; + } + }; @Override public AvailabilityType getAvailability() { @@ -209,20 +215,6 @@ public class CassandraNodeComponent extends JMXServerComponent<ResourceComponent } } - //@Override - public void getValues(MeasurementReport report, Set<MeasurementScheduleRequest> metrics) throws Exception { - for (MeasurementScheduleRequest scheduleRequest : metrics) { - if (scheduleRequest.getName().equals("cluster")) { - Query q = QueryBuilder.select().from("system", "local"); - ResultSet resultSet = this.getCassandraSession().execute(q); - if (!resultSet.isExhausted()) { - report - .addData(new MeasurementDataTrait(scheduleRequest, resultSet.one().getString("cluster_name"))); - } - } - } - } - public Session getCassandraSession() { return this.cassandraSession; } diff --git a/modules/plugins/cassandra/src/main/java/org/rhq/plugins/cassandra/CassandraNodeDiscoveryComponent.java b/modules/plugins/cassandra/src/main/java/org/rhq/plugins/cassandra/CassandraNodeDiscoveryComponent.java index 0b4eab2..febddf1 100644 --- a/modules/plugins/cassandra/src/main/java/org/rhq/plugins/cassandra/CassandraNodeDiscoveryComponent.java +++ b/modules/plugins/cassandra/src/main/java/org/rhq/plugins/cassandra/CassandraNodeDiscoveryComponent.java @@ -23,11 +23,17 @@ package org.rhq.plugins.cassandra; import java.io.File; +import java.io.FileInputStream; +import java.io.InputStream; import java.util.HashSet; import java.util.List; +import java.util.Map; import java.util.Set; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; import org.mc4j.ems.connection.support.metadata.J2SE5ConnectionTypeDescriptor; +import org.yaml.snakeyaml.Yaml; import org.rhq.core.domain.configuration.Configuration; import org.rhq.core.domain.configuration.PropertySimple; @@ -42,6 +48,8 @@ import org.rhq.plugins.jmx.JMXDiscoveryComponent; */ public class CassandraNodeDiscoveryComponent extends JMXDiscoveryComponent { + private static final Log log = LogFactory.getLog(CassandraNodeDiscoveryComponent.class); + @SuppressWarnings({ "rawtypes", "unchecked" }) @Override public Set<DiscoveredResourceDetails> discoverResources(ResourceDiscoveryContext context) { @@ -58,31 +66,83 @@ public class CassandraNodeDiscoveryComponent extends JMXDiscoveryComponent { return details; } + @SuppressWarnings({ "unchecked", "deprecation" }) private DiscoveredResourceDetails getDetails(ResourceDiscoveryContext<?> context, ProcessScanResult processScanResult) { ProcessInfo processInfo = processScanResult.getProcessInfo(); + + Configuration pluginConfig = new Configuration(); + String jmxPort = null; - for (String arg : processInfo.getCommandLine()) { + + String[] arguments = processInfo.getCommandLine(); + int classpathIndex = -1; + for (int i = 0; i < arguments.length; i++) { + String arg = arguments[i]; + if (arg.startsWith("-Dcom.sun.management.jmxremote.port")) { String[] jmxPortArg = arg.split("="); jmxPort = jmxPortArg[1]; - break; + } + if (arg.startsWith("-cp")) { + classpathIndex = i; } } - if (jmxPort == null) { - return null; + if (classpathIndex != -1 && classpathIndex + 1 < arguments.length) { + String[] classpathEntries = arguments[classpathIndex + 1].split(":"); + + String yamlConfigurationPath = null; + for (String classpathEntry : classpathEntries) { + if (classpathEntry.endsWith("conf")) { + yamlConfigurationPath = processInfo.getExecutable().getCwd() + "/" + classpathEntry; + } + } + + if (yamlConfigurationPath != null) { + + InputStream inputStream = null; + try { + inputStream = new FileInputStream(new File(yamlConfigurationPath + "/cassandra.yaml")); + Yaml yaml = new Yaml(); + Map<String, String> parsedProperties = (Map<String, String>) yaml.load(inputStream); + + if (parsedProperties.get("cluster_name") != null) { + pluginConfig.put(new PropertySimple("clusterName", parsedProperties.get("cluster_name"))); + } + + if (parsedProperties.get("listen_address") != null) { + pluginConfig.put(new PropertySimple("host", parsedProperties.get("listen_address"))); + } + + if (parsedProperties.get("native_transport_port") != null) { + pluginConfig.put(new PropertySimple("nativeTransportPort", parsedProperties + .get("native_transport_port"))); + } + } catch (Exception e) { + log.error("YAML Configuration load exception ", e); + } finally { + try { + if ( inputStream != null){ + inputStream.close(); + } + } catch (Exception e) { + log.error("Unable to close stream for yaml configuration", e); + } + } + } } - String resourceKey = "CassandraDaemon:" + jmxPort; - String resourceName = "CassandraDaemon"; + if (jmxPort != null) { + pluginConfig.put(new PropertySimple(JMXDiscoveryComponent.CONNECTION_TYPE, + J2SE5ConnectionTypeDescriptor.class.getName())); + pluginConfig.put(new PropertySimple(JMXDiscoveryComponent.CONNECTOR_ADDRESS_CONFIG_PROPERTY, + "service:jmx:rmi:///jndi/rmi://" + pluginConfig.getSimpleValue("host") + ":" + jmxPort + "/jmxrmi")); + } - Configuration pluginConfig = new Configuration(); - pluginConfig.put(new PropertySimple(JMXDiscoveryComponent.CONNECTION_TYPE, - J2SE5ConnectionTypeDescriptor.class.getName())); - pluginConfig.put(new PropertySimple(JMXDiscoveryComponent.CONNECTOR_ADDRESS_CONFIG_PROPERTY, - "service:jmx:rmi:///jndi/rmi://127.0.0.1:" + jmxPort + "/jmxrmi")); + String resourceKey = "Cassandra (" + pluginConfig.getSimpleValue("host") + ") " + jmxPort; + String resourceName = "Cassandra (" + pluginConfig.getSimpleValue("host") + ")"; String path = processInfo.getExecutable().getCwd(); pluginConfig.put(new PropertySimple("baseDir", new File(path).getParentFile().getAbsolutePath())); diff --git a/modules/plugins/cassandra/src/main/resources/META-INF/rhq-plugin.xml b/modules/plugins/cassandra/src/main/resources/META-INF/rhq-plugin.xml index cf28a8e..905d4b8 100644 --- a/modules/plugins/cassandra/src/main/resources/META-INF/rhq-plugin.xml +++ b/modules/plugins/cassandra/src/main/resources/META-INF/rhq-plugin.xml @@ -45,6 +45,8 @@ <c:simple-property name="commandLine" required="false" type="string" description="the command line of the JVM at the time it was discovered - only used by JVMs with type Local; if the command line of the JVM changes, this property's value will need to be updated accordingly in order for RHQ to connect to the JVM"/> <c:simple-property name="baseDir" displayName="Base Directory" description="The base directory from which the Cassandra Daemon was launched." required="false"/> <c:simple-property name="nativeTransportPort" description="The port on which Cassandra listens for CQL client connections." default="9042" type="integer"/> + <c:simple-property name="host" description="The host on which cassandra listens to CQL client connections" default="localhost"/> + <c:simple-property name="clusterName" description="Cluster name" default="localhost"/> </plugin-configuration> <process-scan name="CassandraDaemon" query="process|basename|match=^java.*,arg|org.apache.cassandra.service.CassandraDaemon|match=.*"/> @@ -65,9 +67,6 @@ </results> </operation> - <metric property="cluster" dataType="trait" description="The name of the cluster. This is used to prevent machines in one logical cluster from joining another." displayType="summary"/> - - <server name="Cassandra Server JVM" sourcePlugin="JMX" sourceType="JMX Server" discovery="org.rhq.plugins.jmx.EmbeddedJMXServerDiscoveryComponent" commit 3a21520a2a9f2805b14f0120f7876b30f3f3901d Author: Thomas Segismont <tsegismo(a)redhat.com&gt; Date: Mon Apr 29 15:23:46 2013 +0200 Add JPA bulk export source diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataJPABulkExportSource.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataJPABulkExportSource.java new file mode 100644 index 0000000..d61fdf1 --- /dev/null +++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataJPABulkExportSource.java @@ -0,0 +1,124 @@ +package org.rhq.server.metrics; + +import java.io.BufferedWriter; +import java.io.File; +import java.io.FileWriter; +import java.sql.Connection; +import java.sql.PreparedStatement; +import java.sql.ResultSet; +import java.util.List; +import java.util.Properties; + +import javax.persistence.EntityManager; +import javax.persistence.EntityManagerFactory; + +import org.apache.commons.lang.time.StopWatch; +import org.apache.log4j.BasicConfigurator; +import org.apache.log4j.Level; +import org.apache.log4j.Logger; +import org.hibernate.ejb.Ejb3Configuration; + +import org.rhq.core.util.jdbc.JDBCUtil; +import org.rhq.core.util.stream.StreamUtil; + +/** + * @author Thomas Segismont + */ +public class ExistingDataJPABulkExportSource extends ExistingDataBulkExportSource { + + public ExistingDataJPABulkExportSource(EntityManager entityManager, String selectNativeQuery, File workDirectory, + String fileName) { + super(entityManager, selectNativeQuery, workDirectory, fileName); + } + + public void exportExistingData() throws Exception { + BufferedWriter fileWriter = null; + Connection connection = null; + PreparedStatement statement = null; + ResultSet resultSet = null; + try { + fileWriter = new BufferedWriter(new FileWriter(getExistingDataFile()), IO_BUFFER_SIZE); + connection = getConnection(); + statement = connection.prepareStatement(getSelectNativeQuery()); + int columnCount = resultSet.getMetaData().getColumnCount(); + resultSet = statement.executeQuery(); + while (resultSet.next()) { + for (int i = 1; i < columnCount + 1; i++) { + if (i > 1) { + fileWriter.write(DELIMITER); + } + fileWriter.write(resultSet.getString(i)); + } + fileWriter.write("\n"); + } + } finally { + StreamUtil.safeClose(fileWriter); + JDBCUtil.safeClose(connection, statement, resultSet); + } + } + + public static void main(String[] args) throws Exception { + BasicConfigurator.configure(); + Logger.getRootLogger().setLevel(Level.INFO); + Logger.getLogger("org.rhq").setLevel(Level.DEBUG); + EntityManagerFactory entityManagerFactory = null; + EntityManager entityManager = null; + ExistingDataBulkExportSource source = null; + try { + entityManagerFactory = createEntityManager(); + entityManager = entityManagerFactory.createEntityManager(); + source = new ExistingPostgresDataBulkExportSource(entityManager, + "SELECT schedule_id, time_stamp, value, minvalue, maxvalue FROM RHQ_MEASUREMENT_DATA_NUM_1D", + // "SELECT count(*) FROM RHQ_MEASUREMENT_DATA_NUM_1D limit 1" + new File(System.getProperty("java.io.tmpdir")), "poupoupidou.txt"); + StopWatch stopWatch = new StopWatch(); + stopWatch.start(); + source.exportExistingData(); + source.startReading(); + int rowIndex = 0; + int maxResults = 30000; + for (;;) { + List<Object[]> existingData = source.getExistingData(rowIndex, maxResults); + if (existingData.size() < maxResults) { + break; + } else { + rowIndex += maxResults; + } + } + stopWatch.stop(); + System.out.println("Execution: " + stopWatch); + } finally { + if (source != null) { + source.stopReading(); + } + if (entityManager != null) { + entityManager.close(); + } + if (entityManagerFactory != null) { + entityManagerFactory.close(); + } + } + } + + private static EntityManagerFactory createEntityManager() throws Exception { + Properties properties = new Properties(); + properties.put("javax.persistence.provider", "org.hibernate.ejb.HibernatePersistence"); + properties.put("hibernate.connection.username", "rhqadmin"); + properties.put("hibernate.connection.password", "rhqadmin"); + properties.put("hibernate.dialect", "org.hibernate.dialect.PostgreSQLDialect"); + String driverClassName = "org.postgresql.Driver"; + try { + //Required to preload the driver manually. + //Without this the driver load will fail due to the packaging. + Class.forName(driverClassName); + } catch (ClassNotFoundException e) { + throw new Exception("Postgres SQL Driver class could not be loaded. Missing class: " + driverClassName); + } + properties.put("hibernate.driver_class", driverClassName); + properties.put("hibernate.connection.url", "jdbc:postgresql://localhost:5432/rhqdev"); + Ejb3Configuration configuration = new Ejb3Configuration(); + configuration.setProperties(properties); + return configuration.buildEntityManagerFactory(); + } + +} commit a6ae7997055dd5e508da3991c08876c3da2fc294 Author: Thomas Segismont <tsegismo(a)redhat.com&gt; Date: Fri Apr 26 09:46:48 2013 +0200 Introduce ExistingDataSource abstraction diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataBulkExportSource.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataBulkExportSource.java new file mode 100644 index 0000000..d817002 --- /dev/null +++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataBulkExportSource.java @@ -0,0 +1,129 @@ +/* + * RHQ Management Platform + * Copyright (C) 2005-2013 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 as published by + * the Free Software Foundation version 2 of the License. + * + * 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 for more details. + * + * You should have received a copy of the GNU 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 java.io.BufferedReader; +import java.io.File; +import java.io.FileReader; +import java.sql.Connection; +import java.sql.SQLException; +import java.util.LinkedList; +import java.util.List; +import java.util.StringTokenizer; + +import javax.persistence.EntityManager; + +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.hibernate.Session; +import org.hibernate.engine.spi.SessionFactoryImplementor; +import org.hibernate.service.jdbc.connections.spi.ConnectionProvider; + +import org.rhq.core.util.stream.StreamUtil; + +/** + * @author Thomas Segismont + */ +public abstract class ExistingDataBulkExportSource implements ExistingDataSource { + + private static final Log LOG = LogFactory.getLog(ExistingDataBulkExportSource.class); + + protected static final int IO_BUFFER_SIZE = 1024 * 64; + + protected static final String DELIMITER = "|"; + + private EntityManager entityManager; + + private String selectNativeQuery; + + private File workDirectory; + + private String fileName; + + private File existingDataFile; + + private BufferedReader existingDataFileReader; + + private int currentIndex; + + public ExistingDataBulkExportSource(EntityManager entityManager, String selectNativeQuery, File workDirectory, + String fileName) { + this.entityManager = entityManager; + this.selectNativeQuery = selectNativeQuery; + this.workDirectory = workDirectory; + this.fileName = fileName; + existingDataFile = new File(workDirectory, fileName); + } + + protected String getSelectNativeQuery() { + return selectNativeQuery; + } + + protected File getExistingDataFile() { + return existingDataFile; + } + + protected Connection getConnection() throws SQLException { + Session session = (Session) entityManager.getDelegate(); + SessionFactoryImplementor sfi = (SessionFactoryImplementor) session.getSessionFactory(); + ConnectionProvider cp = sfi.getConnectionProvider(); + return cp.getConnection(); + } + + public abstract void exportExistingData() throws Exception; + + public void startReading() throws Exception { + if (!existingDataFile.exists() && !existingDataFile.isFile() && !existingDataFile.canRead()) { + throw new IllegalStateException(); + } + existingDataFileReader = new BufferedReader(new FileReader(existingDataFile)); + currentIndex = 0; + } + + public void stopReading() { + StreamUtil.safeClose(existingDataFileReader); + } + + @Override + public List<Object[]> getExistingData(int fromIndex, int maxResults) throws Exception { + if (LOG.isDebugEnabled()) { + LOG.debug("Reading lines " + fromIndex + "-" + (fromIndex + maxResults)); + } + if (fromIndex != currentIndex) { + throw new IllegalStateException(); + } + List<Object[]> results = new LinkedList<Object[]>(); + for (int i = 0; i < maxResults; i++) { + String nextLine = existingDataFileReader.readLine(); + if (nextLine == null) { + break; + } + currentIndex++; + StringTokenizer stringTokenizer = new StringTokenizer(nextLine, DELIMITER); + Object[] row = new Object[stringTokenizer.countTokens()]; + for (int j = 0; j < row.length; j++) { + row[j] = stringTokenizer.nextToken(); + } + results.add(row); + } + return results; + } + +} diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataJPASource.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataJPASource.java new file mode 100644 index 0000000..b072f58 --- /dev/null +++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataJPASource.java @@ -0,0 +1,119 @@ +/* + * RHQ Management Platform + * Copyright (C) 2005-2013 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 as published by + * the Free Software Foundation version 2 of the License. + * + * 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 for more details. + * + * You should have received a copy of the GNU 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 java.util.List; +import java.util.Properties; + +import javax.persistence.EntityManager; +import javax.persistence.EntityManagerFactory; + +import org.apache.commons.lang.time.StopWatch; +import org.apache.commons.logging.Log; +import org.apache.commons.logging.LogFactory; +import org.apache.log4j.BasicConfigurator; +import org.apache.log4j.Level; +import org.apache.log4j.Logger; +import org.hibernate.ejb.Ejb3Configuration; + +/** + * @author Thomas Segismont + */ +public class ExistingDataJPASource implements ExistingDataSource { + + private static final Log LOG = LogFactory.getLog(ExistingDataJPASource.class); + + private EntityManager entityManager; + + private String selectNativeQuery; + + public ExistingDataJPASource(EntityManager entityManager, String selectNativeQuery) { + this.entityManager = entityManager; + this.selectNativeQuery = selectNativeQuery; + } + + @Override + public List<Object[]> getExistingData(int fromIndex, int maxResults) throws Exception { + if (LOG.isDebugEnabled()) { + LOG.debug("Reading lines " + fromIndex + "-" + (fromIndex + maxResults)); + } + return entityManager.createNativeQuery(selectNativeQuery).setFirstResult(fromIndex).setMaxResults(maxResults) + .getResultList(); + } + + public static void main(String[] args) throws Exception { + BasicConfigurator.configure(); + Logger.getRootLogger().setLevel(Level.INFO); + Logger.getLogger("org.rhq").setLevel(Level.DEBUG); + EntityManagerFactory entityManagerFactory = null; + EntityManager entityManager = null; + ExistingDataJPASource source = null; + try { + entityManagerFactory = createEntityManager(); + entityManager = entityManagerFactory.createEntityManager(); + source = new ExistingDataJPASource( + entityManager, + "SELECT schedule_id, time_stamp, value, minvalue, maxvalue FROM RHQ_MEASUREMENT_DATA_NUM_1D"); + StopWatch stopWatch = new StopWatch(); + stopWatch.start(); + int rowIndex = 0; + int maxResults = 30000; + for (; ; ) { + List<Object[]> existingData = source.getExistingData(rowIndex, maxResults); + if (existingData.size() < maxResults) { + break; + } else { + rowIndex += maxResults; + } + } + stopWatch.stop(); + System.out.println("Execution: " + stopWatch); + } finally { + if (entityManager != null) { + entityManager.close(); + } + if (entityManagerFactory != null) { + entityManagerFactory.close(); + } + } + } + + private static EntityManagerFactory createEntityManager() throws Exception { + Properties properties = new Properties(); + properties.put("javax.persistence.provider", "org.hibernate.ejb.HibernatePersistence"); + properties.put("hibernate.connection.username", "rhqadmin"); + properties.put("hibernate.connection.password", "rhqadmin"); + properties.put("hibernate.dialect", "org.hibernate.dialect.PostgreSQLDialect"); + String driverClassName = "org.postgresql.Driver"; + try { + //Required to preload the driver manually. + //Without this the driver load will fail due to the packaging. + Class.forName(driverClassName); + } catch (ClassNotFoundException e) { + throw new Exception("Postgres SQL Driver class could not be loaded. Missing class: " + driverClassName); + } + properties.put("hibernate.driver_class", driverClassName); + properties.put("hibernate.connection.url", "jdbc:postgresql://localhost:5432/rhq"); + Ejb3Configuration configuration = new Ejb3Configuration(); + configuration.setProperties(properties); + return configuration.buildEntityManagerFactory(); + } + +} diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataSource.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataSource.java new file mode 100644 index 0000000..d6e0203 --- /dev/null +++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingDataSource.java @@ -0,0 +1,31 @@ +/* + * RHQ Management Platform + * Copyright (C) 2005-2013 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 as published by + * the Free Software Foundation version 2 of the License. + * + * 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 for more details. + * + * You should have received a copy of the GNU 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 java.util.List; + +/** + * @author Thomas Segismont + */ +public interface ExistingDataSource { + + List<Object[]> getExistingData(int fromIndex, int maxResults) throws Exception; + +} diff --git a/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingPostgresDataBulkExportSource.java b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingPostgresDataBulkExportSource.java new file mode 100644 index 0000000..8d049ed --- /dev/null +++ b/modules/enterprise/server/server-metrics/src/main/java/org/rhq/server/metrics/ExistingPostgresDataBulkExportSource.java @@ -0,0 +1,133 @@ +/* + * RHQ Management Platform + * Copyright (C) 2005-2013 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 as published by + * the Free Software Foundation version 2 of the License. + * + * 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 for more details. + * + * You should have received a copy of the GNU 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 java.io.BufferedWriter; +import java.io.File; +import java.io.FileWriter; +import java.sql.Connection; +import java.util.List; +import java.util.Properties; + +import javax.persistence.EntityManager; +import javax.persistence.EntityManagerFactory; + +import org.apache.commons.lang.time.StopWatch; +import org.apache.log4j.BasicConfigurator; +import org.apache.log4j.Level; +import org.apache.log4j.Logger; +import org.hibernate.ejb.Ejb3Configuration; +import org.postgresql.copy.CopyManager; +import org.postgresql.core.BaseConnection; + +import org.rhq.core.util.jdbc.JDBCUtil; +import org.rhq.core.util.stream.StreamUtil; + +/** + * @author Thomas Segismont + */ +public class ExistingPostgresDataBulkExportSource extends ExistingDataBulkExportSource { + + public ExistingPostgresDataBulkExportSource(EntityManager entityManager, String selectNativeQuery, + File workDirectory, String fileName) { + super(entityManager, selectNativeQuery, workDirectory, fileName); + } + + public void exportExistingData() throws Exception { + BufferedWriter fileWriter = null; + Connection connection = null; + try { + fileWriter = new BufferedWriter(new FileWriter(getExistingDataFile()), IO_BUFFER_SIZE); + connection = getConnection(); + CopyManager copyManager = new CopyManager((BaseConnection) connection); + copyManager.copyOut("COPY (" + getSelectNativeQuery() + ") TO STDOUT WITH DELIMITER '" + DELIMITER + "'", + fileWriter); + } finally { + StreamUtil.safeClose(fileWriter); + JDBCUtil.safeClose(connection); + } + } + + public static void main(String[] args) throws Exception { + BasicConfigurator.configure(); + Logger.getRootLogger().setLevel(Level.INFO); + Logger.getLogger("org.rhq").setLevel(Level.DEBUG); + EntityManagerFactory entityManagerFactory = null; + EntityManager entityManager = null; + ExistingDataBulkExportSource source = null; + try { + entityManagerFactory = createEntityManager(); + entityManager = entityManagerFactory.createEntityManager(); + source = new ExistingPostgresDataBulkExportSource( + entityManager, + "SELECT schedule_id, time_stamp, value, minvalue, maxvalue FROM RHQ_MEASUREMENT_DATA_NUM_1D", + // "SELECT count(*) FROM RHQ_MEASUREMENT_DATA_NUM_1D limit 1" + new File(System.getProperty("java.io.tmpdir")), "poupoupidou.txt"); + StopWatch stopWatch = new StopWatch(); + stopWatch.start(); + source.exportExistingData(); + source.startReading(); + int rowIndex = 0; + int maxResults = 30000; + for (;;) { + List<Object[]> existingData = source.getExistingData(rowIndex, maxResults); + if (existingData.size() < maxResults) { + break; + } else { + rowIndex += maxResults; + } + } + stopWatch.stop(); + System.out.println("Execution: " + stopWatch); + } finally { + if (source != null) { + source.stopReading(); + } + if (entityManager != null) { + entityManager.close(); + } + if (entityManagerFactory != null) { + entityManagerFactory.close(); + } + } + } + + private static EntityManagerFactory createEntityManager() throws Exception { + Properties properties = new Properties(); + properties.put("javax.persistence.provider", "org.hibernate.ejb.HibernatePersistence"); + properties.put("hibernate.connection.username", "rhqadmin"); + properties.put("hibernate.connection.password", "rhqadmin"); + properties.put("hibernate.dialect", "org.hibernate.dialect.PostgreSQLDialect"); + String driverClassName = "org.postgresql.Driver"; + try { + //Required to preload the driver manually. + //Without this the driver load will fail due to the packaging. + Class.forName(driverClassName); + } catch (ClassNotFoundException e) { + throw new Exception("Postgres SQL Driver class could not be loaded. Missing class: " + driverClassName); + } + properties.put("hibernate.driver_class", driverClassName); + properties.put("hibernate.connection.url", "jdbc:postgresql://localhost:5432/rhq"); + Ejb3Configuration configuration = new Ejb3Configuration(); + configuration.setProperties(properties); + return configuration.buildEntityManagerFactory(); + } + +}