Head

GCC Code Coverage Report

Directory:	cvmfs/		Exec	Total	Coverage
File:	cvmfs/swissknife_ingest.cc	Lines:	0	94	0.0 %
Date:	2019-02-03 02:48:13	Branches:	0	52	0.0 %


/**
 * This file is part of the CernVM File System
 */

#include "swissknife_ingest.h"

#include <fcntl.h>
#include <unistd.h>

#include "catalog_virtual.h"
#include "logging.h"
#include "manifest.h"
#include "statistics.h"
#include "sync_mediator.h"
#include "sync_union.h"
#include "sync_union_tarball.h"
#include "util/pointer.h"
#include "util/posix.h"

/*
 * Many of the options possible to set in the ArgumentList are not actually used
 * by the ingest command since they are not part of its interface, hence those
 * unused options cannot be set by the shell script. Of course if there is the
 * necessitty those paramenters can be added and managed.
 * At the moment this approach worked fine and didn't add much complexity,
 * however if yet another command will need to use a similar approach it would
 * be good to consider creating different options handler for each command.
 */
int swissknife::Ingest::Main(const swissknife::ArgumentList &args) {
  SyncParameters params;
  params.dir_rdonly = MakeCanonicalPath(*args.find('c')->second);
  params.dir_temp = MakeCanonicalPath(*args.find('t')->second);
  params.base_hash = shash::MkFromHexPtr(shash::HexPtr(*args.find('b')->second),
                                         shash::kSuffixCatalog);
  params.stratum0 = *args.find('w')->second;
  params.manifest_path = *args.find('o')->second;
  params.spooler_definition = *args.find('r')->second;

  params.public_keys = *args.find('K')->second;
  params.repo_name = *args.find('N')->second;

  if (args.find('T') != args.end()) {
    params.tar_file = *args.find('T')->second;
  }
  if (args.find('B') != args.end()) {
    params.base_directory = *args.find('B')->second;
  }
  if (args.find('D') != args.end()) {
    params.to_delete = *args.find('D')->second;
  }

  if (args.find('O') != args.end()) {
    params.generate_legacy_bulk_chunks = true;
  }
  shash::Algorithms hash_algorithm = shash::kSha1;
  if (args.find('e') != args.end()) {
    hash_algorithm = shash::ParseHashAlgorithm(*args.find('e')->second);
    if (hash_algorithm == shash::kAny) {
      PrintError("unknown hash algorithm");
      return 1;
    }
  }
  if (args.find('Z') != args.end()) {
    params.compression_alg =
        zlib::ParseCompressionAlgorithm(*args.find('Z')->second);
  }

  bool create_catalog = args.find('C') != args.end();

  params.nested_kcatalog_limit = SyncParameters::kDefaultNestedKcatalogLimit;
  params.root_kcatalog_limit = SyncParameters::kDefaultRootKcatalogLimit;
  params.file_mbyte_limit = SyncParameters::kDefaultFileMbyteLimit;

  params.branched_catalog = false;  // could be true?

  if (args.find('P') != args.end()) {
    params.session_token_file = *args.find('P')->second;
  }

  if (args.find('H') != args.end()) {
    params.key_file = *args.find('H')->second;
  }

  perf::StatisticsTemplate publish_statistics("Publish", this->statistics());

  upload::SpoolerDefinition spooler_definition(
      params.spooler_definition, hash_algorithm, params.compression_alg,
      params.generate_legacy_bulk_chunks, params.use_file_chunking,
      params.min_file_chunk_size, params.avg_file_chunk_size,
      params.max_file_chunk_size, params.session_token_file, params.key_file);
  if (params.max_concurrent_write_jobs > 0) {
    spooler_definition.number_of_concurrent_uploads =
        params.max_concurrent_write_jobs;
  }

  upload::SpoolerDefinition spooler_definition_catalogs(
      spooler_definition.Dup2DefaultCompression());

  params.spooler = upload::Spooler::Construct(spooler_definition,
                                              &publish_statistics);
  if (NULL == params.spooler) return 3;
  UniquePtr<upload::Spooler> spooler_catalogs(
      upload::Spooler::Construct(spooler_definition_catalogs));
  if (!spooler_catalogs.IsValid()) return 3;

  const bool follow_redirects = (args.count('L') > 0);
  if (!InitDownloadManager(follow_redirects)) {
    return 3;
  }

  if (!InitVerifyingSignatureManager(params.public_keys,
                                     params.trusted_certs)) {
    return 3;
  }

  bool with_gateway =
      spooler_definition.driver_type == upload::SpoolerDefinition::Gateway;

  UniquePtr<manifest::Manifest> manifest;
  if (params.branched_catalog) {
    // Throw-away manifest
    manifest = new manifest::Manifest(shash::Any(), 0, "");
  } else {
    if (with_gateway) {
      manifest =
          FetchRemoteManifest(params.stratum0, params.repo_name, shash::Any());
    } else {
      manifest = FetchRemoteManifest(params.stratum0, params.repo_name,
                                     params.base_hash);
    }
  }
  if (!manifest) {
    return 3;
  }

  const std::string old_root_hash = manifest->catalog_hash().ToString(true);

  catalog::WritableCatalogManager catalog_manager(
      params.base_hash, params.stratum0, params.dir_temp, spooler_catalogs,
      download_manager(), params.enforce_limits, params.nested_kcatalog_limit,
      params.root_kcatalog_limit, params.file_mbyte_limit, statistics(),
      params.is_balanced, params.max_weight, params.min_weight);
  catalog_manager.Init();

  publish::SyncMediator mediator(&catalog_manager, &params, publish_statistics);

  publish::SyncUnion *sync;

  sync = new publish::SyncUnionTarball(&mediator, params.dir_rdonly,
                                       params.tar_file, params.base_directory,
                                       params.to_delete, create_catalog);
  if (!sync->Initialize()) {
    LogCvmfs(kLogCvmfs, kLogStderr,
             "Initialization of the synchronisation "
             "engine failed");
    return 4;
  }

  sync->Traverse();

  if (!params.authz_file.empty()) {
    LogCvmfs(kLogCvmfs, kLogDebug,
             "Adding contents of authz file %s to"
             " root catalog.",
             params.authz_file.c_str());
    int fd = open(params.authz_file.c_str(), O_RDONLY);
    if (fd == -1) {
      LogCvmfs(kLogCvmfs, kLogStderr,
               "Unable to open authz file (%s)"
               "from the publication process: %s",
               params.authz_file.c_str(), strerror(errno));
      return 7;
    }

    std::string new_authz;
    const bool read_successful = SafeReadToString(fd, &new_authz);
    close(fd);

    if (!read_successful) {
      LogCvmfs(kLogCvmfs, kLogStderr, "Failed to read authz file (%s): %s",
               params.authz_file.c_str(), strerror(errno));
      return 8;
    }

    catalog_manager.SetVOMSAuthz(new_authz);
  }

  if (!mediator.Commit(manifest.weak_ref())) {
    PrintError("something went wrong during sync");
    return 5;
  }

  // finalize the spooler
  LogCvmfs(kLogCvmfs, kLogStdout, "Wait for all uploads to finish");
  params.spooler->WaitForUpload();
  spooler_catalogs->WaitForUpload();
  params.spooler->FinalizeSession(false);

  LogCvmfs(kLogCvmfs, kLogStdout, "Exporting repository manifest");

  // We call FinalizeSession(true) this time, to also trigger the commit
  // operation on the gateway machine (if the upstream is of type "gw").

  // Get the path of the new root catalog
  const std::string new_root_hash = manifest->catalog_hash().ToString(true);

  spooler_catalogs->FinalizeSession(true, old_root_hash, new_root_hash,
                                    params.repo_tag);
  delete params.spooler;

  if (!manifest->Export(params.manifest_path)) {
    PrintError("Failed to create new repository");
    return 6;
  }

  return 0;
}


Generated by: GCOVR (Version 4.1)

Line	Branch	Exec	Source
1			/**
2			* This file is part of the CernVM File System
3			*/
4
5			#include "swissknife_ingest.h"
6
7			#include <fcntl.h>
8			#include <unistd.h>
9
10			#include "catalog_virtual.h"
11			#include "logging.h"
12			#include "manifest.h"
13			#include "statistics.h"
14			#include "sync_mediator.h"
15			#include "sync_union.h"
16			#include "sync_union_tarball.h"
17			#include "util/pointer.h"
18			#include "util/posix.h"
19
20			/*
21			* Many of the options possible to set in the ArgumentList are not actually used
22			* by the ingest command since they are not part of its interface, hence those
23			* unused options cannot be set by the shell script. Of course if there is the
24			* necessitty those paramenters can be added and managed.
25			* At the moment this approach worked fine and didn't add much complexity,
26			* however if yet another command will need to use a similar approach it would
27			* be good to consider creating different options handler for each command.
28			*/
29			int swissknife::Ingest::Main(const swissknife::ArgumentList &args) {
30			SyncParameters params;
31			params.dir_rdonly = MakeCanonicalPath(*args.find('c')->second);
32			params.dir_temp = MakeCanonicalPath(*args.find('t')->second);
33			params.base_hash = shash::MkFromHexPtr(shash::HexPtr(*args.find('b')->second),
34			shash::kSuffixCatalog);
35			params.stratum0 = *args.find('w')->second;
36			params.manifest_path = *args.find('o')->second;
37			params.spooler_definition = *args.find('r')->second;
38
39			params.public_keys = *args.find('K')->second;
40			params.repo_name = *args.find('N')->second;
41
42			if (args.find('T') != args.end()) {
43			params.tar_file = *args.find('T')->second;
44			}
45			if (args.find('B') != args.end()) {
46			params.base_directory = *args.find('B')->second;
47			}
48			if (args.find('D') != args.end()) {
49			params.to_delete = *args.find('D')->second;
50			}
51
52			if (args.find('O') != args.end()) {
53			params.generate_legacy_bulk_chunks = true;
54			}
55			shash::Algorithms hash_algorithm = shash::kSha1;
56			if (args.find('e') != args.end()) {
57			hash_algorithm = shash::ParseHashAlgorithm(*args.find('e')->second);
58			if (hash_algorithm == shash::kAny) {
59			PrintError("unknown hash algorithm");
60			return 1;
61			}
62			}
63			if (args.find('Z') != args.end()) {
64			params.compression_alg =
65			zlib::ParseCompressionAlgorithm(*args.find('Z')->second);
66			}
67
68			bool create_catalog = args.find('C') != args.end();
69
70			params.nested_kcatalog_limit = SyncParameters::kDefaultNestedKcatalogLimit;
71			params.root_kcatalog_limit = SyncParameters::kDefaultRootKcatalogLimit;
72			params.file_mbyte_limit = SyncParameters::kDefaultFileMbyteLimit;
73
74			params.branched_catalog = false; // could be true?
75
76			if (args.find('P') != args.end()) {
77			params.session_token_file = *args.find('P')->second;
78			}
79
80			if (args.find('H') != args.end()) {
81			params.key_file = *args.find('H')->second;
82			}
83
84			perf::StatisticsTemplate publish_statistics("Publish", this->statistics());
85
86			upload::SpoolerDefinition spooler_definition(
87			params.spooler_definition, hash_algorithm, params.compression_alg,
88			params.generate_legacy_bulk_chunks, params.use_file_chunking,
89			params.min_file_chunk_size, params.avg_file_chunk_size,
90			params.max_file_chunk_size, params.session_token_file, params.key_file);
91			if (params.max_concurrent_write_jobs > 0) {
92			spooler_definition.number_of_concurrent_uploads =
93			params.max_concurrent_write_jobs;
94			}
95
96			upload::SpoolerDefinition spooler_definition_catalogs(
97			spooler_definition.Dup2DefaultCompression());
98
99			params.spooler = upload::Spooler::Construct(spooler_definition,
100			&publish_statistics);
101			if (NULL == params.spooler) return 3;
102			UniquePtr<upload::Spooler> spooler_catalogs(
103			upload::Spooler::Construct(spooler_definition_catalogs));
104			if (!spooler_catalogs.IsValid()) return 3;
105
106			const bool follow_redirects = (args.count('L') > 0);
107			if (!InitDownloadManager(follow_redirects)) {
108			return 3;
109			}
110
111			if (!InitVerifyingSignatureManager(params.public_keys,
112			params.trusted_certs)) {
113			return 3;
114			}
115
116			bool with_gateway =
117			spooler_definition.driver_type == upload::SpoolerDefinition::Gateway;
118
119			UniquePtr<manifest::Manifest> manifest;
120			if (params.branched_catalog) {
121			// Throw-away manifest
122			manifest = new manifest::Manifest(shash::Any(), 0, "");
123			} else {
124			if (with_gateway) {
125			manifest =
126			FetchRemoteManifest(params.stratum0, params.repo_name, shash::Any());
127			} else {
128			manifest = FetchRemoteManifest(params.stratum0, params.repo_name,
129			params.base_hash);
130			}
131			}
132			if (!manifest) {
133			return 3;
134			}
135
136			const std::string old_root_hash = manifest->catalog_hash().ToString(true);
137
138			catalog::WritableCatalogManager catalog_manager(
139			params.base_hash, params.stratum0, params.dir_temp, spooler_catalogs,
140			download_manager(), params.enforce_limits, params.nested_kcatalog_limit,
141			params.root_kcatalog_limit, params.file_mbyte_limit, statistics(),
142			params.is_balanced, params.max_weight, params.min_weight);
143			catalog_manager.Init();
144
145			publish::SyncMediator mediator(&catalog_manager, &params, publish_statistics);
146
147			publish::SyncUnion *sync;
148
149			sync = new publish::SyncUnionTarball(&mediator, params.dir_rdonly,
150			params.tar_file, params.base_directory,
151			params.to_delete, create_catalog);
152			if (!sync->Initialize()) {
153			LogCvmfs(kLogCvmfs, kLogStderr,
154			"Initialization of the synchronisation "
155			"engine failed");
156			return 4;
157			}
158
159			sync->Traverse();
160
161			if (!params.authz_file.empty()) {
162			LogCvmfs(kLogCvmfs, kLogDebug,
163			"Adding contents of authz file %s to"
164			" root catalog.",
165			params.authz_file.c_str());
166			int fd = open(params.authz_file.c_str(), O_RDONLY);
167			if (fd == -1) {
168			LogCvmfs(kLogCvmfs, kLogStderr,
169			"Unable to open authz file (%s)"
170			"from the publication process: %s",
171			params.authz_file.c_str(), strerror(errno));
172			return 7;
173			}
174
175			std::string new_authz;
176			const bool read_successful = SafeReadToString(fd, &new_authz);
177			close(fd);
178
179			if (!read_successful) {
180			LogCvmfs(kLogCvmfs, kLogStderr, "Failed to read authz file (%s): %s",
181			params.authz_file.c_str(), strerror(errno));
182			return 8;
183			}
184
185			catalog_manager.SetVOMSAuthz(new_authz);
186			}
187
188			if (!mediator.Commit(manifest.weak_ref())) {
189			PrintError("something went wrong during sync");
190			return 5;
191			}
192
193			// finalize the spooler
194			LogCvmfs(kLogCvmfs, kLogStdout, "Wait for all uploads to finish");
195			params.spooler->WaitForUpload();
196			spooler_catalogs->WaitForUpload();
197			params.spooler->FinalizeSession(false);
198
199			LogCvmfs(kLogCvmfs, kLogStdout, "Exporting repository manifest");
200
201			// We call FinalizeSession(true) this time, to also trigger the commit
202			// operation on the gateway machine (if the upstream is of type "gw").
203
204			// Get the path of the new root catalog
205			const std::string new_root_hash = manifest->catalog_hash().ToString(true);
206
207			spooler_catalogs->FinalizeSession(true, old_root_hash, new_root_hash,
208			params.repo_tag);
209			delete params.spooler;
210
211			if (!manifest->Export(params.manifest_path)) {
212			PrintError("Failed to create new repository");
213			return 6;
214			}
215
216			return 0;
217			}