<commit_before>/**
 * This file is part of the CernVM File System.
 */

#include "swissknife_migrate.h"

#include "catalog_traversal.h"
#include "catalog_sql.h"
#include "catalog_rw.h"

#include "logging.h"

#include <sys/resource.h>
#include <sstream>

using namespace swissknife;
using namespace catalog;

CommandMigrate::CommandMigrate() :
  file_descriptor_limit_(8192),
  catalog_count_(0),
  root_catalog_(NULL)
{
  atomic_init32(&catalogs_processed_);
}


ParameterList CommandMigrate::GetParams() {
  ParameterList result;
  result.push_back(Parameter('r', "repository URL (absolute local path "
                                  "or remote URL)",
                             false, false));
  result.push_back(Parameter('u', "upstream definition string",
                             false, false));
  result.push_back(Parameter('o', "manifest output file",
                             false, false));
  result.push_back(Parameter('t', "temporary directory for catalog decompress",
                             false, false));
  result.push_back(Parameter('n', "fully qualified repository name",
                             true, false));
  result.push_back(Parameter('k', "repository master key(s)",
                             true, false));
  result.push_back(Parameter('f', "fix nested catalog transition points",
                             true, true));
  result.push_back(Parameter('l', "disable linkcount analysis of files",
                             true, true));
  result.push_back(Parameter('s', "enable collection of catalog statistics",
                             true, true));
  return result;
}


void Error(const std::string &message) {
  LogCvmfs(kLogCatalog, kLogStderr, message.c_str());
}


void Error(const std::string                     &message,
           const CommandMigrate::PendingCatalog  *catalog) {
  std::stringstream ss;
  ss << message << std::endl
     << "Catalog: " << catalog->root_path();
  Error(ss.str());
}


void Error(const std::string                     &message,
           const catalog::Sql                    &statement,
           const CommandMigrate::PendingCatalog  *catalog) {
  std::stringstream ss;
  ss << message << std::endl
     << "SQLite:  " << statement.GetLastError() << " - "
                    << statement.GetLastErrorMsg();
  Error(ss.str(), catalog);
}


int CommandMigrate::Main(const ArgumentList &args) {
  // parameter parsing
  const std::string &repo_url           = *args.find('r')->second;
  const std::string &spooler            = *args.find('u')->second;
  const std::string &manifest_path      = *args.find('o')->second;
  const std::string &decompress_tmp_dir = *args.find('t')->second;
  const std::string &repo_name          = (args.count('n') > 0)      ?
                                             *args.find('n')->second :
                                             "";
  const std::string &repo_keys          = (args.count('k') > 0)      ?
                                             *args.find('k')->second :
                                             "";
  const bool fix_transition_points      = (args.count('f') > 0);
  const bool analyze_file_linkcounts    = (args.count('l') == 0);
  const bool collect_catalog_statistics = (args.count('s') > 0);

  // we might need a lot of file descriptors
  if (! RaiseFileDescriptorLimit()) {
    Error("Failed to raise file descriptor limits");
    return 2;
  }

  // load the full catalog hierarchy
  LogCvmfs(kLogCatalog, kLogStdout, "Loading current catalog tree...");
  const bool generate_full_catalog_tree = true;
  CatalogTraversal<CommandMigrate> traversal(
    this,
    &CommandMigrate::CatalogCallback,
    repo_url,
    repo_name,
    repo_keys,
    generate_full_catalog_tree,
    decompress_tmp_dir);
  const bool loading_successful = traversal.Traverse();

  if (!loading_successful) {
    Error("Failed to load catalog tree");
    return 5;
  }

  assert (root_catalog_ != NULL);

  // create an upstream spooler
  const upload::SpoolerDefinition spooler_definition(spooler);
  spooler_ = upload::Spooler::Construct(spooler_definition);
  if (!spooler_) {
    Error("Failed to create upstream Spooler.");
    return 3;
  }
  spooler_->RegisterListener(&CommandMigrate::UploadCallback, this);

  // create a concurrent catalog migration facility
  const unsigned int cpus = GetNumberOfCpuCores();
  MigrationWorker::worker_context context(spooler_definition.temporary_path,
                                          catalog_statistics_list_,
                                          fix_transition_points,
                                          analyze_file_linkcounts,
                                          collect_catalog_statistics);
  concurrent_migration_ = new ConcurrentWorkers<MigrationWorker>(
                                cpus,
                                cpus * 10,
                                &context);
  if (! concurrent_migration_->Initialize()) {
    Error("Failed to initialize worker migration system.");
    return 4;<commit_msg>inform about number of loaded catalogs<commit_after>/**
 * This file is part of the CernVM File System.
 */

#include "swissknife_migrate.h"

#include "catalog_traversal.h"
#include "catalog_sql.h"
#include "catalog_rw.h"

#include "logging.h"

#include <sys/resource.h>
#include <sstream>

using namespace swissknife;
using namespace catalog;

CommandMigrate::CommandMigrate() :
  file_descriptor_limit_(8192),
  catalog_count_(0),
  root_catalog_(NULL)
{
  atomic_init32(&catalogs_processed_);
}


ParameterList CommandMigrate::GetParams() {
  ParameterList result;
  result.push_back(Parameter('r', "repository URL (absolute local path "
                                  "or remote URL)",
                             false, false));
  result.push_back(Parameter('u', "upstream definition string",
                             false, false));
  result.push_back(Parameter('o', "manifest output file",
                             false, false));
  result.push_back(Parameter('t', "temporary directory for catalog decompress",
                             false, false));
  result.push_back(Parameter('n', "fully qualified repository name",
                             true, false));
  result.push_back(Parameter('k', "repository master key(s)",
                             true, false));
  result.push_back(Parameter('f', "fix nested catalog transition points",
                             true, true));
  result.push_back(Parameter('l', "disable linkcount analysis of files",
                             true, true));
  result.push_back(Parameter('s', "enable collection of catalog statistics",
                             true, true));
  return result;
}


void Error(const std::string &message) {
  LogCvmfs(kLogCatalog, kLogStderr, message.c_str());
}


void Error(const std::string                     &message,
           const CommandMigrate::PendingCatalog  *catalog) {
  std::stringstream ss;
  ss << message << std::endl
     << "Catalog: " << catalog->root_path();
  Error(ss.str());
}


void Error(const std::string                     &message,
           const catalog::Sql                    &statement,
           const CommandMigrate::PendingCatalog  *catalog) {
  std::stringstream ss;
  ss << message << std::endl
     << "SQLite:  " << statement.GetLastError() << " - "
                    << statement.GetLastErrorMsg();
  Error(ss.str(), catalog);
}


int CommandMigrate::Main(const ArgumentList &args) {
  // parameter parsing
  const std::string &repo_url           = *args.find('r')->second;
  const std::string &spooler            = *args.find('u')->second;
  const std::string &manifest_path      = *args.find('o')->second;
  const std::string &decompress_tmp_dir = *args.find('t')->second;
  const std::string &repo_name          = (args.count('n') > 0)      ?
                                             *args.find('n')->second :
                                             "";
  const std::string &repo_keys          = (args.count('k') > 0)      ?
                                             *args.find('k')->second :
                                             "";
  const bool fix_transition_points      = (args.count('f') > 0);
  const bool analyze_file_linkcounts    = (args.count('l') == 0);
  const bool collect_catalog_statistics = (args.count('s') > 0);

  // we might need a lot of file descriptors
  if (! RaiseFileDescriptorLimit()) {
    Error("Failed to raise file descriptor limits");
    return 2;
  }

  // load the full catalog hierarchy
  LogCvmfs(kLogCatalog, kLogStdout, "Loading current catalog tree...");
  const bool generate_full_catalog_tree = true;
  CatalogTraversal<CommandMigrate> traversal(
    this,
    &CommandMigrate::CatalogCallback,
    repo_url,
    repo_name,
    repo_keys,
    generate_full_catalog_tree,
    decompress_tmp_dir);
  const bool loading_successful = traversal.Traverse();

  if (!loading_successful) {
    Error("Failed to load catalog tree");
    return 5;
  }

  LogCvmfs(kLogCatalog, kLogStdout, "Loaded %d catalogs", catalog_count_);
  assert (root_catalog_ != NULL);

  // create an upstream spooler
  const upload::SpoolerDefinition spooler_definition(spooler);
  spooler_ = upload::Spooler::Construct(spooler_definition);
  if (!spooler_) {
    Error("Failed to create upstream Spooler.");
    return 3;
  }
  spooler_->RegisterListener(&CommandMigrate::UploadCallback, this);

  // create a concurrent catalog migration facility
  const unsigned int cpus = GetNumberOfCpuCores();
  MigrationWorker::worker_context context(spooler_definition.temporary_path,
                                          catalog_statistics_list_,
                                          fix_transition_points,
                                          analyze_file_linkcounts,
                                          collect_catalog_statistics);
  concurrent_migration_ = new ConcurrentWorkers<MigrationWorker>(
                                cpus,
                                cpus * 10,
                                &context);
  if (! concurrent_migration_->Initialize()) {
    Error("Failed to initialize worker migration system.");
    return 4;
{'fraction_non_alphanumeric': 0.11417632043981936, 'fraction_numerical': 0.0035342627135283725, 'mean_word_length': 2.2463352453792225, 'pattern_counts': {'":': 0, '<': 51, '<?xml version=': 0, '>': 37, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 34, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3366749', 'n_tokens_mistral': 2787, 'n_tokens_neox': 2721, 'n_words': 634}