CernVM-FS  2.12.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 "manifest.h"
12 #include "statistics.h"
13 #include "statistics_database.h"
15 #include "sync_mediator.h"
16 #include "sync_union.h"
17 #include "sync_union_tarball.h"
18 #include "util/logging.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 parameters 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  if (args.find('U') != args.end()) {
72  params.uid = static_cast<uid_t>(String2Int64(*args.find('U')->second));
73  }
74  if (args.find('G') != args.end()) {
75  params.gid = static_cast<gid_t>(String2Int64(*args.find('G')->second));
76  }
77 
78  bool create_catalog = args.find('C') != args.end();
79 
83 
84  params.branched_catalog = false; // could be true?
85 
86  if (args.find('P') != args.end()) {
87  params.session_token_file = *args.find('P')->second;
88  }
89 
90  if (args.find('H') != args.end()) {
91  params.key_file = *args.find('H')->second;
92  }
93 
94  const bool upload_statsdb = (args.count('I') > 0);
95 
96  perf::StatisticsTemplate publish_statistics("publish", this->statistics());
97  StatisticsDatabase *stats_db =
99 
100  upload::SpoolerDefinition spooler_definition(
101  params.spooler_definition, hash_algorithm, params.compression_alg,
104  params.max_file_chunk_size, params.session_token_file, params.key_file);
105  if (params.max_concurrent_write_jobs > 0) {
106  spooler_definition.number_of_concurrent_uploads =
108  }
109 
110  // Sanitize base_directory, removing any leading or trailing slashes
111  // from non-root (!= "/") paths
112  params.base_directory = TrimString(params.base_directory, "/", kTrimAll);
113 
114  upload::SpoolerDefinition spooler_definition_catalogs(
115  spooler_definition.Dup2DefaultCompression());
116 
117  params.spooler = upload::Spooler::Construct(spooler_definition,
118  &publish_statistics);
119  if (NULL == params.spooler) return 3;
121  upload::Spooler::Construct(spooler_definition_catalogs,
122  &publish_statistics));
123  if (!spooler_catalogs.IsValid()) return 3;
124 
125  const bool follow_redirects = (args.count('L') > 0);
126  const string proxy = (args.count('@') > 0) ? *args.find('@')->second : "";
127  if (!InitDownloadManager(follow_redirects, proxy)) {
128  return 3;
129  }
130 
132  params.trusted_certs)) {
133  return 3;
134  }
135 
136  bool with_gateway =
137  spooler_definition.driver_type == upload::SpoolerDefinition::Gateway;
138 
139  // This may fail, in which case a warning is printed and the process continues
141 
143  if (params.branched_catalog) {
144  // Throw-away manifest
145  manifest = new manifest::Manifest(shash::Any(), 0, "");
146  } else {
147  if (with_gateway) {
148  manifest =
150  } else {
151  manifest = FetchRemoteManifest(params.stratum0, params.repo_name,
152  params.base_hash);
153  }
154  }
155  if (!manifest.IsValid()) {
156  return 3;
157  }
158 
159  const std::string old_root_hash = manifest->catalog_hash().ToString(true);
160 
161  catalog::WritableCatalogManager catalog_manager(
162  params.base_hash, params.stratum0, params.dir_temp,
163  spooler_catalogs.weak_ref(),
166  params.is_balanced, params.max_weight, params.min_weight);
167  catalog_manager.Init();
168 
169  publish::SyncMediator mediator(&catalog_manager, &params, publish_statistics);
170  LogCvmfs(kLogPublish, kLogStdout, "Processing changes...");
171 
173  &mediator, params.dir_rdonly, params.tar_file, params.base_directory,
174  params.uid, params.gid, params.to_delete, create_catalog);
175 
176  if (!sync->Initialize()) {
178  "Initialization of the synchronisation "
179  "engine failed");
180  return 4;
181  }
182 
183  sync->Traverse();
184 
185  if (!params.authz_file.empty()) {
187  "Adding contents of authz file %s to"
188  " root catalog.",
189  params.authz_file.c_str());
190  int fd = open(params.authz_file.c_str(), O_RDONLY);
191  if (fd == -1) {
193  "Unable to open authz file (%s)"
194  "from the publication process: %s",
195  params.authz_file.c_str(), strerror(errno));
196  return 7;
197  }
198 
199  std::string new_authz;
200  const bool read_successful = SafeReadToString(fd, &new_authz);
201  close(fd);
202 
203  if (!read_successful) {
204  LogCvmfs(kLogCvmfs, kLogStderr, "Failed to read authz file (%s): %s",
205  params.authz_file.c_str(), strerror(errno));
206  return 8;
207  }
208 
209  catalog_manager.SetVOMSAuthz(new_authz);
210  }
211 
212  if (!mediator.Commit(manifest.weak_ref())) {
213  PrintError("something went wrong during sync");
214  stats_db->StorePublishStatistics(this->statistics(), start_time, false);
215  if (upload_statsdb) {
216  stats_db->UploadStatistics(params.spooler);
217  }
218  return 5;
219  }
220 
221  perf::Counter *revision_counter = statistics()->Register("publish.revision",
222  "Published revision number");
223  revision_counter->Set(catalog_manager.GetRootCatalog()->revision());
224 
225  // finalize the spooler
226  LogCvmfs(kLogCvmfs, kLogStdout, "Wait for all uploads to finish");
227  params.spooler->WaitForUpload();
228  spooler_catalogs->WaitForUpload();
229  params.spooler->FinalizeSession(false);
230 
231  LogCvmfs(kLogCvmfs, kLogStdout, "Exporting repository manifest");
232 
233  // We call FinalizeSession(true) this time, to also trigger the commit
234  // operation on the gateway machine (if the upstream is of type "gw").
235 
236  // Get the path of the new root catalog
237  const std::string new_root_hash = manifest->catalog_hash().ToString(true);
238 
239  if (!spooler_catalogs->FinalizeSession(true, old_root_hash, new_root_hash,
240  params.repo_tag)) {
241  PrintError("Failed to commit the transaction.");
242  stats_db->StorePublishStatistics(this->statistics(), start_time, false);
243  if (upload_statsdb) {
244  stats_db->UploadStatistics(params.spooler);
245  }
246  return 9;
247  }
248 
249  stats_db->StorePublishStatistics(this->statistics(), start_time, true);
250  if (upload_statsdb) {
251  stats_db->UploadStatistics(params.spooler);
252  }
253 
254  delete params.spooler;
255 
256  if (!manifest->Export(params.manifest_path)) {
257  PrintError("Failed to create new repository");
258  return 6;
259  }
260 
261  return 0;
262 }
bool Commit(manifest::Manifest *manifest)
Counter * Register(const std::string &name, const std::string &desc)
Definition: statistics.cc:160
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:125
std::string stratum0
DriverType driver_type
the type of the spooler driver
std::string GetGMTimestamp(const std::string &format)
Definition: string.cc:615
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:41
std::string key_file
std::string dir_temp
int64_t String2Int64(const string &value)
Definition: string.cc:222
std::string base_directory
download::DownloadManager * download_manager() const
Definition: server_tool.cc:108
void Set(const int64_t val)
Definition: statistics.h:33
std::string dir_rdonly
const char kSuffixCatalog
Definition: hash.h:54
const int kTrimAll
Definition: string.h:21
perf::Statistics * statistics()
Definition: server_tool.h:49
shash::Any base_hash
upload::Spooler * spooler
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:2065
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 TrimString(const std::string &path, const std::string &toTrim, const int trimMode)
Definition: string.cc:450
std::string authz_file
Any MkFromHexPtr(const HexPtr hex, const char suffix)
Definition: hash.cc:83
RepositoryTag repo_tag
std::string MakeCanonicalPath(const std::string &path)
Definition: posix.cc:98
std::string tar_file
void PrintError(const string &message)
Definition: logging.cc:543
const upload::Spooler * spooler_catalogs() const
Definition: repository.h:322
bool InitDownloadManager(const bool follow_redirects, const std::string &proxy, const unsigned max_pool_handles=1)
Definition: server_tool.cc:17
std::string public_keys
CVMFS_EXPORT void LogCvmfs(const LogSource source, const int mask, const char *format,...)
Definition: logging.cc:528