CernVM-FS  2.9.0
 All Classes Namespaces Files Functions Variables Typedefs Enumerations Enumerator Friends Macros Pages
commit_processor.cc
Go to the documentation of this file.
1 
5 #include "commit_processor.h"
6 
7 #include <time.h>
8 
9 #include <vector>
10 
11 #include "catalog_diff_tool.h"
12 #include "catalog_merge_tool.h"
13 #include "catalog_mgr_ro.h"
14 #include "catalog_mgr_rw.h"
15 #include "compression.h"
16 #include "download.h"
17 #include "logging.h"
18 #include "manifest.h"
19 #include "manifest_fetch.h"
20 #include "params.h"
21 #include "signing_tool.h"
22 #include "statistics.h"
23 #include "statistics_database.h"
24 #include "swissknife.h"
25 #include "swissknife_history.h"
26 #include "util/algorithm.h"
27 #include "util/pointer.h"
28 #include "util/posix.h"
29 #include "util/raii_temp_dir.h"
30 #include "util/string.h"
31 
32 namespace {
33 
34 PathString RemoveRepoName(const PathString& lease_path) {
35  std::string abs_path = lease_path.ToString();
36  std::string::const_iterator it =
37  std::find(abs_path.begin(), abs_path.end(), '/');
38  if (it != abs_path.end()) {
39  size_t idx = it - abs_path.begin() + 1;
40  return lease_path.Suffix(idx);
41  } else {
42  return lease_path;
43  }
44 }
45 
46 bool CreateNewTag(const RepositoryTag& repo_tag, const std::string& repo_name,
47  const receiver::Params& params, const std::string& temp_dir,
48  const std::string& manifest_path,
49  const std::string& public_key_path) {
51  args['r'].Reset(new std::string(params.spooler_configuration));
52  args['w'].Reset(new std::string(params.stratum0));
53  args['t'].Reset(new std::string(temp_dir));
54  args['m'].Reset(new std::string(manifest_path));
55  args['p'].Reset(new std::string(public_key_path));
56  args['f'].Reset(new std::string(repo_name));
57  args['e'].Reset(new std::string(params.hash_alg_str));
58  args['a'].Reset(new std::string(repo_tag.name_));
59  args['c'].Reset(new std::string(repo_tag.channel_));
60  args['D'].Reset(new std::string(repo_tag.description_));
61  args['x'].Reset(new std::string());
62 
65  const int ret = edit_cmd->Main(args);
66 
67  if (ret) {
68  LogCvmfs(kLogReceiver, kLogSyslogErr, "Error %d creating tag: %s", ret,
69  repo_tag.name_.c_str());
70  return false;
71  }
72 
73  return true;
74 }
75 
76 } // namespace
77 
78 namespace receiver {
79 
80 CommitProcessor::CommitProcessor() : num_errors_(0), statistics_(NULL) {}
81 
83 
101  const std::string& lease_path, const shash::Any& old_root_hash,
102  const shash::Any& new_root_hash, const RepositoryTag& tag,
103  uint64_t *final_revision) {
104  RepositoryTag final_tag = tag;
105  // If tag_name is a generic tag, update the time stamp
106  if (HasPrefix(final_tag.name_, "generic-", false)) {
107  // format time following the ISO 8601 YYYY-MM-DDThh:mm:ss.sssZ
108  // note the millisecond accurracy
109  uint64_t nanoseconds_timestamp = platform_realtime_ns();
110 
111  time_t seconds = nanoseconds_timestamp / 1000000000; // 1E9
112  struct tm timestamp;
113  gmtime_r(&seconds, &timestamp);
114  char seconds_buffer[32];
115  strftime(seconds_buffer, sizeof(seconds_buffer),
116  "generic-%Y-%m-%dT%H:%M:%S", &timestamp);
117 
118  // first we get the raw nanoseconds from the timestamp using the module
119  // and then we divide to extract the millisecond.
120  // the division truncate the number brutally, it should be enough.
121  time_t milliseconds = (nanoseconds_timestamp % 1000000000) / 1000000;
122  char millis_buffer[48];
123  snprintf(millis_buffer, sizeof(millis_buffer), "%s.%03ldZ", seconds_buffer,
124  milliseconds);
125 
126  final_tag.name_ = std::string(millis_buffer);
127  }
128 
130  "CommitProcessor - lease_path: %s, old hash: %s, new hash: %s, "
131  "tag_name: %s, tag_channel: %s, tag_description: %s",
132  lease_path.c_str(), old_root_hash.ToString(true).c_str(),
133  new_root_hash.ToString(true).c_str(), final_tag.name_.c_str(),
134  final_tag.channel_.c_str(), final_tag.description_.c_str());
135 
136  const std::vector<std::string> lease_path_tokens =
137  SplitString(lease_path, '/');
138 
139  const std::string repo_name = lease_path_tokens.front();
140 
141  Params params;
142  if (!GetParamsFromFile(repo_name, &params)) {
143  LogCvmfs(
145  "CommitProcessor - error: Could not get configuration parameters.");
146  return kError;
147  }
148 
149  UniquePtr<ServerTool> server_tool(new ServerTool());
150 
151  if (!server_tool->InitDownloadManager(true)) {
152  LogCvmfs(
154  "CommitProcessor - error: Could not initialize the download manager");
155  return kError;
156  }
157 
158  const std::string public_key = "/etc/cvmfs/keys/" + repo_name + ".pub";
159  const std::string trusted_certs =
160  "/etc/cvmfs/repositories.d/" + repo_name + "/trusted_certs";
161  if (!server_tool->InitVerifyingSignatureManager(public_key, trusted_certs)) {
162  LogCvmfs(
164  "CommitProcessor - error: Could not initialize the signature manager");
165  return kError;
166  }
167 
168  shash::Any manifest_base_hash;
170  params.stratum0, repo_name, manifest_base_hash));
171 
172  // Current catalog from the gateway machine
173  if (!manifest.IsValid()) {
175  "CommitProcessor - error: Could not open repository manifest");
176  return kError;
177  }
178 
180  "CommitProcessor - lease_path: %s, target root hash: %s",
181  lease_path.c_str(),
182  manifest->catalog_hash().ToString(false).c_str());
183 
184  const std::string spooler_temp_dir =
186  assert(!spooler_temp_dir.empty());
187  assert(MkdirDeep(spooler_temp_dir + "/receiver", 0666, true));
188  const std::string temp_dir_root =
189  spooler_temp_dir + "/receiver/commit_processor";
190 
191  const PathString relative_lease_path = RemoveRepoName(PathString(lease_path));
192 
194  "CommitProcessor - lease_path: %s, merging catalogs",
195  lease_path.c_str());
196 
199  merge_tool(params.stratum0, old_root_hash, new_root_hash,
200  relative_lease_path, temp_dir_root,
201  server_tool->download_manager(), manifest.weak_ref(),
202  statistics_);
203  if (!merge_tool.Init()) {
205  "Error: Could not initialize the catalog merge tool");
206  return kError;
207  }
208 
209  std::string new_manifest_path;
210  if (!merge_tool.Run(params, &new_manifest_path, final_revision)) {
212  "CommitProcessor - error: Catalog merge failed");
213  return kMergeFailure;
214  }
215 
216  UniquePtr<RaiiTempDir> raii_temp_dir(RaiiTempDir::Create(temp_dir_root));
217  const std::string temp_dir = raii_temp_dir->dir();
218  const std::string certificate = "/etc/cvmfs/keys/" + repo_name + ".crt";
219  const std::string private_key = "/etc/cvmfs/keys/" + repo_name + ".key";
220 
221  if (!CreateNewTag(final_tag, repo_name, params, temp_dir, new_manifest_path,
222  public_key)) {
223  LogCvmfs(kLogReceiver, kLogSyslogErr, "Error creating tag: %s",
224  final_tag.name_.c_str());
225  return kError;
226  }
227 
228  // We need to re-initialize the ServerTool component for signing
229  server_tool.Destroy();
230  server_tool = new ServerTool();
231 
233  "CommitProcessor - lease_path: %s, signing manifest",
234  lease_path.c_str());
235 
236  // Add C_N root catalog hash to reflog through SigningTool,
237  // so garbage collector can later delete it.
238  std::vector<shash::Any> reflog_catalogs;
239  reflog_catalogs.push_back(new_root_hash);
240 
241  SigningTool signing_tool(server_tool.weak_ref());
242  SigningTool::Result res = signing_tool.Run(
243  new_manifest_path, params.stratum0, params.spooler_configuration,
244  temp_dir, certificate, private_key, repo_name, "", "",
245  "/var/spool/cvmfs/" + repo_name + "/reflog.chksum",
246  params.garbage_collection, false, false, reflog_catalogs);
247  switch (res) {
250  "CommitProcessor - error: missing reflog.chksum");
251  return kMissingReflog;
254  "CommitProcessor - error: missing reflog");
255  return kMissingReflog;
256  case SigningTool::kError:
259  "CommitProcessor - error: signing manifest");
260  return kError;
263  "CommitProcessor - lease_path: %s, success.",
264  lease_path.c_str());
265  }
266 
267  {
268  UniquePtr<ServerTool> server_tool(new ServerTool());
269 
270  if (!server_tool->InitDownloadManager(true)) {
271  LogCvmfs(
273  "CommitProcessor - error: Could not initialize the download manager");
274  return kError;
275  }
276 
277  const std::string public_key = "/etc/cvmfs/keys/" + repo_name + ".pub";
278  const std::string trusted_certs =
279  "/etc/cvmfs/repositories.d/" + repo_name + "/trusted_certs";
280  if (!server_tool->InitVerifyingSignatureManager(public_key,
281  trusted_certs)) {
283  "CommitProcessor - error: Could not initialize the signature "
284  "manager");
285  return kError;
286  }
287 
288  shash::Any manifest_base_hash;
290  params.stratum0, repo_name, manifest_base_hash));
291 
293  "CommitProcessor - lease_path: %s, new root hash: %s",
294  lease_path.c_str(),
295  manifest->catalog_hash().ToString(false).c_str());
296  }
297 
298  // Ensure CVMFS_ROOT_HASH is not set in
299  // /var/spool/cvmfs/<REPO_NAME>/client.local
300  const std::string fname = "/var/spool/cvmfs/" + repo_name + "/client.local";
301  if (truncate(fname.c_str(), 0) < 0) {
302  LogCvmfs(kLogReceiver, kLogSyslogErr, "Could not truncate %s\n",
303  fname.c_str());
304  return kError;
305  }
306 
308  if (stats_db != NULL) {
309  if (!stats_db->StorePublishStatistics(statistics_, start_time_, true)) {
311  "Could not store publish statistics");
312  }
313  if (params.upload_stats_db) {
315  upload::Spooler *spooler = upload::Spooler::Construct(sd);
316  if (!stats_db->UploadStatistics(spooler)) {
318  "Could not upload statistics DB to upstream storage");
319  }
320  delete spooler;
321  }
322  delete stats_db;
323 
324  } else {
325  LogCvmfs(kLogReceiver, kLogSyslogErr, "Could not open statistics DB");
326  }
327 
328  return kSuccess;
329 }
330 
332  std::string start_time)
333 {
334  statistics_ = st;
335  statistics_->Register("publish.revision", "");
336  start_time_ = start_time;
337 }
338 
339 } // namespace receiver
#define LogCvmfs(source, mask,...)
Definition: logging.h:20
Counter * Register(const std::string &name, const std::string &desc)
Definition: statistics.cc:139
const manifest::Manifest * manifest() const
Definition: repository.h:123
std::string spooler_configuration
Definition: params.h:19
bool UploadStatistics(upload::Spooler *spooler, std::string local_path="")
ShortString Suffix(const unsigned start_at) const
Definition: shortstring.h:170
vector< string > SplitString(const string &str, const char delim, const unsigned max_chunks)
Definition: string.cc:287
perf::Statistics * statistics_
Result Run(const std::string &manifest_path, const std::string &repo_url, const std::string &spooler_definition, const std::string &temp_dir, const std::string &certificate="", const std::string &priv_key="", const std::string &repo_name="", const std::string &pwd="", const std::string &meta_info="", const std::string &reflog_chksum_path="", const bool garbage_collectable=false, const bool bootstrap_shortcuts=false, const bool return_early=false, const std::vector< shash::Any > reflog_catalogs=std::vector< shash::Any >())
Definition: signing_tool.cc:27
std::string ToString(const bool with_suffix=false) const
Definition: hash.h:245
std::string name_
int Main(const ArgumentList &args)
bool GetParamsFromFile(const std::string &repo_name, Params *params)
Definition: params.cc:20
perf::Statistics * statistics_
Definition: repository.h:138
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:132
assert((mem||(size==0))&&"Out Of Memory")
std::string GetSpoolerTempDir(const std::string &spooler_config)
Definition: params.cc:14
bool InitVerifyingSignatureManager(const std::string &pubkey_path, const std::string &trusted_certs="")
Definition: server_tool.cc:43
std::string hash_alg_str
Definition: params.h:21
download::DownloadManager * download_manager() const
Definition: server_tool.cc:107
std::string channel_
void SetStatistics(perf::Statistics *st, std::string start_time)
bool InitDownloadManager(const bool follow_redirects, const unsigned max_pool_handles=1, const bool use_system_proxy=true)
Definition: server_tool.cc:21
bool CreateNewTag(const RepositoryTag &repo_tag, const std::string &repo_name, const receiver::Params &params, const std::string &temp_dir, const std::string &manifest_path, const std::string &public_key_path)
uint64_t platform_realtime_ns()
std::string stratum0
Definition: params.h:18
static RaiiTempDir * Create(const std::string &prefix)
Definition: raii_temp_dir.cc:9
bool MkdirDeep(const std::string &path, const mode_t mode, bool verify_writable)
Definition: posix.cc:866
PathString RemoveRepoName(const PathString &lease_path)
shash::Any catalog_hash() const
Definition: manifest.h:124
bool HasPrefix(const string &str, const string &prefix, const bool ignore_case)
Definition: string.cc:264
static StatisticsDatabase * OpenStandardDB(const std::string repo_name)
bool garbage_collection
Definition: params.h:32
bool StorePublishStatistics(const perf::Statistics *statistics, const std::string &start_time, const bool success)
std::string ToString() const
Definition: shortstring.h:113
std::map< char, SharedPtr< std::string > > ArgumentList
Definition: swissknife.h:72
ShortString< kDefaultMaxPath, 0 > PathString
Definition: shortstring.h:189
std::string description_
Result Process(const std::string &lease_path, const shash::Any &old_root_hash, const shash::Any &new_root_hash, const RepositoryTag &tag, uint64_t *final_revision)
bool upload_stats_db
Definition: params.h:36