| Directory: | cvmfs/ |
|---|---|
| File: | cvmfs/receiver/commit_processor.cc |
| Date: | 2026-10-04 02:40:33 |
| Exec | Total | Coverage | |
|---|---|---|---|
| Lines: | 52 | 243 | 21.4% |
| Branches: | 58 | 560 | 10.4% |
| Line | Branch | Exec | Source |
|---|---|---|---|
| 1 | /** | ||
| 2 | * This file is part of the CernVM File System. | ||
| 3 | */ | ||
| 4 | |||
| 5 | #include "commit_processor.h" | ||
| 6 | |||
| 7 | #include <time.h> | ||
| 8 | |||
| 9 | #include <cctype> | ||
| 10 | #include <memory> | ||
| 11 | #include <string> | ||
| 12 | #include <vector> | ||
| 13 | |||
| 14 | #include "catalog_diff_tool.h" | ||
| 15 | #include "catalog_merge_tool.h" | ||
| 16 | #include "catalog_mgr_ro.h" | ||
| 17 | #include "catalog_mgr_rw.h" | ||
| 18 | #include "compression/compression.h" | ||
| 19 | #include "manifest.h" | ||
| 20 | #include "manifest_fetch.h" | ||
| 21 | #include "network/download.h" | ||
| 22 | #include "network/sink_path.h" | ||
| 23 | #include "params.h" | ||
| 24 | #include "signing_tool.h" | ||
| 25 | #include "statistics.h" | ||
| 26 | #include "statistics_database.h" | ||
| 27 | #include "swissknife.h" | ||
| 28 | #include "swissknife_history.h" | ||
| 29 | #include "util/algorithm.h" | ||
| 30 | #include "util/logging.h" | ||
| 31 | #include "util/posix.h" | ||
| 32 | #include "util/raii_temp_dir.h" | ||
| 33 | #include "util/string.h" | ||
| 34 | |||
| 35 | namespace { | ||
| 36 | |||
| 37 | ✗ | PathString RemoveRepoName(const PathString &lease_path) { | |
| 38 | ✗ | std::string abs_path = lease_path.ToString(); | |
| 39 | ✗ | const std::string::const_iterator it = std::find(abs_path.begin(), | |
| 40 | ✗ | abs_path.end(), '/'); | |
| 41 | ✗ | if (it != abs_path.end()) { | |
| 42 | ✗ | const size_t idx = it - abs_path.begin() + 1; | |
| 43 | ✗ | return lease_path.Suffix(idx); | |
| 44 | } else { | ||
| 45 | ✗ | return lease_path; | |
| 46 | } | ||
| 47 | } | ||
| 48 | |||
| 49 | ✗ | bool EditTags(const RepositoryTag &repo_tag, const std::string &repo_name, | |
| 50 | const receiver::Params ¶ms, const std::string &temp_dir, | ||
| 51 | const std::string &manifest_path, | ||
| 52 | const std::string &public_key_path, const std::string &proxy, | ||
| 53 | const time_t auto_tag_threshold, const bool maintain_undo_tags) { | ||
| 54 | ✗ | swissknife::ArgumentList args; | |
| 55 | ✗ | args['r'].Reset(new std::string(params.spooler_configuration)); | |
| 56 | ✗ | args['w'].Reset(new std::string(params.stratum0)); | |
| 57 | ✗ | args['t'].Reset(new std::string(temp_dir)); | |
| 58 | ✗ | args['m'].Reset(new std::string(manifest_path)); | |
| 59 | ✗ | args['p'].Reset(new std::string(public_key_path)); | |
| 60 | ✗ | args['f'].Reset(new std::string(repo_name)); | |
| 61 | ✗ | args['e'].Reset(new std::string(params.hash_alg_str)); | |
| 62 | ✗ | args['a'].Reset(new std::string(repo_tag.name())); | |
| 63 | ✗ | args['D'].Reset(new std::string(repo_tag.description())); | |
| 64 | ✗ | if (maintain_undo_tags) { | |
| 65 | ✗ | args['x'].Reset(new std::string()); | |
| 66 | } | ||
| 67 | ✗ | args['@'].Reset(new std::string(proxy)); | |
| 68 | // Remove the tags requested by `cvmfs_server tag -r` in the same history | ||
| 69 | // transaction as the (possibly empty) new tag, so a single new history | ||
| 70 | // database is published and registered in the reflog for this commit. | ||
| 71 | ✗ | if (!repo_tag.delete_tags().empty()) { | |
| 72 | ✗ | args['d'].Reset(new std::string(repo_tag.delete_tags())); | |
| 73 | } | ||
| 74 | // Remove outdated auto-generated tags in the same history transaction as the | ||
| 75 | // tag we are about to add, so that only a single new history database is | ||
| 76 | // published (and registered in the reflog) for this commit. | ||
| 77 | ✗ | if (auto_tag_threshold > 0) { | |
| 78 | ✗ | args['c'].Reset(new std::string(StringifyInt(auto_tag_threshold))); | |
| 79 | } | ||
| 80 | |||
| 81 | const std::unique_ptr<swissknife::CommandEditTag> edit_cmd( | ||
| 82 | ✗ | new swissknife::CommandEditTag()); | |
| 83 | ✗ | const int ret = edit_cmd->Main(args); | |
| 84 | |||
| 85 | ✗ | if (ret) { | |
| 86 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, "Error %d editing tags (add: '%s')", | |
| 87 | ✗ | ret, repo_tag.name().c_str()); | |
| 88 | ✗ | return false; | |
| 89 | } | ||
| 90 | |||
| 91 | ✗ | return true; | |
| 92 | } | ||
| 93 | |||
| 94 | } // namespace | ||
| 95 | |||
| 96 | namespace receiver { | ||
| 97 | |||
| 98 | // See commit_processor.h for the contract. `now` is injected so the parser is | ||
| 99 | // deterministic and unit-testable. | ||
| 100 | 96 | time_t ParseRelativeTimespan(const std::string ×pan, time_t now) { | |
| 101 | // Tokenize on whitespace, lower-casing as we go. | ||
| 102 | 96 | std::vector<std::string> tokens; | |
| 103 | 96 | std::string current; | |
| 104 |
2/2✓ Branch 1 taken 1028 times.
✓ Branch 2 taken 96 times.
|
1124 | for (size_t i = 0; i < timespan.size(); ++i) { |
| 105 | 1028 | const unsigned char c = static_cast<unsigned char>(timespan[i]); | |
| 106 |
2/2✓ Branch 0 taken 184 times.
✓ Branch 1 taken 844 times.
|
1028 | if (isspace(c)) { |
| 107 |
2/2✓ Branch 1 taken 156 times.
✓ Branch 2 taken 28 times.
|
184 | if (!current.empty()) { |
| 108 |
1/2✓ Branch 1 taken 156 times.
✗ Branch 2 not taken.
|
156 | tokens.push_back(current); |
| 109 | 156 | current.clear(); | |
| 110 | } | ||
| 111 | } else { | ||
| 112 |
1/2✓ Branch 1 taken 844 times.
✗ Branch 2 not taken.
|
844 | current += static_cast<char>(tolower(c)); |
| 113 | } | ||
| 114 | } | ||
| 115 |
2/2✓ Branch 1 taken 88 times.
✓ Branch 2 taken 8 times.
|
96 | if (!current.empty()) { |
| 116 |
1/2✓ Branch 1 taken 88 times.
✗ Branch 2 not taken.
|
88 | tokens.push_back(current); |
| 117 | } | ||
| 118 | |||
| 119 | // Expect exactly "<number> <unit> ago". | ||
| 120 |
6/6✓ Branch 1 taken 72 times.
✓ Branch 2 taken 24 times.
✓ Branch 5 taken 4 times.
✓ Branch 6 taken 68 times.
✓ Branch 7 taken 28 times.
✓ Branch 8 taken 68 times.
|
96 | if (tokens.size() != 3 || tokens[2] != "ago") { |
| 121 | 28 | return 0; | |
| 122 | } | ||
| 123 | 68 | const std::string &number = tokens[0]; | |
| 124 |
1/2✗ Branch 1 not taken.
✓ Branch 2 taken 68 times.
|
68 | if (number.empty()) { |
| 125 | ✗ | return 0; | |
| 126 | } | ||
| 127 |
2/2✓ Branch 1 taken 100 times.
✓ Branch 2 taken 64 times.
|
164 | for (size_t i = 0; i < number.size(); ++i) { |
| 128 |
2/2✓ Branch 1 taken 4 times.
✓ Branch 2 taken 96 times.
|
100 | if (!isdigit(static_cast<unsigned char>(number[i]))) { |
| 129 | 4 | return 0; | |
| 130 | } | ||
| 131 | } | ||
| 132 |
1/2✓ Branch 1 taken 64 times.
✗ Branch 2 not taken.
|
64 | const int64_t count = String2Int64(number); |
| 133 | |||
| 134 | // De-pluralize the unit. | ||
| 135 |
1/2✓ Branch 2 taken 64 times.
✗ Branch 3 not taken.
|
64 | std::string unit = tokens[1]; |
| 136 |
6/8✓ Branch 1 taken 64 times.
✗ Branch 2 not taken.
✓ Branch 5 taken 64 times.
✗ Branch 6 not taken.
✓ Branch 7 taken 48 times.
✓ Branch 8 taken 16 times.
✓ Branch 9 taken 48 times.
✓ Branch 10 taken 16 times.
|
64 | if (!unit.empty() && unit[unit.size() - 1] == 's') { |
| 137 |
1/2✓ Branch 2 taken 48 times.
✗ Branch 3 not taken.
|
48 | unit.resize(unit.size() - 1); |
| 138 | } | ||
| 139 | |||
| 140 | // Fixed-length units can be subtracted directly. | ||
| 141 | 64 | int64_t factor = 0; | |
| 142 |
6/6✓ Branch 1 taken 60 times.
✓ Branch 2 taken 4 times.
✓ Branch 4 taken 8 times.
✓ Branch 5 taken 52 times.
✓ Branch 6 taken 12 times.
✓ Branch 7 taken 52 times.
|
64 | if (unit == "sec" || unit == "second") { |
| 143 | 12 | factor = 1; | |
| 144 |
6/6✓ Branch 1 taken 48 times.
✓ Branch 2 taken 4 times.
✓ Branch 4 taken 4 times.
✓ Branch 5 taken 44 times.
✓ Branch 6 taken 8 times.
✓ Branch 7 taken 44 times.
|
52 | } else if (unit == "min" || unit == "minute") { |
| 145 | 8 | factor = 60; | |
| 146 |
2/2✓ Branch 1 taken 4 times.
✓ Branch 2 taken 40 times.
|
44 | } else if (unit == "hour") { |
| 147 | 4 | factor = 3600; | |
| 148 |
2/2✓ Branch 1 taken 20 times.
✓ Branch 2 taken 20 times.
|
40 | } else if (unit == "day") { |
| 149 | 20 | factor = 86400; | |
| 150 |
2/2✓ Branch 1 taken 4 times.
✓ Branch 2 taken 16 times.
|
20 | } else if (unit == "week") { |
| 151 | 4 | factor = 604800; | |
| 152 | } | ||
| 153 |
2/2✓ Branch 0 taken 48 times.
✓ Branch 1 taken 16 times.
|
64 | if (factor > 0) { |
| 154 | 48 | return now - static_cast<time_t>(count * factor); | |
| 155 | } | ||
| 156 | |||
| 157 | // Calendar units: let mktime() normalize the broken-down time. | ||
| 158 | struct tm broken_time; | ||
| 159 | 16 | localtime_r(&now, &broken_time); | |
| 160 |
2/2✓ Branch 1 taken 8 times.
✓ Branch 2 taken 8 times.
|
16 | if (unit == "month") { |
| 161 | 8 | broken_time.tm_mon -= static_cast<int>(count); | |
| 162 | 8 | return mktime(&broken_time); | |
| 163 | } | ||
| 164 |
2/2✓ Branch 1 taken 4 times.
✓ Branch 2 taken 4 times.
|
8 | if (unit == "year") { |
| 165 | 4 | broken_time.tm_year -= static_cast<int>(count); | |
| 166 | 4 | return mktime(&broken_time); | |
| 167 | } | ||
| 168 | |||
| 169 | 4 | return 0; | |
| 170 | 96 | } | |
| 171 | |||
| 172 | 11 | CommitProcessor::CommitProcessor() : num_errors_(0), statistics_(NULL) { } | |
| 173 | |||
| 174 | 22 | CommitProcessor::~CommitProcessor() { } | |
| 175 | |||
| 176 | /** | ||
| 177 | * Applies the changes from the new catalog onto the repository. | ||
| 178 | * | ||
| 179 | * Let: | ||
| 180 | * + C_O = the root catalog of the repository (given by old_root_hash) at | ||
| 181 | * the beginning of the lease, on the release manager machine | ||
| 182 | * + C_N = the root catalog of the repository (given by new_root_hash), on | ||
| 183 | * the release manager machine, with the changes introduced during the | ||
| 184 | * lease | ||
| 185 | * + C_G = the current root catalog of the repository on the gateway machine. | ||
| 186 | * | ||
| 187 | * This method applies all the changes from C_N, with respect to C_O, onto C_G. | ||
| 188 | * The resulting catalog on the gateway machine (C_GN) is then set as root | ||
| 189 | * catalog in the repository manifest. The method also signs the updated | ||
| 190 | * repository manifest. | ||
| 191 | */ | ||
| 192 | ✗ | CommitProcessor::Result CommitProcessor::Process( | |
| 193 | const std::string &lease_path, const shash::Any &old_root_hash, | ||
| 194 | const shash::Any &new_root_hash, const RepositoryTag &tag, | ||
| 195 | int64_t lease_expiration, uint64_t *final_revision, bool direct_graft) { | ||
| 196 | ✗ | RepositoryTag final_tag = tag; | |
| 197 | // If tag_name is a generic tag, update the time stamp | ||
| 198 | ✗ | if (final_tag.HasGenericName()) { | |
| 199 | ✗ | final_tag.SetGenericName(); | |
| 200 | } | ||
| 201 | |||
| 202 | ✗ | LogCvmfs(kLogReceiver, kLogSyslog, | |
| 203 | "CommitProcessor - lease_path: %s, old hash: %s, new hash: %s, " | ||
| 204 | "tag_name: %s, tag_description: %s", | ||
| 205 | ✗ | lease_path.c_str(), old_root_hash.ToString(true).c_str(), | |
| 206 | ✗ | new_root_hash.ToString(true).c_str(), final_tag.name().c_str(), | |
| 207 | ✗ | final_tag.description().c_str()); | |
| 208 | |||
| 209 | const std::vector<std::string> lease_path_tokens = SplitString(lease_path, | ||
| 210 | ✗ | '/'); | |
| 211 | |||
| 212 | ✗ | const std::string repo_name = lease_path_tokens.front(); | |
| 213 | |||
| 214 | ✗ | Params params; | |
| 215 | ✗ | if (!GetParamsFromFile(repo_name, ¶ms)) { | |
| 216 | ✗ | LogCvmfs( | |
| 217 | kLogReceiver, kLogSyslogErr, | ||
| 218 | "CommitProcessor - error: Could not get configuration parameters."); | ||
| 219 | ✗ | return kError; | |
| 220 | } | ||
| 221 | |||
| 222 | ✗ | const std::unique_ptr<ServerTool> server_tool(new ServerTool()); | |
| 223 | |||
| 224 | ✗ | if (!server_tool->InitDownloadManager(true, params.proxy)) { | |
| 225 | ✗ | LogCvmfs( | |
| 226 | kLogReceiver, kLogSyslogErr, | ||
| 227 | "CommitProcessor - error: Could not initialize the download manager"); | ||
| 228 | ✗ | return kError; | |
| 229 | } | ||
| 230 | |||
| 231 | ✗ | const std::string public_key = "/etc/cvmfs/keys/" + repo_name + ".pub"; | |
| 232 | ✗ | const std::string certificate = "/etc/cvmfs/keys/" + repo_name + ".crt"; | |
| 233 | ✗ | const std::string private_key = "/etc/cvmfs/keys/" + repo_name + ".key"; | |
| 234 | ✗ | if (!server_tool->InitSignatureManager(public_key, certificate, | |
| 235 | private_key)) { | ||
| 236 | ✗ | LogCvmfs( | |
| 237 | kLogReceiver, kLogSyslogErr, | ||
| 238 | "CommitProcessor - error: Could not initialize the signature manager"); | ||
| 239 | ✗ | return kError; | |
| 240 | } | ||
| 241 | |||
| 242 | ✗ | const shash::Any manifest_base_hash; | |
| 243 | const std::unique_ptr<manifest::Manifest> manifest_tgt( | ||
| 244 | server_tool->FetchRemoteManifest(params.stratum0, repo_name, | ||
| 245 | ✗ | manifest_base_hash)); | |
| 246 | |||
| 247 | // Current catalog from the gateway machine | ||
| 248 | ✗ | if (manifest_tgt.get() == nullptr) { | |
| 249 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 250 | "CommitProcessor - error: Could not open repository manifest"); | ||
| 251 | ✗ | return kError; | |
| 252 | } | ||
| 253 | |||
| 254 | ✗ | LogCvmfs(kLogReceiver, kLogSyslog, | |
| 255 | "CommitProcessor - lease_path: %s, target root hash: %s", | ||
| 256 | lease_path.c_str(), | ||
| 257 | ✗ | manifest_tgt->catalog_hash().ToString(false).c_str()); | |
| 258 | |||
| 259 | |||
| 260 | ✗ | std::string cache_dir_; | |
| 261 | ✗ | if (params.use_local_cache) { | |
| 262 | ✗ | cache_dir_ = "/var/spool/cvmfs/" + repo_name + "/cache.server"; | |
| 263 | } | ||
| 264 | |||
| 265 | const std::string spooler_temp_dir = GetSpoolerTempDir( | ||
| 266 | ✗ | params.spooler_configuration); | |
| 267 | ✗ | assert(!spooler_temp_dir.empty()); | |
| 268 | ✗ | assert(MkdirDeep(spooler_temp_dir + "/receiver", 0755, true)); | |
| 269 | const std::string temp_dir_root = spooler_temp_dir | ||
| 270 | ✗ | + "/receiver/commit_processor"; | |
| 271 | |||
| 272 | ✗ | const PathString relative_lease_path = RemoveRepoName(PathString(lease_path)); | |
| 273 | |||
| 274 | ✗ | std::string new_manifest_path; | |
| 275 | ✗ | shash::Any new_manifest_hash; | |
| 276 | |||
| 277 | ✗ | if (direct_graft) { | |
| 278 | // -- Experimental DirectGraft fast path ---------------------------------- | ||
| 279 | // Grafts new_root_hash directly into the parent catalog at | ||
| 280 | // relative_lease_path via WritableCatalogManager::TryGraftNestedCatalog, | ||
| 281 | // bypassing DiffRec entirely. Only valid when lease_path points to a | ||
| 282 | // brand-new directory subtree. Reached only via the experimental dedicated | ||
| 283 | // kCommitGraft reactor request. | ||
| 284 | ✗ | LogCvmfs(kLogReceiver, kLogSyslog, | |
| 285 | "CommitProcessor - lease_path: %s, direct-graft path " | ||
| 286 | "(skipping DiffRec)", | ||
| 287 | lease_path.c_str()); | ||
| 288 | |||
| 289 | const std::unique_ptr<RaiiTempDir> graft_temp_dir( | ||
| 290 | ✗ | RaiiTempDir::Create(temp_dir_root)); | |
| 291 | ✗ | const std::string graft_temp = graft_temp_dir->dir(); | |
| 292 | |||
| 293 | ✗ | perf::StatisticsTemplate stats_tmpl("publish", statistics_); | |
| 294 | // Register the FsCounters (n_files_added, n_directories_added, etc.) that | ||
| 295 | // StorePublishStatistics expects. In the DiffRec path these are created by | ||
| 296 | // CatalogMergeTool::Run(); DirectGraft bypasses that, so we register them | ||
| 297 | // here. The values stay 0 -- accurate for a graft that adds a whole | ||
| 298 | // subtree atomically rather than individual file-level diffs. | ||
| 299 | ✗ | const perf::FsCounters fs_counters(stats_tmpl); | |
| 300 | const upload::SpoolerDefinition definition( | ||
| 301 | params.spooler_configuration, params.hash_alg, params.compression_alg, | ||
| 302 | ✗ | params.generate_legacy_bulk_chunks, params.use_file_chunking, | |
| 303 | params.min_chunk_size, params.avg_chunk_size, params.max_chunk_size, | ||
| 304 | ✗ | "dummy_token", "dummy_key"); | |
| 305 | const std::unique_ptr<upload::Spooler> spooler( | ||
| 306 | ✗ | upload::Spooler::Construct(definition, &stats_tmpl)); | |
| 307 | |||
| 308 | const std::unique_ptr<catalog::WritableCatalogManager> output_mgr( | ||
| 309 | new catalog::WritableCatalogManager( | ||
| 310 | ✗ | manifest_tgt->catalog_hash(), params.stratum0, graft_temp, | |
| 311 | ✗ | spooler.get(), server_tool->download_manager(), | |
| 312 | ✗ | params.enforce_limits, params.nested_kcatalog_limit, | |
| 313 | ✗ | params.root_kcatalog_limit, params.file_mbyte_limit, statistics_, | |
| 314 | ✗ | params.use_autocatalogs, params.max_weight, params.min_weight, | |
| 315 | ✗ | cache_dir_)); | |
| 316 | ✗ | if (!output_mgr->Init()) { | |
| 317 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 318 | "CommitProcessor - error: Could not initialize catalog manager " | ||
| 319 | "for direct-graft"); | ||
| 320 | ✗ | return kError; | |
| 321 | } | ||
| 322 | |||
| 323 | ✗ | if (new_root_hash.IsNull() | |
| 324 | ✗ | || new_root_hash.suffix != shash::kSuffixCatalog) { | |
| 325 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 326 | "CommitProcessor - error: DirectGraft requires a catalog hash"); | ||
| 327 | ✗ | return kMergeFailure; | |
| 328 | } | ||
| 329 | |||
| 330 | // Download new_root_hash to a temp file to obtain the size of the catalog | ||
| 331 | // database. TryGraftNestedCatalog downloads the catalog once more | ||
| 332 | // internally via LoadFreeCatalog; the probe writes outside the local cache | ||
| 333 | // directory, so that second fetch does not hit the cache. | ||
| 334 | ✗ | const std::string catalog_url = params.stratum0 + "/data/" | |
| 335 | ✗ | + new_root_hash.MakePath(); | |
| 336 | ✗ | const std::string catalog_tmp = graft_temp + "/catalog_size"; | |
| 337 | { | ||
| 338 | ✗ | cvmfs::PathSink catalog_sink(catalog_tmp); | |
| 339 | ✗ | const shash::Any expected = new_root_hash; | |
| 340 | // Decompress while downloading (the content hash is still verified | ||
| 341 | // against the compressed stream): nested_catalogs.size holds the size of | ||
| 342 | // the catalog database, not of the compressed CAS object. Compare | ||
| 343 | // CommandCheck::FetchCatalog, which validates this column against the | ||
| 344 | // size of the decompressed catalog. | ||
| 345 | download::JobInfo dl_job(&catalog_url, true, false, &expected, | ||
| 346 | ✗ | &catalog_sink); | |
| 347 | ✗ | const download::Failures dl_ret = server_tool->download_manager()->Fetch( | |
| 348 | &dl_job); | ||
| 349 | ✗ | if (dl_ret != download::kFailOk) { | |
| 350 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 351 | "CommitProcessor - error: failed to download catalog %s " | ||
| 352 | "for size probe (%d)", | ||
| 353 | catalog_url.c_str(), static_cast<int>(dl_ret)); | ||
| 354 | ✗ | unlink(catalog_tmp.c_str()); | |
| 355 | ✗ | return kError; | |
| 356 | } | ||
| 357 | ✗ | } // PathSink destructor closes the file here | |
| 358 | ✗ | const int64_t catalog_size = GetFileSize(catalog_tmp); | |
| 359 | ✗ | unlink(catalog_tmp.c_str()); | |
| 360 | // A zero size would be recorded as "unknown" by swissknife check and | ||
| 361 | // silently disable its size validation, so reject it here. | ||
| 362 | ✗ | if (catalog_size <= 0) { | |
| 363 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 364 | "CommitProcessor - error: empty or unstatable catalog %s", | ||
| 365 | catalog_url.c_str()); | ||
| 366 | ✗ | return kError; | |
| 367 | } | ||
| 368 | |||
| 369 | // Graft: inserts the nested catalog reference into the parent catalog | ||
| 370 | // and propagates the directory entry + counters upward. | ||
| 371 | ✗ | if (!output_mgr->TryGraftNestedCatalog( | |
| 372 | ✗ | relative_lease_path.ToString(), new_root_hash, | |
| 373 | static_cast<uint64_t>(catalog_size))) { | ||
| 374 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 375 | "CommitProcessor - error: DirectGraft validation failed for " | ||
| 376 | "lease_path: %s", | ||
| 377 | lease_path.c_str()); | ||
| 378 | ✗ | return kMergeFailure; | |
| 379 | } | ||
| 380 | |||
| 381 | // Commit updates manifest_tgt in-place (new root hash, revision++, etc.) | ||
| 382 | ✗ | if (!output_mgr->Commit(false, 0, manifest_tgt.get())) { | |
| 383 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 384 | "CommitProcessor - error: Could not commit grafted catalog"); | ||
| 385 | ✗ | return kMergeFailure; | |
| 386 | } | ||
| 387 | |||
| 388 | // Export the updated manifest to a temp file for CreateNewTag/SigningTool. | ||
| 389 | ✗ | new_manifest_path = CreateTempPath(temp_dir_root, 0600); | |
| 390 | ✗ | if (!manifest_tgt->Export(new_manifest_path)) { | |
| 391 | ✗ | LogCvmfs( | |
| 392 | kLogReceiver, kLogSyslogErr, | ||
| 393 | "CommitProcessor - error: Could not export manifest after graft"); | ||
| 394 | ✗ | return kError; | |
| 395 | } | ||
| 396 | ✗ | new_manifest_hash = manifest_tgt->catalog_hash(); | |
| 397 | ✗ | *final_revision = manifest_tgt->revision(); | |
| 398 | |||
| 399 | ✗ | } else { | |
| 400 | // -- Standard DiffRec path via CatalogMergeTool -------------------------- | ||
| 401 | ✗ | LogCvmfs(kLogReceiver, kLogSyslog, | |
| 402 | "CommitProcessor - lease_path: %s, merging catalogs", | ||
| 403 | lease_path.c_str()); | ||
| 404 | |||
| 405 | CatalogMergeTool<catalog::WritableCatalogManager, | ||
| 406 | catalog::SimpleCatalogManager> | ||
| 407 | merge_tool(params.stratum0, old_root_hash, new_root_hash, | ||
| 408 | relative_lease_path, temp_dir_root, | ||
| 409 | server_tool->download_manager(), manifest_tgt.get(), | ||
| 410 | ✗ | statistics_, cache_dir_); | |
| 411 | ✗ | if (!merge_tool.Init()) { | |
| 412 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 413 | "Error: Could not initialize the catalog merge tool"); | ||
| 414 | ✗ | return kError; | |
| 415 | } | ||
| 416 | ✗ | if (!merge_tool.Run(params, &new_manifest_path, &new_manifest_hash, | |
| 417 | final_revision)) { | ||
| 418 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 419 | "CommitProcessor - error: Catalog merge failed"); | ||
| 420 | ✗ | return kMergeFailure; | |
| 421 | } | ||
| 422 | } | ||
| 423 | |||
| 424 | const std::unique_ptr<RaiiTempDir> raii_temp_dir( | ||
| 425 | ✗ | RaiiTempDir::Create(temp_dir_root)); | |
| 426 | ✗ | const std::string temp_dir = raii_temp_dir->dir(); | |
| 427 | |||
| 428 | // Determine the cutoff below which outdated auto-generated tags are removed. | ||
| 429 | // A value sent by the publisher (already an absolute timestamp) takes | ||
| 430 | // precedence over the gateway's local CVMFS_AUTO_TAG_TIMESPAN configuration, | ||
| 431 | // which is a relative "<N> <unit> ago" timespan resolved here. 0 disables | ||
| 432 | // cleanup. | ||
| 433 | ✗ | time_t auto_tag_threshold = final_tag.auto_tag_threshold(); | |
| 434 | ✗ | if (auto_tag_threshold <= 0 && !params.auto_tag_timespan.empty()) { | |
| 435 | ✗ | auto_tag_threshold = ParseRelativeTimespan(params.auto_tag_timespan, | |
| 436 | time(NULL)); | ||
| 437 | ✗ | if (auto_tag_threshold <= 0) { | |
| 438 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 439 | "CommitProcessor - warning: could not parse " | ||
| 440 | "CVMFS_AUTO_TAG_TIMESPAN '%s' (expected \"<N> <unit> ago\")", | ||
| 441 | params.auto_tag_timespan.c_str()); | ||
| 442 | } | ||
| 443 | } | ||
| 444 | ✗ | if (auto_tag_threshold > 0) { | |
| 445 | ✗ | LogCvmfs(kLogReceiver, kLogSyslog, | |
| 446 | "CommitProcessor - lease_path: %s, cleaning up auto tags " | ||
| 447 | "older than %ld", | ||
| 448 | lease_path.c_str(), static_cast<long>(auto_tag_threshold)); | ||
| 449 | } | ||
| 450 | |||
| 451 | // EditTags adds the tag for the new revision, removes any tags requested by | ||
| 452 | // `cvmfs_server tag -r`, and, when a cleanup threshold is set, removes the | ||
| 453 | // outdated auto tags -- all in the same history transaction. A failure here | ||
| 454 | // is fatal: leaving the new revision untagged (or silently keeping stale | ||
| 455 | // tags) would be worse than aborting the commit. | ||
| 456 | // | ||
| 457 | // Only real publish commits should rotate the undo tags (`trunk` and | ||
| 458 | // `trunk-previous`). Pure gateway tag edits reuse the current root hash as | ||
| 459 | // both old and new hash, so updating undo tags there would incorrectly make | ||
| 460 | // `trunk-previous` point at the current HEAD. | ||
| 461 | ✗ | const bool maintain_undo_tags = (old_root_hash != new_root_hash); | |
| 462 | ✗ | if (!EditTags(final_tag, repo_name, params, temp_dir, new_manifest_path, | |
| 463 | public_key, params.proxy, auto_tag_threshold, | ||
| 464 | maintain_undo_tags)) { | ||
| 465 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, "Error editing tags (add: '%s')", | |
| 466 | ✗ | final_tag.name().c_str()); | |
| 467 | ✗ | return kError; | |
| 468 | } | ||
| 469 | |||
| 470 | // Re-check the lease right before the final, repository-modifying step. The | ||
| 471 | // catalog merge and object upload above can be slow, during which the lease | ||
| 472 | // may have expired and an overlapping lease may have been granted to another | ||
| 473 | // publisher. If the deadline has passed we must not publish: the objects | ||
| 474 | // uploaded above stay unreferenced and are reclaimed by garbage collection. | ||
| 475 | // lease_expiration already has the gateway's configured safety margin | ||
| 476 | // subtracted, so this is a plain comparison against the current time. | ||
| 477 | ✗ | if (static_cast<int64_t>(time(NULL)) >= lease_expiration) { | |
| 478 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 479 | "CommitProcessor - lease_path: %s, lease expired during commit; " | ||
| 480 | "skipping publication, uploaded objects will be " | ||
| 481 | "garbage-collected", | ||
| 482 | lease_path.c_str()); | ||
| 483 | ✗ | return kLeaseExpired; | |
| 484 | } | ||
| 485 | |||
| 486 | ✗ | LogCvmfs(kLogReceiver, kLogSyslog, | |
| 487 | "CommitProcessor - lease_path: %s, signing manifest", | ||
| 488 | lease_path.c_str()); | ||
| 489 | |||
| 490 | // Add C_N root catalog hash to reflog through SigningTool, | ||
| 491 | // so garbage collector can later delete it. | ||
| 492 | ✗ | std::vector<shash::Any> reflog_catalogs; | |
| 493 | ✗ | reflog_catalogs.push_back(new_root_hash); | |
| 494 | |||
| 495 | ✗ | SigningTool signing_tool(server_tool.get()); | |
| 496 | ✗ | const SigningTool::Result res = signing_tool.Run( | |
| 497 | new_manifest_path, params.stratum0, params.spooler_configuration, | ||
| 498 | temp_dir, certificate, private_key, repo_name, "", "", | ||
| 499 | ✗ | "/var/spool/cvmfs/" + repo_name + "/reflog.chksum", params.proxy, | |
| 500 | ✗ | params.garbage_collection, false, false, reflog_catalogs); | |
| 501 | ✗ | switch (res) { | |
| 502 | ✗ | case SigningTool::kReflogChecksumMissing: | |
| 503 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 504 | "CommitProcessor - error: missing reflog.chksum"); | ||
| 505 | ✗ | return kMissingReflog; | |
| 506 | ✗ | case SigningTool::kReflogMissing: | |
| 507 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 508 | "CommitProcessor - error: missing reflog"); | ||
| 509 | ✗ | return kMissingReflog; | |
| 510 | ✗ | case SigningTool::kError: | |
| 511 | case SigningTool::kInitError: | ||
| 512 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 513 | "CommitProcessor - error: signing manifest"); | ||
| 514 | ✗ | return kError; | |
| 515 | ✗ | case SigningTool::kSuccess: | |
| 516 | ✗ | LogCvmfs(kLogReceiver, kLogSyslog, | |
| 517 | "CommitProcessor - lease_path: %s, success.", | ||
| 518 | lease_path.c_str()); | ||
| 519 | } | ||
| 520 | |||
| 521 | ✗ | LogCvmfs(kLogReceiver, kLogSyslog, | |
| 522 | "CommitProcessor - lease_path: %s, new root hash: %s", | ||
| 523 | ✗ | lease_path.c_str(), new_manifest_hash.ToString(false).c_str()); | |
| 524 | |||
| 525 | // Ensure CVMFS_ROOT_HASH is not set in | ||
| 526 | // /var/spool/cvmfs/<REPO_NAME>/client.local | ||
| 527 | ✗ | const std::string fname = "/var/spool/cvmfs/" + repo_name + "/client.local"; | |
| 528 | ✗ | if (truncate(fname.c_str(), 0) < 0) { | |
| 529 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, "Could not truncate %s\n", | |
| 530 | fname.c_str()); | ||
| 531 | ✗ | return kError; | |
| 532 | } | ||
| 533 | |||
| 534 | ✗ | StatisticsDatabase *stats_db = StatisticsDatabase::OpenStandardDB(repo_name); | |
| 535 | ✗ | if (stats_db != NULL) { | |
| 536 | ✗ | if (!stats_db->StorePublishStatistics(statistics_, start_time_, true)) { | |
| 537 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 538 | "Could not store publish statistics"); | ||
| 539 | } | ||
| 540 | ✗ | if (params.upload_stats_db) { | |
| 541 | const upload::SpoolerDefinition sd(params.spooler_configuration, | ||
| 542 | ✗ | shash::kAny); | |
| 543 | ✗ | upload::Spooler *spooler = upload::Spooler::Construct(sd); | |
| 544 | ✗ | if (!stats_db->UploadStatistics(spooler)) { | |
| 545 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, | |
| 546 | "Could not upload statistics DB to upstream storage"); | ||
| 547 | } | ||
| 548 | ✗ | delete spooler; | |
| 549 | } | ||
| 550 | ✗ | delete stats_db; | |
| 551 | |||
| 552 | } else { | ||
| 553 | ✗ | LogCvmfs(kLogReceiver, kLogSyslogErr, "Could not open statistics DB"); | |
| 554 | } | ||
| 555 | |||
| 556 | ✗ | return kSuccess; | |
| 557 | } | ||
| 558 | |||
| 559 | 11 | void CommitProcessor::SetStatistics(perf::Statistics *st, | |
| 560 | const std::string &start_time) { | ||
| 561 | 11 | statistics_ = st; | |
| 562 |
3/6✓ Branch 2 taken 11 times.
✗ Branch 3 not taken.
✓ Branch 6 taken 11 times.
✗ Branch 7 not taken.
✓ Branch 9 taken 11 times.
✗ Branch 10 not taken.
|
11 | statistics_->Register("publish.revision", ""); |
| 563 | 11 | start_time_ = start_time; | |
| 564 | 11 | } | |
| 565 | |||
| 566 | } // namespace receiver | ||
| 567 |