gerrit.onap Code Review - sdc.git/blob

   1 /**
   2  * Copyright © 2016-2017 European Support Limited.
   3  */
   4 package org.openecomp.core.tools.exportinfo;
   5
   6 import com.datastax.driver.core.ResultSet;
   7 import com.datastax.driver.core.ResultSetFuture;
   8 import com.datastax.driver.core.Session;
   9 import com.google.common.collect.Sets;
  10 import com.google.common.util.concurrent.FutureCallback;
  11 import com.google.common.util.concurrent.Futures;
  12 import org.apache.commons.io.FileUtils;
  13 import org.openecomp.core.nosqldb.impl.cassandra.CassandraSessionFactory;
  14 import org.openecomp.core.tools.importinfo.ImportProperties;
  15 import org.openecomp.core.tools.util.Utils;
  16 import org.openecomp.core.tools.util.ZipUtils;
  17 import org.openecomp.core.zusammen.impl.CassandraConnectionInitializer;
  18 import org.openecomp.sdc.logging.api.Logger;
  19 import org.openecomp.sdc.logging.api.LoggerFactory;
  20 import org.yaml.snakeyaml.Yaml;
  21
  22 import java.io.File;
  23 import java.io.IOException;
  24 import java.io.InputStream;
  25 import java.nio.file.Path;
  26 import java.nio.file.Paths;
  27 import java.time.LocalDateTime;
  28 import java.time.format.DateTimeFormatter;
  29 import java.util.HashSet;
  30 import java.util.List;
  31 import java.util.Map;
  32 import java.util.Set;
  33 import java.util.concurrent.CountDownLatch;
  34 import java.util.concurrent.Executor;
  35 import java.util.concurrent.ExecutorService;
  36 import java.util.concurrent.Executors;
  37 import java.util.stream.Collectors;
  38
  39 import static java.nio.file.Files.createDirectories;
  40
  41
  42 public final class ExportDataCommand {
  43     private static final Logger logger = LoggerFactory.getLogger(ExportDataCommand.class);
  44     public static final String JOIN_DELIMITER = "$#";
  45     public static final String JOIN_DELIMITER_SPLITTER = "\\$\\#";
  46     public static final String MAP_DELIMITER = "!@";
  47     public static final String MAP_DELIMITER_SPLITTER = "\\!\\@";
  48     public static final int THREAD_POOL_SIZE = 6;
  49
  50     private ExportDataCommand() {
  51     }
  52
  53     public static void exportData(String filterItem) {
  54         ExecutorService executor = null;
  55         try {
  56             CassandraConnectionInitializer.setCassandraConnectionPropertiesToSystem();
  57             Path rootDir = Paths.get(ImportProperties.ROOT_DIRECTORY);
  58             initDir(rootDir);
  59             try (Session session = CassandraSessionFactory.getSession()) {
  60                 final Set<String> filteredItems = Sets.newHashSet(filterItem);
  61                 Set<String> fis = filteredItems.stream().map(fi -> fi.replaceAll("\\r", "")).collect(Collectors.toSet());
  62                 Map<String, List<String>> queries;
  63                 Yaml yaml = new Yaml();
  64                 try (InputStream is = ExportDataCommand.class.getResourceAsStream("/queries.yaml")) {
  65                     queries = (Map<String, List<String>>) yaml.load(is);
  66                 }
  67                 List<String> queriesList = queries.get("queries");
  68                 List<String> itemsColumns = queries.get("item_columns");
  69                 Set<String> vlms = new HashSet<>();
  70                 CountDownLatch doneQueries = new CountDownLatch(queriesList.size());
  71                 executor = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
  72                 for (int i = 0; i < queriesList.size(); i++) {
  73                     executeQuery(session, queriesList.get(i), fis, itemsColumns.get(i), vlms, doneQueries, executor);
  74                 }
  75                 doneQueries.await();
  76                 if (!vlms.isEmpty()) {
  77                     CountDownLatch doneVmls = new CountDownLatch(queriesList.size());
  78                     for (int i = 0; i < queriesList.size(); i++) {
  79                         executeQuery(session, queriesList.get(i), vlms, itemsColumns.get(i), null, doneVmls, executor);
  80                     }
  81
  82                     doneVmls.await();
  83                 }
  84             }
  85             zipPath(rootDir);
  86             FileUtils.forceDelete(rootDir.toFile());
  87         } catch (Exception ex) {
  88             Utils.logError(logger, ex);
  89         } finally {
  90             if (executor != null) {
  91                 executor.shutdown();
  92             }
  93         }
  94
  95     }
  96
  97
  98     private static void executeQuery(final Session session, final String query, final Set<String> filteredItems, final String filteredColumn,
  99                                         final Set<String> vlms, final CountDownLatch donequerying, Executor executor) {
 100         ResultSetFuture resultSetFuture = session.executeAsync(query);
 101         Futures.addCallback(resultSetFuture, new FutureCallback<ResultSet>() {
 102             @Override
 103             public void onSuccess(ResultSet resultSet) {
 104                 try {
 105                     Utils.printMessage(logger, "Start to serialize " + query);
 106                     new ExportSerializer().serializeResult(resultSet, filteredItems, filteredColumn, vlms);
 107                     donequerying.countDown();
 108                 } catch (Exception e){
 109                     Utils.logError(logger, "Serialization failed :" + query, e);
 110                     System.exit(-1);
 111                 }
 112             }
 113
 114             @Override
 115             public void onFailure(Throwable t) {
 116                 Utils.logError(logger, "Query failed :" + query, t);
 117                 System.exit(-1);
 118             }
 119         }, executor);
 120     }
 121
 122     private static void zipPath(Path rootDir) throws IOException {
 123         LocalDateTime date = LocalDateTime.now();
 124         DateTimeFormatter formatter = DateTimeFormatter.ISO_LOCAL_DATE_TIME;
 125         String dateStr = date.format(formatter);
 126         dateStr = dateStr.replaceAll(":", "_");
 127         String zipFile = System.getProperty("user.home") + File.separatorChar + "onboarding_import" + dateStr + ".zip";
 128         ZipUtils.createZip(zipFile, rootDir);
 129         Utils.printMessage(logger, "Exported file :" + zipFile);
 130     }
 131
 132
 133     public static void initDir(Path rootDir) throws IOException {
 134         if (rootDir.toFile().exists()) {
 135             FileUtils.forceDelete(rootDir.toFile());
 136         }
 137         createDirectories(rootDir);
 138     }
 139
 140 }