ambari-metrics-timelineservice/src/main/java/org/apache/ambari/metrics/core/timeline/upgrade/core/MetricsDataMigrationLauncher.java - ambari-metrics - Git at Google

 /**
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.ambari.metrics.core.timeline.upgrade.core;

 import com.google.common.collect.Iterables;
 import com.google.common.collect.Sets;
 import org.apache.ambari.metrics.core.timeline.PhoenixHBaseAccessor;
 import org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration;
 import org.apache.ambari.metrics.core.timeline.discovery.TimelineMetricMetadataManager;
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;

 import java.io.BufferedReader;
 import java.io.FileInputStream;
 import java.io.FileWriter;
 import java.io.IOException;
 import java.io.InputStreamReader;
 import java.net.MalformedURLException;
 import java.net.URISyntaxException;
 import java.nio.file.Files;
 import java.nio.file.Paths;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.TimeUnit;
 import java.util.stream.Stream;

 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_AGGREGATE_DAILY_TABLE_NAME;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_AGGREGATE_DAILY_V1_TABLE_NAME;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_AGGREGATE_HOURLY_TABLE_NAME;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_AGGREGATE_HOURLY_V1_TABLE_NAME;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_AGGREGATE_MINUTE_TABLE_NAME;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_AGGREGATE_MINUTE_V1_TABLE_NAME;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_CLUSTER_AGGREGATE_DAILY_TABLE_NAME;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_CLUSTER_AGGREGATE_DAILY_V1_TABLE_NAME;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_CLUSTER_AGGREGATE_HOURLY_TABLE_NAME;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_CLUSTER_AGGREGATE_HOURLY_V1_TABLE_NAME;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_CLUSTER_AGGREGATE_MINUTE_TABLE_NAME;
 import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_CLUSTER_AGGREGATE_MINUTE_V1_TABLE_NAME;

 public class MetricsDataMigrationLauncher {
   private static final Log LOG = LogFactory.getLog(MetricsDataMigrationLauncher.class);
   private static final Long DEFAULT_TIMEOUT_MINUTES = 60*24L;
   private static String patternPrefix = "._p_";
   private static final int DEFAULT_BATCH_SIZE = 5;
   public static final Map<String, String> CLUSTER_AGGREGATE_TABLES_MAPPING = new HashMap<>();
   public static final Map<String, String> HOST_AGGREGATE_TABLES_MAPPING = new HashMap<>();
   public static final String DEFAULT_PROCESSED_METRICS_FILE_LOCATION = "/var/log/ambari-metrics-collector/ambari-metrics-migration-state.txt";
   public static final int DEFAULT_NUMBER_OF_THREADS = 3;
   public static final long ONE_MONTH_MILLIS = 2592000000L;
   public static final long DEFAULT_START_TIME = System.currentTimeMillis() - ONE_MONTH_MILLIS; //Last month

   static {
     CLUSTER_AGGREGATE_TABLES_MAPPING.put(METRICS_CLUSTER_AGGREGATE_MINUTE_V1_TABLE_NAME, METRICS_CLUSTER_AGGREGATE_MINUTE_TABLE_NAME);
     CLUSTER_AGGREGATE_TABLES_MAPPING.put(METRICS_CLUSTER_AGGREGATE_HOURLY_V1_TABLE_NAME, METRICS_CLUSTER_AGGREGATE_HOURLY_TABLE_NAME);
     CLUSTER_AGGREGATE_TABLES_MAPPING.put(METRICS_CLUSTER_AGGREGATE_DAILY_V1_TABLE_NAME, METRICS_CLUSTER_AGGREGATE_DAILY_TABLE_NAME);

     HOST_AGGREGATE_TABLES_MAPPING.put(METRICS_AGGREGATE_MINUTE_V1_TABLE_NAME, METRICS_AGGREGATE_MINUTE_TABLE_NAME);
     HOST_AGGREGATE_TABLES_MAPPING.put(METRICS_AGGREGATE_HOURLY_V1_TABLE_NAME, METRICS_AGGREGATE_HOURLY_TABLE_NAME);
     HOST_AGGREGATE_TABLES_MAPPING.put(METRICS_AGGREGATE_DAILY_V1_TABLE_NAME, METRICS_AGGREGATE_DAILY_TABLE_NAME);
   }

   private final Set<Set<String>> metricNamesBatches;
   private final String processedMetricsFilePath;
   private Set<String> metricNames;

   private Long startTime;
   private Integer batchSize;
   private Integer numberOfThreads;
   private TimelineMetricConfiguration timelineMetricConfiguration;
   private PhoenixHBaseAccessor hBaseAccessor;
   private Map<String, Set<String>> processedMetrics;

   public MetricsDataMigrationLauncher(String whitelistedFilePath, String processedMetricsFilePath, Long startTime, Integer numberOfThreads, Integer batchSize) throws Exception {
     this.startTime = startTime == null? DEFAULT_START_TIME : startTime;
     this.numberOfThreads = numberOfThreads == null? DEFAULT_NUMBER_OF_THREADS : numberOfThreads;
     this.batchSize = batchSize == null? DEFAULT_BATCH_SIZE : batchSize;
     this.processedMetricsFilePath = processedMetricsFilePath == null? DEFAULT_PROCESSED_METRICS_FILE_LOCATION : processedMetricsFilePath;

     initializeHbaseAccessor();

     LOG.info("Looking for whitelisted metric names...");

     if (whitelistedFilePath != null) {
       this.metricNames = readMetricWhitelistFromFile(whitelistedFilePath);
     } else if (timelineMetricConfiguration.isWhitelistingEnabled()) {
       String whitelistFile = timelineMetricConfiguration.getMetricsConf().get(TimelineMetricConfiguration.TIMELINE_METRICS_WHITELIST_FILE, TimelineMetricConfiguration.TIMELINE_METRICS_WHITELIST_FILE_LOCATION_DEFAULT);
       metricNames = readMetricWhitelistFromFile(whitelistFile);
     } else {
       LOG.error("No whitelisted metrics specified. Exiting...");
       throw new Exception("List of whitelisted metrics must be provided");
     }

     readProcessedMetricsMap();

     LOG.info("Setting up batches...");
     this.metricNamesBatches = new HashSet<>();

     Iterables.partition(metricNames, this.batchSize)
       .forEach(batch -> metricNamesBatches.add(new HashSet<>(batch)));
     LOG.info(String.format("Split metric names into %s batches with size of %s", metricNamesBatches.size(), this.batchSize));
   }


   private void readProcessedMetricsMap() {
     Map<String, Set<String>> result = new HashMap<>();
     if (!Files.exists(Paths.get(processedMetricsFilePath))) {
       LOG.info(String.format("The processed metrics file %s is missing, assuming there were no metrics processed.", processedMetricsFilePath));
       this.processedMetrics = new HashMap<>();
     }
     LOG.info(String.format("Reading the list of already copied metrics from %s", processedMetricsFilePath));
     try {
       try (Stream<String> stream = Files.lines(Paths.get(processedMetricsFilePath))) {
         stream.forEach( line -> {
           String [] lineSplit = line.split(":");
           if (!result.containsKey(lineSplit[0])) {
             result.put(lineSplit[0], new HashSet<>(Collections.singletonList(lineSplit[1])));
           } else {
             result.get(lineSplit[0]).add(lineSplit[1]);
           }
         });
       }
     } catch (IOException e) {
       LOG.error(e);
     }
     this.processedMetrics = result;
   }

   public void runMigration(Long timeoutInMinutes) throws IOException {

     FileWriter processedMetricsFileWriter = new FileWriter(processedMetricsFilePath, true);
     LOG.info("Setting up copiers...");
     Set<AbstractPhoenixMetricsCopier> copiers = new HashSet<>();
     for (Set<String> batch : metricNamesBatches) {
       for (Map.Entry<String, String> entry : CLUSTER_AGGREGATE_TABLES_MAPPING.entrySet()) {
         Set<String> filteredMetrics = filterProcessedMetrics(batch, this.processedMetrics, entry.getKey());
         if (!filteredMetrics.isEmpty()) {
           copiers.add(new PhoenixClusterMetricsCopier(entry.getKey(), entry.getValue(), hBaseAccessor,
             filteredMetrics, startTime, processedMetricsFileWriter));
         }
       }

       for (Map.Entry<String, String> entry : HOST_AGGREGATE_TABLES_MAPPING.entrySet()) {
         Set<String> filteredMetrics = filterProcessedMetrics(batch, processedMetrics, entry.getKey());
         if (!filteredMetrics.isEmpty()) {
           copiers.add(new PhoenixHostMetricsCopier(entry.getKey(), entry.getValue(), hBaseAccessor,
             filteredMetrics, startTime, processedMetricsFileWriter));
         }
       }
     }

     if (copiers.isEmpty()) {
       LOG.info("No copy threads to run, looks like all metrics have been copied.");
       processedMetricsFileWriter.close();
       return;
     }

     LOG.info("Running the copy threads...");
     long startTimer = System.currentTimeMillis();
     ExecutorService executorService = Executors.newFixedThreadPool(numberOfThreads == null ? DEFAULT_NUMBER_OF_THREADS : numberOfThreads);
     for (AbstractPhoenixMetricsCopier copier : copiers) {
       executorService.submit(copier);
     }

     executorService.shutdown();

     try {
       executorService.awaitTermination(timeoutInMinutes, TimeUnit.MINUTES);
     } catch (InterruptedException e) {
       LOG.error(e);
     }

     long estimatedTime = System.currentTimeMillis() - startTimer;
     LOG.info(String.format("Copying took %s seconds", estimatedTime/1000.0));

     processedMetricsFileWriter.close();
   }

   private void initializeHbaseAccessor() throws MalformedURLException, URISyntaxException {
     this.hBaseAccessor = new PhoenixHBaseAccessor(null);
     this.timelineMetricConfiguration = TimelineMetricConfiguration.getInstance();
     timelineMetricConfiguration.initialize();

     TimelineMetricMetadataManager timelineMetricMetadataManager = new TimelineMetricMetadataManager(hBaseAccessor);
     timelineMetricConfiguration.initialize();
     hBaseAccessor.setMetadataInstance(timelineMetricMetadataManager);
   }

   private static Set<String> filterProcessedMetrics(Set<String> metricNames, Map<String, Set<String>> processedMetrics, String tableName) {
     if (!processedMetrics.containsKey(tableName)) {
       return metricNames;
     }
     return Sets.difference(metricNames, processedMetrics.get(tableName));
   }

   /**
    * reads metric names from given file
    * replaces all * with % and removes all ._p_
    */
   private static Set<String> readMetricWhitelistFromFile(String whitelistFile) {
     LOG.info(String.format("Reading metric names from %s", whitelistFile));
     Set<String> whitelistedMetrics = new HashSet<>();

     BufferedReader br = null;
     String strLine;

     try(FileInputStream fstream = new FileInputStream(whitelistFile)) {
       br = new BufferedReader(new InputStreamReader(fstream));

       while ((strLine = br.readLine()) != null)   {
         strLine = strLine.trim();
         if (StringUtils.isEmpty(strLine)) {
           continue;
         }
         if (strLine.startsWith(patternPrefix)) {
           strLine = strLine.replace(patternPrefix, "");
         }
         if (strLine.contains("*")) {
           strLine = strLine.replaceAll("\\*", "%");
         }
         whitelistedMetrics.add(strLine);
       }
     } catch (IOException ioEx) {
       LOG.error(ioEx);
     }
     return whitelistedMetrics;
   }


   /**
    *
    * @param args
    * REQUIRED args[0] - processedMetricsFilePath - full path to the file where processed metric are/will be stored
    *
    * OPTIONAL args[1] - whitelistedFilePath      - full path to the file with whitelisted metrics filenames
    *                                               if not provided the default whitelist file location will be used if configured
    *                                               if not configured - will result in error
    *          args[2] - startTime                - default value is set to the last 30 days
    *          args[3] - numberOfThreads          - default value is 3
    *          args[4] - batchSize                - default value is 5
    *          args[5] - timeoutInMinutes         - default value is set to the equivalent of 24 hours
    */
   public static void main(String[] args) {
     String processedMetricsFilePath = null;
     String whitelistedFilePath = null;
     Long startTime = null;
     Integer numberOfThreads = null;
     Integer batchSize = null;
     Long timeoutInMinutes = DEFAULT_TIMEOUT_MINUTES;

     if (args.length>0) {
       processedMetricsFilePath = args[0];
     }
     if (args.length>1) {
       whitelistedFilePath = args[1];
     }
     if (args.length>2) {
       startTime = Long.valueOf(args[2]);
     }
     if (args.length>3) {
       numberOfThreads = Integer.valueOf(args[3]);
     }
     if (args.length>4) {
       batchSize = Integer.valueOf(args[4]);
     }
     if (args.length>5) {
       timeoutInMinutes = Long.valueOf(args[5]);
     }

     MetricsDataMigrationLauncher dataMigrationLauncher = null;
     try {
       LOG.info("Initializing system...");
       dataMigrationLauncher = new MetricsDataMigrationLauncher(whitelistedFilePath, processedMetricsFilePath, startTime, numberOfThreads, batchSize);
     } catch (Exception e) {
       LOG.error("Exception during system setup, exiting...", e);
       System.exit(1);
     }

     try {
       dataMigrationLauncher.runMigration(timeoutInMinutes);
     } catch (IOException e) {
       LOG.error("Exception during data migration, exiting...", e);
       System.exit(1);
     }

     System.exit(0);

   }
 }
	/**
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.ambari.metrics.core.timeline.upgrade.core;

	import com.google.common.collect.Iterables;
	import com.google.common.collect.Sets;
	import org.apache.ambari.metrics.core.timeline.PhoenixHBaseAccessor;
	import org.apache.ambari.metrics.core.timeline.TimelineMetricConfiguration;
	import org.apache.ambari.metrics.core.timeline.discovery.TimelineMetricMetadataManager;
	import org.apache.commons.lang.StringUtils;
	import org.apache.commons.logging.Log;
	import org.apache.commons.logging.LogFactory;

	import java.io.BufferedReader;
	import java.io.FileInputStream;
	import java.io.FileWriter;
	import java.io.IOException;
	import java.io.InputStreamReader;
	import java.net.MalformedURLException;
	import java.net.URISyntaxException;
	import java.nio.file.Files;
	import java.nio.file.Paths;
	import java.util.Collections;
	import java.util.HashMap;
	import java.util.HashSet;
	import java.util.Map;
	import java.util.Set;
	import java.util.concurrent.ExecutorService;
	import java.util.concurrent.Executors;
	import java.util.concurrent.TimeUnit;
	import java.util.stream.Stream;

	import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_AGGREGATE_DAILY_TABLE_NAME;
	import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_AGGREGATE_DAILY_V1_TABLE_NAME;
	import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_AGGREGATE_HOURLY_TABLE_NAME;
	import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_AGGREGATE_HOURLY_V1_TABLE_NAME;
	import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_AGGREGATE_MINUTE_TABLE_NAME;
	import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_AGGREGATE_MINUTE_V1_TABLE_NAME;
	import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_CLUSTER_AGGREGATE_DAILY_TABLE_NAME;
	import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_CLUSTER_AGGREGATE_DAILY_V1_TABLE_NAME;
	import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_CLUSTER_AGGREGATE_HOURLY_TABLE_NAME;
	import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_CLUSTER_AGGREGATE_HOURLY_V1_TABLE_NAME;
	import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_CLUSTER_AGGREGATE_MINUTE_TABLE_NAME;
	import static org.apache.ambari.metrics.core.timeline.query.PhoenixTransactSQL.METRICS_CLUSTER_AGGREGATE_MINUTE_V1_TABLE_NAME;

	public class MetricsDataMigrationLauncher {
	private static final Log LOG = LogFactory.getLog(MetricsDataMigrationLauncher.class);
	private static final Long DEFAULT_TIMEOUT_MINUTES = 60*24L;
	private static String patternPrefix = "._p_";
	private static final int DEFAULT_BATCH_SIZE = 5;
	public static final Map<String, String> CLUSTER_AGGREGATE_TABLES_MAPPING = new HashMap<>();
	public static final Map<String, String> HOST_AGGREGATE_TABLES_MAPPING = new HashMap<>();
	public static final String DEFAULT_PROCESSED_METRICS_FILE_LOCATION = "/var/log/ambari-metrics-collector/ambari-metrics-migration-state.txt";
	public static final int DEFAULT_NUMBER_OF_THREADS = 3;
	public static final long ONE_MONTH_MILLIS = 2592000000L;
	public static final long DEFAULT_START_TIME = System.currentTimeMillis() - ONE_MONTH_MILLIS; //Last month

	static {
	CLUSTER_AGGREGATE_TABLES_MAPPING.put(METRICS_CLUSTER_AGGREGATE_MINUTE_V1_TABLE_NAME, METRICS_CLUSTER_AGGREGATE_MINUTE_TABLE_NAME);
	CLUSTER_AGGREGATE_TABLES_MAPPING.put(METRICS_CLUSTER_AGGREGATE_HOURLY_V1_TABLE_NAME, METRICS_CLUSTER_AGGREGATE_HOURLY_TABLE_NAME);
	CLUSTER_AGGREGATE_TABLES_MAPPING.put(METRICS_CLUSTER_AGGREGATE_DAILY_V1_TABLE_NAME, METRICS_CLUSTER_AGGREGATE_DAILY_TABLE_NAME);

	HOST_AGGREGATE_TABLES_MAPPING.put(METRICS_AGGREGATE_MINUTE_V1_TABLE_NAME, METRICS_AGGREGATE_MINUTE_TABLE_NAME);
	HOST_AGGREGATE_TABLES_MAPPING.put(METRICS_AGGREGATE_HOURLY_V1_TABLE_NAME, METRICS_AGGREGATE_HOURLY_TABLE_NAME);
	HOST_AGGREGATE_TABLES_MAPPING.put(METRICS_AGGREGATE_DAILY_V1_TABLE_NAME, METRICS_AGGREGATE_DAILY_TABLE_NAME);
	}

	private final Set<Set<String>> metricNamesBatches;
	private final String processedMetricsFilePath;
	private Set<String> metricNames;

	private Long startTime;
	private Integer batchSize;
	private Integer numberOfThreads;
	private TimelineMetricConfiguration timelineMetricConfiguration;
	private PhoenixHBaseAccessor hBaseAccessor;
	private Map<String, Set<String>> processedMetrics;

	public MetricsDataMigrationLauncher(String whitelistedFilePath, String processedMetricsFilePath, Long startTime, Integer numberOfThreads, Integer batchSize) throws Exception {
	this.startTime = startTime == null? DEFAULT_START_TIME : startTime;
	this.numberOfThreads = numberOfThreads == null? DEFAULT_NUMBER_OF_THREADS : numberOfThreads;
	this.batchSize = batchSize == null? DEFAULT_BATCH_SIZE : batchSize;
	this.processedMetricsFilePath = processedMetricsFilePath == null? DEFAULT_PROCESSED_METRICS_FILE_LOCATION : processedMetricsFilePath;

	initializeHbaseAccessor();

	LOG.info("Looking for whitelisted metric names...");

	if (whitelistedFilePath != null) {
	this.metricNames = readMetricWhitelistFromFile(whitelistedFilePath);
	} else if (timelineMetricConfiguration.isWhitelistingEnabled()) {
	String whitelistFile = timelineMetricConfiguration.getMetricsConf().get(TimelineMetricConfiguration.TIMELINE_METRICS_WHITELIST_FILE, TimelineMetricConfiguration.TIMELINE_METRICS_WHITELIST_FILE_LOCATION_DEFAULT);
	metricNames = readMetricWhitelistFromFile(whitelistFile);
	} else {
	LOG.error("No whitelisted metrics specified. Exiting...");
	throw new Exception("List of whitelisted metrics must be provided");
	}

	readProcessedMetricsMap();

	LOG.info("Setting up batches...");
	this.metricNamesBatches = new HashSet<>();

	Iterables.partition(metricNames, this.batchSize)
	.forEach(batch -> metricNamesBatches.add(new HashSet<>(batch)));
	LOG.info(String.format("Split metric names into %s batches with size of %s", metricNamesBatches.size(), this.batchSize));
	}


	private void readProcessedMetricsMap() {
	Map<String, Set<String>> result = new HashMap<>();
	if (!Files.exists(Paths.get(processedMetricsFilePath))) {
	LOG.info(String.format("The processed metrics file %s is missing, assuming there were no metrics processed.", processedMetricsFilePath));
	this.processedMetrics = new HashMap<>();
	}
	LOG.info(String.format("Reading the list of already copied metrics from %s", processedMetricsFilePath));
	try {
	try (Stream<String> stream = Files.lines(Paths.get(processedMetricsFilePath))) {
	stream.forEach( line -> {
	String [] lineSplit = line.split(":");
	if (!result.containsKey(lineSplit[0])) {
	result.put(lineSplit[0], new HashSet<>(Collections.singletonList(lineSplit[1])));
	} else {
	result.get(lineSplit[0]).add(lineSplit[1]);
	}
	});
	}
	} catch (IOException e) {
	LOG.error(e);
	}
	this.processedMetrics = result;
	}

	public void runMigration(Long timeoutInMinutes) throws IOException {

	FileWriter processedMetricsFileWriter = new FileWriter(processedMetricsFilePath, true);
	LOG.info("Setting up copiers...");
	Set<AbstractPhoenixMetricsCopier> copiers = new HashSet<>();
	for (Set<String> batch : metricNamesBatches) {
	for (Map.Entry<String, String> entry : CLUSTER_AGGREGATE_TABLES_MAPPING.entrySet()) {
	Set<String> filteredMetrics = filterProcessedMetrics(batch, this.processedMetrics, entry.getKey());
	if (!filteredMetrics.isEmpty()) {
	copiers.add(new PhoenixClusterMetricsCopier(entry.getKey(), entry.getValue(), hBaseAccessor,
	filteredMetrics, startTime, processedMetricsFileWriter));
	}
	}

	for (Map.Entry<String, String> entry : HOST_AGGREGATE_TABLES_MAPPING.entrySet()) {
	Set<String> filteredMetrics = filterProcessedMetrics(batch, processedMetrics, entry.getKey());
	if (!filteredMetrics.isEmpty()) {
	copiers.add(new PhoenixHostMetricsCopier(entry.getKey(), entry.getValue(), hBaseAccessor,
	filteredMetrics, startTime, processedMetricsFileWriter));
	}
	}
	}

	if (copiers.isEmpty()) {
	LOG.info("No copy threads to run, looks like all metrics have been copied.");
	processedMetricsFileWriter.close();
	return;
	}

	LOG.info("Running the copy threads...");
	long startTimer = System.currentTimeMillis();
	ExecutorService executorService = Executors.newFixedThreadPool(numberOfThreads == null ? DEFAULT_NUMBER_OF_THREADS : numberOfThreads);
	for (AbstractPhoenixMetricsCopier copier : copiers) {
	executorService.submit(copier);
	}

	executorService.shutdown();

	try {
	executorService.awaitTermination(timeoutInMinutes, TimeUnit.MINUTES);
	} catch (InterruptedException e) {
	LOG.error(e);
	}

	long estimatedTime = System.currentTimeMillis() - startTimer;
	LOG.info(String.format("Copying took %s seconds", estimatedTime/1000.0));

	processedMetricsFileWriter.close();
	}

	private void initializeHbaseAccessor() throws MalformedURLException, URISyntaxException {
	this.hBaseAccessor = new PhoenixHBaseAccessor(null);
	this.timelineMetricConfiguration = TimelineMetricConfiguration.getInstance();
	timelineMetricConfiguration.initialize();

	TimelineMetricMetadataManager timelineMetricMetadataManager = new TimelineMetricMetadataManager(hBaseAccessor);
	timelineMetricConfiguration.initialize();
	hBaseAccessor.setMetadataInstance(timelineMetricMetadataManager);
	}

	private static Set<String> filterProcessedMetrics(Set<String> metricNames, Map<String, Set<String>> processedMetrics, String tableName) {
	if (!processedMetrics.containsKey(tableName)) {
	return metricNames;
	}
	return Sets.difference(metricNames, processedMetrics.get(tableName));
	}

	/**
	* reads metric names from given file
	* replaces all * with % and removes all ._p_
	*/
	private static Set<String> readMetricWhitelistFromFile(String whitelistFile) {
	LOG.info(String.format("Reading metric names from %s", whitelistFile));
	Set<String> whitelistedMetrics = new HashSet<>();

	BufferedReader br = null;
	String strLine;

	try(FileInputStream fstream = new FileInputStream(whitelistFile)) {
	br = new BufferedReader(new InputStreamReader(fstream));

	while ((strLine = br.readLine()) != null) {
	strLine = strLine.trim();
	if (StringUtils.isEmpty(strLine)) {
	continue;
	}
	if (strLine.startsWith(patternPrefix)) {
	strLine = strLine.replace(patternPrefix, "");
	}
	if (strLine.contains("*")) {
	strLine = strLine.replaceAll("\\*", "%");
	}
	whitelistedMetrics.add(strLine);
	}
	} catch (IOException ioEx) {
	LOG.error(ioEx);
	}
	return whitelistedMetrics;
	}


	/**
	*
	* @param args
	* REQUIRED args[0] - processedMetricsFilePath - full path to the file where processed metric are/will be stored
	*
	* OPTIONAL args[1] - whitelistedFilePath - full path to the file with whitelisted metrics filenames
	* if not provided the default whitelist file location will be used if configured
	* if not configured - will result in error
	* args[2] - startTime - default value is set to the last 30 days
	* args[3] - numberOfThreads - default value is 3
	* args[4] - batchSize - default value is 5
	* args[5] - timeoutInMinutes - default value is set to the equivalent of 24 hours
	*/
	public static void main(String[] args) {
	String processedMetricsFilePath = null;
	String whitelistedFilePath = null;
	Long startTime = null;
	Integer numberOfThreads = null;
	Integer batchSize = null;
	Long timeoutInMinutes = DEFAULT_TIMEOUT_MINUTES;

	if (args.length>0) {
	processedMetricsFilePath = args[0];
	}
	if (args.length>1) {
	whitelistedFilePath = args[1];
	}
	if (args.length>2) {
	startTime = Long.valueOf(args[2]);
	}
	if (args.length>3) {
	numberOfThreads = Integer.valueOf(args[3]);
	}
	if (args.length>4) {
	batchSize = Integer.valueOf(args[4]);
	}
	if (args.length>5) {
	timeoutInMinutes = Long.valueOf(args[5]);
	}

	MetricsDataMigrationLauncher dataMigrationLauncher = null;
	try {
	LOG.info("Initializing system...");
	dataMigrationLauncher = new MetricsDataMigrationLauncher(whitelistedFilePath, processedMetricsFilePath, startTime, numberOfThreads, batchSize);
	} catch (Exception e) {
	LOG.error("Exception during system setup, exiting...", e);
	System.exit(1);
	}

	try {
	dataMigrationLauncher.runMigration(timeoutInMinutes);
	} catch (IOException e) {
	LOG.error("Exception during data migration, exiting...", e);
	System.exit(1);
	}

	System.exit(0);

	}
	}