GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/upload_s3.cc
Date: 2026-08-30 02:40:36
Exec Total Coverage
Lines: 239 350 68.3%
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 15360 void S3Uploader::RequestCtrl::WaitFor() {
40 char c;
41
1/2
✓ Branch 1 taken 15360 times.
✗ Branch 2 not taken.
 15360 ReadPipe(pipe_wait[0], &c, 1);
42
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 15360 times.
 15360 assert(c == 'c');
43
1/2
✓ Branch 1 taken 15360 times.
✗ Branch 2 not taken.
 15360 ClosePipe(pipe_wait);
44 15360 }
45
46
47 390 S3Uploader::S3Uploader(const SpoolerDefinition &spooler_definition)
48 : AbstractUploader(spooler_definition)
49 390 , dns_buckets_(true)
50 390 , num_parallel_uploads_(kDefaultNumParallelUploads)
51 390 , num_retries_(kDefaultNumRetries)
52 390 , timeout_sec_(kDefaultTimeoutSec)
53 390 , authz_method_(s3fanout::kAuthzAwsV2)
54 390 , peek_before_put_(true)
55 390 , use_https_(false)
56 390 , batch_delete_enabled_(true)
57 390 , batch_delete_size_(kDefaultBatchDeleteSize)
58
1/2
✓ Branch 2 taken 390 times.
✗ Branch 3 not taken.
 390 , proxy_("")
59
1/2
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
 390 , temporary_path_(spooler_definition.temporary_path)
60
1/2
✓ Branch 12 taken 390 times.
✗ Branch 13 not taken.
 780 , x_amz_acl_("public-read") {
61
2/4
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 390 times.
✗ Branch 4 not taken.
 390 assert(spooler_definition.IsValid()
62 && spooler_definition.driver_type == SpoolerDefinition::S3);
63
64 390 atomic_init32(&io_errors_);
65 390 const int mutex_ret = pthread_mutex_init(&delete_batch_mutex_, NULL);
66
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 390 times.
 390 assert(mutex_ret == 0);
67
68
2/4
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 390 times.
 390 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 390 times.
 390 if (authz_method_ == s3fanout::kAuthzAzure)
74 ✗ batch_delete_enabled_ = false;
75
76
1/2
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
 390 s3fanout::S3FanoutManager::S3Config s3config;
77
1/2
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
 390 s3config.access_key = access_key_;
78
1/2
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
 390 s3config.secret_key = secret_key_;
79
1/2
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
 390 s3config.hostname_port = host_name_port_;
80 390 s3config.authz_method = authz_method_;
81
1/2
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
 390 s3config.region = region_;
82
1/2
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
 390 s3config.flavor = flavor_;
83
1/2
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
 390 s3config.bucket = bucket_;
84 390 s3config.dns_buckets = dns_buckets_;
85 390 s3config.pool_max_handles = num_parallel_uploads_;
86 390 s3config.opt_timeout_sec = timeout_sec_;
87 390 s3config.opt_max_retries = num_retries_;
88 390 s3config.opt_backoff_init_ms = kDefaultBackoffInitMs;
89 390 s3config.opt_backoff_max_ms = kDefaultBackoffMaxMs;
90
1/2
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
 390 s3config.x_amz_acl = x_amz_acl_;
91
92
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 390 times.
 390 if (use_https_) {
93 ✗ s3config.protocol = "https";
94 } else {
95
1/2
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
 390 s3config.protocol = "http";
96 }
97
1/2
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
 390 s3config.proxy = proxy_;
98
99 780 s3fanout_mgr_ = std::unique_ptr<s3fanout::S3FanoutManager>(
100
2/4
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 390 times.
✗ Branch 5 not taken.
 780 new s3fanout::S3FanoutManager(s3config));
101
1/2
✓ Branch 2 taken 390 times.
✗ Branch 3 not taken.
 390 s3fanout_mgr_->Spawn();
102
103 390 const int retval = pthread_create(&thread_collect_results_, NULL,
104 MainCollectResults, this);
105
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 390 times.
 390 assert(retval == 0);
106 390 }
107
108
109 2340 S3Uploader::~S3Uploader() {
110 // Signal termination to our own worker thread
111 780 s3fanout_mgr_->PushCompletedJob(NULL);
112 780 pthread_join(thread_collect_results_, NULL);
113 780 pthread_mutex_destroy(&delete_batch_mutex_);
114 1560 }
115
116
117 390 bool S3Uploader::ParseSpoolerDefinition(
118 const SpoolerDefinition &spooler_definition) {
119 const std::vector<std::string> config = SplitString(
120
1/2
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
 390 spooler_definition.spooler_configuration, '@');
121
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 390 times.
 390 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 390 times.
✗ Branch 3 not taken.
 390 repository_alias_ = config[0];
129 390 const std::string &config_path = config[1];
130
131
2/4
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 390 times.
 390 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 390 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 390 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 390 times.
✗ Branch 8 not taken.
✓ Branch 10 taken 390 times.
✗ Branch 11 not taken.
 390 new DefaultOptionsTemplateManager(repository_alias_));
140
1/2
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
 390 options_manager.ParsePath(config_path, false);
141 390 std::string parameter;
142
143
3/6
✓ Branch 2 taken 390 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 390 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 390 times.
 390 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 390 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 390 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 390 times.
 390 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 390 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 390 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 390 times.
 390 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 390 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 390 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 390 times.
 390 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 390 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 390 times.
✗ Branch 6 not taken.
✓ Branch 9 taken 390 times.
✗ Branch 10 not taken.
 390 if (options_manager.GetValue("CVMFS_S3_DNS_BUCKETS", &parameter)) {
166
1/2
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
 390 if (parameter == "false") {
167 390 dns_buckets_ = false;
168 }
169 }
170
3/6
✓ Branch 2 taken 390 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 390 times.
✗ Branch 6 not taken.
✓ Branch 9 taken 390 times.
✗ Branch 10 not taken.
 390 if (options_manager.GetValue("CVMFS_S3_MAX_NUMBER_OF_PARALLEL_CONNECTIONS",
171 &parameter)) {
172
1/2
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
 390 num_parallel_uploads_ = String2Uint64(parameter);
173 }
174
3/6
✓ Branch 2 taken 390 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 390 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 390 times.
 390 if (options_manager.GetValue("CVMFS_S3_MAX_RETRIES", &parameter)) {
175 ✗ num_retries_ = String2Uint64(parameter);
176 }
177
3/6
✓ Branch 2 taken 390 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 390 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 390 times.
 390 if (options_manager.GetValue("CVMFS_S3_TIMEOUT", &parameter)) {
178 ✗ timeout_sec_ = String2Uint64(parameter);
179 }
180
3/6
✓ Branch 2 taken 390 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 390 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 390 times.
 390 if (options_manager.GetValue("CVMFS_S3_REGION", &region_)) {
181 ✗ authz_method_ = s3fanout::kAuthzAwsV4;
182 }
183
3/6
✓ Branch 2 taken 390 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 390 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 390 times.
 390 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 390 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 390 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 390 times.
 390 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 390 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 390 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 390 times.
 390 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 390 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 390 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 390 times.
 390 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 390 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 390 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 390 times.
 390 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 390 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 390 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 390 times.
 390 if (options_manager.GetValue("CVMFS_S3_USE_HTTPS", &parameter)) {
243 ✗ use_https_ = options_manager.IsOn(parameter);
244 }
245
246
3/6
✓ Branch 2 taken 390 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 390 times.
✗ Branch 6 not taken.
✓ Branch 9 taken 390 times.
✗ Branch 10 not taken.
 390 if (options_manager.GetValue("CVMFS_S3_PORT", &parameter)) {
247
2/4
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 390 times.
✗ Branch 5 not taken.
 390 host_name_port_ = host_name_ + ":" + parameter;
248 } else {
249 ✗ host_name_port_ = host_name_;
250 }
251
252
3/6
✓ Branch 2 taken 390 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 390 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 390 times.
 390 if (options_manager.IsDefined("CVMFS_S3_PROXY")) {
253 ✗ options_manager.GetValue("CVMFS_S3_PROXY", &proxy_);
254 }
255
256 390 return true;
257 390 }
258
259
260 489 bool S3Uploader::WillHandle(const SpoolerDefinition &spooler_definition) {
261 489 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 60 unsigned int S3Uploader::GetNumberOfErrors() const {
297 60 return atomic_read32(&io_errors_);
298 }
299
300
301 /**
302 * Worker thread takes care of requesting new jobs and cleaning old ones.
303 */
304 390 void *S3Uploader::MainCollectResults(void *data) {
305 390 LogCvmfs(kLogUploadS3, kLogDebug, "Upload_S3 WorkerThread started.");
306 390 S3Uploader *uploader = reinterpret_cast<S3Uploader *>(data);
307
308 while (true) {
309 18990 s3fanout::JobInfo *info = uploader->s3fanout_mgr_->PopCompletedJob();
310
2/2
✓ Branch 0 taken 390 times.
✓ Branch 1 taken 18600 times.
 18990 if (!info)
311 390 break;
312 // Report completed job
313 18600 int reply_code = 0;
314
2/2
✓ Branch 0 taken 60 times.
✓ Branch 1 taken 18540 times.
 18600 if (info->error_code != s3fanout::kFailOk) {
315
1/2
✓ Branch 0 taken 60 times.
✗ Branch 1 not taken.
 60 if ((info->request != s3fanout::JobInfo::kReqHeadOnly)
316
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 60 times.
 60 || (info->error_code != s3fanout::kFailNotFound)) {
317 ✗ if (info->request == s3fanout::JobInfo::kReqDeleteMulti) {
318 ✗ LogCvmfs(kLogUploadS3, kLogStderr,
319 "Batch delete of %lu objects failed. (error code: %d - %s)",
320 ✗ info->multi_delete_keys.size(), info->error_code,
321 s3fanout::Code2Ascii(info->error_code));
322 } else {
323 ✗ LogCvmfs(kLogUploadS3, kLogStderr,
324 "Upload job for '%s' failed. (error code: %d - %s)",
325 ✗ info->object_key.c_str(), info->error_code,
326 s3fanout::Code2Ascii(info->error_code));
327 }
328 ✗ reply_code = 99;
329 ✗ atomic_inc32(&uploader->io_errors_);
330 }
331 }
332
2/2
✓ Branch 0 taken 210 times.
✓ Branch 1 taken 18390 times.
 18600 if (info->request == s3fanout::JobInfo::kReqDeleteMulti) {
333 // Parse response for per-key errors
334 210 std::set<std::string> failed_keys;
335 420 if (info->error_code == s3fanout::kFailOk
336
3/6
✓ Branch 0 taken 210 times.
✗ Branch 1 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 210 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 210 times.
 210 && !info->response_body.empty()) {
337 ✗ std::vector<std::string> error_keys, error_codes, error_messages;
338 ✗ const unsigned num_errors = s3fanout::ParseDeleteMultiResponse(
339 ✗ info->response_body, &error_keys, &error_codes, &error_messages);
340 ✗ for (unsigned i = 0; i < num_errors; ++i) {
341 ✗ LogCvmfs(kLogUploadS3, kLogStderr,
342 "S3 multi-delete error for key '%s': %s - %s",
343 ✗ error_keys[i].c_str(), error_codes[i].c_str(),
344 ✗ error_messages[i].c_str());
345 ✗ atomic_inc32(&uploader->io_errors_);
346 ✗ failed_keys.insert(error_keys[i]);
347 }
348 }
349 // Decrement jobs_in_flight_ once for the entire batch.
350
1/2
✓ Branch 1 taken 210 times.
✗ Branch 2 not taken.
 210 uploader->Respond(
351 NULL,
352
1/2
✓ Branch 1 taken 210 times.
✗ Branch 2 not taken.
 420 UploaderResults(UploaderResults::kRemove,
353
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 210 times.
 210 (info->error_code != s3fanout::kFailOk) ? 99 : 0));
354
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 18390 times.
 18600 } else if (info->request == s3fanout::JobInfo::kReqDelete) {
355 ✗ uploader->Respond(NULL, UploaderResults());
356
2/2
✓ Branch 0 taken 150 times.
✓ Branch 1 taken 18240 times.
 18390 } else if (info->request == s3fanout::JobInfo::kReqHeadOnly) {
357
2/2
✓ Branch 0 taken 60 times.
✓ Branch 1 taken 90 times.
 150 if (info->error_code == s3fanout::kFailNotFound)
358 60 reply_code = 1;
359
1/2
✓ Branch 1 taken 150 times.
✗ Branch 2 not taken.
 150 uploader->Respond(static_cast<CallbackTN *>(info->callback),
360 300 UploaderResults(UploaderResults::kLookup, reply_code));
361 } else {
362
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 18240 times.
 18240 if (info->request == s3fanout::JobInfo::kReqHeadPut) {
363 // The HEAD request was not transformed into a PUT request, thus this
364 // was a duplicate
365 // Uploaded catalogs are always unique ->
366 // assume this was a regular file and decrease appropriate counters
367 ✗ uploader->CountDuplicates();
368 ✗ uploader->DecUploadedChunks();
369 ✗ uploader->CountUploadedBytes(-(info->payload_size));
370 }
371 36480 uploader->Respond(
372
1/2
✓ Branch 1 taken 18240 times.
✗ Branch 2 not taken.
 18240 static_cast<CallbackTN *>(info->callback),
373 36480 UploaderResults(UploaderResults::kChunkCommit, reply_code));
374
375
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 18240 times.
 18240 assert(info->origin.get() == nullptr);
376 }
377
1/2
✓ Branch 0 taken 18600 times.
✗ Branch 1 not taken.
 18600 delete info;
378 18600 }
379
380 390 LogCvmfs(kLogUploadS3, kLogDebug, "Upload_S3 WorkerThread finished.");
381 390 return NULL;
382 }
383
384
385 15210 void S3Uploader::DoUpload(const std::string &remote_path,
386 IngestionSource *source,
387 const CallbackTN *callback) {
388
1/2
✓ Branch 1 taken 15210 times.
✗ Branch 2 not taken.
 15210 bool rvb = source->Open();
389
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 15210 times.
 15210 if (!rvb) {
390 ✗ Respond(callback, UploaderResults(100, source->GetPath()));
391 ✗ return;
392 }
393 uint64_t size;
394
1/2
✓ Branch 1 taken 15210 times.
✗ Branch 2 not taken.
 15210 rvb = source->GetSize(&size);
395
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 15210 times.
 15210 assert(rvb);
396
397 15210 FileBackedBuffer *origin = FileBackedBuffer::Create(
398
1/2
✓ Branch 2 taken 15210 times.
✗ Branch 3 not taken.
 15210 kInMemoryObjectThreshold, spooler_definition().temporary_path);
399
400 unsigned char buffer[kPageSize];
401 ssize_t nbytes;
402 do {
403
1/2
✓ Branch 1 taken 5975010 times.
✗ Branch 2 not taken.
 5975010 nbytes = source->Read(buffer, kPageSize);
404
2/2
✓ Branch 0 taken 5964930 times.
✓ Branch 1 taken 10080 times.
 5975010 if (nbytes > 0)
405
1/2
✓ Branch 1 taken 5964930 times.
✗ Branch 2 not taken.
 5964930 origin->Append(buffer, nbytes);
406
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 5975010 times.
 5975010 if (nbytes < 0) {
407 ✗ source->Close();
408 ✗ delete origin;
409 ✗ Respond(callback, UploaderResults(100, source->GetPath()));
410 ✗ return;
411 }
412
2/2
✓ Branch 0 taken 5959800 times.
✓ Branch 1 taken 15210 times.
 5975010 } while (nbytes == kPageSize);
413
1/2
✓ Branch 1 taken 15210 times.
✗ Branch 2 not taken.
 15210 source->Close();
414
1/2
✓ Branch 1 taken 15210 times.
✗ Branch 2 not taken.
 15210 origin->Commit();
415
416 s3fanout::JobInfo *info = new s3fanout::JobInfo(
417
2/4
✓ Branch 1 taken 15210 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 15210 times.
✗ Branch 5 not taken.
 30420 repository_alias_ + "/" + remote_path,
418 const_cast<void *>(static_cast<void const *>(callback)),
419
2/4
✓ Branch 1 taken 15210 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 15210 times.
✗ Branch 5 not taken.
 15210 origin);
420
421
3/6
✓ Branch 2 taken 15210 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 15210 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 15210 times.
 15210 if (HasPrefix(remote_path, ".cvmfs", false /*ignore_case*/)) {
422 ✗ info->request = s3fanout::JobInfo::kReqPutDotCvmfs;
423
3/6
✓ Branch 2 taken 15210 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 15210 times.
✗ Branch 6 not taken.
✗ Branch 9 not taken.
✓ Branch 10 taken 15210 times.
 15210 } else if (HasSuffix(remote_path, ".html", false)) {
424 ✗ info->request = s3fanout::JobInfo::kReqPutHtml;
425 } else {
426
1/2
✓ Branch 0 taken 15210 times.
✗ Branch 1 not taken.
 15210 if (peek_before_put_)
427 15210 info->request = s3fanout::JobInfo::kReqHeadPut;
428 }
429
430 15210 RequestCtrl req_ctrl;
431
1/2
✓ Branch 1 taken 15210 times.
✗ Branch 2 not taken.
 15210 MakePipe(req_ctrl.pipe_wait);
432 15210 req_ctrl.callback_forward = callback;
433
1/2
✓ Branch 1 taken 15210 times.
✗ Branch 2 not taken.
 15210 req_ctrl.original_path = source->GetPath();
434 15210 info->callback = const_cast<void *>(static_cast<void const *>(
435
1/2
✓ Branch 1 taken 15210 times.
✗ Branch 2 not taken.
 15210 MakeClosure(&S3Uploader::OnReqComplete, this, &req_ctrl)));
436
437
1/2
✓ Branch 1 taken 15210 times.
✗ Branch 2 not taken.
 15210 UploadJobInfo(info);
438
1/2
✓ Branch 1 taken 15210 times.
✗ Branch 2 not taken.
 15210 req_ctrl.WaitFor();
439
1/2
✓ Branch 2 taken 15210 times.
✗ Branch 3 not taken.
 15210 LogCvmfs(kLogUploadS3, kLogDebug, "Uploading from source finished: %s",
440
1/2
✓ Branch 1 taken 15210 times.
✗ Branch 2 not taken.
 30420 source->GetPath().c_str());
441 15210 }
442
443
444 18390 void S3Uploader::UploadJobInfo(s3fanout::JobInfo *info) {
445 18390 LogCvmfs(kLogUploadS3, kLogDebug,
446 "Uploading:\n"
447 "--> Object: '%s'\n"
448 "--> Bucket: '%s'\n"
449 "--> Host: '%s'\n",
450 info->object_key.c_str(), bucket_.c_str(), host_name_port_.c_str());
451
452 18390 s3fanout_mgr_->PushNewJob(info);
453 18390 }
454
455
456 3030 UploadStreamHandle *S3Uploader::InitStreamedUpload(const CallbackTN *callback) {
457 return new S3StreamHandle(callback, kInMemoryObjectThreshold,
458
1/2
✓ Branch 3 taken 3030 times.
✗ Branch 4 not taken.
 3030 spooler_definition().temporary_path);
459 }
460
461
462 24060 void S3Uploader::StreamedUpload(UploadStreamHandle *handle,
463 UploadBuffer buffer,
464 const CallbackTN *callback) {
465 24060 S3StreamHandle *s3_handle = static_cast<S3StreamHandle *>(handle);
466
467 24060 s3_handle->buffer->Append(buffer.data, buffer.size);
468
1/2
✓ Branch 2 taken 24060 times.
✗ Branch 3 not taken.
 24060 Respond(callback, UploaderResults(UploaderResults::kBufferUpload, 0));
469 24060 }
470
471
472 3030 void S3Uploader::FinalizeStreamedUpload(UploadStreamHandle *handle,
473 const shash::Any &content_hash) {
474 3030 S3StreamHandle *s3_handle = static_cast<S3StreamHandle *>(handle);
475
476 // New file name based on content hash or remote_path override
477 3030 std::string final_path;
478
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 3030 times.
 3030 if (s3_handle->remote_path != "") {
479 ✗ final_path = repository_alias_ + "/" + s3_handle->remote_path;
480 } else {
481
3/6
✓ Branch 1 taken 3030 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 3030 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 3030 times.
✗ Branch 8 not taken.
 3030 final_path = repository_alias_ + "/data/" + content_hash.MakePath();
482 }
483
484
1/2
✓ Branch 2 taken 3030 times.
✗ Branch 3 not taken.
 3030 s3_handle->buffer->Commit();
485
486
1/2
✓ Branch 2 taken 3030 times.
✗ Branch 3 not taken.
 3030 const size_t bytes_uploaded = s3_handle->buffer->GetSize();
487
488 s3fanout::JobInfo *info = new s3fanout::JobInfo(
489 final_path,
490 3030 const_cast<void *>(static_cast<void const *>(handle->commit_callback)),
491
2/4
✓ Branch 2 taken 3030 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 3030 times.
✗ Branch 6 not taken.
 3030 s3_handle->buffer.release());
492
493
1/2
✓ Branch 0 taken 3030 times.
✗ Branch 1 not taken.
 3030 if (peek_before_put_)
494 3030 info->request = s3fanout::JobInfo::kReqHeadPut;
495
1/2
✓ Branch 1 taken 3030 times.
✗ Branch 2 not taken.
 3030 UploadJobInfo(info);
496
497 // Remove the temporary file
498
1/2
✓ Branch 0 taken 3030 times.
✗ Branch 1 not taken.
 3030 delete s3_handle;
499
500 // Update statistics counters
501 3030 if (!content_hash.HasSuffix()
502
5/6
✓ Branch 0 taken 30 times.
✓ Branch 1 taken 3000 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 30 times.
✓ Branch 4 taken 3000 times.
✓ Branch 5 taken 30 times.
 3030 || content_hash.suffix == shash::kSuffixPartial) {
503
1/2
✓ Branch 1 taken 3000 times.
✗ Branch 2 not taken.
 3000 CountUploadedChunks();
504
1/2
✓ Branch 1 taken 3000 times.
✗ Branch 2 not taken.
 3000 CountUploadedBytes(bytes_uploaded);
505
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 30 times.
 30 } else if (content_hash.suffix == shash::kSuffixCatalog) {
506 ✗ CountUploadedCatalogs();
507 ✗ CountUploadedCatalogBytes(bytes_uploaded);
508 }
509 3030 }
510
511
512 150 s3fanout::JobInfo *S3Uploader::CreateJobInfo(const std::string &path) const {
513
2/4
✓ Branch 2 taken 150 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 150 times.
✗ Branch 6 not taken.
 150 FileBackedBuffer *buf = FileBackedBuffer::Create(kInMemoryObjectThreshold);
514
1/2
✓ Branch 2 taken 150 times.
✗ Branch 3 not taken.
 150 return new s3fanout::JobInfo(path, NULL, buf);
515 }
516
517
518 18030 void S3Uploader::DoRemoveAsync(const std::string &file_to_delete) {
519
2/4
✓ Branch 1 taken 18030 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 18030 times.
✗ Branch 5 not taken.
 18030 const std::string mangled_path = repository_alias_ + "/" + file_to_delete;
520
521
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 18030 times.
 18030 if (!batch_delete_enabled_) {
522 ✗ s3fanout::JobInfo *info = CreateJobInfo(mangled_path);
523 ✗ info->request = s3fanout::JobInfo::kReqDelete;
524 ✗ LogCvmfs(kLogUploadS3, kLogDebug, "Asynchronously removing %s/%s",
525 bucket_.c_str(), info->object_key.c_str());
526 ✗ s3fanout_mgr_->PushNewJob(info);
527 ✗ return;
528 }
529
530 // Batch delete: collect keys and flush when the S3 batch limit is reached.
531 //
532 // The caller (RemoveAsync) already incremented jobs_in_flight_ for this key.
533 // We undo that immediately: for batched deletes, jobs_in_flight_ is managed
534 // per-batch rather than per-key. FlushDeleteBatch() increments once when
535 // the batch is dispatched, and MainCollectResults decrements once when the
536 // batch response arrives. This avoids a deadlock where jobs_in_flight_
537 // saturates (e.g. at 512) before the batch threshold (1000) is reached,
538 // blocking forever on a flush that never happens.
539
1/2
✓ Branch 1 taken 18030 times.
✗ Branch 2 not taken.
 18030 DecJobsInFlight();
540 18030 const MutexLockGuard guard(delete_batch_mutex_);
541
1/2
✓ Branch 1 taken 18030 times.
✗ Branch 2 not taken.
 18030 pending_deletes_.push_back(mangled_path);
542
2/2
✓ Branch 1 taken 180 times.
✓ Branch 2 taken 17850 times.
 18030 if (pending_deletes_.size() >= batch_delete_size_) {
543
1/2
✓ Branch 1 taken 180 times.
✗ Branch 2 not taken.
 180 FlushDeleteBatch();
544 }
545
1/2
✓ Branch 2 taken 18030 times.
✗ Branch 3 not taken.
 18030 }
546
547
548 570 void S3Uploader::FlushDeleteBatch() const {
549 // Caller must hold delete_batch_mutex_
550
2/2
✓ Branch 1 taken 360 times.
✓ Branch 2 taken 210 times.
 570 if (pending_deletes_.empty())
551 360 return;
552
553
1/2
✓ Branch 2 taken 210 times.
✗ Branch 3 not taken.
 210 LogCvmfs(kLogUploadS3, kLogDebug, "Flushing batch delete of %lu objects",
554 pending_deletes_.size());
555
556 // Build XML request body
557
1/2
✓ Branch 1 taken 210 times.
✗ Branch 2 not taken.
 210 const std::string xml = s3fanout::ComposeDeleteMultiXml(pending_deletes_);
558
2/4
✓ Branch 2 taken 210 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 210 times.
✗ Branch 6 not taken.
 210 FileBackedBuffer *buf = FileBackedBuffer::Create(kInMemoryObjectThreshold);
559
1/2
✓ Branch 3 taken 210 times.
✗ Branch 4 not taken.
 210 buf->Append(xml.data(), xml.length());
560
1/2
✓ Branch 1 taken 210 times.
✗ Branch 2 not taken.
 210 buf->Commit();
561
562 // The object_key for multi-delete is empty (URL is bucket root + ?delete)
563
3/6
✓ Branch 2 taken 210 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 210 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 210 times.
✗ Branch 9 not taken.
 210 s3fanout::JobInfo *info = new s3fanout::JobInfo("", NULL, buf);
564 210 info->request = s3fanout::JobInfo::kReqDeleteMulti;
565 210 info->multi_delete_keys.swap(pending_deletes_);
566
567 // Track this batch as a single job in flight. MainCollectResults will
568 // call Respond() exactly once for kReqDeleteMulti to balance it.
569
1/2
✓ Branch 1 taken 210 times.
✗ Branch 2 not taken.
 210 IncJobsInFlight();
570
1/2
✓ Branch 2 taken 210 times.
✗ Branch 3 not taken.
 210 s3fanout_mgr_->PushNewJob(info);
571 210 }
572
573
574 390 void S3Uploader::WaitForUpload() const {
575 {
576 390 const MutexLockGuard guard(delete_batch_mutex_);
577
1/2
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
 390 FlushDeleteBatch();
578 390 }
579 390 AbstractUploader::WaitForUpload();
580 390 }
581
582
583 15360 void S3Uploader::OnReqComplete(const upload::UploaderResults &results,
584 RequestCtrl *ctrl) {
585 15360 ctrl->return_code = results.return_code;
586
2/2
✓ Branch 0 taken 15210 times.
✓ Branch 1 taken 150 times.
 15360 if (ctrl->callback_forward != NULL) {
587 // We are already in Respond() so we must not call it again
588 15210 const upload::UploaderResults fix_path(results.return_code,
589
1/2
✓ Branch 1 taken 15210 times.
✗ Branch 2 not taken.
 15210 ctrl->original_path);
590
1/2
✓ Branch 1 taken 15210 times.
✗ Branch 2 not taken.
 15210 (*(ctrl->callback_forward))(fix_path);
591
1/2
✓ Branch 0 taken 15210 times.
✗ Branch 1 not taken.
 15210 delete ctrl->callback_forward;
592 15210 ctrl->callback_forward = NULL;
593 15210 }
594 15360 char c = 'c';
595
1/2
✓ Branch 1 taken 15360 times.
✗ Branch 2 not taken.
 15360 WritePipe(ctrl->pipe_wait[1], &c, 1);
596 15360 }
597
598
599 150 bool S3Uploader::Peek(const std::string &path) {
600
2/4
✓ Branch 1 taken 150 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 150 times.
✗ Branch 5 not taken.
 150 const std::string mangled_path = repository_alias_ + "/" + path;
601
1/2
✓ Branch 1 taken 150 times.
✗ Branch 2 not taken.
 150 s3fanout::JobInfo *info = CreateJobInfo(mangled_path);
602
603 150 RequestCtrl req_ctrl;
604
1/2
✓ Branch 1 taken 150 times.
✗ Branch 2 not taken.
 150 MakePipe(req_ctrl.pipe_wait);
605 150 info->request = s3fanout::JobInfo::kReqHeadOnly;
606 150 info->callback = const_cast<void *>(static_cast<void const *>(
607
1/2
✓ Branch 1 taken 150 times.
✗ Branch 2 not taken.
 150 MakeClosure(&S3Uploader::OnReqComplete, this, &req_ctrl)));
608
609
1/2
✓ Branch 1 taken 150 times.
✗ Branch 2 not taken.
 150 IncJobsInFlight();
610
1/2
✓ Branch 1 taken 150 times.
✗ Branch 2 not taken.
 150 UploadJobInfo(info);
611
1/2
✓ Branch 1 taken 150 times.
✗ Branch 2 not taken.
 150 req_ctrl.WaitFor();
612
613 150 return req_ctrl.return_code == 0;
614 150 }
615
616
617 // noop: no mkdir needed in S3 storage
618 ✗ bool S3Uploader::Mkdir(const std::string &path) { return true; }
619
620
621 ✗ bool S3Uploader::PlaceBootstrappingShortcut(const shash::Any &object) {
622 ✗ return false; // TODO(rmeusel): implement
623 }
624
625
626 ✗ int64_t S3Uploader::DoGetObjectSize(const std::string &file_name) {
627 // TODO(dosarudaniel): use a head request for byte count
628 // Re-enable 661 integration test when done
629 ✗ return -EOPNOTSUPP;
630 }
631
632 } // namespace upload
633