CernVM-FS  2.9.0
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Pages
swissknife_ingest.cc
Go to the documentation of this file.
1 
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 "statistics_database.h"
16 #include "sync_mediator.h"
17 #include "sync_union.h"
18 #include "sync_union_tarball.h"
19 #include "util/pointer.h"
20 #include "util/posix.h"
21 
22 /*
23  * Many of the options possible to set in the ArgumentList are not actually used
24  * by the ingest command since they are not part of its interface, hence those
25  * unused options cannot be set by the shell script. Of course if there is the
26  * necessitty those paramenters can be added and managed.
27  * At the moment this approach worked fine and didn't add much complexity,
28  * however if yet another command will need to use a similar approach it would
29  * be good to consider creating different options handler for each command.
30  */
32  std::string start_time = GetGMTimestamp();
33 
34  SyncParameters params;
35  params.dir_rdonly = MakeCanonicalPath(*args.find('c')->second);
36  params.dir_temp = MakeCanonicalPath(*args.find('t')->second);
37  params.base_hash = shash::MkFromHexPtr(shash::HexPtr(*args.find('b')->second),
39  params.stratum0 = *args.find('w')->second;
40  params.manifest_path = *args.find('o')->second;
41  params.spooler_definition = *args.find('r')->second;
42 
43  params.public_keys = *args.find('K')->second;
44  params.repo_name = *args.find('N')->second;
45 
46  if (args.find('T') != args.end()) {
47  params.tar_file = *args.find('T')->second;
48  }
49  if (args.find('B') != args.end()) {
50  params.base_directory = *args.find('B')->second;
51  }
52  if (args.find('D') != args.end()) {
53  params.to_delete = *args.find('D')->second;
54  }
55 
56  if (args.find('O') != args.end()) {
57  params.generate_legacy_bulk_chunks = true;
58  }
59  shash::Algorithms hash_algorithm = shash::kSha1;
60  if (args.find('e') != args.end()) {
61  hash_algorithm = shash::ParseHashAlgorithm(*args.find('e')->second);
62  if (hash_algorithm == shash::kAny) {
63  PrintError("unknown hash algorithm");
64  return 1;
65  }
66  }
67  if (args.find('Z') != args.end()) {
68  params.compression_alg =
69  zlib::ParseCompressionAlgorithm(*args.find('Z')->second);
70  }
71 
72  bool create_catalog = args.find('C') != args.end();
73 
77 
78  params.branched_catalog = false; // could be true?
79 
80  if (args.find('P') != args.end()) {
81  params.session_token_file = *args.find('P')->second;
82  }
83 
84  if (args.find('H') != args.end()) {
85  params.key_file = *args.find('H')->second;
86  }
87 
88  const bool upload_statsdb = (args.count('I') > 0);
89 
90  perf::StatisticsTemplate publish_statistics("publish", this->statistics());
91  StatisticsDatabase *stats_db =
93 
94  upload::SpoolerDefinition spooler_definition(
95  params.spooler_definition, hash_algorithm, params.compression_alg,
98  params.max_file_chunk_size, params.session_token_file, params.key_file);
99  if (params.max_concurrent_write_jobs > 0) {
100  spooler_definition.number_of_concurrent_uploads =
102  }
103 
104  upload::SpoolerDefinition spooler_definition_catalogs(
105  spooler_definition.Dup2DefaultCompression());
106 
107  params.spooler = upload::Spooler::Construct(spooler_definition,
108  &publish_statistics);
109  if (NULL == params.spooler) return 3;
111  upload::Spooler::Construct(spooler_definition_catalogs,
112  &publish_statistics));
113  if (!spooler_catalogs.IsValid()) return 3;
114 
115  const bool follow_redirects = (args.count('L') > 0);
116  if (!InitDownloadManager(follow_redirects)) {
117  return 3;
118  }
119 
121  params.trusted_certs)) {
122  return 3;
123  }
124 
125  bool with_gateway =
126  spooler_definition.driver_type == upload::SpoolerDefinition::Gateway;
127 
128  // This may fail, in which case a warning is printed and the process continues
130 
132  if (params.branched_catalog) {
133  // Throw-away manifest
134  manifest = new manifest::Manifest(shash::Any(), 0, "");
135  } else {
136  if (with_gateway) {
137  manifest =
139  } else {
140  manifest = FetchRemoteManifest(params.stratum0, params.repo_name,
141  params.base_hash);
142  }
143  }
144  if (!manifest.IsValid()) {
145  return 3;
146  }
147 
148  const std::string old_root_hash = manifest->catalog_hash().ToString(true);
149 
150  catalog::WritableCatalogManager catalog_manager(
151  params.base_hash, params.stratum0, params.dir_temp,
152  spooler_catalogs.weak_ref(),
155  params.is_balanced, params.max_weight, params.min_weight);
156  catalog_manager.Init();
157 
158  publish::SyncMediator mediator(&catalog_manager, &params, publish_statistics);
159  LogCvmfs(kLogPublish, kLogStdout, "Processing changes...");
160 
161  publish::SyncUnion *sync;
162 
163  sync = new publish::SyncUnionTarball(&mediator, params.dir_rdonly,
164  params.tar_file, params.base_directory,
165  params.to_delete, create_catalog);
166  if (!sync->Initialize()) {
168  "Initialization of the synchronisation "
169  "engine failed");
170  return 4;
171  }
172 
173  sync->Traverse();
174 
175  if (!params.authz_file.empty()) {
177  "Adding contents of authz file %s to"
178  " root catalog.",
179  params.authz_file.c_str());
180  int fd = open(params.authz_file.c_str(), O_RDONLY);
181  if (fd == -1) {
183  "Unable to open authz file (%s)"
184  "from the publication process: %s",
185  params.authz_file.c_str(), strerror(errno));
186  return 7;
187  }
188 
189  std::string new_authz;
190  const bool read_successful = SafeReadToString(fd, &new_authz);
191  close(fd);
192 
193  if (!read_successful) {
194  LogCvmfs(kLogCvmfs, kLogStderr, "Failed to read authz file (%s): %s",
195  params.authz_file.c_str(), strerror(errno));
196  return 8;
197  }
198 
199  catalog_manager.SetVOMSAuthz(new_authz);
200  }
201 
202  if (!mediator.Commit(manifest.weak_ref())) {
203  PrintError("something went wrong during sync");
204  stats_db->StorePublishStatistics(this->statistics(), start_time, false);
205  if (upload_statsdb) {
206  stats_db->UploadStatistics(params.spooler);
207  }
208  return 5;
209  }
210 
211  perf::Counter *revision_counter = statistics()->Register("publish.revision",
212  "Published revision number");
213  revision_counter->Set(catalog_manager.GetRootCatalog()->revision());
214 
215  // finalize the spooler
216  LogCvmfs(kLogCvmfs, kLogStdout, "Wait for all uploads to finish");
217  params.spooler->WaitForUpload();
218  spooler_catalogs->WaitForUpload();
219  params.spooler->FinalizeSession(false);
220 
221  LogCvmfs(kLogCvmfs, kLogStdout, "Exporting repository manifest");
222 
223  // We call FinalizeSession(true) this time, to also trigger the commit
224  // operation on the gateway machine (if the upstream is of type "gw").
225 
226  // Get the path of the new root catalog
227  const std::string new_root_hash = manifest->catalog_hash().ToString(true);
228 
229  if (!spooler_catalogs->FinalizeSession(true, old_root_hash, new_root_hash,
230  params.repo_tag)) {
231  PrintError("Failed to commit the transaction.");
232  stats_db->StorePublishStatistics(this->statistics(), start_time, false);
233  if (upload_statsdb) {
234  stats_db->UploadStatistics(params.spooler);
235  }
236  return 9;
237  }
238 
239  stats_db->StorePublishStatistics(this->statistics(), start_time, true);
240  if (upload_statsdb) {
241  stats_db->UploadStatistics(params.spooler);
242  }
243 
244  delete params.spooler;
245 
246  if (!manifest->Export(params.manifest_path)) {
247  PrintError("Failed to create new repository");
248  return 6;
249  }
250 
251  return 0;
252 }
bool Commit(manifest::Manifest *manifest)
#define LogCvmfs(source, mask,...)
Definition: logging.h:20
Counter * Register(const std::string &name, const std::string &desc)
Definition: statistics.cc:139
Algorithms ParseCompressionAlgorithm(const std::string &algorithm_option)
Definition: compression.cc:148
size_t avg_file_chunk_size
const manifest::Manifest * manifest() const
Definition: repository.h:123
std::string stratum0
DriverType driver_type
the type of the spooler driver
std::string GetGMTimestamp(const std::string &format)
Definition: string.cc:585
bool UploadStatistics(upload::Spooler *spooler, std::string local_path="")
T * weak_ref() const
Definition: pointer.h:42
SpoolerDefinition Dup2DefaultCompression() const
bool generate_legacy_bulk_chunks
unsigned file_mbyte_limit
uint64_t max_concurrent_write_jobs
static const unsigned kDefaultFileMbyteLimit
virtual bool Initialize()
Definition: sync_union.cc:24
std::string to_delete
static const unsigned kDefaultNestedKcatalogLimit
static const unsigned kDefaultRootKcatalogLimit
size_t min_file_chunk_size
std::string spooler_definition
std::string manifest_path
manifest::Manifest * FetchRemoteManifest(const std::string &repository_url, const std::string &repository_name, const shash::Any &base_hash=shash::Any()) const
Definition: server_tool.cc:133
zlib::Algorithms compression_alg
unsigned nested_kcatalog_limit
std::string repo_name
int Main(const ArgumentList &args)
bool InitVerifyingSignatureManager(const std::string &pubkey_path, const std::string &trusted_certs="")
Definition: server_tool.cc:44
Algorithms
Definition: hash.h:39
std::string key_file
std::string dir_temp
std::string base_directory
download::DownloadManager * download_manager() const
Definition: server_tool.cc:108
void Set(const int64_t val)
Definition: statistics.h:33
bool InitDownloadManager(const bool follow_redirects, const unsigned max_pool_handles=1, const bool use_system_proxy=true)
Definition: server_tool.cc:21
std::string dir_rdonly
const char kSuffixCatalog
Definition: hash.h:52
perf::Statistics * statistics()
Definition: server_tool.h:49
shash::Any base_hash
upload::Spooler * spooler
Any MkFromHexPtr(const HexPtr hex, const char suffix)
Definition: hash.cc:83
bool IsValid() const
Definition: pointer.h:43
static StatisticsDatabase * OpenStandardDB(const std::string repo_name)
bool SafeReadToString(int fd, std::string *final_result)
Definition: posix.cc:2006
unsigned root_kcatalog_limit
size_t max_file_chunk_size
bool StorePublishStatistics(const perf::Statistics *statistics, const std::string &start_time, const bool success)
std::string session_token_file
std::string trusted_certs
bool ObtainDacReadSearchCapability()
virtual void Traverse()=0
std::map< char, SharedPtr< std::string > > ArgumentList
Definition: swissknife.h:72
Algorithms ParseHashAlgorithm(const string &algorithm_option)
Definition: hash.cc:72
std::string authz_file
RepositoryTag repo_tag
std::string MakeCanonicalPath(const std::string &path)
Definition: posix.cc:96
std::string tar_file
void PrintError(const string &message)
Definition: logging.cc:465
const upload::Spooler * spooler_catalogs() const
Definition: repository.h:311
std::string public_keys