GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/upload_s3.cc
Date: 2026-10-11 02:40:02
Exec Total Coverage
Lines: 238 349 68.2%
Branches: 209 524 39.9%
Line Branch Exec Source
1 /**
2 * This file is part of the CernVM File System.
3 */
4
5 #include "upload_s3.h"
6
7 #include <errno.h>
8 #include <fcntl.h>
9 #include <inttypes.h>
10 #include <unistd.h>
11
12 #include <set>
13 #include <string>
14 #include <vector>
15
16 #include "compression/compression.h"
17 #include "network/s3fanout.h"
18 #include "options.h"
19 #include "util/exception.h"
20 #include "util/logging.h"
21 #include "util/mutex.h"
22 #include "util/posix.h"
23 #include "util/string.h"
24
25 namespace upload {
26
27 /*
28 * Allowed values of x-amz-acl according to S3 API
29 */
30 static const char *x_amz_acl_allowed_values_[8] = {"private",
31 "public-read",
32 "public-write",
33 "authenticated-read",
34 "aws-exec-read",
35 "bucket-owner-read",
36 "bucket-owner-full-control",
37 ""};
38
39 22528 void S3Uploader::RequestCtrl::WaitFor() {
40 char c;
41
1/2
✓ Branch 1 taken 22528 times.
✗ Branch 2 not taken.
 22528 ReadPipe(pipe_wait[0], &c, 1);
42
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 22528 times.
 22528 assert(c == 'c');
43
1/2
✓ Branch 1 taken 22528 times.
✗ Branch 2 not taken.
 22528 ClosePipe(pipe_wait);
44 22528 }
45
46
47 572 S3Uploader::S3Uploader(const SpoolerDefinition &spooler_definition)
48 : AbstractUploader(spooler_definition)
49 572 , dns_buckets_(true)
50 572 , num_parallel_uploads_(kDefaultNumParallelUploads)
51 572 , num_retries_(kDefaultNumRetries)
52 572 , timeout_sec_(kDefaultTimeoutSec)
53 572 , authz_method_(s3fanout::kAuthzAwsV2)
54 572 , peek_before_put_(true)
55 572 , use_https_(false)
56 572 , batch_delete_enabled_(true)
57 572 , batch_delete_size_(kDefaultBatchDeleteSize)
58
1/2
✓ Branch 2 taken 572 times.
✗ Branch 3 not taken.
 572 , proxy_("")
59
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 , temporary_path_(spooler_definition.temporary_path)
60
1/2
✓ Branch 12 taken 572 times.
✗ Branch 13 not taken.
 1144 , x_amz_acl_("public-read") {
61
2/4
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 572 times.
✗ Branch 4 not taken.
 572 assert(spooler_definition.IsValid()
62 && spooler_definition.driver_type == SpoolerDefinition::S3);
63
64 572 io_errors_.store(0);
65 572 const int mutex_ret = pthread_mutex_init(&delete_batch_mutex_, NULL);
66
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 572 times.
 572 assert(mutex_ret == 0);
67
68
2/4
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 572 times.
 572 if (!ParseSpoolerDefinition(spooler_definition)) {
69 ✗ PANIC(kLogStderr, "Error in parsing the spooler definition");
70 }
71
72 // Disable batch delete for Azure (not supported)
73
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 572 times.
 572 if (authz_method_ == s3fanout::kAuthzAzure)
74 ✗ batch_delete_enabled_ = false;
75
76
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 s3fanout::S3FanoutManager::S3Config s3config;
77
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 s3config.access_key = access_key_;
78
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 s3config.secret_key = secret_key_;
79
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 s3config.hostname_port = host_name_port_;
80 572 s3config.authz_method = authz_method_;
81
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 s3config.region = region_;
82
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 s3config.flavor = flavor_;
83
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 s3config.bucket = bucket_;
84 572 s3config.dns_buckets = dns_buckets_;
85 572 s3config.pool_max_handles = num_parallel_uploads_;
86 572 s3config.opt_timeout_sec = timeout_sec_;
87 572 s3config.opt_max_retries = num_retries_;
88 572 s3config.opt_backoff_init_ms = kDefaultBackoffInitMs;
89 572 s3config.opt_backoff_max_ms = kDefaultBackoffMaxMs;
90
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 s3config.x_amz_acl = x_amz_acl_;
91
92
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 572 times.
 572 if (use_https_) {
93 ✗ s3config.protocol = "https";
94 } else {
95
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 s3config.protocol = "http";
96 }
97
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 s3config.proxy = proxy_;
98
99 1144 s3fanout_mgr_ = std::unique_ptr<s3fanout::S3FanoutManager>(
100
2/4
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 572 times.
✗ Branch 5 not taken.
 1144 new s3fanout::S3FanoutManager(s3config));
101
1/2
✓ Branch 2 taken 572 times.
✗ Branch 3 not taken.
 572 s3fanout_mgr_->Spawn();
102
103 572 const int retval = pthread_create(&thread_collect_results_, NULL,
104 MainCollectResults, this);
105
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 572 times.
 572 assert(retval == 0);
106 572 }
107
108
109 3432 S3Uploader::~S3Uploader() {
110 // Signal termination to our own worker thread
111 1144 s3fanout_mgr_->PushCompletedJob(NULL);
112 1144 pthread_join(thread_collect_results_, NULL);
113 1144 pthread_mutex_destroy(&delete_batch_mutex_);
114 2288 }
115
116
117 572 bool S3Uploader::ParseSpoolerDefinition(
118 const SpoolerDefinition &spooler_definition) {
119 const std::vector<std::string> config = SplitString(
120
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 spooler_definition.spooler_configuration, '@');
121
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 572 times.
 572 if (config.size() != 2) {
122 ✗ LogCvmfs(kLogUploadS3, kLogStderr,
123 "Failed to parse spooler configuration string '%s'.\n"
124 "Provide: <repo_alias>@/path/to/s3.conf",
125 spooler_definition.spooler_configuration.c_str());
126 ✗ return false;
127 }
128
1/2
✓ Branch 2 taken 572 times.
✗ Branch 3 not taken.
 572 repository_alias_ = config[0];
129 572 const std::string &config_path = config[1];
130
131
2/4
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 572 times.
 572 if (!FileExists(config_path)) {
132 ✗ LogCvmfs(kLogUploadS3, kLogStderr, "Cannot find S3 config file at '%s'",
133 config_path.c_str());
134 ✗ return false;
135 }
136
137 // Parse S3 configuration
138 BashOptionsManager options_manager = BashOptionsManager(
139
4/8
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 572 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 572 times.
✗ Branch 8 not taken.
✓ Branch 10 taken 572 times.
✗ Branch 11 not taken.
 572 new DefaultOptionsTemplateManager(repository_alias_));
140
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 options_manager.ParsePath(config_path, false);
141 572 std::string parameter;
142
143
3/6
✓ Branch 2 taken 572 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 572 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 572 times.
 572 if (!options_manager.GetValue("CVMFS_S3_HOST", &host_name_)) {
144 ✗ LogCvmfs(kLogUploadS3, kLogStderr,
145 "Failed to parse CVMFS_S3_HOST from '%s'", config_path.c_str());
146 ✗ return false;
147 }
148
3/6
✓ Branch 2 taken 572 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 572 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 572 times.
 572 if (!options_manager.GetValue("CVMFS_S3_ACCESS_KEY", &access_key_)) {
149 ✗ LogCvmfs(kLogUploadS3, kLogStderr,
150 "Failed to parse CVMFS_S3_ACCESS_KEY from '%s'.",
151 config_path.c_str());
152 ✗ return false;
153 }
154
3/6
✓ Branch 2 taken 572 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 572 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 572 times.
 572 if (!options_manager.GetValue("CVMFS_S3_SECRET_KEY", &secret_key_)) {
155 ✗ LogCvmfs(kLogUploadS3, kLogStderr,
156 "Failed to parse CVMFS_S3_SECRET_KEY from '%s'.",
157 config_path.c_str());
158 ✗ return false;
159 }
160
3/6
✓ Branch 2 taken 572 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 572 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 572 times.
 572 if (!options_manager.GetValue("CVMFS_S3_BUCKET", &bucket_)) {
161 ✗ LogCvmfs(kLogUploadS3, kLogStderr,
162 "Failed to parse CVMFS_S3_BUCKET from '%s'.", config_path.c_str());
163 ✗ return false;
164 }
165
3/6
✓ Branch 2 taken 572 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 572 times.
✗ Branch 6 not taken.
✓ Branch 9 taken 572 times.
✗ Branch 10 not taken.
 572 if (options_manager.GetValue("CVMFS_S3_DNS_BUCKETS", &parameter)) {
166
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 if (parameter == "false") {
167 572 dns_buckets_ = false;
168 }
169 }
170
3/6
✓ Branch 2 taken 572 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 572 times.
✗ Branch 6 not taken.
✓ Branch 9 taken 572 times.
✗ Branch 10 not taken.
 572 if (options_manager.GetValue("CVMFS_S3_MAX_NUMBER_OF_PARALLEL_CONNECTIONS",
171 &parameter)) {
172
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 num_parallel_uploads_ = String2Uint64(parameter);
173 }
174
3/6
✓ Branch 2 taken 572 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 572 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 572 times.
 572 if (options_manager.GetValue("CVMFS_S3_MAX_RETRIES", &parameter)) {
175 ✗ num_retries_ = String2Uint64(parameter);
176 }
177
3/6
✓ Branch 2 taken 572 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 572 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 572 times.
 572 if (options_manager.GetValue("CVMFS_S3_TIMEOUT", &parameter)) {
178 ✗ timeout_sec_ = String2Uint64(parameter);
179 }
180
3/6
✓ Branch 2 taken 572 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 572 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 572 times.
 572 if (options_manager.GetValue("CVMFS_S3_REGION", &region_)) {
181 ✗ authz_method_ = s3fanout::kAuthzAwsV4;
182 }
183
3/6
✓ Branch 2 taken 572 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 572 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 572 times.
 572 if (options_manager.GetValue("CVMFS_S3_FLAVOR", &flavor_)) {
184 ✗ if (flavor_ == "azure") {
185 ✗ authz_method_ = s3fanout::kAuthzAzure;
186 ✗ } else if (flavor_ == "awsv2") {
187 ✗ authz_method_ = s3fanout::kAuthzAwsV2;
188 ✗ } else if (flavor_ == "awsv4") {
189 ✗ authz_method_ = s3fanout::kAuthzAwsV4;
190 } else {
191 ✗ LogCvmfs(kLogUploadS3, kLogStderr,
192 "Failed to parse CVMFS_S3_FLAVOR from '%s', "
193 "valid options are azure, awsv2 or awsv4",
194 config_path.c_str());
195 ✗ return false;
196 }
197 }
198
3/6
✓ Branch 2 taken 572 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 572 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 572 times.
 572 if (options_manager.GetValue("CVMFS_S3_PEEK_BEFORE_PUT", &parameter)) {
199 ✗ peek_before_put_ = options_manager.IsOn(parameter);
200 }
201
3/6
✓ Branch 2 taken 572 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 572 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 572 times.
 572 if (options_manager.GetValue("CVMFS_S3_X_AMZ_ACL", &parameter)) {
202 ✗ bool isAllowed = false;
203 ✗ size_t const len = sizeof(x_amz_acl_allowed_values_)
204 / sizeof(x_amz_acl_allowed_values_[0]);
205 ✗ for (size_t i = 0; i < len; i++) {
206 ✗ if (x_amz_acl_allowed_values_[i] == parameter) {
207 ✗ isAllowed = true;
208 ✗ break;
209 }
210 }
211 ✗ if (!isAllowed) {
212 ✗ LogCvmfs(kLogUploadS3, kLogStderr,
213 "%s is not an allowed value for CVMFS_S3_X_AMZ_ACL",
214 parameter.c_str());
215 ✗ return false;
216 }
217 ✗ x_amz_acl_ = parameter;
218 }
219
220
3/6
✓ Branch 2 taken 572 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 572 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 572 times.
 572 if (options_manager.GetValue("CVMFS_S3_BATCH_DELETE", &parameter)) {
221 ✗ batch_delete_enabled_ = options_manager.IsOn(parameter);
222 }
223
224
3/6
✓ Branch 2 taken 572 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 572 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 572 times.
 572 if (options_manager.GetValue("CVMFS_S3_BATCH_DELETE_SIZE", &parameter)) {
225 ✗ const unsigned requested = String2Uint64(parameter);
226 ✗ if (requested == 0) {
227 ✗ LogCvmfs(kLogUploadS3, kLogStderr,
228 "CVMFS_S3_BATCH_DELETE_SIZE must be > 0, using default %u",
229 kDefaultBatchDeleteSize);
230 ✗ batch_delete_size_ = kDefaultBatchDeleteSize;
231 ✗ } else if (requested > kMaxBatchDeleteSize) {
232 ✗ LogCvmfs(kLogUploadS3, kLogStderr,
233 "Warning: CVMFS_S3_BATCH_DELETE_SIZE=%u exceeds the S3 "
234 "multi-object DELETE limit of %u, clamping to %u",
235 requested, kMaxBatchDeleteSize, kMaxBatchDeleteSize);
236 ✗ batch_delete_size_ = kMaxBatchDeleteSize;
237 } else {
238 ✗ batch_delete_size_ = requested;
239 }
240 }
241
242
3/6
✓ Branch 2 taken 572 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 572 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 572 times.
 572 if (options_manager.GetValue("CVMFS_S3_USE_HTTPS", &parameter)) {
243 ✗ use_https_ = options_manager.IsOn(parameter);
244 }
245
246
3/6
✓ Branch 2 taken 572 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 572 times.
✗ Branch 6 not taken.
✓ Branch 9 taken 572 times.
✗ Branch 10 not taken.
 572 if (options_manager.GetValue("CVMFS_S3_PORT", &parameter)) {
247
2/4
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 572 times.
✗ Branch 5 not taken.
 572 host_name_port_ = host_name_ + ":" + parameter;
248 } else {
249 ✗ host_name_port_ = host_name_;
250 }
251
252
3/6
✓ Branch 2 taken 572 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 572 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 572 times.
 572 if (options_manager.IsDefined("CVMFS_S3_PROXY")) {
253 ✗ options_manager.GetValue("CVMFS_S3_PROXY", &proxy_);
254 }
255
256 572 return true;
257 572 }
258
259
260 676 bool S3Uploader::WillHandle(const SpoolerDefinition &spooler_definition) {
261 676 return spooler_definition.driver_type == SpoolerDefinition::S3;
262 }
263
264
265 ✗ bool S3Uploader::Create() {
266 ✗ if (!dns_buckets_)
267 ✗ return false;
268
269 ✗ s3fanout::JobInfo *info = CreateJobInfo("");
270 ✗ info->request = s3fanout::JobInfo::kReqPutBucket;
271 ✗ std::string request_content;
272 ✗ if (!region_.empty()) {
273 ✗ request_content = std::string("<CreateBucketConfiguration xmlns="
274 "\"http://s3.amazonaws.com/doc/2006-03-01/\">"
275 "<LocationConstraint>")
276 ✗ + region_
277 ✗ + "</LocationConstraint>"
278 ✗ "</CreateBucketConfiguration>";
279 ✗ info->origin->Append(request_content.data(), request_content.length());
280 ✗ info->origin->Commit();
281 }
282
283 ✗ RequestCtrl req_ctrl;
284 ✗ MakePipe(req_ctrl.pipe_wait);
285 ✗ info->callback = const_cast<void *>(static_cast<void const *>(
286 ✗ MakeClosure(&S3Uploader::OnReqComplete, this, &req_ctrl)));
287
288 ✗ IncJobsInFlight();
289 ✗ UploadJobInfo(info);
290 ✗ req_ctrl.WaitFor();
291
292 ✗ return req_ctrl.return_code == 0;
293 }
294
295
296 176 unsigned int S3Uploader::GetNumberOfErrors() const { return io_errors_.load(); }
297
298
299 /**
300 * Worker thread takes care of requesting new jobs and cleaning old ones.
301 */
302 572 void *S3Uploader::MainCollectResults(void *data) {
303 572 LogCvmfs(kLogUploadS3, kLogDebug, "Upload_S3 WorkerThread started.");
304 572 S3Uploader *uploader = reinterpret_cast<S3Uploader *>(data);
305
306 while (true) {
307 27852 s3fanout::JobInfo *info = uploader->s3fanout_mgr_->PopCompletedJob();
308
2/2
✓ Branch 0 taken 572 times.
✓ Branch 1 taken 27280 times.
 27852 if (!info)
309 572 break;
310 // Report completed job
311 27280 int reply_code = 0;
312
2/2
✓ Branch 0 taken 88 times.
✓ Branch 1 taken 27192 times.
 27280 if (info->error_code != s3fanout::kFailOk) {
313
1/2
✓ Branch 0 taken 88 times.
✗ Branch 1 not taken.
 88 if ((info->request != s3fanout::JobInfo::kReqHeadOnly)
314
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 88 times.
 88 || (info->error_code != s3fanout::kFailNotFound)) {
315 ✗ if (info->request == s3fanout::JobInfo::kReqDeleteMulti) {
316 ✗ LogCvmfs(kLogUploadS3, kLogStderr,
317 "Batch delete of %lu objects failed. (error code: %d - %s)",
318 ✗ info->multi_delete_keys.size(), info->error_code,
319 s3fanout::Code2Ascii(info->error_code));
320 } else {
321 ✗ LogCvmfs(kLogUploadS3, kLogStderr,
322 "Upload job for '%s' failed. (error code: %d - %s)",
323 ✗ info->object_key.c_str(), info->error_code,
324 s3fanout::Code2Ascii(info->error_code));
325 }
326 ✗ reply_code = 99;
327 ✗ uploader->io_errors_.fetch_add(1);
328 }
329 }
330
2/2
✓ Branch 0 taken 308 times.
✓ Branch 1 taken 26972 times.
 27280 if (info->request == s3fanout::JobInfo::kReqDeleteMulti) {
331 // Parse response for per-key errors
332 308 std::set<std::string> failed_keys;
333 616 if (info->error_code == s3fanout::kFailOk
334
3/6
✓ Branch 0 taken 308 times.
✗ Branch 1 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 308 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 308 times.
 308 && !info->response_body.empty()) {
335 ✗ std::vector<std::string> error_keys, error_codes, error_messages;
336 ✗ const unsigned num_errors = s3fanout::ParseDeleteMultiResponse(
337 ✗ info->response_body, &error_keys, &error_codes, &error_messages);
338 ✗ for (unsigned i = 0; i < num_errors; ++i) {
339 ✗ LogCvmfs(kLogUploadS3, kLogStderr,
340 "S3 multi-delete error for key '%s': %s - %s",
341 ✗ error_keys[i].c_str(), error_codes[i].c_str(),
342 ✗ error_messages[i].c_str());
343 ✗ uploader->io_errors_.fetch_add(1);
344 ✗ failed_keys.insert(error_keys[i]);
345 }
346 }
347 // Decrement jobs_in_flight_ once for the entire batch.
348
1/2
✓ Branch 1 taken 308 times.
✗ Branch 2 not taken.
 308 uploader->Respond(
349 NULL,
350
1/2
✓ Branch 1 taken 308 times.
✗ Branch 2 not taken.
 616 UploaderResults(UploaderResults::kRemove,
351
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 308 times.
 308 (info->error_code != s3fanout::kFailOk) ? 99 : 0));
352
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 26972 times.
 27280 } else if (info->request == s3fanout::JobInfo::kReqDelete) {
353 ✗ uploader->Respond(NULL, UploaderResults());
354
2/2
✓ Branch 0 taken 220 times.
✓ Branch 1 taken 26752 times.
 26972 } else if (info->request == s3fanout::JobInfo::kReqHeadOnly) {
355
2/2
✓ Branch 0 taken 88 times.
✓ Branch 1 taken 132 times.
 220 if (info->error_code == s3fanout::kFailNotFound)
356 88 reply_code = 1;
357
1/2
✓ Branch 1 taken 220 times.
✗ Branch 2 not taken.
 220 uploader->Respond(static_cast<CallbackTN *>(info->callback),
358 440 UploaderResults(UploaderResults::kLookup, reply_code));
359 } else {
360
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 26752 times.
 26752 if (info->request == s3fanout::JobInfo::kReqHeadPut) {
361 // The HEAD request was not transformed into a PUT request, thus this
362 // was a duplicate
363 // Uploaded catalogs are always unique ->
364 // assume this was a regular file and decrease appropriate counters
365 ✗ uploader->CountDuplicates();
366 ✗ uploader->DecUploadedChunks();
367 ✗ uploader->CountUploadedBytes(-(info->payload_size));
368 }
369 53504 uploader->Respond(
370
1/2
✓ Branch 1 taken 26752 times.
✗ Branch 2 not taken.
 26752 static_cast<CallbackTN *>(info->callback),
371 53504 UploaderResults(UploaderResults::kChunkCommit, reply_code));
372
373
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 26752 times.
 26752 assert(info->origin.get() == nullptr);
374 }
375
1/2
✓ Branch 0 taken 27280 times.
✗ Branch 1 not taken.
 27280 delete info;
376 27280 }
377
378 572 LogCvmfs(kLogUploadS3, kLogDebug, "Upload_S3 WorkerThread finished.");
379 572 return NULL;
380 }
381
382
383 22308 void S3Uploader::DoUpload(const std::string &remote_path,
384 IngestionSource *source,
385 const CallbackTN *callback) {
386
1/2
✓ Branch 1 taken 22308 times.
✗ Branch 2 not taken.
 22308 bool rvb = source->Open();
387
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 22308 times.
 22308 if (!rvb) {
388 ✗ Respond(callback, UploaderResults(100, source->GetPath()));
389 ✗ return;
390 }
391 uint64_t size;
392
1/2
✓ Branch 1 taken 22308 times.
✗ Branch 2 not taken.
 22308 rvb = source->GetSize(&size);
393
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 22308 times.
 22308 assert(rvb);
394
395 22308 FileBackedBuffer *origin = FileBackedBuffer::Create(
396
1/2
✓ Branch 2 taken 22308 times.
✗ Branch 3 not taken.
 22308 kInMemoryObjectThreshold, spooler_definition().temporary_path);
397
398 unsigned char buffer[kPageSize];
399 ssize_t nbytes;
400 do {
401
1/2
✓ Branch 1 taken 8763348 times.
✗ Branch 2 not taken.
 8763348 nbytes = source->Read(buffer, kPageSize);
402
2/2
✓ Branch 0 taken 8748564 times.
✓ Branch 1 taken 14784 times.
 8763348 if (nbytes > 0)
403
1/2
✓ Branch 1 taken 8748564 times.
✗ Branch 2 not taken.
 8748564 origin->Append(buffer, nbytes);
404
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 8763348 times.
 8763348 if (nbytes < 0) {
405 ✗ source->Close();
406 ✗ delete origin;
407 ✗ Respond(callback, UploaderResults(100, source->GetPath()));
408 ✗ return;
409 }
410
2/2
✓ Branch 0 taken 8741040 times.
✓ Branch 1 taken 22308 times.
 8763348 } while (nbytes == kPageSize);
411
1/2
✓ Branch 1 taken 22308 times.
✗ Branch 2 not taken.
 22308 source->Close();
412
1/2
✓ Branch 1 taken 22308 times.
✗ Branch 2 not taken.
 22308 origin->Commit();
413
414 s3fanout::JobInfo *info = new s3fanout::JobInfo(
415
2/4
✓ Branch 1 taken 22308 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 22308 times.
✗ Branch 5 not taken.
 44616 repository_alias_ + "/" + remote_path,
416 const_cast<void *>(static_cast<void const *>(callback)),
417
2/4
✓ Branch 1 taken 22308 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 22308 times.
✗ Branch 5 not taken.
 22308 origin);
418
419
3/6
✓ Branch 2 taken 22308 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 22308 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 22308 times.
 22308 if (HasPrefix(remote_path, ".cvmfs", false /*ignore_case*/)) {
420 ✗ info->request = s3fanout::JobInfo::kReqPutDotCvmfs;
421
3/6
✓ Branch 2 taken 22308 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 22308 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 22308 times.
 22308 } else if (HasSuffix(remote_path, ".html", false)) {
422 ✗ info->request = s3fanout::JobInfo::kReqPutHtml;
423 } else {
424
1/2
✓ Branch 0 taken 22308 times.
✗ Branch 1 not taken.
 22308 if (peek_before_put_)
425 22308 info->request = s3fanout::JobInfo::kReqHeadPut;
426 }
427
428 22308 RequestCtrl req_ctrl;
429
1/2
✓ Branch 1 taken 22308 times.
✗ Branch 2 not taken.
 22308 MakePipe(req_ctrl.pipe_wait);
430 22308 req_ctrl.callback_forward = callback;
431
1/2
✓ Branch 1 taken 22308 times.
✗ Branch 2 not taken.
 22308 req_ctrl.original_path = source->GetPath();
432 22308 info->callback = const_cast<void *>(static_cast<void const *>(
433
1/2
✓ Branch 1 taken 22308 times.
✗ Branch 2 not taken.
 22308 MakeClosure(&S3Uploader::OnReqComplete, this, &req_ctrl)));
434
435
1/2
✓ Branch 1 taken 22308 times.
✗ Branch 2 not taken.
 22308 UploadJobInfo(info);
436
1/2
✓ Branch 1 taken 22308 times.
✗ Branch 2 not taken.
 22308 req_ctrl.WaitFor();
437
1/2
✓ Branch 2 taken 22308 times.
✗ Branch 3 not taken.
 22308 LogCvmfs(kLogUploadS3, kLogDebug, "Uploading from source finished: %s",
438
1/2
✓ Branch 1 taken 22308 times.
✗ Branch 2 not taken.
 44616 source->GetPath().c_str());
439 22308 }
440
441
442 26972 void S3Uploader::UploadJobInfo(s3fanout::JobInfo *info) {
443 26972 LogCvmfs(kLogUploadS3, kLogDebug,
444 "Uploading:\n"
445 "--> Object: '%s'\n"
446 "--> Bucket: '%s'\n"
447 "--> Host: '%s'\n",
448 info->object_key.c_str(), bucket_.c_str(), host_name_port_.c_str());
449
450 26972 s3fanout_mgr_->PushNewJob(info);
451 26972 }
452
453
454 4444 UploadStreamHandle *S3Uploader::InitStreamedUpload(const CallbackTN *callback) {
455 return new S3StreamHandle(callback, kInMemoryObjectThreshold,
456
1/2
✓ Branch 3 taken 4444 times.
✗ Branch 4 not taken.
 4444 spooler_definition().temporary_path);
457 }
458
459
460 35200 void S3Uploader::StreamedUpload(UploadStreamHandle *handle,
461 UploadBuffer buffer,
462 const CallbackTN *callback) {
463 35200 S3StreamHandle *s3_handle = static_cast<S3StreamHandle *>(handle);
464
465 35200 s3_handle->buffer->Append(buffer.data, buffer.size);
466
1/2
✓ Branch 2 taken 35288 times.
✗ Branch 3 not taken.
 35288 Respond(callback, UploaderResults(UploaderResults::kBufferUpload, 0));
467 35288 }
468
469
470 4444 void S3Uploader::FinalizeStreamedUpload(UploadStreamHandle *handle,
471 const shash::Any &content_hash) {
472 4444 S3StreamHandle *s3_handle = static_cast<S3StreamHandle *>(handle);
473
474 // New file name based on content hash or remote_path override
475 4444 std::string final_path;
476
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 4444 times.
 4444 if (s3_handle->remote_path != "") {
477 ✗ final_path = repository_alias_ + "/" + s3_handle->remote_path;
478 } else {
479
3/6
✓ Branch 1 taken 4444 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 4444 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 4444 times.
✗ Branch 8 not taken.
 4444 final_path = repository_alias_ + "/data/" + content_hash.MakePath();
480 }
481
482
1/2
✓ Branch 2 taken 4444 times.
✗ Branch 3 not taken.
 4444 s3_handle->buffer->Commit();
483
484
1/2
✓ Branch 2 taken 4444 times.
✗ Branch 3 not taken.
 4444 const size_t bytes_uploaded = s3_handle->buffer->GetSize();
485
486 s3fanout::JobInfo *info = new s3fanout::JobInfo(
487 final_path,
488 4444 const_cast<void *>(static_cast<void const *>(handle->commit_callback)),
489
2/4
✓ Branch 2 taken 4444 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 4444 times.
✗ Branch 6 not taken.
 4444 s3_handle->buffer.release());
490
491
1/2
✓ Branch 0 taken 4444 times.
✗ Branch 1 not taken.
 4444 if (peek_before_put_)
492 4444 info->request = s3fanout::JobInfo::kReqHeadPut;
493
1/2
✓ Branch 1 taken 4444 times.
✗ Branch 2 not taken.
 4444 UploadJobInfo(info);
494
495 // Remove the temporary file
496
1/2
✓ Branch 0 taken 4444 times.
✗ Branch 1 not taken.
 4444 delete s3_handle;
497
498 // Update statistics counters
499 4444 if (!content_hash.HasSuffix()
500
5/6
✓ Branch 0 taken 44 times.
✓ Branch 1 taken 4400 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 44 times.
✓ Branch 4 taken 4400 times.
✓ Branch 5 taken 44 times.
 4444 || content_hash.suffix == shash::kSuffixPartial) {
501
1/2
✓ Branch 1 taken 4400 times.
✗ Branch 2 not taken.
 4400 CountUploadedChunks();
502
1/2
✓ Branch 1 taken 4400 times.
✗ Branch 2 not taken.
 4400 CountUploadedBytes(bytes_uploaded);
503
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 44 times.
 44 } else if (content_hash.suffix == shash::kSuffixCatalog) {
504 ✗ CountUploadedCatalogs();
505 ✗ CountUploadedCatalogBytes(bytes_uploaded);
506 }
507 4444 }
508
509
510 220 s3fanout::JobInfo *S3Uploader::CreateJobInfo(const std::string &path) const {
511
2/4
✓ Branch 2 taken 220 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 220 times.
✗ Branch 6 not taken.
 220 FileBackedBuffer *buf = FileBackedBuffer::Create(kInMemoryObjectThreshold);
512
1/2
✓ Branch 2 taken 220 times.
✗ Branch 3 not taken.
 220 return new s3fanout::JobInfo(path, NULL, buf);
513 }
514
515
516 26444 void S3Uploader::DoRemoveAsync(const std::string &file_to_delete) {
517
2/4
✓ Branch 1 taken 26444 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 26444 times.
✗ Branch 5 not taken.
 26444 const std::string mangled_path = repository_alias_ + "/" + file_to_delete;
518
519
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 26444 times.
 26444 if (!batch_delete_enabled_) {
520 ✗ s3fanout::JobInfo *info = CreateJobInfo(mangled_path);
521 ✗ info->request = s3fanout::JobInfo::kReqDelete;
522 ✗ LogCvmfs(kLogUploadS3, kLogDebug, "Asynchronously removing %s/%s",
523 bucket_.c_str(), info->object_key.c_str());
524 ✗ s3fanout_mgr_->PushNewJob(info);
525 ✗ return;
526 }
527
528 // Batch delete: collect keys and flush when the S3 batch limit is reached.
529 //
530 // The caller (RemoveAsync) already incremented jobs_in_flight_ for this key.
531 // We undo that immediately: for batched deletes, jobs_in_flight_ is managed
532 // per-batch rather than per-key. FlushDeleteBatch() increments once when
533 // the batch is dispatched, and MainCollectResults decrements once when the
534 // batch response arrives. This avoids a deadlock where jobs_in_flight_
535 // saturates (e.g. at 512) before the batch threshold (1000) is reached,
536 // blocking forever on a flush that never happens.
537
1/2
✓ Branch 1 taken 26444 times.
✗ Branch 2 not taken.
 26444 DecJobsInFlight();
538 26444 const MutexLockGuard guard(delete_batch_mutex_);
539
1/2
✓ Branch 1 taken 26444 times.
✗ Branch 2 not taken.
 26444 pending_deletes_.push_back(mangled_path);
540
2/2
✓ Branch 1 taken 264 times.
✓ Branch 2 taken 26180 times.
 26444 if (pending_deletes_.size() >= batch_delete_size_) {
541
1/2
✓ Branch 1 taken 264 times.
✗ Branch 2 not taken.
 264 FlushDeleteBatch();
542 }
543
1/2
✓ Branch 2 taken 26444 times.
✗ Branch 3 not taken.
 26444 }
544
545
546 836 void S3Uploader::FlushDeleteBatch() const {
547 // Caller must hold delete_batch_mutex_
548
2/2
✓ Branch 1 taken 528 times.
✓ Branch 2 taken 308 times.
 836 if (pending_deletes_.empty())
549 528 return;
550
551
1/2
✓ Branch 2 taken 308 times.
✗ Branch 3 not taken.
 308 LogCvmfs(kLogUploadS3, kLogDebug, "Flushing batch delete of %lu objects",
552 pending_deletes_.size());
553
554 // Build XML request body
555
1/2
✓ Branch 1 taken 308 times.
✗ Branch 2 not taken.
 308 const std::string xml = s3fanout::ComposeDeleteMultiXml(pending_deletes_);
556
2/4
✓ Branch 2 taken 308 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 308 times.
✗ Branch 6 not taken.
 308 FileBackedBuffer *buf = FileBackedBuffer::Create(kInMemoryObjectThreshold);
557
1/2
✓ Branch 3 taken 308 times.
✗ Branch 4 not taken.
 308 buf->Append(xml.data(), xml.length());
558
1/2
✓ Branch 1 taken 308 times.
✗ Branch 2 not taken.
 308 buf->Commit();
559
560 // The object_key for multi-delete is empty (URL is bucket root + ?delete)
561
3/6
✓ Branch 2 taken 308 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 308 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 308 times.
✗ Branch 9 not taken.
 308 s3fanout::JobInfo *info = new s3fanout::JobInfo("", NULL, buf);
562 308 info->request = s3fanout::JobInfo::kReqDeleteMulti;
563 308 info->multi_delete_keys.swap(pending_deletes_);
564
565 // Track this batch as a single job in flight. MainCollectResults will
566 // call Respond() exactly once for kReqDeleteMulti to balance it.
567
1/2
✓ Branch 1 taken 308 times.
✗ Branch 2 not taken.
 308 IncJobsInFlight();
568
1/2
✓ Branch 2 taken 308 times.
✗ Branch 3 not taken.
 308 s3fanout_mgr_->PushNewJob(info);
569 308 }
570
571
572 572 void S3Uploader::WaitForUpload() const {
573 {
574 572 const MutexLockGuard guard(delete_batch_mutex_);
575
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 FlushDeleteBatch();
576 572 }
577 572 AbstractUploader::WaitForUpload();
578 572 }
579
580
581 22528 void S3Uploader::OnReqComplete(const upload::UploaderResults &results,
582 RequestCtrl *ctrl) {
583 22528 ctrl->return_code = results.return_code;
584
2/2
✓ Branch 0 taken 22308 times.
✓ Branch 1 taken 220 times.
 22528 if (ctrl->callback_forward != NULL) {
585 // We are already in Respond() so we must not call it again
586 22308 const upload::UploaderResults fix_path(results.return_code,
587
1/2
✓ Branch 1 taken 22308 times.
✗ Branch 2 not taken.
 22308 ctrl->original_path);
588
1/2
✓ Branch 1 taken 22308 times.
✗ Branch 2 not taken.
 22308 (*(ctrl->callback_forward))(fix_path);
589
1/2
✓ Branch 0 taken 22308 times.
✗ Branch 1 not taken.
 22308 delete ctrl->callback_forward;
590 22308 ctrl->callback_forward = NULL;
591 22308 }
592 22528 char c = 'c';
593
1/2
✓ Branch 1 taken 22528 times.
✗ Branch 2 not taken.
 22528 WritePipe(ctrl->pipe_wait[1], &c, 1);
594 22528 }
595
596
597 220 bool S3Uploader::Peek(const std::string &path) {
598
2/4
✓ Branch 1 taken 220 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 220 times.
✗ Branch 5 not taken.
 220 const std::string mangled_path = repository_alias_ + "/" + path;
599
1/2
✓ Branch 1 taken 220 times.
✗ Branch 2 not taken.
 220 s3fanout::JobInfo *info = CreateJobInfo(mangled_path);
600
601 220 RequestCtrl req_ctrl;
602
1/2
✓ Branch 1 taken 220 times.
✗ Branch 2 not taken.
 220 MakePipe(req_ctrl.pipe_wait);
603 220 info->request = s3fanout::JobInfo::kReqHeadOnly;
604 220 info->callback = const_cast<void *>(static_cast<void const *>(
605
1/2
✓ Branch 1 taken 220 times.
✗ Branch 2 not taken.
 220 MakeClosure(&S3Uploader::OnReqComplete, this, &req_ctrl)));
606
607
1/2
✓ Branch 1 taken 220 times.
✗ Branch 2 not taken.
 220 IncJobsInFlight();
608
1/2
✓ Branch 1 taken 220 times.
✗ Branch 2 not taken.
 220 UploadJobInfo(info);
609
1/2
✓ Branch 1 taken 220 times.
✗ Branch 2 not taken.
 220 req_ctrl.WaitFor();
610
611 220 return req_ctrl.return_code == 0;
612 220 }
613
614
615 // noop: no mkdir needed in S3 storage
616 ✗ bool S3Uploader::Mkdir(const std::string &path) { return true; }
617
618
619 ✗ bool S3Uploader::PlaceBootstrappingShortcut(const shash::Any &object) {
620 ✗ return false; // TODO(rmeusel): implement
621 }
622
623
624 ✗ int64_t S3Uploader::DoGetObjectSize(const std::string &file_name) {
625 // TODO(dosarudaniel): use a head request for byte count
626 // Re-enable 661 integration test when done
627 ✗ return -EOPNOTSUPP;
628 }
629
630 } // namespace upload
631