CernVM-FS  2.10.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,
50  const std::string& proxy) {
52  args['r'].Reset(new std::string(params.spooler_configuration));
53  args['w'].Reset(new std::string(params.stratum0));
54  args['t'].Reset(new std::string(temp_dir));
55  args['m'].Reset(new std::string(manifest_path));
56  args['p'].Reset(new std::string(public_key_path));
57  args['f'].Reset(new std::string(repo_name));
58  args['e'].Reset(new std::string(params.hash_alg_str));
59  args['a'].Reset(new std::string(repo_tag.name_));
60  args['c'].Reset(new std::string(repo_tag.channel_));
61  args['D'].Reset(new std::string(repo_tag.description_));
62  args['x'].Reset(new std::string());
63  args['@'].Reset(new std::string(proxy));
64 
67  const int ret = edit_cmd->Main(args);
68 
69  if (ret) {
70  LogCvmfs(kLogReceiver, kLogSyslogErr, "Error %d creating tag: %s", ret,
71  repo_tag.name_.c_str());
72  return false;
73  }
74 
75  return true;
76 }
77 
78 } // namespace
79 
80 namespace receiver {
81 
82 CommitProcessor::CommitProcessor() : num_errors_(0), statistics_(NULL) {}
83 
85 
103  const std::string& lease_path, const shash::Any& old_root_hash,
104  const shash::Any& new_root_hash, const RepositoryTag& tag,
105  uint64_t *final_revision) {
106  RepositoryTag final_tag = tag;
107  // If tag_name is a generic tag, update the time stamp
108  if (HasPrefix(final_tag.name_, "generic-", false)) {
109  // format time following the ISO 8601 YYYY-MM-DDThh:mm:ss.sssZ
110  // note the millisecond accurracy
111  uint64_t nanoseconds_timestamp = platform_realtime_ns();
112 
113  time_t seconds = static_cast<time_t>(
114  nanoseconds_timestamp / 1000000000); // 1E9
115  struct tm timestamp;
116  gmtime_r(&seconds, &timestamp);
117  char seconds_buffer[32];
118  strftime(seconds_buffer, sizeof(seconds_buffer),
119  "generic-%Y-%m-%dT%H:%M:%S", &timestamp);
120 
121  // first we get the raw nanoseconds from the timestamp using the module
122  // and then we divide to extract the millisecond.
123  // the division truncate the number brutally, it should be enough.
124  time_t milliseconds = static_cast<time_t>(
125  (nanoseconds_timestamp % 1000000000) / 1000000);
126  char millis_buffer[48];
127  snprintf(millis_buffer, sizeof(millis_buffer), "%s.%03ldZ", seconds_buffer,
128  milliseconds);
129 
130  final_tag.name_ = std::string(millis_buffer);
131  }
132 
134  "CommitProcessor - lease_path: %s, old hash: %s, new hash: %s, "
135  "tag_name: %s, tag_channel: %s, tag_description: %s",
136  lease_path.c_str(), old_root_hash.ToString(true).c_str(),
137  new_root_hash.ToString(true).c_str(), final_tag.name_.c_str(),
138  final_tag.channel_.c_str(), final_tag.description_.c_str());
139 
140  const std::vector<std::string> lease_path_tokens =
141  SplitString(lease_path, '/');
142 
143  const std::string repo_name = lease_path_tokens.front();
144 
145  Params params;
146  if (!GetParamsFromFile(repo_name, &params)) {
147  LogCvmfs(
149  "CommitProcessor - error: Could not get configuration parameters.");
150  return kError;
151  }
152 
153  UniquePtr<ServerTool> server_tool(new ServerTool());
154 
155  if (!server_tool->InitDownloadManager(true, params.proxy)) {
156  LogCvmfs(
158  "CommitProcessor - error: Could not initialize the download manager");
159  return kError;
160  }
161 
162  const std::string public_key = "/etc/cvmfs/keys/" + repo_name + ".pub";
163  const std::string trusted_certs =
164  "/etc/cvmfs/repositories.d/" + repo_name + "/trusted_certs";
165  if (!server_tool->InitVerifyingSignatureManager(public_key, trusted_certs)) {
166  LogCvmfs(
168  "CommitProcessor - error: Could not initialize the signature manager");
169  return kError;
170  }
171 
172  shash::Any manifest_base_hash;
174  params.stratum0, repo_name, manifest_base_hash));
175 
176  // Current catalog from the gateway machine
177  if (!manifest.IsValid()) {
179  "CommitProcessor - error: Could not open repository manifest");
180  return kError;
181  }
182 
184  "CommitProcessor - lease_path: %s, target root hash: %s",
185  lease_path.c_str(),
186  manifest->catalog_hash().ToString(false).c_str());
187 
188  const std::string spooler_temp_dir =
190  assert(!spooler_temp_dir.empty());
191  assert(MkdirDeep(spooler_temp_dir + "/receiver", 0666, true));
192  const std::string temp_dir_root =
193  spooler_temp_dir + "/receiver/commit_processor";
194 
195  const PathString relative_lease_path = RemoveRepoName(PathString(lease_path));
196 
198  "CommitProcessor - lease_path: %s, merging catalogs",
199  lease_path.c_str());
200 
203  merge_tool(params.stratum0, old_root_hash, new_root_hash,
204  relative_lease_path, temp_dir_root,
205  server_tool->download_manager(), manifest.weak_ref(),
206  statistics_);
207  if (!merge_tool.Init()) {
209  "Error: Could not initialize the catalog merge tool");
210  return kError;
211  }
212 
213  std::string new_manifest_path;
214  if (!merge_tool.Run(params, &new_manifest_path, final_revision)) {
216  "CommitProcessor - error: Catalog merge failed");
217  return kMergeFailure;
218  }
219 
220  UniquePtr<RaiiTempDir> raii_temp_dir(RaiiTempDir::Create(temp_dir_root));
221  const std::string temp_dir = raii_temp_dir->dir();
222  const std::string certificate = "/etc/cvmfs/keys/" + repo_name + ".crt";
223  const std::string private_key = "/etc/cvmfs/keys/" + repo_name + ".key";
224 
225  if (!CreateNewTag(final_tag, repo_name, params, temp_dir, new_manifest_path,
226  public_key, params.proxy)) {
227  LogCvmfs(kLogReceiver, kLogSyslogErr, "Error creating tag: %s",
228  final_tag.name_.c_str());
229  return kError;
230  }
231 
232  // We need to re-initialize the ServerTool component for signing
233  server_tool.Destroy();
234  server_tool = new ServerTool();
235 
237  "CommitProcessor - lease_path: %s, signing manifest",
238  lease_path.c_str());
239 
240  // Add C_N root catalog hash to reflog through SigningTool,
241  // so garbage collector can later delete it.
242  std::vector<shash::Any> reflog_catalogs;
243  reflog_catalogs.push_back(new_root_hash);
244 
245  SigningTool signing_tool(server_tool.weak_ref());
246  SigningTool::Result res = signing_tool.Run(
247  new_manifest_path, params.stratum0, params.spooler_configuration,
248  temp_dir, certificate, private_key, repo_name, "", "",
249  "/var/spool/cvmfs/" + repo_name + "/reflog.chksum", params.proxy,
250  params.garbage_collection, false, false, reflog_catalogs);
251  switch (res) {
254  "CommitProcessor - error: missing reflog.chksum");
255  return kMissingReflog;
258  "CommitProcessor - error: missing reflog");
259  return kMissingReflog;
260  case SigningTool::kError:
263  "CommitProcessor - error: signing manifest");
264  return kError;
267  "CommitProcessor - lease_path: %s, success.",
268  lease_path.c_str());
269  }
270 
271  {
272  UniquePtr<ServerTool> server_tool(new ServerTool());
273 
274  if (!server_tool->InitDownloadManager(true, params.proxy)) {
275  LogCvmfs(
277  "CommitProcessor - error: Could not initialize the download manager");
278  return kError;
279  }
280 
281  const std::string public_key = "/etc/cvmfs/keys/" + repo_name + ".pub";
282  const std::string trusted_certs =
283  "/etc/cvmfs/repositories.d/" + repo_name + "/trusted_certs";
284  if (!server_tool->InitVerifyingSignatureManager(public_key,
285  trusted_certs)) {
287  "CommitProcessor - error: Could not initialize the signature "
288  "manager");
289  return kError;
290  }
291 
292  shash::Any manifest_base_hash;
294  params.stratum0, repo_name, manifest_base_hash));
295 
297  "CommitProcessor - lease_path: %s, new root hash: %s",
298  lease_path.c_str(),
299  manifest->catalog_hash().ToString(false).c_str());
300  }
301 
302  // Ensure CVMFS_ROOT_HASH is not set in
303  // /var/spool/cvmfs/<REPO_NAME>/client.local
304  const std::string fname = "/var/spool/cvmfs/" + repo_name + "/client.local";
305  if (truncate(fname.c_str(), 0) < 0) {
306  LogCvmfs(kLogReceiver, kLogSyslogErr, "Could not truncate %s\n",
307  fname.c_str());
308  return kError;
309  }
310 
312  if (stats_db != NULL) {
313  if (!stats_db->StorePublishStatistics(statistics_, start_time_, true)) {
315  "Could not store publish statistics");
316  }
317  if (params.upload_stats_db) {
319  upload::Spooler *spooler = upload::Spooler::Construct(sd);
320  if (!stats_db->UploadStatistics(spooler)) {
322  "Could not upload statistics DB to upstream storage");
323  }
324  delete spooler;
325  }
326  delete stats_db;
327 
328  } else {
329  LogCvmfs(kLogReceiver, kLogSyslogErr, "Could not open statistics DB");
330  }
331 
332  return kSuccess;
333 }
334 
336  const std::string &start_time)
337 {
338  statistics_ = st;
339  statistics_->Register("publish.revision", "");
340  start_time_ = start_time;
341 }
342 
343 } // 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
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 std::string &proxy="", 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 spooler_configuration
Definition: params.h:20
bool UploadStatistics(upload::Spooler *spooler, std::string local_path="")
ShortString Suffix(const unsigned start_at) const
Definition: shortstring.h:171
vector< string > SplitString(const string &str, const char delim, const unsigned max_chunks)
Definition: string.cc:288
perf::Statistics * statistics_
std::string ToString(const bool with_suffix=false) const
Definition: hash.h:248
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:138
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:49
std::string hash_alg_str
Definition: params.h:22
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, const std::string &proxy)
download::DownloadManager * download_manager() const
Definition: server_tool.cc:113
std::string channel_
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:871
PathString RemoveRepoName(const PathString &lease_path)
shash::Any catalog_hash() const
Definition: manifest.h:124
bool InitDownloadManager(const bool follow_redirects, const std::string &proxy="", const unsigned max_pool_handles=1)
Definition: server_tool.cc:21
bool HasPrefix(const string &str, const string &prefix, const bool ignore_case)
Definition: string.cc:265
static StatisticsDatabase * OpenStandardDB(const std::string repo_name)
bool garbage_collection
Definition: params.h:33
void SetStatistics(perf::Statistics *st, const std::string &start_time)
bool StorePublishStatistics(const perf::Statistics *statistics, const std::string &start_time, const bool success)
std::string ToString() const
Definition: shortstring.h:114
std::map< char, SharedPtr< std::string > > ArgumentList
Definition: swissknife.h:72
ShortString< kDefaultMaxPath, 0 > PathString
Definition: shortstring.h:190
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)
std::string proxy
Definition: params.h:19
bool upload_stats_db
Definition: params.h:37