gerrit.onap Code Review - sdc.git/blob

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