GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/swissknife_ingestsql.cc
Date: 2026-10-04 02:40:33
Exec Total Coverage
Lines: 0 903 0.0%
Branches: 0 594 0.0%
Line Branch Exec Source
1 #include "swissknife_ingestsql.h"
2
3 #include <fcntl.h>
4 #include <grp.h>
5 #include <pwd.h>
6 #include <stdio.h>
7 #include <sys/resource.h>
8 #include <sys/time.h>
9 #include <sys/types.h>
10 #include <unistd.h>
11
12 #include <csignal>
13 #include <cstdlib>
14 #include <fstream>
15 #include <sstream>
16 #include <stack>
17 #include <unordered_map>
18 #include <unordered_set>
19
20 #include "acl.h"
21 #include "catalog_downloader.h"
22 #include "catalog_mgr_rw.h"
23 #include "curl/curl.h"
24 #include "gateway_util.h"
25 #include "shortstring.h"
26 #include "swissknife_lease_curl.h"
27 #include "swissknife_lease_json.h"
28 #include "swissknife_sync.h"
29 #include "upload.h"
30 #include "util/logging.h"
31
32 #define CHECK_SQLITE_ERROR(ret, expected) \
33 do { \
34 const int sqlite_result = (ret); \
35 if (sqlite_result != (expected)) { \
36 LogCvmfs(kLogCvmfs, kLogStderr, "SQLite error: %d", \
37 sqlite_result); \
38 abort(); \
39 } \
40 } while (0)
41
42 #define CUSTOM_ASSERT(check, msg, ...) \
43 do { \
44 if (!(check)) { \
45 LogCvmfs(kLogCvmfs, kLogStderr, msg, ##__VA_ARGS__); \
46 abort(); \
47 } \
48 } while (0)
49
50 #define SHOW_PROGRESS(item, freq, curr, total) \
51 do { \
52 if ((curr) % freq == 0 || (curr) == total) { \
53 LogCvmfs(kLogCvmfs, kLogStdout, "Processed %d/%d %s", (curr), total, \
54 item); \
55 } \
56 } while (0)
57
58
59 static const unsigned kExternalChunkSize = 24 * 1024 * 1024;
60 static const unsigned kInternalChunkSize = 6 * 1024 * 1024;
61 static const unsigned kDefaultLeaseBusyRetryInterval = 10;
62 static const unsigned kLeaseRefreshInterval = 90; // seconds
63
64
65 static bool g_lease_acquired = false;
66 static string g_gateway_url;
67 static string g_gateway_key_id;
68 static string g_gateway_secret;
69 static string g_session_token;
70 static string g_session_token_file;
71 static string g_s3_file;
72 static time_t g_last_lease_refresh = 0;
73 static bool g_stop_refresh = false;
74 static string g_wait_for_update;
75 static int64_t g_priority = 0;
76 static bool g_add_missing_catalogs = false;
77 static string get_lease_from_paths(vector<string> paths);
78 static vector<string> get_all_dirs_from_sqlite(vector<string> &sqlite_db_vec,
79 bool include_additions,
80 bool include_deletions);
81 static string get_parent(const string &path);
82 static string get_basename(const string &path);
83
84 static XattrList marshal_xattrs(const char *acl);
85 static string sanitise_name(const char *name_cstr, bool allow_leading_slash);
86 static void on_signal(int sig);
87 static string acquire_lease(const string &key_id, const string &secret,
88 const string &lease_path,
89 const string &repo_service_url,
90 bool force_cancel_lease, uint64_t *current_revision,
91 string &current_root_hash,
92 unsigned int refresh_interval);
93 static void cancel_lease();
94 static void refresh_lease();
95 static vector<string> get_file_list(string &path);
96 static int check_hash(const char *hash);
97 static void recursively_delete_directory(
98 PathString &path, catalog::WritableCatalogManager &catalog_manager);
99 static void create_empty_database(string &filename);
100 static void relax_db_locking(sqlite3 *db);
101 static bool check_prefix(const std::string &path, const std::string &prefix);
102
103 static bool isDatabaseMarkedComplete(const char *dbfile);
104 static void setDatabaseMarkedComplete(const char *dbfile);
105
106 static void invalidate_manifest(std::string proxy_list, std::string url);
107 static void wait_for_update(std::string path, long revision);
108
109 extern "C" void *lease_refresh_thread(void *payload);
110
111 extern long g_final_revision;
112
113
114 ✗ static string sanitise_name(const char *name_cstr,
115 bool allow_leading_slash = false) {
116 ✗ int reason = 0;
117 ✗ const char *c = name_cstr;
118 ✗ while (*c == '/') {
119 ✗ c++;
120 } // strip any leading slashes
121 ✗ string const name = string(c);
122 ✗ bool ok = true;
123
124 ✗ if (!allow_leading_slash && HasPrefix(name, "/", true)) {
125 ✗ reason = 1;
126 ✗ ok = false;
127 }
128 ✗ if (HasSuffix(name, "/", true)) {
129 ✗ if (!(allow_leading_slash
130 ✗ && name.size() == 1)) { // account for the case where name=="/"
131 ✗ reason = 2;
132 ✗ ok = false;
133 }
134 }
135 ✗ if (name.find("//") != string::npos) {
136 ✗ reason = 3;
137 ✗ ok = false;
138 }
139 ✗ if (HasPrefix(name, "./", true) || HasPrefix(name, "../", true)) {
140 ✗ reason = 4;
141 ✗ ok = false;
142 }
143 ✗ if (HasSuffix(name, "/.", true) || HasSuffix(name, "/..", true)) {
144 ✗ reason = 5;
145 ✗ ok = false;
146 }
147 ✗ if (name.find("/./") != string::npos || name.find("/../") != string::npos) {
148 ✗ reason = 6;
149 ✗ ok = false;
150 }
151 ✗ if (name == "") {
152 ✗ reason = 7;
153 ✗ ok = false;
154 }
155 ✗ CUSTOM_ASSERT(ok, "Name [%s] is invalid (reason %d)", name.c_str(), reason);
156 ✗ return string(name);
157 }
158
159 ✗ static string get_parent(const string &path) {
160 ✗ size_t const found = path.find_last_of('/');
161 ✗ if (found == string::npos) {
162 ✗ return string("");
163 }
164 ✗ return path.substr(0, found);
165 }
166
167 ✗ static string get_basename(const string &path) {
168 ✗ const size_t found = path.find_last_of('/');
169 ✗ if (found == string::npos) {
170 ✗ return path;
171 }
172 ✗ return path.substr(found + 1);
173 }
174
175 // this is copied from MakeRelativePath
176 ✗ static string MakeCatalogPath(const std::string &relative_path) {
177 ✗ return (relative_path == "") ? "" : "/" + relative_path;
178 }
179
180 ✗ static string acquire_lease(const string &key_id, const string &secret,
181 const string &lease_path,
182 const string &repo_service_url,
183 bool force_cancel_lease, uint64_t *current_revision,
184 string &current_root_hash,
185 unsigned int refresh_interval) {
186 ✗ const CURLcode ret = curl_global_init(CURL_GLOBAL_ALL);
187 ✗ CUSTOM_ASSERT(ret == CURLE_OK, "failed to init curl");
188
189 ✗ string gateway_metadata_str;
190 ✗ char *gateway_metadata = getenv("CVMFS_GATEWAY_METADATA");
191 ✗ if (gateway_metadata != NULL)
192 ✗ gateway_metadata_str = gateway_metadata;
193
194 while (true) {
195 ✗ CurlBuffer buffer;
196 ✗ if (MakeAcquireRequest(key_id, secret, lease_path, repo_service_url,
197 &buffer, gateway_metadata_str)) {
198 ✗ string session_token;
199
200 ✗ const LeaseReply rep = ParseAcquireReplyWithRevision(
201 buffer, &session_token, current_revision, current_root_hash);
202 ✗ switch (rep) {
203 ✗ case kLeaseReplySuccess:
204 ✗ g_lease_acquired = true;
205 ✗ g_last_lease_refresh = time(NULL);
206 ✗ return session_token;
207 break;
208 ✗ case kLeaseReplyBusy:
209 ✗ if (force_cancel_lease) {
210 ✗ LogCvmfs(kLogCvmfs, kLogStderr,
211 "Lease busy, forcing cancellation (TODO");
212 }
213 ✗ LogCvmfs(kLogCvmfs, kLogStderr, "Lease busy, retrying in %d sec",
214 refresh_interval);
215 ✗ sleep(refresh_interval);
216 ✗ break;
217 ✗ default:
218 ✗ LogCvmfs(kLogCvmfs, kLogStderr,
219 "Error acquiring lease: %s. Retrying in %d sec",
220 buffer.data.c_str(), refresh_interval);
221 ✗ sleep(refresh_interval);
222 }
223 ✗ } else {
224 ✗ LogCvmfs(kLogCvmfs, kLogStderr,
225 "Error making lease acquisition request. Retrying in %d sec",
226 refresh_interval);
227 ✗ sleep(refresh_interval);
228 }
229 }
230 assert(false);
231 return "";
232 }
233
234 ✗ static uint64_t make_commit_on_gateway(const std::string &old_root_hash,
235 const std::string &new_root_hash,
236 int64_t priority) {
237 ✗ CurlBuffer buffer;
238 char priorityStr[100];
239 ✗ (void)sprintf(priorityStr, "%" PRId64,
240 priority); // skipping return value check; no way such large
241 // buffer will overflow
242 ✗ buffer.data = "";
243
244 ✗ const std::string payload = "{\n\"old_root_hash\": \"" + old_root_hash
245 ✗ + "\",\n\"new_root_hash\": \"" + new_root_hash
246 ✗ + "\",\n\"priority\": " + priorityStr + "}";
247
248 ✗ return MakeEndRequest("POST", g_gateway_key_id, g_gateway_secret,
249 g_session_token, g_gateway_url, payload, &buffer,
250 ✗ true /*expect_final_revision*/);
251 }
252
253 ✗ static void refresh_lease() {
254 ✗ CurlBuffer buffer;
255 ✗ buffer.data = "";
256 ✗ if ((time(NULL) - g_last_lease_refresh) < kLeaseRefreshInterval) {
257 ✗ return;
258 }
259
260 ✗ if (MakeEndRequest("PATCH", g_gateway_key_id, g_gateway_secret,
261 g_session_token, g_gateway_url, "", &buffer,
262 false /*expect_final_revision*/)) {
263 ✗ const int ret = ParseDropReply(buffer);
264 ✗ if (kLeaseReplySuccess == ret) {
265 ✗ LogCvmfs(kLogCvmfs, kLogVerboseMsg, "Lease refreshed");
266 ✗ g_last_lease_refresh = time(NULL);
267 } else {
268 ✗ LogCvmfs(kLogCvmfs, kLogStderr, "Lease refresh failed: %d", ret);
269 }
270 } else {
271 ✗ LogCvmfs(kLogCvmfs, kLogStderr, "Lease refresh request failed");
272 ✗ if (buffer.data == "Method Not Allowed\n") {
273 ✗ g_last_lease_refresh = time(NULL);
274 ✗ LogCvmfs(kLogCvmfs, kLogStderr,
275 "This gateway does not support lease refresh");
276 }
277 }
278 }
279
280
281 ✗ static void cancel_lease() {
282 ✗ CurlBuffer buffer;
283 ✗ if (MakeEndRequest("DELETE", g_gateway_key_id, g_gateway_secret,
284 g_session_token, g_gateway_url, "", &buffer,
285 false /*expect_final_revision*/)) {
286 ✗ const int ret = ParseDropReply(buffer);
287 ✗ if (kLeaseReplySuccess == ret) {
288 ✗ LogCvmfs(kLogCvmfs, kLogStdout, "Lease cancelled");
289 } else {
290 ✗ LogCvmfs(kLogCvmfs, kLogStderr, "Lease cancellation failed: %d", ret);
291 }
292 } else {
293 ✗ LogCvmfs(kLogCvmfs, kLogStderr, "Lease cancellation request failed");
294 }
295 ✗ g_stop_refresh = true;
296 }
297
298 ✗ static void on_signal(int sig) {
299 ✗ (void)signal(sig, SIG_DFL);
300 ✗ if (g_lease_acquired) {
301 ✗ LogCvmfs(kLogCvmfs, kLogStdout, "Cancelling lease");
302 ✗ cancel_lease();
303 ✗ unlink(g_session_token_file.c_str());
304 }
305 ✗ if (sig == SIGINT || sig == SIGTERM)
306 ✗ exit(1);
307 }
308
309 ✗ static vector<string> get_all_dirs_from_sqlite(vector<string> &sqlite_db_vec,
310 bool include_additions,
311 bool include_deletions) {
312 int ret;
313 ✗ vector<string> paths;
314
315 ✗ for (vector<string>::iterator it = sqlite_db_vec.begin();
316 ✗ it != sqlite_db_vec.end();
317 ✗ it++) {
318 sqlite3 *db;
319 ✗ ret = sqlite3_open_v2((*it).c_str(), &db, SQLITE_OPEN_READONLY, NULL);
320 ✗ CHECK_SQLITE_ERROR(ret, SQLITE_OK);
321 ✗ relax_db_locking(db);
322
323 ✗ vector<string> tables;
324 ✗ if (include_additions) {
325 ✗ tables.push_back("dirs");
326 ✗ tables.push_back("links");
327 ✗ tables.push_back("files");
328 }
329 ✗ if (include_deletions) {
330 ✗ tables.push_back("deletions");
331 }
332
333 // get all the paths from the DB
334 ✗ for (vector<string>::iterator it = tables.begin(); it != tables.end();
335 ✗ it++) {
336 sqlite3_stmt *stmt;
337 ✗ const string query = "SELECT name FROM " + *it;
338 ✗ ret = sqlite3_prepare_v2(db, query.c_str(), -1, &stmt, NULL);
339 ✗ CHECK_SQLITE_ERROR(ret, SQLITE_OK);
340 ✗ while (sqlite3_step(stmt) == SQLITE_ROW) {
341 const char *name = reinterpret_cast<const char *>(
342 ✗ sqlite3_column_text(stmt, 0));
343 ✗ const string names = sanitise_name(name);
344 ✗ if (*it == "dirs") {
345 ✗ paths.push_back(names);
346 } else {
347 ✗ paths.push_back(get_parent(names));
348 }
349 }
350 ✗ ret = sqlite3_finalize(stmt);
351 ✗ CHECK_SQLITE_ERROR(ret, SQLITE_OK);
352 }
353 ✗ ret = sqlite3_close_v2(db);
354 ✗ CHECK_SQLITE_ERROR(ret, SQLITE_OK);
355 }
356 ✗ return paths;
357 }
358
359 ✗ static int get_db_schema_revision(sqlite3 *db,
360 const std::string &db_name = "") {
361 sqlite3_stmt *stmt;
362 ✗ std::ostringstream stmt_str;
363 stmt_str << "SELECT value FROM " << db_name
364 ✗ << "properties WHERE key = 'schema_revision'";
365 ✗ int ret = sqlite3_prepare_v2(db, stmt_str.str().c_str(), -1, &stmt, NULL);
366 ✗ CHECK_SQLITE_ERROR(ret, SQLITE_OK);
367
368 ✗ ret = sqlite3_step(stmt);
369 // if table exists, we require that it must have a schema_revision row
370 ✗ CHECK_SQLITE_ERROR(ret, SQLITE_ROW);
371 const std::string schema_revision_str(
372 ✗ reinterpret_cast<const char *>(sqlite3_column_text(stmt, 0)));
373 ✗ CHECK_SQLITE_ERROR(sqlite3_finalize(stmt), SQLITE_OK);
374 ✗ return std::stoi(schema_revision_str);
375 }
376
377 ✗ static int get_row_count(sqlite3 *db, const std::string &table_name) {
378 sqlite3_stmt *stmt;
379 ✗ std::ostringstream stmt_str;
380 ✗ stmt_str << "SELECT COUNT(*) FROM " << table_name;
381 ✗ int ret = sqlite3_prepare_v2(db, stmt_str.str().c_str(), -1, &stmt, NULL);
382 ✗ CHECK_SQLITE_ERROR(ret, SQLITE_OK);
383
384 ✗ ret = sqlite3_step(stmt);
385 ✗ CHECK_SQLITE_ERROR(ret, SQLITE_ROW);
386 const std::string count_str(
387 ✗ reinterpret_cast<const char *>(sqlite3_column_text(stmt, 0)));
388 ✗ CHECK_SQLITE_ERROR(sqlite3_finalize(stmt), SQLITE_OK);
389 ✗ return std::stoi(count_str);
390 }
391
392 ✗ static int calculate_print_frequency(int total) {
393 ✗ int base = 1000;
394 ✗ while (base * 50 < total)
395 ✗ base *= 10;
396 ✗ return base;
397 }
398
399 // compute a common path among for paths for use as the lease path
400 ✗ static string get_lease_from_paths(vector<string> paths) {
401 ✗ CUSTOM_ASSERT(!paths.empty(), "no paths are provided");
402
403 // we'd have to ensure path is relative
404 // (probably best to check this elsewhere as it is not just a requirement for
405 // this function)
406 ✗ auto lease = PathString(paths.at(0));
407 ✗ for (auto it = paths.begin() + 1; it != paths.end(); ++it) {
408 ✗ auto path = PathString(*it);
409 // shrink the lease path until it is a parent of "path"
410 ✗ while (!IsSubPath(lease, path)) {
411 ✗ auto i = lease.GetLength() - 1;
412 ✗ for (; i >= 0; --i) {
413 ✗ if (lease.GetChars()[i] == '/' || i == 0) {
414 ✗ lease.Truncate(i);
415 ✗ break;
416 }
417 }
418 }
419 ✗ if (lease.IsEmpty())
420 ✗ break; // early stop if lease is already at the root
421 }
422
423 ✗ auto prefix = "/" + lease.ToString();
424
425 ✗ LogCvmfs(kLogCvmfs, kLogStdout, "Longest prefix is %s", prefix.c_str());
426 ✗ return prefix;
427 }
428
429 ✗ static XattrList marshal_xattrs(const char *acl_string) {
430 ✗ XattrList aclobj;
431
432 ✗ if (acl_string == NULL || acl_string[0] == '\0') {
433 ✗ return aclobj;
434 }
435
436 bool equiv_mode;
437 size_t binary_size;
438 char *binary_acl;
439 ✗ const int ret = acl_from_text_to_xattr_value(string(acl_string), binary_acl,
440 binary_size, equiv_mode);
441 ✗ if (ret) {
442 ✗ LogCvmfs(kLogCvmfs, kLogStderr,
443 "failure of acl_from_text_to_xattr_value(%s)", acl_string);
444 ✗ abort();
445 }
446 ✗ if (!equiv_mode) {
447 ✗ CUSTOM_ASSERT(
448 aclobj.Set("system.posix_acl_access", string(binary_acl, binary_size)),
449 "failed to set system.posix_acl_access (ACL size %ld)", binary_size);
450 ✗ free(binary_acl);
451 }
452
453 ✗ return aclobj;
454 }
455
456 ✗ std::unordered_map<string, string> load_config(const string &config_file) {
457 ✗ std::unordered_map<string, string> config_map;
458 ✗ ifstream input(config_file);
459 ✗ if (!input) {
460 ✗ LogCvmfs(kLogCvmfs, kLogStderr, "could not open config file %s",
461 config_file.c_str());
462 ✗ return config_map;
463 }
464 ✗ vector<string> lines;
465 ✗ for (string line; getline(input, line);) {
466 ✗ lines.push_back(line);
467 }
468
469 ✗ for (auto it = lines.begin(); it != lines.end(); it++) {
470 ✗ const string l = *it;
471 ✗ const size_t p = l.find('=', 0);
472 ✗ if (p != string::npos) {
473 ✗ const string key = l.substr(0, p);
474 ✗ string val = l.substr(p + 1);
475 // trim any double quotes
476 ✗ if (val.front() == '"') {
477 ✗ val = val.substr(1, val.length() - 2);
478 }
479 ✗ config_map[key] = val;
480 }
481 }
482
483 ✗ return config_map;
484 }
485
486 ✗ string retrieve_config(std::unordered_map<string, string> &config_map,
487 const string &key) {
488 ✗ auto kv = config_map.find(key);
489 ✗ CUSTOM_ASSERT(kv != config_map.end(), "Parameter %s not found in config",
490 key.c_str());
491 ✗ return kv->second;
492 }
493
494 ✗ static vector<string> get_file_list(string &path) {
495 ✗ vector<string> paths;
496 ✗ const char *cpath = path.c_str();
497 struct stat st;
498 ✗ const int ret = stat(cpath, &st);
499 ✗ CUSTOM_ASSERT(ret == 0, "failed to stat file %s", cpath);
500
501 ✗ if (S_ISDIR(st.st_mode)) {
502 DIR *d;
503 struct dirent *dir;
504 ✗ d = opendir(cpath);
505 ✗ if (d) {
506 ✗ while ((dir = readdir(d)) != NULL) {
507 ✗ const char *t = strrchr(dir->d_name, '.');
508 ✗ if (t && !strcmp(t, ".db")) {
509 ✗ paths.push_back(path + "/" + dir->d_name);
510 }
511 }
512 ✗ closedir(d);
513 }
514 } else {
515 ✗ paths.push_back(path);
516 }
517 ✗ return paths;
518 }
519
520 extern bool g_log_with_time;
521
522 ✗ int swissknife::IngestSQL::Main(const swissknife::ArgumentList &args) {
523 // the catalog code uses assert() liberally.
524 // install ABRT signal handler to catch an abort and cancel lease
525 ✗ if (signal(SIGABRT, &on_signal) == SIG_ERR
526 ✗ || signal(SIGINT, &on_signal) == SIG_ERR
527 ✗ || signal(SIGTERM, &on_signal) == SIG_ERR) {
528 ✗ LogCvmfs(kLogCvmfs, kLogStdout, "Setting signal handlers failed");
529 ✗ exit(1);
530 }
531
532 ✗ const bool enable_corefiles = (args.find('c') != args.end());
533 ✗ if (!enable_corefiles) {
534 struct rlimit rlim;
535 ✗ rlim.rlim_cur = rlim.rlim_max = 0;
536 ✗ setrlimit(RLIMIT_CORE, &rlim);
537 }
538
539
540 ✗ if (args.find('n') != args.end()) {
541 ✗ create_empty_database(*args.find('n')->second);
542 ✗ exit(0);
543 }
544
545 ✗ if (args.find('B') != args.end()) {
546 ✗ g_wait_for_update = *args.find('B')->second;
547 }
548
549 ✗ if (args.find('P') != args.end()) {
550 ✗ const char *arg = (*args.find('P')->second).c_str();
551 char *at_null_terminator_if_number;
552 ✗ g_priority = strtoll(arg, &at_null_terminator_if_number, 10);
553 ✗ if (*at_null_terminator_if_number != '\0') {
554 ✗ LogCvmfs(kLogCvmfs, kLogStderr,
555 "Priority parameter value '%s' parsing failed", arg);
556 ✗ return 1;
557 }
558 } else {
559 ✗ g_priority = -time(NULL);
560 }
561
562
563 ✗ unsigned int lease_busy_retry_interval = kDefaultLeaseBusyRetryInterval;
564 ✗ if (args.find('r') != args.end()) {
565 ✗ lease_busy_retry_interval = atoi((*args.find('r')->second).c_str());
566 }
567
568 ✗ string dir_temp = "";
569 const char *env_tmpdir;
570 ✗ if (args.find('t') != args.end()) {
571 ✗ dir_temp = MakeCanonicalPath(*args.find('t')->second);
572 ✗ } else if ((env_tmpdir = getenv("TMPDIR"))) {
573 ✗ dir_temp = MakeCanonicalPath(env_tmpdir);
574 } else {
575 ✗ LogCvmfs(kLogCvmfs, kLogStderr, "-t or TMPDIR required");
576 ✗ return 1;
577 }
578
579 ✗ string kConfigDir("/etc/cvmfs/gateway-client/");
580 ✗ if (args.find('C') != args.end()) {
581 ✗ kConfigDir = MakeCanonicalPath(*args.find('C')->second);
582 ✗ kConfigDir += "/";
583 ✗ LogCvmfs(kLogCvmfs, kLogStdout, "Overriding configuration dir prefix to %s",
584 kConfigDir.c_str());
585 }
586
587 // mandatory arguments
588 ✗ string const repo_name = *args.find('N')->second;
589 ✗ string sqlite_db_path = *args.find('D')->second;
590
591 ✗ vector<string> sqlite_db_vec = get_file_list(sqlite_db_path);
592
593 // optional arguments
594 ✗ bool const allow_deletions = (args.find('d') != args.end());
595 ✗ bool const force_cancel_lease = (args.find('x') != args.end());
596 ✗ bool const allow_additions = !allow_deletions
597 ✗ || (args.find('a') != args.end());
598 ✗ g_add_missing_catalogs = (args.find('z') != args.end());
599 ✗ bool const check_completed_graft_property = (args.find('Z') != args.end());
600 ✗ if (args.find('v') != args.end()) {
601 ✗ SetLogVerbosity(kLogVerbose);
602 }
603
604 ✗ if (check_completed_graft_property) {
605 ✗ if (sqlite_db_vec.size() != 1) {
606 ✗ LogCvmfs(kLogCvmfs, kLogStderr, "-Z requires a single DB file");
607 ✗ exit(1);
608 }
609 ✗ if (isDatabaseMarkedComplete(sqlite_db_vec[0].c_str())) {
610 ✗ LogCvmfs(kLogCvmfs, kLogStderr,
611 "DB file is already marked as completed_graft");
612 ✗ exit(0);
613 } else {
614 ✗ LogCvmfs(kLogCvmfs, kLogStderr,
615 "DB file is not marked as completed_graft");
616 }
617 }
618
619 ✗ string const config_file = kConfigDir + repo_name + "/config";
620 ✗ string stratum0;
621 ✗ string proxy;
622
623 ✗ string additional_prefix = "";
624 ✗ bool has_additional_prefix = false;
625 ✗ if (args.find('p') != args.end()) {
626 ✗ additional_prefix = *args.find('p')->second;
627 ✗ additional_prefix = sanitise_name(additional_prefix.c_str(), true);
628 ✗ if (additional_prefix.back() != '/') {
629 ✗ additional_prefix += "/";
630 }
631 ✗ has_additional_prefix = true;
632 ✗ LogCvmfs(kLogCvmfs, kLogStdout,
633 "Adding additional prefix %s to lease and all paths",
634 additional_prefix.c_str());
635 // now we are confident that any additional prefix has no leading / and does
636 // have a tailing /
637 }
638 ✗ auto config_map = load_config(config_file);
639
640 ✗ if (args.find('g') != args.end()) {
641 ✗ g_gateway_url = *args.find('g')->second;
642 } else {
643 ✗ g_gateway_url = retrieve_config(config_map, "CVMFS_GATEWAY");
644 }
645 ✗ if (args.find('w') != args.end()) {
646 ✗ stratum0 = *args.find('w')->second;
647 } else {
648 ✗ stratum0 = retrieve_config(config_map, "CVMFS_STRATUM0");
649 }
650
651 ✗ if (args.find('@') != args.end()) {
652 ✗ proxy = *args.find('@')->second;
653 } else {
654 ✗ proxy = retrieve_config(config_map, "CVMFS_HTTP_PROXY");
655 }
656
657 ✗ string lease_path = "";
658 // bool lease_autodetected = false;
659 ✗ if (args.find('l') != args.end()) {
660 ✗ lease_path = *args.find('l')->second;
661 } else {
662 // lease path wasn't specified, so try to autodetect it
663 vector<string> const paths = get_all_dirs_from_sqlite(
664 ✗ sqlite_db_vec, allow_additions, allow_deletions);
665 ✗ if (paths.size() == 0) {
666 ✗ LogCvmfs(kLogCvmfs, kLogStdout, "Database is empty, nothing to do");
667 ✗ return 0; // treat it as a success
668 }
669 ✗ lease_path = get_lease_from_paths(paths);
670 // lease_autodetected = true;
671 }
672
673 ✗ if (has_additional_prefix) {
674 ✗ if (lease_path == "/") {
675 ✗ lease_path = "/" + additional_prefix;
676 } else {
677 ✗ if (lease_path.substr(0, 1) == "/") {
678 ✗ lease_path = "/" + additional_prefix
679 ✗ + lease_path.substr(1, lease_path.size() - 1);
680 } else {
681 ✗ lease_path = "/" + additional_prefix
682 ✗ + lease_path; // prefix is certain to have a trailing /
683 }
684 }
685 }
686 ✗ if (lease_path.substr(0, 1) != "/") {
687 ✗ lease_path = "/" + lease_path;
688 }
689 ✗ LogCvmfs(kLogCvmfs, kLogStdout, "Lease path is %s", lease_path.c_str());
690
691
692 ✗ string public_keys = kConfigDir + repo_name + "/pubkey";
693 ✗ string key_file = kConfigDir + repo_name + "/gatewaykey";
694 ✗ string s3_file = kConfigDir + repo_name + "/s3.conf";
695
696 ✗ if (args.find('k') != args.end()) {
697 ✗ public_keys = *args.find('k')->second;
698 }
699 ✗ if (args.find('s') != args.end()) {
700 ✗ key_file = *args.find('s')->second;
701 }
702 ✗ if (args.find('3') != args.end()) {
703 ✗ s3_file = *args.find('3')->second;
704 }
705
706 ✗ CUSTOM_ASSERT(access(public_keys.c_str(), R_OK) == 0, "%s is not readable",
707 public_keys.c_str());
708 ✗ CUSTOM_ASSERT(access(key_file.c_str(), R_OK) == 0, "%s is not readable",
709 key_file.c_str());
710
711 // string spooler_definition_string = string("gw,,") + g_gateway_url;
712 // create a spooler that will upload to S3
713 ✗ string const spooler_definition_string = string("S3,") + dir_temp + ","
714 ✗ + repo_name + "@" + s3_file;
715
716 // load gateway lease
717 ✗ if (!gateway::ReadKeys(key_file, &g_gateway_key_id, &g_gateway_secret)) {
718 ✗ LogCvmfs(kLogCvmfs, kLogStderr, "gateway::ReadKeys failed");
719 ✗ return 1;
720 }
721
722 ✗ uint64_t current_revision = 0;
723 ✗ std::string current_root_hash = "";
724
725 // acquire lease and save token to a file in the tmpdir
726 ✗ LogCvmfs(kLogCvmfs, kLogStdout, "Acquiring gateway lease on %s",
727 lease_path.c_str());
728 ✗ g_session_token = acquire_lease(g_gateway_key_id, g_gateway_secret,
729 ✗ repo_name + lease_path, g_gateway_url,
730 force_cancel_lease, &current_revision,
731 ✗ current_root_hash, lease_busy_retry_interval);
732
733
734 ✗ char *_tmpfile = strdup((dir_temp + "/gateway_session_token_XXXXXX").c_str());
735 ✗ int const temp_fd = mkstemp(_tmpfile);
736 ✗ g_session_token_file = string(_tmpfile);
737 ✗ free(_tmpfile);
738
739 ✗ FILE *fout = fdopen(temp_fd, "wb");
740 ✗ CUSTOM_ASSERT(fout != NULL,
741 "failed to open session token file %s for writing",
742 g_session_token_file.c_str());
743 ✗ fputs(g_session_token.c_str(), fout);
744 ✗ fclose(fout);
745
746 // now start the lease refresh thread
747 pthread_t lease_thread;
748 ✗ if (0 != pthread_create(&lease_thread, NULL, lease_refresh_thread, NULL)) {
749 ✗ LogCvmfs(kLogCvmfs, kLogStderr, "Unable to start lease refresh thread");
750 ✗ cancel_lease();
751 ✗ return 1;
752 }
753
754 // now initialise the various bits we need
755
756 upload::SpoolerDefinition spooler_definition(
757 spooler_definition_string, shash::kSha1, zlib::kZlibDefault, false, true,
758 SyncParameters::kDefaultMinFileChunkSize,
759 SyncParameters::kDefaultAvgFileChunkSize,
760 ✗ SyncParameters::kDefaultMaxFileChunkSize, g_session_token_file, key_file);
761
762 ✗ if (args.find('q') != args.end()) {
763 ✗ spooler_definition.number_of_concurrent_uploads = String2Uint64(
764 ✗ *args.find('q')->second);
765 }
766
767 upload::SpoolerDefinition const spooler_definition_catalogs(
768 ✗ spooler_definition.Dup2DefaultCompression());
769
770 std::unique_ptr<upload::Spooler> const spooler_catalogs(
771 ✗ upload::Spooler::Construct(spooler_definition_catalogs, nullptr));
772
773 ✗ if (spooler_catalogs.get() == nullptr) {
774 ✗ LogCvmfs(kLogCvmfs, kLogStderr, "spooler_catalogs invalid");
775 ✗ cancel_lease();
776 ✗ return 1;
777 }
778 ✗ if (!InitDownloadManager(true, proxy, kCatalogDownloadMultiplier)) {
779 ✗ LogCvmfs(kLogCvmfs, kLogStderr, "download manager init failed");
780 ✗ cancel_lease();
781 ✗ return 1;
782 }
783 ✗ if (!InitSignatureManager(public_keys, "")) {
784 ✗ LogCvmfs(kLogCvmfs, kLogStderr, "signature manager init failed");
785 ✗ cancel_lease();
786 ✗ return 1;
787 }
788
789 ✗ std::unique_ptr<manifest::Manifest> manifest;
790
791 ✗ manifest.reset(FetchRemoteManifest(stratum0, repo_name, shash::Any()));
792
793 ✗ if (manifest.get() == nullptr) {
794 ✗ LogCvmfs(kLogCvmfs, kLogStderr, "manifest invalid");
795 ✗ cancel_lease();
796 ✗ return 1;
797 }
798
799 ✗ if (current_revision > 0) {
800 ✗ if (current_revision == manifest->revision()) {
801 ✗ if (current_root_hash != manifest->catalog_hash().ToString()) {
802 ✗ LogCvmfs(kLogCvmfs, kLogStderr,
803 "Mismatch between cvmfspublished and gateway hash for "
804 "revision %lu (%s!=%s)",
805 current_revision, current_root_hash.c_str(),
806 manifest->catalog_hash().ToString().c_str());
807 ✗ cancel_lease();
808 ✗ return 1;
809 } else {
810 ✗ LogCvmfs(kLogCvmfs, kLogStdout,
811 "Gateway and .cvmfspublished agree on repo version %lu",
812 current_revision);
813 }
814 }
815 ✗ if (current_revision > manifest->revision()) {
816 ✗ LogCvmfs(kLogCvmfs, kLogStdout,
817 "Gateway has supplied a newer revision than the current "
818 ".cvmfspublished %lu > %lu",
819 current_revision, manifest->revision());
820 ✗ manifest->set_revision(current_revision);
821 ✗ manifest->set_catalog_hash(shash::MkFromHexPtr(
822 shash::HexPtr(current_root_hash), shash::kSuffixCatalog));
823 ✗ } else if (current_revision < manifest->revision()) {
824 ✗ LogCvmfs(kLogCvmfs, kLogStdout,
825 "Gateway has supplied an older revision than the current "
826 ".cvmfspublished %lu < %lu",
827 current_revision, manifest->revision());
828 }
829 } else {
830 ✗ LogCvmfs(kLogCvmfs, kLogStdout,
831 "Gateway has not supplied a revision. Using .cvmfspublished");
832 }
833
834
835 // get hash of current root catalog, remove terminal "C", encode it
836 ✗ string const old_root_hash = manifest->catalog_hash().ToString(true);
837 ✗ string const hash = old_root_hash.substr(0, old_root_hash.length() - 1);
838 ✗ shash::Any const base_hash = shash::MkFromHexPtr(shash::HexPtr(hash),
839 shash::kSuffixCatalog);
840 ✗ LogCvmfs(kLogCvmfs, kLogStdout, "old_root_hash: %s", old_root_hash.c_str());
841
842 ✗ bool const is_balanced = false;
843
844 catalog::WritableCatalogManager catalog_manager(
845 base_hash, stratum0, dir_temp, spooler_catalogs.get(), download_manager(),
846 false, SyncParameters::kDefaultNestedKcatalogLimit,
847 SyncParameters::kDefaultRootKcatalogLimit,
848 SyncParameters::kDefaultFileMbyteLimit, statistics(), is_balanced,
849 SyncParameters::kDefaultMaxWeight, SyncParameters::kDefaultMinWeight,
850 ✗ dir_temp /* dir_cache */);
851
852 ✗ catalog_manager.Init();
853
854
855 // now graft the contents of the DB
856 ✗ vector<sqlite3 *> open_dbs;
857 ✗ for (auto &&db_file : sqlite_db_vec) {
858 sqlite3 *db;
859 ✗ CHECK_SQLITE_ERROR(
860 sqlite3_open_v2(db_file.c_str(), &db, SQLITE_OPEN_READONLY, NULL),
861 SQLITE_OK);
862 ✗ relax_db_locking(db);
863 ✗ open_dbs.push_back(db);
864 }
865 ✗ process_sqlite(open_dbs, catalog_manager, allow_additions, allow_deletions,
866 ✗ lease_path.substr(1), additional_prefix);
867 ✗ for (auto &&db : open_dbs) {
868 ✗ CHECK_SQLITE_ERROR(sqlite3_close_v2(db), SQLITE_OK);
869 }
870
871 // commit changes
872 ✗ LogCvmfs(kLogCvmfs, kLogStdout, "Committing changes...");
873 ✗ if (!catalog_manager.Commit(false, false, manifest.get())) {
874 ✗ LogCvmfs(kLogCvmfs, kLogStderr, "something went wrong during sync");
875 ✗ cancel_lease();
876 ✗ return 1;
877 }
878
879 // finalize the spooler
880 ✗ LogCvmfs(kLogCvmfs, kLogStdout, "Waiting for all uploads to finish...");
881 ✗ spooler_catalogs->WaitForUpload();
882
883 ✗ LogCvmfs(kLogCvmfs, kLogStdout, "Exporting repository manifest");
884
885 // We call FinalizeSession(true) this time, to also trigger the commit
886 // operation on the gateway machine (if the upstream is of type "gw").
887
888 // Get the path of the new root catalog
889 ✗ const string new_root_hash = manifest->catalog_hash().ToString(true);
890
891 // if (!spooler_catalogs->FinalizeSession(true, old_root_hash, new_root_hash,
892 // RepositoryTag())) {
893 // LogCvmfs(kLogCvmfs, kLogStderr, "Failed to commit the transaction");
894 // // lease is only released on success
895 // cancel_lease();
896 // return 1;
897 // }
898
899 ✗ LogCvmfs(kLogCvmfs, kLogStdout, "Committing with priority %" PRId64,
900 g_priority);
901
902 ✗ bool const ok = make_commit_on_gateway(old_root_hash, new_root_hash,
903 ✗ g_priority);
904 ✗ if (!ok) {
905 ✗ LogCvmfs(kLogCvmfs, kLogStderr,
906 "something went wrong during commit on gateway");
907 ✗ cancel_lease();
908 ✗ exit(1);
909 }
910
911
912 ✗ unlink(g_session_token_file.c_str());
913
914 ✗ g_stop_refresh = true;
915
916 ✗ if (g_wait_for_update != "") {
917 ✗ invalidate_manifest(proxy, stratum0 + "/.cvmfspublished");
918 ✗ wait_for_update(g_wait_for_update, g_final_revision);
919 }
920
921 ✗ if (check_completed_graft_property) {
922 ✗ setDatabaseMarkedComplete(sqlite_db_vec[0].c_str());
923 }
924
925
926 ✗ return 0;
927 }
928
929 ✗ size_t writeFunction(void *ptr, size_t size, size_t nmemb, std::string *data) {
930 ✗ return size * nmemb;
931 }
932
933 ✗ void replaceAllSubstrings(std::string &str, const std::string &from,
934 const std::string &to) {
935 ✗ if (from.empty()) {
936 ✗ return; // Avoid infinite loop if 'from' is an empty string.
937 }
938 ✗ size_t startPos = 0;
939 ✗ while ((startPos = str.find(from, startPos)) != std::string::npos) {
940 ✗ str.replace(startPos, from.length(), to);
941 ✗ startPos += to.length(); // Advance startPos to avoid replacing the
942 // substring just inserted.
943 }
944 }
945
946
947 ✗ void swissknife::IngestSQL::process_sqlite(
948 const std::vector<sqlite3 *> &dbs,
949 catalog::WritableCatalogManager &catalog_manager, bool allow_additions,
950 bool allow_deletions, const std::string &lease_path,
951 const std::string &additional_prefix) {
952 ✗ std::map<std::string, Directory> all_dirs;
953 ✗ std::map<std::string, std::vector<File> > all_files;
954 ✗ std::map<std::string, std::vector<Symlink> > all_symlinks;
955
956 ✗ for (auto &&db : dbs) {
957 ✗ load_dirs(db, lease_path, additional_prefix, all_dirs);
958 }
959
960 // put in a nested scope so we can free up memory of `dir_names`
961 {
962 ✗ LogCvmfs(kLogCvmfs, kLogStdout,
963 "Precaching existing directories (starting from %s)",
964 lease_path.c_str());
965 ✗ std::unordered_set<std::string> dir_names;
966 ✗ std::transform(all_dirs.begin(), all_dirs.end(),
967 std::inserter(dir_names, dir_names.end()),
968 ✗ [](const std::pair<std::string, Directory> &pair) {
969 ✗ return MakeCatalogPath(pair.first);
970 });
971 ✗ catalog_manager.LoadCatalogs(MakeCatalogPath(lease_path), dir_names);
972 }
973
974 ✗ for (auto &&db : dbs) {
975 ✗ load_files(db, lease_path, additional_prefix, all_files);
976 ✗ load_symlinks(db, lease_path, additional_prefix, all_symlinks);
977 }
978
979 // perform all deletions first
980 ✗ if (allow_deletions) {
981 ✗ LogCvmfs(kLogCvmfs, kLogStdout, "Processing deletions...");
982 ✗ for (auto &&db : dbs) {
983 ✗ CHECK_SQLITE_ERROR(
984 do_deletions(db, catalog_manager, lease_path, additional_prefix),
985 SQLITE_OK);
986 }
987 }
988
989 ✗ if (allow_additions) {
990 ✗ LogCvmfs(kLogCvmfs, kLogStdout, "Processing additions...");
991 // first ensure all directories are present and create missing ones
992 ✗ do_additions(all_dirs, all_files, all_symlinks, lease_path,
993 catalog_manager);
994 }
995 }
996
997 ✗ void add_dir_to_tree(
998 std::string path,
999 std::unordered_map<std::string, std::set<std::string> > &tree,
1000 const std::string &lease_path) {
1001 ✗ tree[path];
1002 ✗ std::string parent_path = get_parent(path);
1003 // recursively create any missing parents in the tree
1004 // avoid creating a loop when we insert the root path
1005 ✗ while (path != parent_path && path != lease_path
1006 ✗ && !tree[parent_path].count(path)) {
1007 ✗ tree[parent_path].insert(path);
1008 ✗ path = parent_path;
1009 ✗ parent_path = get_parent(path);
1010 }
1011 }
1012
1013 ✗ int swissknife::IngestSQL::do_additions(
1014 const DirMap &all_dirs, const FileMap &all_files,
1015 const SymlinkMap &all_symlinks, const std::string &lease_path,
1016 catalog::WritableCatalogManager &catalog_manager) {
1017 // STEP 1:
1018 // - collect all the dirs/symlinks/files we need to process from the DB
1019 // - build a tree of paths for DFS traversal
1020 // - note the tree will contain all parent dirs of symlinks/files even if
1021 // those are not
1022 // explicitly added to the dirs table
1023 ✗ std::unordered_map<std::string, std::set<std::string> > tree;
1024 ✗ for (auto &&p : all_dirs) {
1025 ✗ add_dir_to_tree(p.first, tree, lease_path);
1026 }
1027 ✗ for (auto &&p : all_files) {
1028 ✗ add_dir_to_tree(p.first, tree, lease_path);
1029 }
1030 ✗ for (auto &&p : all_symlinks) {
1031 ✗ add_dir_to_tree(p.first, tree, lease_path);
1032 }
1033 ✗ int const row_count = static_cast<int>(tree.size());
1034 ✗ int const print_every = calculate_print_frequency(row_count);
1035 ✗ int curr_row = 0;
1036 ✗ LogCvmfs(kLogCvmfs, kLogStdout,
1037 "Changeset: %ld dirs, %ld files, %ld symlinks", tree.size(),
1038 all_files.size(), all_symlinks.size());
1039
1040 // STEP 2:
1041 // - process all the changes with DFS traversal
1042 // - make directories in pre-order
1043 // - add files/symlinks and schedule upload in post-order
1044 ✗ catalog_manager.SetupSingleCatalogUploadCallback();
1045 ✗ std::stack<string> dfs_stack;
1046 ✗ for (auto &&p : tree) {
1047 // figure out the starting point by checking whose parent is missing from
1048 // the tree
1049 ✗ if (p.first == "" || !tree.count(get_parent(p.first))) {
1050 ✗ CUSTOM_ASSERT(dfs_stack.empty(),
1051 "provided DB input forms more than one path trees");
1052 ✗ dfs_stack.push(p.first);
1053 }
1054 }
1055 ✗ std::set<string> visited;
1056 ✗ while (!dfs_stack.empty()) {
1057 ✗ string const curr_dir = dfs_stack.top();
1058 // add content for the dir in post-order traversal
1059 ✗ if (visited.count(curr_dir)) {
1060 ✗ curr_row++;
1061 ✗ if (all_symlinks.count(curr_dir)) {
1062 ✗ add_symlinks(catalog_manager, all_symlinks.at(curr_dir));
1063 }
1064 ✗ if (all_files.count(curr_dir)) {
1065 ✗ add_files(catalog_manager, all_files.at(curr_dir));
1066 }
1067 // snapshot the dir (if it's a nested catalog mountpoint)
1068 ✗ catalog::DirectoryEntry dir_entry;
1069 ✗ bool exists = false;
1070 ✗ exists = catalog_manager.LookupDirEntry(
1071 ✗ MakeCatalogPath(curr_dir), catalog::kLookupDefault, &dir_entry);
1072 ✗ CUSTOM_ASSERT(exists, "Directory %s is missing from the catalog",
1073 curr_dir.c_str());
1074 ✗ if (dir_entry.IsNestedCatalogMountpoint()
1075 ✗ || dir_entry.IsNestedCatalogRoot()) {
1076 ✗ catalog_manager.AddCatalogToQueue(curr_dir);
1077 ✗ catalog_manager.ScheduleReadyCatalogs();
1078 }
1079 ✗ dfs_stack.pop();
1080 ✗ SHOW_PROGRESS("directories", print_every, curr_row, row_count);
1081 ✗ } else {
1082 ✗ visited.insert(curr_dir);
1083 // push children to the stack
1084 ✗ auto it = tree.find(curr_dir);
1085 ✗ if (it != tree.end()) {
1086 ✗ for (auto &&child : it->second) {
1087 ✗ dfs_stack.push(child);
1088 }
1089 ✗ tree.erase(it);
1090 }
1091 ✗ if (!all_dirs.count(curr_dir))
1092 ✗ continue;
1093
1094 // create the dir first in pre-order traversal
1095 ✗ const IngestSQL::Directory &dir = all_dirs.at(curr_dir);
1096 ✗ catalog::DirectoryEntry dir_entry;
1097
1098 ✗ bool exists = false;
1099 ✗ exists = catalog_manager.LookupDirEntry(
1100 ✗ MakeCatalogPath(curr_dir), catalog::kLookupDefault, &dir_entry);
1101 ✗ CUSTOM_ASSERT(
1102 !(exists && !S_ISDIR(dir_entry.mode_)),
1103 "Refusing to replace existing file/symlink at %s with a directory",
1104 dir.name.c_str());
1105
1106 ✗ dir_entry.name_ = NameString(get_basename(dir.name));
1107 ✗ dir_entry.mtime_ = dir.mtime / 1000000000;
1108 ✗ dir_entry.mode_ = dir.mode | S_IFDIR;
1109 ✗ dir_entry.mode_ &= (S_IFDIR | 0777);
1110 ✗ dir_entry.uid_ = dir.owner;
1111 ✗ dir_entry.gid_ = dir.grp;
1112 ✗ dir_entry.has_xattrs_ = !dir.xattr.IsEmpty();
1113
1114 ✗ bool add_nested_catalog = false;
1115
1116 ✗ if (exists) {
1117 ✗ catalog_manager.TouchDirectory(dir_entry, dir.xattr, dir.name);
1118 ✗ if ((!dir_entry.IsNestedCatalogMountpoint()
1119 ✗ && !dir_entry.IsNestedCatalogRoot())
1120 ✗ && (g_add_missing_catalogs || dir.nested)) {
1121 ✗ add_nested_catalog = true;
1122 ✗ LogCvmfs(kLogCvmfs, kLogVerboseMsg,
1123 "Touching existing directory %s and adding nested catalog",
1124 dir.name.c_str());
1125 } else {
1126 ✗ LogCvmfs(kLogCvmfs, kLogVerboseMsg, "Touching existing directory %s",
1127 dir.name.c_str());
1128 }
1129 } else {
1130 ✗ LogCvmfs(kLogCvmfs, kLogVerboseMsg, "Adding directory [%s]",
1131 dir.name.c_str());
1132 ✗ catalog_manager.AddDirectory(dir_entry, dir.xattr,
1133 ✗ get_parent(dir.name));
1134 ✗ if (dir.nested) {
1135 ✗ add_nested_catalog = true;
1136 }
1137 }
1138 ✗ if (add_nested_catalog) {
1139 // now add a .cvmfscatalog file
1140 // so that manual changes won't remove the nested catalog
1141 ✗ LogCvmfs(kLogCvmfs, kLogVerboseMsg,
1142 "Placing .cvmfscatalog file in [%s]", dir.name.c_str());
1143 ✗ catalog::DirectoryEntryBase dir2;
1144 ✗ dir2.name_ = NameString(".cvmfscatalog");
1145 ✗ dir2.mtime_ = dir.mtime / 1000000000;
1146 ✗ dir2.mode_ = (S_IFREG | 0666);
1147 ✗ dir2.uid_ = 0;
1148 ✗ dir2.gid_ = 0;
1149 ✗ dir2.has_xattrs_ = 0;
1150 ✗ dir2.checksum_ = shash::MkFromHexPtr(
1151 shash::HexPtr("da39a3ee5e6b4b0d3255bfef95601890afd80709"),
1152 shash::kSuffixNone); // hash of ""
1153 ✗ XattrList const xattr2;
1154 ✗ catalog_manager.AddFile(dir2, xattr2, dir.name);
1155
1156 ✗ LogCvmfs(kLogCvmfs, kLogVerboseMsg, "Creating Nested Catalog [%s]",
1157 dir.name.c_str());
1158 ✗ catalog_manager.CreateNestedCatalog(dir.name);
1159 }
1160 }
1161 }
1162
1163 // sanity check that we have processed all the input
1164 ✗ CUSTOM_ASSERT(tree.empty(),
1165 "not all directories are processed, malformed input DB?");
1166 ✗ catalog_manager.RemoveSingleCatalogUploadCallback();
1167 ✗ return 0;
1168 }
1169
1170 ✗ int swissknife::IngestSQL::add_symlinks(
1171 catalog::WritableCatalogManager &catalog_manager,
1172 const std::vector<Symlink> &symlinks) {
1173 ✗ for (auto &&symlink : symlinks) {
1174 ✗ catalog::DirectoryEntry dir;
1175 ✗ catalog::DirectoryEntryBase dir2;
1176 ✗ XattrList const xattr;
1177 ✗ bool exists = false;
1178 ✗ exists = catalog_manager.LookupDirEntry(MakeCatalogPath(symlink.name),
1179 catalog::kLookupDefault, &dir);
1180
1181 ✗ dir2.name_ = NameString(get_basename(symlink.name));
1182 ✗ dir2.mtime_ = symlink.mtime / 1000000000;
1183 ✗ dir2.uid_ = symlink.owner;
1184 ✗ dir2.gid_ = symlink.grp;
1185 ✗ dir2.has_xattrs_ = false;
1186 ✗ dir2.symlink_ = LinkString(symlink.target);
1187 ✗ dir2.mode_ = S_IFLNK | 0777;
1188
1189 ✗ int noop = false;
1190
1191 ✗ if (exists) {
1192 ✗ if (symlink.skip_if_file_or_dir) {
1193 ✗ if (S_ISDIR(dir.mode_) || S_ISREG(dir.mode_)) {
1194 ✗ LogCvmfs(kLogCvmfs, kLogVerboseMsg,
1195 "File or directory for symlink [%s] exists, skipping "
1196 "symlink creation",
1197 symlink.name.c_str());
1198 ✗ noop = true;
1199 ✗ } else if (S_ISLNK(dir.mode_)) {
1200 ✗ LogCvmfs(kLogCvmfs, kLogVerboseMsg, "Removing existing symlink [%s]",
1201 symlink.name.c_str());
1202 ✗ catalog_manager.RemoveFile(symlink.name);
1203 } else {
1204 ✗ CUSTOM_ASSERT(0, "unknown mode for dirent: %d", dir.mode_);
1205 }
1206 } else {
1207 ✗ CUSTOM_ASSERT(!S_ISDIR(dir.mode_),
1208 "Not removing directory [%s] to create symlink",
1209 symlink.name.c_str());
1210 ✗ LogCvmfs(kLogCvmfs, kLogVerboseMsg,
1211 "Removing existing file/symlink [%s]", symlink.name.c_str());
1212 ✗ catalog_manager.RemoveFile(symlink.name);
1213 }
1214 }
1215 ✗ if (!noop) {
1216 ✗ string const parent = get_parent(symlink.name);
1217 ✗ LogCvmfs(kLogCvmfs, kLogVerboseMsg, "Adding symlink [%s] -> [%s]",
1218 symlink.name.c_str(), symlink.target.c_str());
1219 ✗ catalog_manager.AddFile(dir2, xattr, parent);
1220 }
1221 }
1222 ✗ return 0;
1223 }
1224
1225 ✗ static int check_hash(const char *hash) {
1226 ✗ if (strlen(hash) != 40) {
1227 ✗ return 1;
1228 }
1229 ✗ for (int i = 0; i < 40; i++) {
1230 // < '0' || > 'f' || ( > '9' && < 'a' )
1231 ✗ if (hash[i] < 0x30 || hash[i] > 0x66
1232 ✗ || (hash[i] > 0x39 && hash[i] < 0x61)) {
1233 ✗ return 1;
1234 }
1235 }
1236 ✗ return 0;
1237 }
1238
1239 ✗ bool check_prefix(const std::string &path, const std::string &prefix) {
1240 ✗ if (prefix == "" || prefix == "/") {
1241 ✗ return true;
1242 }
1243 ✗ if ("/" + path == prefix) {
1244 ✗ return true;
1245 }
1246 ✗ if (!HasPrefix(path, prefix, false)) {
1247 ✗ LogCvmfs(kLogCvmfs, kLogVerboseMsg, "Entry %s is outside lease path: %s",
1248 path.c_str(), prefix.c_str());
1249 ✗ return false;
1250 }
1251 ✗ return true;
1252 }
1253
1254 ✗ void swissknife::IngestSQL::load_dirs(
1255 sqlite3 *db, const std::string &lease_path,
1256 const std::string &additional_prefix,
1257 std::map<std::string, Directory> &all_dirs) {
1258 sqlite3_stmt *stmt;
1259 ✗ int const schema_revision = get_db_schema_revision(db);
1260 string select_stmt = "SELECT name, mode, mtime, owner, grp, acl, nested FROM "
1261 ✗ "dirs";
1262 ✗ if (schema_revision <= 3) {
1263 ✗ select_stmt = "SELECT name, mode, mtime, owner, grp, acl FROM dirs";
1264 }
1265 ✗ int const ret = sqlite3_prepare_v2(db, select_stmt.c_str(), -1, &stmt, NULL);
1266 ✗ CHECK_SQLITE_ERROR(ret, SQLITE_OK);
1267 ✗ while (sqlite3_step(stmt) == SQLITE_ROW) {
1268 ✗ char *name_cstr = (char *)sqlite3_column_text(stmt, 0);
1269 ✗ mode_t const mode = sqlite3_column_int(stmt, 1);
1270 ✗ time_t const mtime = sqlite3_column_int64(stmt, 2);
1271 ✗ uid_t const owner = sqlite3_column_int(stmt, 3);
1272 ✗ gid_t const grp = sqlite3_column_int(stmt, 4);
1273 ✗ int const nested = schema_revision <= 3 ? 1 : sqlite3_column_int(stmt, 6);
1274
1275 ✗ string const name = additional_prefix + sanitise_name(name_cstr);
1276 ✗ CUSTOM_ASSERT(check_prefix(name, lease_path),
1277 "%s is not below lease path %s", name.c_str(),
1278 lease_path.c_str());
1279
1280 ✗ Directory dir(name, mtime, mode, owner, grp, nested);
1281 ✗ char *acl = (char *)sqlite3_column_text(stmt, 5);
1282 ✗ dir.xattr = marshal_xattrs(acl);
1283 ✗ all_dirs.insert(std::make_pair(name, dir));
1284 }
1285 ✗ CHECK_SQLITE_ERROR(sqlite3_finalize(stmt), SQLITE_OK);
1286 }
1287
1288 ✗ void swissknife::IngestSQL::load_files(
1289 sqlite3 *db, const std::string &lease_path,
1290 const std::string &additional_prefix,
1291 std::map<std::string, std::vector<File> > &all_files) {
1292 sqlite3_stmt *stmt;
1293 ✗ int const schema_revision = get_db_schema_revision(db);
1294 string select_stmt = "SELECT name, mode, mtime, owner, grp, size, hashes, "
1295 ✗ "internal, compressed FROM files";
1296 ✗ if (schema_revision <= 2) {
1297 select_stmt = "SELECT name, mode, mtime, owner, grp, size, hashes, "
1298 ✗ "internal FROM files";
1299 }
1300 ✗ int const ret = sqlite3_prepare_v2(db, select_stmt.c_str(), -1, &stmt, NULL);
1301 ✗ CHECK_SQLITE_ERROR(ret, SQLITE_OK);
1302 ✗ while (sqlite3_step(stmt) == SQLITE_ROW) {
1303 ✗ char *name = (char *)sqlite3_column_text(stmt, 0);
1304 ✗ mode_t const mode = sqlite3_column_int(stmt, 1);
1305 ✗ time_t const mtime = sqlite3_column_int64(stmt, 2);
1306 ✗ uid_t const owner = sqlite3_column_int(stmt, 3);
1307 ✗ gid_t const grp = sqlite3_column_int(stmt, 4);
1308 ✗ size_t const size = sqlite3_column_int64(stmt, 5);
1309 ✗ char *hashes_cstr = (char *)sqlite3_column_text(stmt, 6);
1310 ✗ int const internal = sqlite3_column_int(stmt, 7);
1311 ✗ int const compressed = schema_revision <= 2 ? 0
1312 ✗ : sqlite3_column_int(stmt, 8);
1313
1314 ✗ string names = additional_prefix + sanitise_name(name);
1315 ✗ CUSTOM_ASSERT(check_prefix(names, lease_path),
1316 "%s is not below lease path %s", names.c_str(),
1317 lease_path.c_str());
1318 ✗ string const parent_dir = get_parent(names);
1319
1320 ✗ if (!all_files.count(parent_dir)) {
1321 ✗ all_files[parent_dir] = vector<swissknife::IngestSQL::File>();
1322 }
1323 ✗ all_files[parent_dir].emplace_back(std::move(names), mtime, size, owner,
1324 grp, mode, internal, compressed);
1325
1326 // tokenize hashes
1327 char *ref;
1328 char *tok;
1329 ✗ tok = strtok_r(hashes_cstr, ",", &ref);
1330 ✗ vector<off_t> offsets;
1331 ✗ vector<size_t> sizes;
1332 ✗ vector<shash::Any> hashes;
1333 ✗ off_t offset = 0;
1334
1335 CUSTOM_ASSERT(size >= 0, "file size cannot be negative [%s]",
1336 names.c_str());
1337 ✗ size_t const kChunkSize = internal ? kInternalChunkSize
1338 : kExternalChunkSize;
1339
1340 ✗ while (tok) {
1341 ✗ offsets.push_back(offset);
1342 // TODO: check the hash format
1343 ✗ CUSTOM_ASSERT(check_hash(tok) == 0,
1344 "provided hash for [%s] is invalid: %s", names.c_str(),
1345 tok);
1346 ✗ hashes.push_back(
1347 ✗ shash::MkFromHexPtr(shash::HexPtr(tok), shash::kSuffixNone));
1348 ✗ tok = strtok_r(NULL, ",", &ref);
1349 ✗ offset += kChunkSize; // in the future we might want variable chunk
1350 // sizes specified in the DB
1351 }
1352 ✗ size_t expected_num_chunks = size / kChunkSize;
1353 ✗ if (expected_num_chunks * (size_t)kChunkSize < (size_t)size || size == 0) {
1354 ✗ expected_num_chunks++;
1355 }
1356 ✗ CUSTOM_ASSERT(
1357 offsets.size() == expected_num_chunks,
1358 "offsets size %ld does not match expected number of chunks %ld",
1359 offsets.size(), expected_num_chunks);
1360 ✗ for (size_t i = 0; i < offsets.size() - 1; i++) {
1361 ✗ sizes.push_back(size_t(offsets[i + 1] - offsets[i]));
1362 }
1363
1364 ✗ sizes.push_back(size_t(size - offsets[offsets.size() - 1]));
1365 ✗ for (size_t i = 0; i < offsets.size(); i++) {
1366 ✗ FileChunk const chunk = FileChunk(hashes[i], offsets[i], sizes[i]);
1367 ✗ all_files[parent_dir].back().chunks.PushBack(chunk);
1368 }
1369 }
1370 ✗ CHECK_SQLITE_ERROR(sqlite3_finalize(stmt), SQLITE_OK);
1371 }
1372
1373 ✗ void swissknife::IngestSQL::load_symlinks(
1374 sqlite3 *db, const std::string &lease_path,
1375 const std::string &additional_prefix,
1376 std::map<std::string, std::vector<Symlink> > &all_symlinks) {
1377 sqlite3_stmt *stmt;
1378 string const select_stmt = "SELECT name, target, mtime, owner, grp, "
1379 ✗ "skip_if_file_or_dir FROM links";
1380 ✗ int const ret = sqlite3_prepare_v2(db, select_stmt.c_str(), -1, &stmt, NULL);
1381 ✗ CHECK_SQLITE_ERROR(ret, SQLITE_OK);
1382 ✗ while (sqlite3_step(stmt) == SQLITE_ROW) {
1383 ✗ char *name_cstr = (char *)sqlite3_column_text(stmt, 0);
1384 ✗ char *target_cstr = (char *)sqlite3_column_text(stmt, 1);
1385 ✗ time_t const mtime = sqlite3_column_int64(stmt, 2);
1386 ✗ uid_t const owner = sqlite3_column_int(stmt, 3);
1387 ✗ gid_t const grp = sqlite3_column_int(stmt, 4);
1388 ✗ int const skip_if_file_or_dir = sqlite3_column_int(stmt, 5);
1389
1390 ✗ string names = additional_prefix + sanitise_name(name_cstr);
1391 ✗ CUSTOM_ASSERT(check_prefix(names, lease_path),
1392 "%s is not below lease path %s", names.c_str(),
1393 lease_path.c_str());
1394 ✗ string target = target_cstr;
1395 ✗ string const parent_dir = get_parent(names);
1396
1397 ✗ if (!all_symlinks.count(parent_dir)) {
1398 ✗ all_symlinks[parent_dir] = vector<swissknife::IngestSQL::Symlink>();
1399 }
1400 ✗ all_symlinks[parent_dir].emplace_back(std::move(names), std::move(target),
1401 mtime, owner, grp,
1402 skip_if_file_or_dir);
1403 }
1404 ✗ CHECK_SQLITE_ERROR(sqlite3_finalize(stmt), SQLITE_OK);
1405 }
1406
1407 ✗ int swissknife::IngestSQL::add_files(
1408 catalog::WritableCatalogManager &catalog_manager,
1409 const std::vector<File> &files) {
1410 ✗ for (auto &&file : files) {
1411 ✗ catalog::DirectoryEntry dir;
1412 ✗ XattrList const xattr;
1413 ✗ bool exists = false;
1414 ✗ exists = catalog_manager.LookupDirEntry(MakeCatalogPath(file.name),
1415 catalog::kLookupDefault, &dir);
1416
1417 ✗ dir.name_ = NameString(get_basename(file.name));
1418 ✗ dir.mtime_ = file.mtime / 1000000000;
1419 ✗ dir.mode_ = file.mode | S_IFREG;
1420 ✗ dir.mode_ &= (S_IFREG | 0777);
1421 ✗ dir.uid_ = file.owner;
1422 ✗ dir.gid_ = file.grp;
1423 ✗ dir.size_ = file.size;
1424 ✗ dir.has_xattrs_ = false;
1425 ✗ dir.is_external_file_ = !file.internal;
1426 ✗ dir.set_is_chunked_file(true);
1427 ✗ dir.checksum_ = shash::MkFromHexPtr(
1428 shash::HexPtr("0000000000000000000000000000000000000000"),
1429 shash::kSuffixNone);
1430
1431 // compression is permitted only for internal data
1432 ✗ CUSTOM_ASSERT(file.internal || (!file.internal && file.compressed < 2),
1433 "compression is only allowed for internal data [%s]",
1434 file.name.c_str());
1435
1436 ✗ switch (file.compressed) {
1437 ✗ case 1: // Uncompressed
1438 ✗ dir.compression_algorithm_ = zlib::kNoCompression;
1439 ✗ break;
1440 ✗ case 2: // Compressed with Zlib
1441 ✗ dir.compression_algorithm_ = zlib::kZlibDefault;
1442 ✗ break;
1443 // future cases: different compression schemes
1444 ✗ default: // default behaviour: compressed if internal, content-addressed.
1445 // Uncompressed if external
1446 ✗ dir.compression_algorithm_ = file.internal ? zlib::kZlibDefault
1447 : zlib::kNoCompression;
1448 }
1449
1450 ✗ if (exists) {
1451 ✗ CUSTOM_ASSERT(
1452 !S_ISDIR(dir.mode()) && !S_ISLNK(dir.mode()),
1453 "Refusing to replace existing dir/symlink at %s with a file",
1454 file.name.c_str());
1455 ✗ LogCvmfs(kLogCvmfs, kLogVerboseMsg, "Removing existing file [%s]",
1456 file.name.c_str());
1457 ✗ catalog_manager.RemoveFile(file.name);
1458 }
1459 ✗ string const parent = get_parent(file.name);
1460 ✗ LogCvmfs(kLogCvmfs, kLogVerboseMsg, "Adding chunked file [%s]",
1461 file.name.c_str());
1462 ✗ catalog_manager.AddChunkedFile(dir, xattr, parent, file.chunks);
1463 }
1464
1465 ✗ return 0;
1466 }
1467
1468 ✗ int swissknife::IngestSQL::do_deletions(
1469 sqlite3 *db, catalog::WritableCatalogManager &catalog_manager,
1470 const std::string &lease_path, const std::string &additional_prefix) {
1471 sqlite3_stmt *stmt;
1472 ✗ int const row_count = get_row_count(db, "deletions");
1473 ✗ int const print_every = calculate_print_frequency(row_count);
1474 ✗ int curr_row = 0;
1475 ✗ int ret = sqlite3_prepare_v2(db,
1476 "SELECT name, directory, file, link FROM "
1477 "deletions ORDER BY length(name) DESC",
1478 -1, &stmt, NULL);
1479 ✗ CHECK_SQLITE_ERROR(ret, SQLITE_OK);
1480 ✗ while (sqlite3_step(stmt) == SQLITE_ROW) {
1481 ✗ curr_row++;
1482
1483 ✗ char *name = (char *)sqlite3_column_text(stmt, 0);
1484 ✗ int64_t const isdir = sqlite3_column_int64(stmt, 1);
1485 ✗ int64_t const isfile = sqlite3_column_int64(stmt, 2);
1486 ✗ int64_t const islink = sqlite3_column_int64(stmt, 3);
1487
1488 ✗ string const names = additional_prefix + sanitise_name(name);
1489 ✗ CUSTOM_ASSERT(check_prefix(names, lease_path),
1490 "%s is not below lease path %s", names.c_str(),
1491 lease_path.c_str());
1492
1493 ✗ catalog::DirectoryEntry dirent;
1494 ✗ bool exists = false;
1495 ✗ exists = catalog_manager.LookupDirEntry(MakeCatalogPath(names),
1496 catalog::kLookupDefault, &dirent);
1497 ✗ if (exists) {
1498 ✗ if ((isdir && S_ISDIR(dirent.mode()))
1499 ✗ || (islink && S_ISLNK(dirent.mode()))
1500 ✗ || (isfile && S_ISREG(dirent.mode()))) {
1501 ✗ if (S_ISDIR(dirent.mode())) {
1502 ✗ PathString names_path(names);
1503 ✗ recursively_delete_directory(names_path, catalog_manager);
1504 ✗ } else {
1505 ✗ LogCvmfs(kLogCvmfs, kLogVerboseMsg, "Removing link/file [%s]",
1506 names.c_str());
1507 ✗ catalog_manager.RemoveFile(names);
1508 }
1509 } else {
1510 ✗ LogCvmfs(kLogCvmfs, kLogVerboseMsg,
1511 "Mismatch in deletion type, not deleting: [%s] (dir %ld/%d , "
1512 "link %ld/%d, file %ld/%d)",
1513 names.c_str(), isdir, S_ISDIR(dirent.mode()), islink,
1514 S_ISLNK(dirent.mode()), isfile, S_ISREG(dirent.mode()));
1515 }
1516 } else {
1517 ✗ LogCvmfs(kLogCvmfs, kLogVerboseMsg, "Not Removing non-existent [%s]",
1518 names.c_str());
1519 }
1520
1521 ✗ SHOW_PROGRESS("deletions", print_every, curr_row, row_count);
1522 }
1523 ✗ ret = sqlite3_finalize(stmt);
1524 ✗ return ret;
1525 }
1526
1527 const char *schema[] = {"PRAGMA journal_mode=WAL;",
1528
1529 "CREATE TABLE IF NOT EXISTS dirs ( \
1530 name TEXT PRIMARY KEY, \
1531 mode INTEGER NOT NULL DEFAULT 493,\
1532 mtime INTEGER NOT NULL DEFAULT 0,\
1533 owner INTEGER NOT NULL DEFAULT 0, \
1534 grp INTEGER NOT NULL DEFAULT 0, \
1535 acl TEXT NOT NULL DEFAULT '', \
1536 nested INTEGER DEFAULT 1);",
1537
1538 "CREATE TABLE IF NOT EXISTS files ( \
1539 name TEXT PRIMARY KEY, \
1540 mode INTEGER NOT NULL DEFAULT 420, \
1541 mtime INTEGER NOT NULL DEFAULT 0,\
1542 owner INTEGER NOT NULL DEFAULT 0,\
1543 grp INTEGER NOT NULL DEFAULT 0,\
1544 size INTEGER NOT NULL DEFAULT 0,\
1545 hashes TEXT NOT NULL DEFAULT '',\
1546 internal INTEGER NOT NULL DEFAULT 0,\
1547 compressed INTEGER NOT NULL DEFAULT 0\
1548 );",
1549
1550 "CREATE TABLE IF NOT EXISTS links (\
1551 name TEXT PRIMARY KEY,\
1552 target TEXT NOT NULL DEFAULT '',\
1553 mtime INTEGER NOT NULL DEFAULT 0,\
1554 owner INTEGER NOT NULL DEFAULT 0,\
1555 grp INTEGER NOT NULL DEFAULT 0,\
1556 skip_if_file_or_dir INTEGER NOT NULL DEFAULT 0\
1557 );",
1558
1559 "CREATE TABLE IF NOT EXISTS deletions (\
1560 name TEXT PRIMARY KEY,\
1561 directory INTEGER NOT NULL DEFAULT 0,\
1562 file INTEGER NOT NULL DEFAULT 0,\
1563 link INTEGER NOT NULL DEFAULT 0\
1564 );",
1565
1566 "CREATE TABLE IF NOT EXISTS properties (\
1567 key TEXT PRIMARY KEY,\
1568 value TEXT NOT NULL\
1569 );",
1570
1571 "INSERT INTO properties VALUES ('schema_revision', "
1572 "'4') ON CONFLICT DO NOTHING;",
1573 NULL};
1574
1575 ✗ static void create_empty_database(string &filename) {
1576 sqlite3 *db_out;
1577 ✗ LogCvmfs(kLogCvmfs, kLogStdout, "Creating empty database file %s",
1578 filename.c_str());
1579 ✗ int ret = sqlite3_open_v2(filename.c_str(), &db_out,
1580 SQLITE_OPEN_READWRITE | SQLITE_OPEN_CREATE, NULL);
1581 ✗ CHECK_SQLITE_ERROR(ret, SQLITE_OK);
1582 ✗ relax_db_locking(db_out);
1583
1584 ✗ const char **ptr = schema;
1585 ✗ while (*ptr != NULL) {
1586 ✗ ret = sqlite3_exec(db_out, *ptr, NULL, NULL, NULL);
1587 ✗ CHECK_SQLITE_ERROR(ret, SQLITE_OK);
1588 ✗ ptr++;
1589 }
1590 ✗ sqlite3_close(db_out);
1591 }
1592
1593 ✗ static void recursively_delete_directory(
1594 PathString &path, catalog::WritableCatalogManager &catalog_manager) {
1595 ✗ catalog::DirectoryEntryList const listing;
1596
1597 // Add all names
1598 ✗ catalog::StatEntryList listing_from_catalog;
1599 ✗ bool const retval = catalog_manager.ListingStat(
1600 ✗ PathString("/" + path.ToString()), &listing_from_catalog);
1601
1602 ✗ CUSTOM_ASSERT(retval, "failed to call ListingStat for %s", path.c_str());
1603
1604 ✗ if (!catalog_manager.IsTransitionPoint(path.ToString())) {
1605 ✗ for (unsigned i = 0; i < listing_from_catalog.size(); ++i) {
1606 ✗ PathString entry_path;
1607 ✗ entry_path.Assign(path);
1608 ✗ entry_path.Append("/", 1);
1609 ✗ entry_path.Append(listing_from_catalog.AtPtr(i)->name.GetChars(),
1610 ✗ listing_from_catalog.AtPtr(i)->name.GetLength());
1611
1612 ✗ if (S_ISDIR(listing_from_catalog.AtPtr(i)->info.st_mode)) {
1613 ✗ LogCvmfs(kLogCvmfs, kLogVerboseMsg, "Recursing into %s/",
1614 entry_path.ToString().c_str());
1615 ✗ recursively_delete_directory(entry_path, catalog_manager);
1616
1617
1618 } else {
1619 ✗ LogCvmfs(kLogCvmfs, kLogVerboseMsg, " Recursively removing %s",
1620 entry_path.ToString().c_str());
1621 ✗ catalog_manager.RemoveFile(entry_path.ToString());
1622 }
1623 }
1624 } else {
1625 ✗ LogCvmfs(kLogCvmfs, kLogVerboseMsg, "Removing nested catalog %s",
1626 path.ToString().c_str());
1627 ✗ catalog_manager.RemoveNestedCatalog(path.ToString(), false);
1628 }
1629 ✗ LogCvmfs(kLogCvmfs, kLogVerboseMsg, "Removing directory %s",
1630 path.ToString().c_str());
1631 ✗ catalog_manager.RemoveDirectory(path.ToString());
1632 }
1633
1634
1635 ✗ static void relax_db_locking(sqlite3 *db) {
1636 ✗ int ret = 0;
1637 ✗ ret = sqlite3_exec(db, "PRAGMA temp_store=2", NULL, NULL, NULL);
1638 ✗ CHECK_SQLITE_ERROR(ret, SQLITE_OK);
1639 ✗ ret = sqlite3_exec(db, "PRAGMA synchronous=OFF", NULL, NULL, NULL);
1640 ✗ CHECK_SQLITE_ERROR(ret, SQLITE_OK);
1641 }
1642
1643
1644 ✗ extern "C" void *lease_refresh_thread(void *payload) {
1645 ✗ while (!g_stop_refresh) {
1646 ✗ sleep(2);
1647 ✗ refresh_lease();
1648 }
1649 ✗ return NULL;
1650 }
1651
1652 ✗ static bool isDatabaseMarkedComplete(const char *dbfile) {
1653 int ret;
1654 sqlite3 *db;
1655 sqlite3_stmt *stmt;
1656 ✗ bool retval = false;
1657
1658 ✗ ret = sqlite3_open(dbfile, &db);
1659 ✗ if (ret != SQLITE_OK) {
1660 ✗ return false;
1661 }
1662
1663 ✗ const char *req = "SELECT value FROM properties WHERE key='completed_graft'";
1664 ✗ ret = sqlite3_prepare_v2(db, req, -1, &stmt, NULL);
1665 ✗ if (ret != SQLITE_OK) {
1666 ✗ return false;
1667 }
1668 ✗ if (sqlite3_step(stmt) == SQLITE_ROW) {
1669 ✗ int const id = sqlite3_column_int(stmt, 0);
1670 ✗ if (id > 0) {
1671 ✗ retval = true;
1672 }
1673 }
1674 ✗ sqlite3_close(db);
1675 ✗ return retval;
1676 }
1677
1678 ✗ static void setDatabaseMarkedComplete(const char *dbfile) {
1679 int ret;
1680 sqlite3 *db;
1681 char *err;
1682
1683 ✗ ret = sqlite3_open(dbfile, &db);
1684 ✗ if (ret != SQLITE_OK) {
1685 ✗ return;
1686 }
1687
1688 ✗ const char *req = "INSERT INTO properties (key, value) VALUES "
1689 "('completed_graft',1) ON CONFLICT(key) DO UPDATE SET "
1690 "value=1 WHERE key='completed_graft'";
1691
1692 ✗ ret = sqlite3_exec(db, req, 0, 0, &err);
1693 ✗ if (ret != SQLITE_OK) {
1694 ✗ return;
1695 }
1696 ✗ sqlite3_close(db);
1697 }
1698
1699 ✗ static void invalidate_manifest(std::string proxy_list, std::string url) {
1700 // split the proxy string -- remove any '"' and split on '|' or ';'
1701 ✗ size_t pos = 0;
1702 // replace any ';' with '|' to simplify subsequent split
1703 ✗ while ((pos = proxy_list.find(';', pos)) != std::string::npos) {
1704 ✗ proxy_list.replace(pos, 1, 1, '|');
1705 ✗ ++pos;
1706 }
1707 // remove any leading or trailing '"'
1708 ✗ if (HasPrefix(proxy_list, "\"", true)) {
1709 ✗ proxy_list = proxy_list.substr(1);
1710 }
1711 ✗ if (HasSuffix(proxy_list, "\"", true)) {
1712 ✗ proxy_list = proxy_list.substr(0, proxy_list.size() - 1);
1713 }
1714 // rewrite the port from 6086 to 6081 to ensure the invalidation works
1715 ✗ replaceAllSubstrings(proxy_list, ":6086", ":6081");
1716
1717 ✗ std::vector<std::string> proxies = SplitString(proxy_list, '|');
1718
1719 // now iterate over all the proxies
1720 // for the first, force a no-cache GET back to google
1721 // for the remainder just PURGE
1722 ✗ bool first = true;
1723 ✗ for (auto p = proxies.begin(); p != proxies.end(); p++) {
1724 ✗ bool ok = true;
1725 ✗ const string proxy = *p;
1726 ✗ CURL *curl = NULL;
1727 ✗ CURLcode res = CURLE_OK;
1728 ✗ struct curl_slist *headers = NULL;
1729 ✗ curl = curl_easy_init();
1730 ✗ if (!curl) {
1731 ✗ LogCvmfs(kLogCvmfs, kLogStdout, "Unable to init curl!");
1732 ✗ return;
1733 }
1734 ✗ res = curl_easy_setopt(curl, CURLOPT_URL, url.c_str());
1735 ✗ if (proxy != "DIRECT") {
1736 ✗ res = curl_easy_setopt(curl, CURLOPT_PROXY, proxy.c_str());
1737 }
1738 ✗ res = curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT, 1l);
1739 ✗ res = curl_easy_setopt(curl, CURLOPT_TIMEOUT, 3l);
1740 ✗ res = curl_easy_setopt(curl, CURLOPT_WRITEFUNCTION, writeFunction);
1741
1742 ✗ if (first) {
1743 ✗ headers = curl_slist_append(headers, "Cache-Control: no-cache");
1744 ✗ curl_easy_setopt(curl, CURLOPT_HTTPHEADER, headers);
1745 } else {
1746 ✗ curl_easy_setopt(curl, CURLOPT_CUSTOMREQUEST, "PURGE");
1747 }
1748
1749 ✗ res = curl_easy_perform(curl);
1750 ✗ if (res != CURLE_OK) {
1751 ✗ LogCvmfs(kLogCvmfs, kLogStdout,
1752 "Manifest invalidation failed: curl error = [%d] [%s] url = %s "
1753 "proxy = %s",
1754 res, curl_easy_strerror(res), url.c_str(), proxy.c_str());
1755 ✗ ok = false;
1756 }
1757 ✗ if (headers) {
1758 ✗ curl_slist_free_all(headers);
1759 }
1760 ✗ curl_easy_cleanup(curl);
1761 ✗ if (ok) {
1762 ✗ first = false;
1763 }
1764 }
1765 }
1766
1767
1768 ✗ static void wait_for_update(std::string path, long revision) {
1769 char val[101];
1770 ✗ memset(val, 0, 101);
1771 ✗ long current = -1;
1772 DIR *d;
1773 ✗ while (-1 != getxattr(path.c_str(), "user.revision", val, 100)) {
1774 ✗ const long x = atol(val);
1775 ✗ if (x >= revision) {
1776 ✗ LogCvmfs(kLogCvmfs, kLogStdout, "Mount reached revision %ld", x);
1777 ✗ return;
1778 ✗ } else if (x != current) {
1779 ✗ current = x;
1780 ✗ LogCvmfs(kLogCvmfs, kLogStdout, "Mount at revision %ld, waiting..", x);
1781 }
1782 ✗ sleep(1);
1783 ✗ d = opendir(path.c_str());
1784 ✗ if (d) {
1785 ✗ closedir(d);
1786 }
1787 }
1788 ✗ LogCvmfs(kLogCvmfs, kLogStdout,
1789 "Unable to query user.revision xattr of [%s]: errno: %d",
1790 path.c_str(), errno);
1791 }
1792