GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/network/s3fanout.cc
Date: 2026-10-11 02:40:02
Exec Total Coverage
Lines: 649 919 70.6%
Branches: 469 1337 35.1%
Line Branch Exec Source
1 /**
2 * This file is part of the CernVM File System.
3 *
4 * Runs a thread using libcurls asynchronous I/O mode to push data to S3
5 */
6
7 #include "s3fanout.h"
8
9 #include <pthread.h>
10
11 #include <algorithm>
12 #include <cassert>
13 #include <cerrno>
14 #include <cstring>
15 #include <utility>
16
17 #include "crypto/hash.h"
18 #include "util/exception.h"
19 #include "util/platform.h"
20 #include "util/posix.h"
21 #include "util/string.h"
22
23 using namespace std; // NOLINT
24
25 namespace s3fanout {
26
27 /**
28 * Escapes characters that are not allowed in XML text content.
29 * Only & and < need escaping; >, ', " are safe in text nodes.
30 */
31 26577 static string XmlEscape(const string &input) {
32 26577 string result;
33
1/2
✓ Branch 2 taken 26577 times.
✗ Branch 3 not taken.
 26577 result.reserve(input.size());
34
2/2
✓ Branch 1 taken 605658 times.
✓ Branch 2 taken 26577 times.
 632235 for (unsigned i = 0; i < input.size(); ++i) {
35
3/3
✓ Branch 1 taken 19 times.
✓ Branch 2 taken 19 times.
✓ Branch 3 taken 605620 times.
 605658 switch (input[i]) {
36 19 case '&':
37
1/2
✓ Branch 1 taken 19 times.
✗ Branch 2 not taken.
 19 result += "&amp;";
38 19 break;
39 19 case '<':
40
1/2
✓ Branch 1 taken 19 times.
✗ Branch 2 not taken.
 19 result += "&lt;";
41 19 break;
42 605620 default:
43
1/2
✓ Branch 2 taken 605620 times.
✗ Branch 3 not taken.
 605620 result += input[i];
44 605620 break;
45 }
46 }
47 26577 return result;
48 }
49
50
51 /**
52 * The CanonicalizedResource of an S3 V2 signature. It must mirror MkUrl():
53 * the server rebuilds the resource from the request path, so any difference,
54 * down to a trailing slash, yields SignatureDoesNotMatch.
55 *
56 * With DNS buckets the bucket lives in the hostname and the path is
57 * "/" + object_key, so an empty key gives "/<bucket>/". In path style the
58 * path already carries the bucket and an empty key gives "/<bucket>".
59 */
60 54146 string MkV2CanonicalResource(const string &bucket, const string &object_key,
61 bool dns_buckets, bool multi_delete) {
62 54146 string resource = "/" + bucket;
63
6/6
✓ Branch 0 taken 54089 times.
✓ Branch 1 taken 57 times.
✓ Branch 3 taken 53743 times.
✓ Branch 4 taken 346 times.
✓ Branch 5 taken 53800 times.
✓ Branch 6 taken 346 times.
 54146 if (dns_buckets || !object_key.empty())
64
2/4
✓ Branch 1 taken 53800 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 53800 times.
✗ Branch 5 not taken.
 53800 resource += "/" + object_key;
65 // V2 requires subresources to be signed; multi-delete posts to "?delete"
66
2/2
✓ Branch 0 taken 346 times.
✓ Branch 1 taken 53800 times.
 54146 if (multi_delete)
67
1/2
✓ Branch 1 taken 346 times.
✗ Branch 2 not taken.
 346 resource += "?delete";
68 54146 return resource;
69 }
70
71
72 /**
73 * Builds an S3 DeleteObjects XML request body for multi-object delete.
74 * Uses Quiet mode so the response only contains errors, not successes.
75 */
76 384 string ComposeDeleteMultiXml(const vector<string> &keys) {
77 string xml = "<?xml version=\"1.0\" "
78
1/2
✓ Branch 2 taken 384 times.
✗ Branch 3 not taken.
 384 "encoding=\"UTF-8\"?><Delete><Quiet>true</Quiet>";
79 // ~70 bytes per <Object><Key>...</Key></Object> entry
80
1/2
✓ Branch 3 taken 384 times.
✗ Branch 4 not taken.
 384 xml.reserve(xml.size() + keys.size() * 70 + 10);
81
2/2
✓ Branch 1 taken 26577 times.
✓ Branch 2 taken 384 times.
 26961 for (unsigned i = 0; i < keys.size(); ++i) {
82
4/8
✓ Branch 2 taken 26577 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 26577 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 26577 times.
✗ Branch 9 not taken.
✓ Branch 11 taken 26577 times.
✗ Branch 12 not taken.
 26577 xml += "<Object><Key>" + XmlEscape(keys[i]) + "</Key></Object>";
83 }
84
1/2
✓ Branch 1 taken 384 times.
✗ Branch 2 not taken.
 384 xml += "</Delete>";
85 384 return xml;
86 }
87
88
89 /**
90 * Parses the S3 DeleteObjects error response XML.
91 * In Quiet mode, the response only contains <Error> elements for failed keys.
92 * Returns the number of errors found.
93 */
94 76 unsigned ParseDeleteMultiResponse(const string &response,
95 vector<string> *error_keys,
96 vector<string> *error_codes,
97 vector<string> *error_messages) {
98 76 unsigned num_errors = 0;
99 76 string::size_type pos = 0;
100
101 while (true) {
102 133 const string::size_type err_start = response.find("<Error>", pos);
103
2/2
✓ Branch 0 taken 57 times.
✓ Branch 1 taken 76 times.
 133 if (err_start == string::npos)
104 57 break;
105 76 const string::size_type err_end = response.find("</Error>", err_start);
106
2/2
✓ Branch 0 taken 19 times.
✓ Branch 1 taken 57 times.
 76 if (err_end == string::npos)
107 19 break;
108
109
1/2
✓ Branch 1 taken 57 times.
✗ Branch 2 not taken.
 57 const string error_block = response.substr(err_start, err_end - err_start);
110 57 num_errors++;
111
112 // Extract <Key>...</Key>
113 57 const string::size_type key_start = error_block.find("<Key>");
114 57 const string::size_type key_end = error_block.find("</Key>");
115
2/4
✓ Branch 0 taken 57 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 57 times.
✗ Branch 3 not taken.
 57 if (key_start != string::npos && key_end != string::npos) {
116
1/2
✓ Branch 1 taken 57 times.
✗ Branch 2 not taken.
 57 error_keys->push_back(
117
1/2
✓ Branch 1 taken 57 times.
✗ Branch 2 not taken.
 114 error_block.substr(key_start + 5, key_end - key_start - 5));
118 } else {
119 ✗ error_keys->push_back("");
120 }
121
122 // Extract <Code>...</Code>
123 57 const string::size_type code_start = error_block.find("<Code>");
124 57 const string::size_type code_end = error_block.find("</Code>");
125
2/4
✓ Branch 0 taken 57 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 57 times.
✗ Branch 3 not taken.
 57 if (code_start != string::npos && code_end != string::npos) {
126
1/2
✓ Branch 1 taken 57 times.
✗ Branch 2 not taken.
 57 error_codes->push_back(
127
1/2
✓ Branch 1 taken 57 times.
✗ Branch 2 not taken.
 114 error_block.substr(code_start + 6, code_end - code_start - 6));
128 } else {
129 ✗ error_codes->push_back("");
130 }
131
132 // Extract <Message>...</Message>
133 57 const string::size_type msg_start = error_block.find("<Message>");
134 57 const string::size_type msg_end = error_block.find("</Message>");
135
2/4
✓ Branch 0 taken 57 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 57 times.
✗ Branch 3 not taken.
 57 if (msg_start != string::npos && msg_end != string::npos) {
136
1/2
✓ Branch 1 taken 57 times.
✗ Branch 2 not taken.
 57 error_messages->push_back(
137
1/2
✓ Branch 1 taken 57 times.
✗ Branch 2 not taken.
 114 error_block.substr(msg_start + 9, msg_end - msg_start - 9));
138 } else {
139 ✗ error_messages->push_back("");
140 }
141
142 57 pos = err_end + 8; // length of "</Error>"
143 57 }
144
145 76 return num_errors;
146 }
147
148 const char *S3FanoutManager::kCacheControlCas = "Cache-Control: max-age=259200";
149 const unsigned S3FanoutManager::kDefault429ThrottleMs = 250;
150 const unsigned S3FanoutManager::kMax429ThrottleMs = 10000;
151 const unsigned S3FanoutManager::kThrottleReportIntervalSec = 10;
152 const unsigned S3FanoutManager::kDefaultHTTPPort = 80;
153 const unsigned S3FanoutManager::kDefaultHTTPSPort = 443;
154
155 572 std::string S3FanoutManager::MkDotCvmfsCacheControlHeader(
156 unsigned defaultMaxAge, int overrideMaxAge) {
157 const char *var;
158 int max_age_sec;
159 char *at_null_terminator_if_number;
160 572 bool value_determined = false;
161
162
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 572 times.
 572 if (overrideMaxAge >= 0) {
163 ✗ max_age_sec = overrideMaxAge;
164 ✗ value_determined = true;
165 }
166
167
1/2
✓ Branch 0 taken 572 times.
✗ Branch 1 not taken.
 572 if (!value_determined) {
168 572 var = getenv("CVMFS_MAX_TTL_SECS");
169
1/4
✗ Branch 0 not taken.
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 572 if (var && var[0]) {
170 ✗ max_age_sec = strtoll(var, &at_null_terminator_if_number, 10);
171 ✗ if (*at_null_terminator_if_number == '\0' && max_age_sec >= 0) {
172 ✗ value_determined = true;
173 }
174 }
175 }
176
177
1/2
✓ Branch 0 taken 572 times.
✗ Branch 1 not taken.
 572 if (!value_determined) {
178 572 var = getenv("CVMFS_MAX_TTL");
179
1/4
✗ Branch 0 not taken.
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 572 if (var && var[0]) {
180 ✗ max_age_sec = strtoll(var, &at_null_terminator_if_number, 10);
181 ✗ if (*at_null_terminator_if_number == '\0' && max_age_sec >= 0) {
182 ✗ max_age_sec *= 60;
183 ✗ value_determined = true;
184 }
185 }
186 }
187
188
1/2
✓ Branch 0 taken 572 times.
✗ Branch 1 not taken.
 572 if (!value_determined) {
189 572 max_age_sec = defaultMaxAge;
190 }
191
192
2/4
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 572 times.
✗ Branch 5 not taken.
 1144 return "Cache-Control: max-age=" + std::to_string(max_age_sec);
193 }
194
195 /**
196 * Parses Retry-After and X-Retry-In headers attached to HTTP 429 responses
197 */
198 794 void S3FanoutManager::DetectThrottleIndicator(const std::string &header,
199 JobInfo *info) {
200 794 std::string value_str;
201
4/6
✓ Branch 2 taken 794 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 794 times.
✗ Branch 6 not taken.
✓ Branch 9 taken 309 times.
✓ Branch 10 taken 485 times.
 794 if (HasPrefix(header, "retry-after:", true))
202
1/2
✓ Branch 1 taken 309 times.
✗ Branch 2 not taken.
 309 value_str = header.substr(12);
203
4/6
✓ Branch 2 taken 794 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 794 times.
✗ Branch 6 not taken.
✓ Branch 9 taken 95 times.
✓ Branch 10 taken 699 times.
 794 if (HasPrefix(header, "x-retry-in:", true))
204
1/2
✓ Branch 1 taken 95 times.
✗ Branch 2 not taken.
 95 value_str = header.substr(11);
205
206
1/2
✓ Branch 1 taken 794 times.
✗ Branch 2 not taken.
 794 value_str = Trim(value_str, true /* trim_newline */);
207
2/2
✓ Branch 1 taken 366 times.
✓ Branch 2 taken 428 times.
 794 if (!value_str.empty()) {
208
1/2
✓ Branch 1 taken 366 times.
✗ Branch 2 not taken.
 366 const unsigned value_numeric = String2Uint64(value_str);
209
2/4
✓ Branch 2 taken 366 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 366 times.
✗ Branch 6 not taken.
 732 const unsigned value_ms = HasSuffix(value_str, "ms", true /* ignore_case */)
210
2/2
✓ Branch 0 taken 95 times.
✓ Branch 1 taken 271 times.
 366 ? value_numeric
211 366 : (value_numeric * 1000);
212
2/2
✓ Branch 0 taken 347 times.
✓ Branch 1 taken 19 times.
 366 if (value_ms > 0)
213 347 info->throttle_ms = std::min(value_ms, kMax429ThrottleMs);
214 }
215 794 }
216
217
218 /**
219 * Called by curl for every HTTP header. Not called for file:// transfers.
220 */
221 162800 static size_t CallbackCurlHeader(void *ptr, size_t size, size_t nmemb,
222 void *info_link) {
223 162800 const size_t num_bytes = size * nmemb;
224
1/2
✓ Branch 2 taken 162800 times.
✗ Branch 3 not taken.
 162800 const string header_line(static_cast<const char *>(ptr), num_bytes);
225 162800 JobInfo *info = static_cast<JobInfo *>(info_link);
226
227 // Check for http status code errors
228
4/6
✓ Branch 2 taken 162800 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 162800 times.
✗ Branch 6 not taken.
✓ Branch 9 taken 54208 times.
✓ Branch 10 taken 108592 times.
 162800 if (HasPrefix(header_line, "HTTP/1.", false)) {
229
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 54208 times.
 54208 if (header_line.length() < 10)
230 ✗ return 0;
231
232 unsigned i;
233
5/6
✓ Branch 1 taken 108416 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 54208 times.
✓ Branch 5 taken 54208 times.
✓ Branch 6 taken 54208 times.
✓ Branch 7 taken 54208 times.
 108416 for (i = 8; (i < header_line.length()) && (header_line[i] == ' '); ++i) {
234 }
235
236
2/2
✓ Branch 1 taken 27192 times.
✓ Branch 2 taken 27016 times.
 54208 if (header_line[i] == '2') {
237 27192 return num_bytes;
238 } else {
239
1/2
✓ Branch 2 taken 27016 times.
✗ Branch 3 not taken.
 27016 LogCvmfs(kLogS3Fanout, kLogDebug, "http status error code [info %p]: %s",
240 info, header_line.c_str());
241
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 27016 times.
 27016 if (header_line.length() < i + 3) {
242 ✗ LogCvmfs(kLogS3Fanout, kLogStderr, "S3: invalid HTTP response '%s'",
243 header_line.c_str());
244 ✗ info->error_code = kFailOther;
245 ✗ return 0;
246 }
247
2/4
✓ Branch 3 taken 27016 times.
✗ Branch 4 not taken.
✓ Branch 6 taken 27016 times.
✗ Branch 7 not taken.
 27016 info->http_error = String2Int64(string(&header_line[i], 3));
248
249
2/7
✓ Branch 0 taken 176 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✓ Branch 5 taken 26840 times.
✗ Branch 6 not taken.
 27016 switch (info->http_error) {
250 176 case 429:
251 176 info->error_code = kFailRetry;
252 176 info->throttle_ms = S3FanoutManager::kDefault429ThrottleMs;
253 176 info->throttle_timestamp = platform_monotonic_time();
254 176 return num_bytes;
255 ✗ case 507: // Insufficient Storage
256 ✗ info->error_code = kFailInsufficientStorage;
257 ✗ break;
258 ✗ case 503:
259 case 502: // Can happen if the S3 gateway-backend connection breaks
260 case 500: // sometimes see this as a transient error from S3
261 ✗ info->error_code = kFailServiceUnavailable;
262 ✗ break;
263 ✗ case 501:
264 case 400:
265 ✗ info->error_code = kFailBadRequest;
266 ✗ break;
267 ✗ case 403:
268 ✗ info->error_code = kFailForbidden;
269 ✗ break;
270 26840 case 404:
271 26840 info->error_code = kFailNotFound;
272 26840 return num_bytes;
273 ✗ default:
274 ✗ info->error_code = kFailOther;
275 }
276 ✗ return 0;
277 }
278 }
279
280
2/2
✓ Branch 0 taken 528 times.
✓ Branch 1 taken 108064 times.
 108592 if (info->error_code == kFailRetry) {
281
1/2
✓ Branch 1 taken 528 times.
✗ Branch 2 not taken.
 528 S3FanoutManager::DetectThrottleIndicator(header_line, info);
282 }
283
284 108592 return num_bytes;
285 162800 }
286
287
288 /**
289 * Called by curl for every new chunk to upload.
290 */
291 689744 static size_t CallbackCurlData(void *ptr, size_t size, size_t nmemb,
292 void *info_link) {
293 689744 const size_t num_bytes = size * nmemb;
294 689744 JobInfo *info = static_cast<JobInfo *>(info_link);
295
296 689744 LogCvmfs(kLogS3Fanout, kLogDebug, "Data callback with %zu bytes", num_bytes);
297
298
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 689744 times.
 689744 if (num_bytes == 0)
299 ✗ return 0;
300
301 689744 const uint64_t read_bytes = info->origin->Read(ptr, num_bytes);
302
303 689744 LogCvmfs(kLogS3Fanout, kLogDebug, "source buffer pushed out %lu bytes",
304 read_bytes);
305
306 689744 return read_bytes;
307 }
308
309
310 /**
311 * Captures the HTTP response body. For kReqDeleteMulti, the response contains
312 * XML error information. For other requests, the body is ignored.
313 */
314 ✗ static size_t CallbackCurlBody(char *ptr, size_t size, size_t nmemb,
315 void *userdata) {
316 ✗ const size_t num_bytes = size * nmemb;
317 ✗ JobInfo *info = static_cast<JobInfo *>(userdata);
318 ✗ if (info != NULL && info->request == JobInfo::kReqDeleteMulti) {
319 ✗ info->response_body.append(ptr, num_bytes);
320 }
321 ✗ return num_bytes;
322 }
323
324
325 /**
326 * Called when new curl sockets arrive or existing curl sockets depart.
327 */
328 120164 int S3FanoutManager::CallbackCurlSocket(CURL *easy, curl_socket_t s, int action,
329 void *userp, void *socketp) {
330 120164 S3FanoutManager *s3fanout_mgr = static_cast<S3FanoutManager *>(userp);
331 120164 LogCvmfs(kLogS3Fanout, kLogDebug,
332 "CallbackCurlSocket called with easy "
333 "handle %p, socket %d, action %d, up %p, "
334 "sp %p, fds_inuse %d, jobs %d",
335 easy, s, action, userp, socketp, s3fanout_mgr->watch_fds_inuse_,
336 120164 s3fanout_mgr->available_jobs_->Get());
337
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 120164 times.
 120164 if (action == CURL_POLL_NONE)
338 ✗ return 0;
339
340 // Find s in watch_fds_
341 // First 2 fds are job and terminate pipes (not curl related)
342 unsigned index;
343
2/2
✓ Branch 0 taken 187968 times.
✓ Branch 1 taken 54208 times.
 242176 for (index = 2; index < s3fanout_mgr->watch_fds_inuse_; ++index) {
344
2/2
✓ Branch 0 taken 65956 times.
✓ Branch 1 taken 122012 times.
 187968 if (s3fanout_mgr->watch_fds_[index].fd == s)
345 65956 break;
346 }
347 // Or create newly
348
2/2
✓ Branch 0 taken 54208 times.
✓ Branch 1 taken 65956 times.
 120164 if (index == s3fanout_mgr->watch_fds_inuse_) {
349 // Extend array if necessary
350
2/2
✓ Branch 0 taken 132 times.
✓ Branch 1 taken 54076 times.
 54208 if (s3fanout_mgr->watch_fds_inuse_ == s3fanout_mgr->watch_fds_size_) {
351 132 s3fanout_mgr->watch_fds_size_ *= 2;
352 132 s3fanout_mgr->watch_fds_ = static_cast<struct pollfd *>(
353 132 srealloc(s3fanout_mgr->watch_fds_,
354 132 s3fanout_mgr->watch_fds_size_ * sizeof(struct pollfd)));
355 }
356 54208 s3fanout_mgr->watch_fds_[s3fanout_mgr->watch_fds_inuse_].fd = s;
357 54208 s3fanout_mgr->watch_fds_[s3fanout_mgr->watch_fds_inuse_].events = 0;
358 54208 s3fanout_mgr->watch_fds_[s3fanout_mgr->watch_fds_inuse_].revents = 0;
359 54208 s3fanout_mgr->watch_fds_inuse_++;
360 }
361
362
4/5
✓ Branch 0 taken 54208 times.
✓ Branch 1 taken 132 times.
✓ Branch 2 taken 11616 times.
✓ Branch 3 taken 54208 times.
✗ Branch 4 not taken.
 120164 switch (action) {
363 54208 case CURL_POLL_IN:
364 54208 s3fanout_mgr->watch_fds_[index].events = POLLIN | POLLPRI;
365 54208 break;
366 132 case CURL_POLL_OUT:
367 132 s3fanout_mgr->watch_fds_[index].events = POLLOUT | POLLWRBAND;
368 132 break;
369 11616 case CURL_POLL_INOUT:
370 11616 s3fanout_mgr->watch_fds_[index].events = POLLIN | POLLPRI | POLLOUT
371 | POLLWRBAND;
372 11616 break;
373 54208 case CURL_POLL_REMOVE:
374
2/2
✓ Branch 0 taken 8712 times.
✓ Branch 1 taken 45496 times.
 54208 if (index < s3fanout_mgr->watch_fds_inuse_ - 1)
375 s3fanout_mgr
376 8712 ->watch_fds_[index] = s3fanout_mgr->watch_fds_
377 8712 [s3fanout_mgr->watch_fds_inuse_ - 1];
378 54208 s3fanout_mgr->watch_fds_inuse_--;
379 // Shrink array if necessary
380
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 54208 times.
 54208 if ((s3fanout_mgr->watch_fds_inuse_ > s3fanout_mgr->watch_fds_max_)
381 ✗ && (s3fanout_mgr->watch_fds_inuse_
382 ✗ < s3fanout_mgr->watch_fds_size_ / 2)) {
383 ✗ s3fanout_mgr->watch_fds_size_ /= 2;
384 ✗ s3fanout_mgr->watch_fds_ = static_cast<struct pollfd *>(
385 ✗ srealloc(s3fanout_mgr->watch_fds_,
386 ✗ s3fanout_mgr->watch_fds_size_ * sizeof(struct pollfd)));
387 }
388 54208 break;
389 ✗ default:
390 ✗ PANIC(NULL);
391 }
392
393 120164 return 0;
394 }
395
396
397 /**
398 * Worker thread event loop.
399 */
400 572 void *S3FanoutManager::MainUpload(void *data) {
401
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 LogCvmfs(kLogS3Fanout, kLogDebug, "Upload I/O thread started");
402 572 S3FanoutManager *s3fanout_mgr = static_cast<S3FanoutManager *>(data);
403
404 572 s3fanout_mgr->InitPipeWatchFds();
405
406 // Don't schedule more jobs into the multi handle than the maximum number of
407 // parallel connections. This should prevent starvation and thus a timeout
408 // of the authorization header (CVM-1339).
409 572 unsigned jobs_in_flight = 0;
410
411 while (true) {
412 // Check events with 100ms timeout
413 731412 const int timeout_ms = 100;
414
1/2
✓ Branch 1 taken 731412 times.
✗ Branch 2 not taken.
 731412 int retval = poll(s3fanout_mgr->watch_fds_, s3fanout_mgr->watch_fds_inuse_,
415 timeout_ms);
416
2/2
✓ Branch 0 taken 4356 times.
✓ Branch 1 taken 727056 times.
 731412 if (retval == 0) {
417 // Handle timeout
418 4356 int still_running = 0;
419
1/2
✓ Branch 1 taken 4356 times.
✗ Branch 2 not taken.
 4356 retval = curl_multi_socket_action(s3fanout_mgr->curl_multi_,
420 CURL_SOCKET_TIMEOUT, 0, &still_running);
421
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 4356 times.
 4356 if (retval != CURLM_OK) {
422 ✗ LogCvmfs(kLogS3Fanout, kLogStderr, "Error, timeout due to: %d", retval);
423 ✗ assert(retval == CURLM_OK);
424 }
425
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 727056 times.
 727056 } else if (retval < 0) {
426 ✗ assert(errno == EINTR);
427 ✗ continue;
428 }
429
430 // Terminate I/O thread
431
2/2
✓ Branch 0 taken 572 times.
✓ Branch 1 taken 730840 times.
 731412 if (s3fanout_mgr->watch_fds_[0].revents)
432 572 break;
433
434 // New job incoming
435
2/2
✓ Branch 0 taken 27280 times.
✓ Branch 1 taken 703560 times.
 730840 if (s3fanout_mgr->watch_fds_[1].revents) {
436 27280 s3fanout_mgr->watch_fds_[1].revents = 0;
437 JobInfo *info;
438
1/2
✓ Branch 1 taken 27280 times.
✗ Branch 2 not taken.
 27280 ReadPipe(s3fanout_mgr->pipe_jobs_[0], &info, sizeof(info));
439
1/2
✓ Branch 1 taken 27280 times.
✗ Branch 2 not taken.
 27280 CURL *handle = s3fanout_mgr->AcquireCurlHandle();
440
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 27280 times.
 27280 if (handle == NULL) {
441 ✗ PANIC(kLogStderr, "Failed to acquire CURL handle.");
442 }
443
1/2
✓ Branch 1 taken 27280 times.
✗ Branch 2 not taken.
 27280 const s3fanout::Failures init_failure = s3fanout_mgr->InitializeRequest(
444 info, handle);
445
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 27280 times.
 27280 if (init_failure != s3fanout::kFailOk) {
446 ✗ PANIC(kLogStderr,
447 "Failed to initialize CURL handle (error: %d - %s | errno: %d)",
448 init_failure, Code2Ascii(init_failure), errno);
449 }
450
1/2
✓ Branch 1 taken 27280 times.
✗ Branch 2 not taken.
 27280 s3fanout_mgr->SetUrlOptions(info);
451
452
1/2
✓ Branch 1 taken 27280 times.
✗ Branch 2 not taken.
 27280 curl_multi_add_handle(s3fanout_mgr->curl_multi_, handle);
453
1/2
✓ Branch 1 taken 27280 times.
✗ Branch 2 not taken.
 27280 s3fanout_mgr->active_requests_->insert(info);
454 27280 jobs_in_flight++;
455 27280 int still_running = 0, retval = 0;
456
1/2
✓ Branch 1 taken 27280 times.
✗ Branch 2 not taken.
 27280 retval = curl_multi_socket_action(s3fanout_mgr->curl_multi_,
457 CURL_SOCKET_TIMEOUT, 0, &still_running);
458
459
1/2
✓ Branch 1 taken 27280 times.
✗ Branch 2 not taken.
 27280 LogCvmfs(kLogS3Fanout, kLogDebug, "curl_multi_socket_action: %d - %d",
460 retval, still_running);
461 }
462
463
464 // Activity on curl sockets
465 // Within this loop the curl_multi_socket_action() may cause socket(s)
466 // to be removed from watch_fds_. If a socket is removed it is replaced
467 // by the socket at the end of the array and the inuse count is decreased.
468 // Therefore loop over the array in reverse order.
469 // First 2 fds are job and terminate pipes (not curl related)
470
2/2
✓ Branch 0 taken 1607628 times.
✓ Branch 1 taken 730840 times.
 2338468 for (int32_t i = s3fanout_mgr->watch_fds_inuse_ - 1; i >= 2; --i) {
471
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1607628 times.
 1607628 if (static_cast<uint32_t>(i) >= s3fanout_mgr->watch_fds_inuse_) {
472 ✗ continue;
473 }
474
2/2
✓ Branch 0 taken 751476 times.
✓ Branch 1 taken 856152 times.
 1607628 if (s3fanout_mgr->watch_fds_[i].revents) {
475 751476 int ev_bitmask = 0;
476
2/2
✓ Branch 0 taken 81268 times.
✓ Branch 1 taken 670208 times.
 751476 if (s3fanout_mgr->watch_fds_[i].revents & (POLLIN | POLLPRI))
477 81268 ev_bitmask |= CURL_CSELECT_IN;
478
2/2
✓ Branch 0 taken 670208 times.
✓ Branch 1 taken 81268 times.
 751476 if (s3fanout_mgr->watch_fds_[i].revents & (POLLOUT | POLLWRBAND))
479 670208 ev_bitmask |= CURL_CSELECT_OUT;
480 751476 if (s3fanout_mgr->watch_fds_[i].revents
481
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 751476 times.
 751476 & (POLLERR | POLLHUP | POLLNVAL))
482 ✗ ev_bitmask |= CURL_CSELECT_ERR;
483 751476 s3fanout_mgr->watch_fds_[i].revents = 0;
484
485 751476 int still_running = 0;
486 751476 retval = curl_multi_socket_action(s3fanout_mgr->curl_multi_,
487
1/2
✓ Branch 1 taken 751476 times.
✗ Branch 2 not taken.
 751476 s3fanout_mgr->watch_fds_[i].fd,
488 ev_bitmask,
489 &still_running);
490 }
491 }
492
493 // Check if transfers are completed
494 CURLMsg *curl_msg;
495 int msgs_in_queue;
496
3/4
✓ Branch 1 taken 785048 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 54208 times.
✓ Branch 4 taken 730840 times.
 785048 while ((curl_msg = curl_multi_info_read(s3fanout_mgr->curl_multi_,
497 &msgs_in_queue))) {
498
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 54208 times.
 54208 assert(curl_msg->msg == CURLMSG_DONE);
499
500 54208 s3fanout_mgr->statistics_->num_requests++;
501 JobInfo *info;
502 54208 CURL *easy_handle = curl_msg->easy_handle;
503 54208 const int curl_error = curl_msg->data.result;
504
1/2
✓ Branch 1 taken 54208 times.
✗ Branch 2 not taken.
 54208 curl_easy_getinfo(easy_handle, CURLINFO_PRIVATE, &info);
505
506
1/2
✓ Branch 1 taken 54208 times.
✗ Branch 2 not taken.
 54208 curl_multi_remove_handle(s3fanout_mgr->curl_multi_, easy_handle);
507
3/4
✓ Branch 1 taken 54208 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 26928 times.
✓ Branch 4 taken 27280 times.
 54208 if (s3fanout_mgr->VerifyAndFinalize(curl_error, info)) {
508
1/2
✓ Branch 1 taken 26928 times.
✗ Branch 2 not taken.
 26928 curl_multi_add_handle(s3fanout_mgr->curl_multi_, easy_handle);
509 26928 int still_running = 0;
510
1/2
✓ Branch 1 taken 26928 times.
✗ Branch 2 not taken.
 26928 curl_multi_socket_action(s3fanout_mgr->curl_multi_, CURL_SOCKET_TIMEOUT,
511 0, &still_running);
512 } else {
513 // Return easy handle into pool and write result back
514 27280 jobs_in_flight--;
515
1/2
✓ Branch 1 taken 27280 times.
✗ Branch 2 not taken.
 27280 s3fanout_mgr->active_requests_->erase(info);
516
1/2
✓ Branch 1 taken 27280 times.
✗ Branch 2 not taken.
 27280 s3fanout_mgr->ReleaseCurlHandle(info, easy_handle);
517
1/2
✓ Branch 1 taken 27280 times.
✗ Branch 2 not taken.
 27280 s3fanout_mgr->available_jobs_->Decrement();
518
519 // Add to list of completed jobs
520
1/2
✓ Branch 1 taken 27280 times.
✗ Branch 2 not taken.
 27280 s3fanout_mgr->PushCompletedJob(info);
521 }
522 }
523 730840 }
524
525 572 set<CURL *>::iterator i = s3fanout_mgr->pool_handles_inuse_->begin();
526 572 const set<CURL *>::const_iterator i_end = s3fanout_mgr->pool_handles_inuse_
527 572 ->end();
528
1/2
✗ Branch 2 not taken.
✓ Branch 3 taken 572 times.
 572 for (; i != i_end; ++i) {
529 ✗ curl_multi_remove_handle(s3fanout_mgr->curl_multi_, *i);
530 ✗ curl_easy_cleanup(*i);
531 }
532 572 s3fanout_mgr->pool_handles_inuse_->clear();
533 572 free(s3fanout_mgr->watch_fds_);
534
535
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 LogCvmfs(kLogS3Fanout, kLogDebug, "Upload I/O thread terminated");
536 572 return NULL;
537 }
538
539
540 /**
541 * Gets an idle CURL handle from the pool. Creates a new one and adds it to
542 * the pool if necessary.
543 */
544 27280 CURL *S3FanoutManager::AcquireCurlHandle() const {
545 CURL *handle;
546
547 27280 const MutexLockGuard guard(curl_handle_lock_);
548
549
2/2
✓ Branch 1 taken 2420 times.
✓ Branch 2 taken 24860 times.
 27280 if (pool_handles_idle_->empty()) {
550 CURLcode retval;
551
552 // Create a new handle
553
1/2
✓ Branch 1 taken 2420 times.
✗ Branch 2 not taken.
 2420 handle = curl_easy_init();
554
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2420 times.
 2420 assert(handle != NULL);
555
556 // Other settings
557
1/2
✓ Branch 1 taken 2420 times.
✗ Branch 2 not taken.
 2420 retval = curl_easy_setopt(handle, CURLOPT_NOSIGNAL, 1);
558
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2420 times.
 2420 assert(retval == CURLE_OK);
559
1/2
✓ Branch 1 taken 2420 times.
✗ Branch 2 not taken.
 2420 retval = curl_easy_setopt(handle, CURLOPT_HEADERFUNCTION,
560 CallbackCurlHeader);
561
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2420 times.
 2420 assert(retval == CURLE_OK);
562
1/2
✓ Branch 1 taken 2420 times.
✗ Branch 2 not taken.
 2420 retval = curl_easy_setopt(handle, CURLOPT_READFUNCTION, CallbackCurlData);
563
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2420 times.
 2420 assert(retval == CURLE_OK);
564
1/2
✓ Branch 1 taken 2420 times.
✗ Branch 2 not taken.
 2420 retval = curl_easy_setopt(handle, CURLOPT_WRITEFUNCTION, CallbackCurlBody);
565
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 2420 times.
 2420 assert(retval == CURLE_OK);
566 // WRITEDATA is set per-request in InitializeRequest
567 } else {
568 24860 handle = *(pool_handles_idle_->begin());
569
1/2
✓ Branch 2 taken 24860 times.
✗ Branch 3 not taken.
 24860 pool_handles_idle_->erase(pool_handles_idle_->begin());
570 }
571
572
1/2
✓ Branch 1 taken 27280 times.
✗ Branch 2 not taken.
 27280 pool_handles_inuse_->insert(handle);
573
574 27280 return handle;
575 27280 }
576
577
578 27280 void S3FanoutManager::ReleaseCurlHandle(JobInfo *info, CURL *handle) const {
579
1/2
✓ Branch 0 taken 27280 times.
✗ Branch 1 not taken.
 27280 if (info->http_headers) {
580
1/2
✓ Branch 1 taken 27280 times.
✗ Branch 2 not taken.
 27280 curl_slist_free_all(info->http_headers);
581 27280 info->http_headers = NULL;
582 }
583
584 27280 const MutexLockGuard guard(curl_handle_lock_);
585
586
1/2
✓ Branch 1 taken 27280 times.
✗ Branch 2 not taken.
 27280 const set<CURL *>::iterator elem = pool_handles_inuse_->find(handle);
587
1/2
✗ Branch 2 not taken.
✓ Branch 3 taken 27280 times.
 27280 assert(elem != pool_handles_inuse_->end());
588
589
2/2
✓ Branch 1 taken 1276 times.
✓ Branch 2 taken 26004 times.
 27280 if (pool_handles_idle_->size() > config_.pool_max_handles) {
590
1/2
✓ Branch 1 taken 1276 times.
✗ Branch 2 not taken.
 1276 const CURLcode retval = curl_easy_setopt(handle, CURLOPT_SHARE, NULL);
591
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1276 times.
 1276 assert(retval == CURLE_OK);
592
1/2
✓ Branch 1 taken 1276 times.
✗ Branch 2 not taken.
 1276 curl_easy_cleanup(handle);
593 const std::map<CURL *, S3FanOutDnsEntry *>::size_type
594
1/2
✓ Branch 1 taken 1276 times.
✗ Branch 2 not taken.
 1276 retitems = curl_sharehandles_->erase(handle);
595
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1276 times.
 1276 assert(retitems == 1);
596 } else {
597
1/2
✓ Branch 1 taken 26004 times.
✗ Branch 2 not taken.
 26004 pool_handles_idle_->insert(handle);
598 }
599
600
1/2
✓ Branch 1 taken 27280 times.
✗ Branch 2 not taken.
 27280 pool_handles_inuse_->erase(elem);
601 27280 }
602
603 572 void S3FanoutManager::InitPipeWatchFds() {
604
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 572 times.
 572 assert(watch_fds_inuse_ == 0);
605
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 572 times.
 572 assert(watch_fds_size_ >= 2);
606 572 watch_fds_[0].fd = pipe_terminate_[0];
607 572 watch_fds_[0].events = POLLIN | POLLPRI;
608 572 watch_fds_[0].revents = 0;
609 572 ++watch_fds_inuse_;
610 572 watch_fds_[1].fd = pipe_jobs_[0];
611 572 watch_fds_[1].events = POLLIN | POLLPRI;
612 572 watch_fds_[1].revents = 0;
613 572 ++watch_fds_inuse_;
614 572 }
615
616 /**
617 * The Amazon AWS 2 authorization header according to
618 * http://docs.aws.amazon.com/AmazonS3/latest/dev/RESTAuthentication.html#ConstructingTheAuthenticationHeader
619 */
620 54032 bool S3FanoutManager::MkV2Authz(const JobInfo &info,
621 vector<string> *headers) const {
622 54032 string payload_hash;
623
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 const bool retval = MkPayloadHash(info, &payload_hash);
624
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 54032 times.
 54032 if (!retval)
625 ✗ return false;
626
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 const string content_type = GetContentType(info);
627
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 const string request = GetRequestString(info);
628
629
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 const string timestamp = RfcTimestamp();
630
5/10
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 54032 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 54032 times.
✗ Branch 8 not taken.
✓ Branch 10 taken 54032 times.
✗ Branch 11 not taken.
✓ Branch 13 taken 54032 times.
✗ Branch 14 not taken.
 108064 string to_sign = request + "\n" + payload_hash + "\n" + content_type + "\n"
631
2/4
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 54032 times.
✗ Branch 5 not taken.
 108064 + timestamp + "\n";
632
2/4
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 54032 times.
✗ Branch 4 not taken.
 54032 if (config_.x_amz_acl != "")
633
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 108064 to_sign += "x-amz-acl:" + config_.x_amz_acl
634
2/4
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 54032 times.
✗ Branch 5 not taken.
 54032 + "\n"; // CanonicalizedAmzHeaders
635 54032 to_sign += MkV2CanonicalResource(config_.bucket, info.object_key,
636 54032 config_.dns_buckets,
637
2/4
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 54032 times.
✗ Branch 5 not taken.
 54032 info.request == JobInfo::kReqDeleteMulti);
638
1/2
✓ Branch 3 taken 54032 times.
✗ Branch 4 not taken.
 54032 LogCvmfs(kLogS3Fanout, kLogDebug, "%s string to sign: %s", request.c_str(),
639 to_sign.c_str());
640
641
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 shash::Any hmac;
642 54032 hmac.algorithm = shash::kSha1;
643
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 shash::Hmac(config_.secret_key,
644 54032 reinterpret_cast<const unsigned char *>(to_sign.data()),
645 54032 to_sign.length(), &hmac);
646
647
3/6
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 54032 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 54032 times.
✗ Branch 8 not taken.
 162096 headers->push_back("Authorization: AWS " + config_.access_key + ":"
648
3/6
✓ Branch 2 taken 54032 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 54032 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 54032 times.
✗ Branch 9 not taken.
 270160 + Base64(string(reinterpret_cast<char *>(hmac.digest),
649 54032 hmac.GetDigestSize())));
650
2/4
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 54032 times.
✗ Branch 5 not taken.
 54032 headers->push_back("Date: " + timestamp);
651
2/4
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 54032 times.
✗ Branch 5 not taken.
 54032 headers->push_back("X-Amz-Acl: " + config_.x_amz_acl);
652
2/2
✓ Branch 1 taken 27060 times.
✓ Branch 2 taken 26972 times.
 54032 if (!payload_hash.empty())
653
2/4
✓ Branch 1 taken 27060 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 27060 times.
✗ Branch 5 not taken.
 27060 headers->push_back("Content-MD5: " + payload_hash);
654
2/2
✓ Branch 1 taken 27060 times.
✓ Branch 2 taken 26972 times.
 54032 if (!content_type.empty())
655
2/4
✓ Branch 1 taken 27060 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 27060 times.
✗ Branch 5 not taken.
 27060 headers->push_back("Content-Type: " + content_type);
656 54032 return true;
657 54032 }
658
659
660 ✗ string S3FanoutManager::GetUriEncode(const string &val,
661 bool encode_slash) const {
662 ✗ string result;
663 ✗ const unsigned len = val.length();
664 ✗ result.reserve(len);
665 ✗ for (unsigned i = 0; i < len; ++i) {
666 ✗ const char c = val[i];
667 ✗ if ((c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z')
668 ✗ || (c >= '0' && c <= '9') || c == '_' || c == '-' || c == '~'
669 ✗ || c == '.') {
670 ✗ result.push_back(c);
671 ✗ } else if (c == '/') {
672 ✗ if (encode_slash) {
673 ✗ result += "%2F";
674 } else {
675 ✗ result.push_back(c);
676 }
677 } else {
678 ✗ result.push_back('%');
679 ✗ result.push_back((c / 16) + ((c / 16 <= 9) ? '0' : 'A' - 10));
680 ✗ result.push_back((c % 16) + ((c % 16 <= 9) ? '0' : 'A' - 10));
681 }
682 }
683 ✗ return result;
684 }
685
686
687 ✗ string S3FanoutManager::GetAwsV4SigningKey(const string &date) const {
688 ✗ if (last_signing_key_.first == date)
689 ✗ return last_signing_key_.second;
690
691 ✗ const string date_key = shash::Hmac256("AWS4" + config_.secret_key, date,
692 ✗ true);
693 ✗ const string date_region_key = shash::Hmac256(date_key, config_.region, true);
694 const string date_region_service_key = shash::Hmac256(date_region_key, "s3",
695 ✗ true);
696 string signing_key = shash::Hmac256(date_region_service_key, "aws4_request",
697 ✗ true);
698 ✗ last_signing_key_.first = date;
699 ✗ last_signing_key_.second = signing_key;
700 ✗ return signing_key;
701 }
702
703
704 /**
705 * The Amazon AWS4 authorization header according to
706 * http://docs.aws.amazon.com/AmazonS3/latest/API/sigv4-auth-using-authorization-header.html
707 */
708 ✗ bool S3FanoutManager::MkV4Authz(const JobInfo &info,
709 vector<string> *headers) const {
710 ✗ string payload_hash;
711 ✗ const bool retval = MkPayloadHash(info, &payload_hash);
712 ✗ if (!retval)
713 ✗ return false;
714 ✗ const string content_type = GetContentType(info);
715 ✗ const string timestamp = IsoTimestamp();
716 ✗ const string date = timestamp.substr(0, 8);
717 ✗ vector<string> tokens = SplitString(complete_hostname_, ':');
718 ✗ assert(tokens.size() <= 2);
719 ✗ string canonical_hostname = tokens[0];
720
721 // if we could split the hostname in two and if the port is *NOT* a default
722 // one
723 ✗ if (tokens.size() == 2
724 ✗ && !((String2Uint64(tokens[1]) == kDefaultHTTPPort)
725 ✗ || (String2Uint64(tokens[1]) == kDefaultHTTPSPort)))
726 ✗ canonical_hostname += ":" + tokens[1];
727
728 ✗ string signed_headers;
729 ✗ string canonical_headers;
730 ✗ if (!content_type.empty()) {
731 ✗ signed_headers += "content-type;";
732 ✗ headers->push_back("Content-Type: " + content_type);
733 ✗ canonical_headers += "content-type:" + content_type + "\n";
734 }
735 ✗ if (config_.x_amz_acl != "") {
736 ✗ signed_headers += "host;x-amz-acl;x-amz-content-sha256;x-amz-date";
737 } else {
738 ✗ signed_headers += "host;x-amz-content-sha256;x-amz-date";
739 }
740 ✗ canonical_headers += "host:" + canonical_hostname + "\n";
741 ✗ if (config_.x_amz_acl != "") {
742 ✗ canonical_headers += "x-amz-acl:" + config_.x_amz_acl + "\n";
743 }
744 ✗ canonical_headers += "x-amz-content-sha256:" + payload_hash + "\n"
745 ✗ + "x-amz-date:" + timestamp + "\n";
746
747 ✗ const string scope = date + "/" + config_.region + "/s3/aws4_request";
748 ✗ string uri;
749 ✗ if (config_.dns_buckets) {
750 ✗ uri = string("/") + info.object_key;
751 ✗ } else if (info.object_key.empty()) {
752 ✗ uri = string("/") + config_.bucket;
753 } else {
754 ✗ uri = string("/") + config_.bucket + "/" + info.object_key;
755 }
756
757 // V4 canonical query string: empty for most requests, "delete=" for
758 // multi-object delete (the S3 ?delete parameter has no value)
759 ✗ const string canonical_query = (info.request == JobInfo::kReqDeleteMulti)
760 ? "delete="
761 ✗ : "";
762
763 ✗ const string canonical_request = GetRequestString(info) + "\n"
764 ✗ + GetUriEncode(uri, false) + "\n"
765 ✗ + canonical_query + "\n" + canonical_headers
766 ✗ + "\n" + signed_headers + "\n"
767 ✗ + payload_hash;
768
769 ✗ const string hash_request = shash::Sha256String(canonical_request.c_str());
770
771 ✗ const string string_to_sign = "AWS4-HMAC-SHA256\n" + timestamp + "\n" + scope
772 ✗ + "\n" + hash_request;
773
774 ✗ const string signing_key = GetAwsV4SigningKey(date);
775 ✗ const string signature = shash::Hmac256(signing_key, string_to_sign);
776
777 ✗ headers->push_back("X-Amz-Acl: " + config_.x_amz_acl);
778 ✗ headers->push_back("X-Amz-Content-Sha256: " + payload_hash);
779 ✗ headers->push_back("X-Amz-Date: " + timestamp);
780 ✗ headers->push_back("Authorization: AWS4-HMAC-SHA256 "
781 "Credential="
782 ✗ + config_.access_key + "/" + scope
783 ✗ + ","
784 "SignedHeaders="
785 ✗ + signed_headers
786 ✗ + ","
787 "Signature="
788 ✗ + signature);
789
790 // S3 requires Content-MD5 for multi-object delete
791 ✗ if (info.request == JobInfo::kReqDeleteMulti) {
792 ✗ shash::Any md5_hash(shash::kMd5);
793 unsigned char *data;
794 ✗ const unsigned int nbytes = info.origin->Data(
795 ✗ reinterpret_cast<void **>(&data), info.origin->GetSize(), 0);
796 ✗ assert(nbytes == info.origin->GetSize());
797 ✗ shash::HashMem(data, nbytes, &md5_hash);
798 ✗ headers->push_back(
799 "Content-MD5: "
800 ✗ + Base64(string(reinterpret_cast<char *>(md5_hash.digest),
801 ✗ md5_hash.GetDigestSize())));
802 }
803 ✗ return true;
804 }
805
806 /**
807 * The Azure Blob authorization header according to
808 * https://docs.microsoft.com/en-us/rest/api/storageservices/authorize-with-shared-key
809 */
810 ✗ bool S3FanoutManager::MkAzureAuthz(const JobInfo &info,
811 vector<string> *headers) const {
812 ✗ const string timestamp = RfcTimestamp();
813 const string canonical_headers = "x-ms-blob-type:BlockBlob\nx-ms-date:"
814 ✗ + timestamp + "\nx-ms-version:2011-08-18";
815 ✗ const string canonical_resource = "/" + config_.access_key + "/"
816 ✗ + config_.bucket + "/" + info.object_key;
817
818 ✗ string string_to_sign;
819 ✗ if ((info.request == JobInfo::kReqHeadOnly)
820 ✗ || (info.request == JobInfo::kReqHeadPut)
821 ✗ || (info.request == JobInfo::kReqDelete)) {
822 ✗ string_to_sign = GetRequestString(info) + string("\n\n\n")
823 ✗ + "\n\n\n\n\n\n\n\n\n" + canonical_headers + "\n"
824 ✗ + canonical_resource;
825 } else {
826 ✗ string_to_sign = GetRequestString(info) + string("\n\n\n")
827 ✗ + string(StringifyInt(info.origin->GetSize()))
828 ✗ + "\n\n\n\n\n\n\n\n\n" + canonical_headers + "\n"
829 ✗ + canonical_resource;
830 }
831
832 ✗ string signing_key;
833 ✗ const int retval = Debase64(config_.secret_key, &signing_key);
834 ✗ if (!retval)
835 ✗ return false;
836
837 ✗ const string signature = shash::Hmac256(signing_key, string_to_sign, true);
838
839 ✗ headers->push_back("x-ms-date: " + timestamp);
840 ✗ headers->push_back("x-ms-version: 2011-08-18");
841 ✗ headers->push_back("Authorization: SharedKey " + config_.access_key + ":"
842 ✗ + Base64(signature));
843 ✗ headers->push_back("x-ms-blob-type: BlockBlob");
844 ✗ return true;
845 }
846
847 54032 void S3FanoutManager::InitializeDnsSettingsCurl(CURL *handle,
848 CURLSH *sharehandle,
849 curl_slist *clist) const {
850 54032 CURLcode retval = curl_easy_setopt(handle, CURLOPT_SHARE, sharehandle);
851
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 54032 times.
 54032 assert(retval == CURLE_OK);
852 54032 retval = curl_easy_setopt(handle, CURLOPT_RESOLVE, clist);
853
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 54032 times.
 54032 assert(retval == CURLE_OK);
854 54032 }
855
856
857 54032 int S3FanoutManager::InitializeDnsSettings(CURL *handle,
858 std::string host_with_port) const {
859 // Use existing handle
860 const std::map<CURL *, S3FanOutDnsEntry *>::const_iterator
861
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 it = curl_sharehandles_->find(handle);
862
2/2
✓ Branch 3 taken 51612 times.
✓ Branch 4 taken 2420 times.
 54032 if (it != curl_sharehandles_->end()) {
863
1/2
✓ Branch 1 taken 51612 times.
✗ Branch 2 not taken.
 51612 InitializeDnsSettingsCurl(handle, it->second->sharehandle,
864 51612 it->second->clist);
865 51612 return 0;
866 }
867
868 // Add protocol information for extraction of fields for DNS
869
2/4
✓ Branch 1 taken 2420 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 2420 times.
✗ Branch 4 not taken.
 2420 if (!IsHttpUrl(host_with_port))
870
2/4
✓ Branch 1 taken 2420 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 2420 times.
✗ Branch 5 not taken.
 2420 host_with_port = config_.protocol + "://" + host_with_port;
871
1/2
✓ Branch 1 taken 2420 times.
✗ Branch 2 not taken.
 2420 const std::string remote_host = dns::ExtractHost(host_with_port);
872
1/2
✓ Branch 1 taken 2420 times.
✗ Branch 2 not taken.
 2420 const std::string remote_port = dns::ExtractPort(host_with_port);
873
874 // If we have the name already resolved, use the least used IP
875 2420 S3FanOutDnsEntry *useme = NULL;
876 2420 unsigned int usemin = UINT_MAX;
877 2420 std::set<S3FanOutDnsEntry *>::iterator its3 = sharehandles_->begin();
878
2/2
✓ Branch 3 taken 1936 times.
✓ Branch 4 taken 2420 times.
 4356 for (; its3 != sharehandles_->end(); ++its3) {
879
1/2
✓ Branch 2 taken 1936 times.
✗ Branch 3 not taken.
 1936 if ((*its3)->dns_name == remote_host) {
880
1/2
✓ Branch 1 taken 1936 times.
✗ Branch 2 not taken.
 1936 if (usemin >= (*its3)->counter) {
881 1936 usemin = (*its3)->counter;
882 1936 useme = (*its3);
883 }
884 }
885 }
886
2/2
✓ Branch 0 taken 1936 times.
✓ Branch 1 taken 484 times.
 2420 if (useme != NULL) {
887
1/2
✓ Branch 1 taken 1936 times.
✗ Branch 2 not taken.
 1936 curl_sharehandles_->insert(
888 1936 std::pair<CURL *, S3FanOutDnsEntry *>(handle, useme));
889 1936 useme->counter++;
890
1/2
✓ Branch 1 taken 1936 times.
✗ Branch 2 not taken.
 1936 InitializeDnsSettingsCurl(handle, useme->sharehandle, useme->clist);
891 1936 return 0;
892 }
893
894 // We need to resolve the hostname
895 // TODO(ssheikki): support ipv6 also... if (opt_ipv4_only_)
896
1/2
✓ Branch 1 taken 484 times.
✗ Branch 2 not taken.
 484 const dns::Host host = resolver_->Resolve(remote_host);
897
1/2
✓ Branch 2 taken 484 times.
✗ Branch 3 not taken.
 484 set<string> const ipv4_addresses = host.ipv4_addresses();
898 484 std::set<string>::iterator its = ipv4_addresses.begin();
899 484 S3FanOutDnsEntry *dnse = NULL;
900
2/2
✓ Branch 3 taken 484 times.
✓ Branch 4 taken 484 times.
 968 for (; its != ipv4_addresses.end(); ++its) {
901
2/4
✓ Branch 1 taken 484 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 484 times.
✗ Branch 5 not taken.
 484 dnse = new S3FanOutDnsEntry();
902 484 dnse->counter = 0;
903
1/2
✓ Branch 1 taken 484 times.
✗ Branch 2 not taken.
 484 dnse->dns_name = remote_host;
904
4/12
✗ Branch 1 not taken.
✓ Branch 2 taken 484 times.
✗ Branch 5 not taken.
✗ Branch 6 not taken.
✓ Branch 8 taken 484 times.
✗ Branch 9 not taken.
✓ Branch 11 taken 484 times.
✗ Branch 12 not taken.
✗ Branch 14 not taken.
✓ Branch 15 taken 484 times.
✗ Branch 18 not taken.
✗ Branch 19 not taken.
 484 dnse->port = remote_port.size() == 0 ? "80" : remote_port;
905
1/2
✓ Branch 2 taken 484 times.
✗ Branch 3 not taken.
 484 dnse->ip = *its;
906 484 dnse->clist = NULL;
907
1/2
✓ Branch 2 taken 484 times.
✗ Branch 3 not taken.
 484 dnse->clist = curl_slist_append(
908 dnse->clist,
909
4/8
✓ Branch 1 taken 484 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 484 times.
✗ Branch 5 not taken.
✓ Branch 7 taken 484 times.
✗ Branch 8 not taken.
✓ Branch 10 taken 484 times.
✗ Branch 11 not taken.
 968 (dnse->dns_name + ":" + dnse->port + ":" + dnse->ip).c_str());
910
1/2
✓ Branch 1 taken 484 times.
✗ Branch 2 not taken.
 484 dnse->sharehandle = curl_share_init();
911
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 484 times.
 484 assert(dnse->sharehandle != NULL);
912
1/2
✓ Branch 1 taken 484 times.
✗ Branch 2 not taken.
 484 const CURLSHcode share_retval = curl_share_setopt(
913 dnse->sharehandle, CURLSHOPT_SHARE, CURL_LOCK_DATA_DNS);
914
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 484 times.
 484 assert(share_retval == CURLSHE_OK);
915
1/2
✓ Branch 1 taken 484 times.
✗ Branch 2 not taken.
 484 sharehandles_->insert(dnse);
916 }
917
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 484 times.
 484 if (dnse == NULL) {
918 ✗ LogCvmfs(kLogS3Fanout, kLogStderr | kLogSyslogErr,
919 "Error: DNS resolve failed for address '%s'.",
920 remote_host.c_str());
921 ✗ assert(dnse != NULL);
922 ✗ return -1;
923 }
924
1/2
✓ Branch 1 taken 484 times.
✗ Branch 2 not taken.
 484 curl_sharehandles_->insert(
925 484 std::pair<CURL *, S3FanOutDnsEntry *>(handle, dnse));
926 484 dnse->counter++;
927
1/2
✓ Branch 1 taken 484 times.
✗ Branch 2 not taken.
 484 InitializeDnsSettingsCurl(handle, dnse->sharehandle, dnse->clist);
928
929 484 return 0;
930 2420 }
931
932
933 54032 bool S3FanoutManager::MkPayloadHash(const JobInfo &info,
934 string *hex_hash) const {
935
2/2
✓ Branch 0 taken 53812 times.
✓ Branch 1 taken 220 times.
 54032 if (info.request == JobInfo::kReqHeadOnly
936
2/2
✓ Branch 0 taken 27060 times.
✓ Branch 1 taken 26752 times.
 53812 || info.request == JobInfo::kReqHeadPut
937
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 27060 times.
 27060 || info.request == JobInfo::kReqDelete) {
938
1/4
✓ Branch 0 taken 26972 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 26972 switch (config_.authz_method) {
939 26972 case kAuthzAwsV2:
940 26972 hex_hash->clear();
941 26972 break;
942 ✗ case kAuthzAwsV4:
943 // Sha256 over empty string
944 *hex_hash = "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b78"
945 ✗ "52b855";
946 ✗ break;
947 ✗ case kAuthzAzure:
948 // no payload hash required for Azure signature
949 ✗ hex_hash->clear();
950 ✗ break;
951 ✗ default:
952 ✗ PANIC(NULL);
953 }
954 26972 return true;
955 }
956
957 // kReqDeleteMulti is a POST with a body (XML payload), same hashing as PUT
958 // falls through intentionally.
959
960 // PUT or POST with payload
961
1/2
✓ Branch 1 taken 27060 times.
✗ Branch 2 not taken.
 27060 shash::Any payload_hash(shash::kMd5);
962
963 unsigned char *data;
964 27060 const unsigned int nbytes = info.origin->Data(
965
2/4
✓ Branch 2 taken 27060 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 27060 times.
✗ Branch 6 not taken.
 27060 reinterpret_cast<void **>(&data), info.origin->GetSize(), 0);
966
2/4
✓ Branch 2 taken 27060 times.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✓ Branch 5 taken 27060 times.
 27060 assert(nbytes == info.origin->GetSize());
967
968
1/4
✓ Branch 0 taken 27060 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 27060 switch (config_.authz_method) {
969 27060 case kAuthzAwsV2:
970
1/2
✓ Branch 1 taken 27060 times.
✗ Branch 2 not taken.
 27060 shash::HashMem(data, nbytes, &payload_hash);
971
2/4
✓ Branch 2 taken 27060 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 27060 times.
✗ Branch 6 not taken.
 81180 *hex_hash = Base64(string(reinterpret_cast<char *>(payload_hash.digest),
972 54120 payload_hash.GetDigestSize()));
973 27060 return true;
974 ✗ case kAuthzAwsV4:
975 ✗ *hex_hash = shash::Sha256Mem(data, nbytes);
976 ✗ return true;
977 ✗ case kAuthzAzure:
978 // no payload hash required for Azure signature
979 ✗ hex_hash->clear();
980 ✗ return true;
981 ✗ default:
982 ✗ PANIC(NULL);
983 }
984 }
985
986 54032 string S3FanoutManager::GetRequestString(const JobInfo &info) const {
987
3/5
✓ Branch 0 taken 26972 times.
✓ Branch 1 taken 26752 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 308 times.
✗ Branch 4 not taken.
 54032 switch (info.request) {
988 26972 case JobInfo::kReqHeadOnly:
989 case JobInfo::kReqHeadPut:
990
1/2
✓ Branch 2 taken 26972 times.
✗ Branch 3 not taken.
 26972 return "HEAD";
991 26752 case JobInfo::kReqPutCas:
992 case JobInfo::kReqPutDotCvmfs:
993 case JobInfo::kReqPutHtml:
994 case JobInfo::kReqPutBucket:
995
1/2
✓ Branch 2 taken 26752 times.
✗ Branch 3 not taken.
 26752 return "PUT";
996 ✗ case JobInfo::kReqDelete:
997 ✗ return "DELETE";
998 308 case JobInfo::kReqDeleteMulti:
999
1/2
✓ Branch 2 taken 308 times.
✗ Branch 3 not taken.
 308 return "POST";
1000 ✗ default:
1001 ✗ PANIC(NULL);
1002 }
1003 }
1004
1005
1006 54032 string S3FanoutManager::GetContentType(const JobInfo &info) const {
1007
3/6
✓ Branch 0 taken 26972 times.
✓ Branch 1 taken 26752 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 308 times.
✗ Branch 5 not taken.
 54032 switch (info.request) {
1008 26972 case JobInfo::kReqHeadOnly:
1009 case JobInfo::kReqHeadPut:
1010 case JobInfo::kReqDelete:
1011
1/2
✓ Branch 2 taken 26972 times.
✗ Branch 3 not taken.
 26972 return "";
1012 26752 case JobInfo::kReqPutCas:
1013
1/2
✓ Branch 2 taken 26752 times.
✗ Branch 3 not taken.
 26752 return "application/octet-stream";
1014 ✗ case JobInfo::kReqPutDotCvmfs:
1015 ✗ return "application/x-cvmfs";
1016 ✗ case JobInfo::kReqPutHtml:
1017 ✗ return "text/html";
1018 308 case JobInfo::kReqPutBucket:
1019 case JobInfo::kReqDeleteMulti:
1020
1/2
✓ Branch 2 taken 308 times.
✗ Branch 3 not taken.
 308 return "text/xml";
1021 ✗ default:
1022 ✗ PANIC(NULL);
1023 }
1024 }
1025
1026
1027 /**
1028 * Request parameters set the URL and other options such as timeout and
1029 * proxy.
1030 */
1031 54032 Failures S3FanoutManager::InitializeRequest(JobInfo *info, CURL *handle) const {
1032 // Initialize internal download state
1033 54032 info->curl_handle = handle;
1034 54032 info->error_code = kFailOk;
1035 54032 info->http_error = 0;
1036 54032 info->num_retries = 0;
1037 54032 info->backoff_ms = 0;
1038 54032 info->throttle_ms = 0;
1039 54032 info->throttle_timestamp = 0;
1040 54032 info->http_headers = NULL;
1041 // info->payload_size is needed in S3Uploader::MainCollectResults,
1042 // where info->origin is already destroyed.
1043
1/2
✓ Branch 2 taken 54032 times.
✗ Branch 3 not taken.
 54032 info->payload_size = info->origin->GetSize();
1044
1045
2/4
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 54032 times.
✗ Branch 5 not taken.
 54032 InitializeDnsSettings(handle, complete_hostname_);
1046
1047 CURLcode retval;
1048
2/2
✓ Branch 0 taken 53812 times.
✓ Branch 1 taken 220 times.
 54032 if (info->request == JobInfo::kReqHeadOnly
1049
2/2
✓ Branch 0 taken 27060 times.
✓ Branch 1 taken 26752 times.
 53812 || info->request == JobInfo::kReqHeadPut
1050
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 27060 times.
 27060 || info->request == JobInfo::kReqDelete) {
1051
1/2
✓ Branch 1 taken 26972 times.
✗ Branch 2 not taken.
 26972 retval = curl_easy_setopt(handle, CURLOPT_UPLOAD, 0);
1052
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 26972 times.
 26972 assert(retval == CURLE_OK);
1053
1/2
✓ Branch 1 taken 26972 times.
✗ Branch 2 not taken.
 26972 retval = curl_easy_setopt(handle, CURLOPT_NOBODY, 1);
1054
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 26972 times.
 26972 assert(retval == CURLE_OK);
1055
1056
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 26972 times.
 26972 if (info->request == JobInfo::kReqDelete) {
1057 ✗ retval = curl_easy_setopt(handle, CURLOPT_CUSTOMREQUEST,
1058 GetRequestString(*info).c_str());
1059 ✗ assert(retval == CURLE_OK);
1060 } else {
1061
1/2
✓ Branch 1 taken 26972 times.
✗ Branch 2 not taken.
 26972 retval = curl_easy_setopt(handle, CURLOPT_CUSTOMREQUEST, NULL);
1062
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 26972 times.
 26972 assert(retval == CURLE_OK);
1063 }
1064
2/2
✓ Branch 0 taken 308 times.
✓ Branch 1 taken 26752 times.
 27060 } else if (info->request == JobInfo::kReqDeleteMulti) {
1065 // POST request with XML body read from origin buffer
1066
1/2
✓ Branch 1 taken 308 times.
✗ Branch 2 not taken.
 308 retval = curl_easy_setopt(handle, CURLOPT_UPLOAD, 1);
1067
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 308 times.
 308 assert(retval == CURLE_OK);
1068
1/2
✓ Branch 1 taken 308 times.
✗ Branch 2 not taken.
 308 retval = curl_easy_setopt(handle, CURLOPT_NOBODY, 0);
1069
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 308 times.
 308 assert(retval == CURLE_OK);
1070
1/2
✓ Branch 1 taken 308 times.
✗ Branch 2 not taken.
 308 retval = curl_easy_setopt(handle, CURLOPT_CUSTOMREQUEST, "POST");
1071
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 308 times.
 308 assert(retval == CURLE_OK);
1072
2/4
✓ Branch 2 taken 308 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 308 times.
✗ Branch 6 not taken.
 308 retval = curl_easy_setopt(handle, CURLOPT_INFILESIZE_LARGE,
1073 static_cast<curl_off_t>(info->origin->GetSize()));
1074
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 308 times.
 308 assert(retval == CURLE_OK);
1075 308 info->response_body.clear();
1076 } else {
1077
1/2
✓ Branch 1 taken 26752 times.
✗ Branch 2 not taken.
 26752 retval = curl_easy_setopt(handle, CURLOPT_CUSTOMREQUEST, NULL);
1078
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 26752 times.
 26752 assert(retval == CURLE_OK);
1079
1/2
✓ Branch 1 taken 26752 times.
✗ Branch 2 not taken.
 26752 retval = curl_easy_setopt(handle, CURLOPT_UPLOAD, 1);
1080
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 26752 times.
 26752 assert(retval == CURLE_OK);
1081
1/2
✓ Branch 1 taken 26752 times.
✗ Branch 2 not taken.
 26752 retval = curl_easy_setopt(handle, CURLOPT_NOBODY, 0);
1082
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 26752 times.
 26752 assert(retval == CURLE_OK);
1083
2/4
✓ Branch 2 taken 26752 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 26752 times.
✗ Branch 6 not taken.
 26752 retval = curl_easy_setopt(handle, CURLOPT_INFILESIZE_LARGE,
1084 static_cast<curl_off_t>(info->origin->GetSize()));
1085
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 26752 times.
 26752 assert(retval == CURLE_OK);
1086
1087
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 26752 times.
 26752 if (info->request == JobInfo::kReqPutDotCvmfs) {
1088 ✗ info->http_headers = curl_slist_append(
1089 info->http_headers, dot_cvmfs_cache_control_header.c_str());
1090
1/2
✓ Branch 0 taken 26752 times.
✗ Branch 1 not taken.
 26752 } else if (info->request == JobInfo::kReqPutCas) {
1091
1/2
✓ Branch 1 taken 26752 times.
✗ Branch 2 not taken.
 26752 info->http_headers = curl_slist_append(info->http_headers,
1092 kCacheControlCas);
1093 }
1094 }
1095
1096 bool retval_b;
1097
1098 // Authorization
1099 54032 vector<string> authz_headers;
1100
1/4
✓ Branch 0 taken 54032 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
 54032 switch (config_.authz_method) {
1101 54032 case kAuthzAwsV2:
1102
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 retval_b = MkV2Authz(*info, &authz_headers);
1103 54032 break;
1104 ✗ case kAuthzAwsV4:
1105 ✗ retval_b = MkV4Authz(*info, &authz_headers);
1106 ✗ break;
1107 ✗ case kAuthzAzure:
1108 ✗ retval_b = MkAzureAuthz(*info, &authz_headers);
1109 ✗ break;
1110 ✗ default:
1111 ✗ PANIC(NULL);
1112 }
1113
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 54032 times.
 54032 if (!retval_b)
1114 ✗ return kFailLocalIO;
1115
2/2
✓ Branch 1 taken 216216 times.
✓ Branch 2 taken 54032 times.
 270248 for (unsigned i = 0; i < authz_headers.size(); ++i) {
1116
1/2
✓ Branch 2 taken 216216 times.
✗ Branch 3 not taken.
 216216 info->http_headers = curl_slist_append(info->http_headers,
1117 216216 authz_headers[i].c_str());
1118 }
1119
1120 // Common headers
1121
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 info->http_headers = curl_slist_append(info->http_headers,
1122 "Connection: Keep-Alive");
1123
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 info->http_headers = curl_slist_append(info->http_headers, "Pragma:");
1124 // No 100-continue
1125
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 info->http_headers = curl_slist_append(info->http_headers, "Expect:");
1126 // Strip unnecessary header
1127
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 info->http_headers = curl_slist_append(info->http_headers, "Accept:");
1128
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 info->http_headers = curl_slist_append(info->http_headers,
1129 54032 user_agent_->c_str());
1130
1131 // Set curl parameters
1132
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 retval = curl_easy_setopt(handle, CURLOPT_PRIVATE, static_cast<void *>(info));
1133
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 54032 times.
 54032 assert(retval == CURLE_OK);
1134
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 retval = curl_easy_setopt(handle, CURLOPT_HEADERDATA,
1135 static_cast<void *>(info));
1136
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 54032 times.
 54032 assert(retval == CURLE_OK);
1137
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 retval = curl_easy_setopt(handle, CURLOPT_READDATA,
1138 static_cast<void *>(info));
1139
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 54032 times.
 54032 assert(retval == CURLE_OK);
1140
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 retval = curl_easy_setopt(handle, CURLOPT_WRITEDATA,
1141 static_cast<void *>(info));
1142
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 54032 times.
 54032 assert(retval == CURLE_OK);
1143
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 retval = curl_easy_setopt(handle, CURLOPT_HTTPHEADER, info->http_headers);
1144
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 54032 times.
 54032 assert(retval == CURLE_OK);
1145
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 54032 times.
 54032 if (opt_ipv4_only_) {
1146 ✗ retval = curl_easy_setopt(handle, CURLOPT_IPRESOLVE, CURL_IPRESOLVE_V4);
1147 ✗ assert(retval == CURLE_OK);
1148 }
1149 // Follow HTTP redirects
1150
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 retval = curl_easy_setopt(handle, CURLOPT_FOLLOWLOCATION, 1L);
1151
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 54032 times.
 54032 assert(retval == CURLE_OK);
1152
1153
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 retval = curl_easy_setopt(handle, CURLOPT_ERRORBUFFER, info->errorbuffer);
1154
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 54032 times.
 54032 assert(retval == CURLE_OK);
1155
1156
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 54032 times.
 54032 if (config_.protocol == "https") {
1157 ✗ retval = curl_easy_setopt(handle, CURLOPT_SSL_VERIFYPEER, 1L);
1158 ✗ assert(retval == CURLE_OK);
1159 ✗ retval = curl_easy_setopt(handle, CURLOPT_PROXY_SSL_VERIFYPEER, 1L);
1160 ✗ assert(retval == CURLE_OK);
1161 ✗ const bool add_cert = ssl_certificate_store_.ApplySslCertificatePath(
1162 handle);
1163 ✗ assert(add_cert);
1164 }
1165
1166 54032 return kFailOk;
1167 54032 }
1168
1169
1170 /**
1171 * Sets the URL specific options such as host to use and timeout.
1172 */
1173 54032 void S3FanoutManager::SetUrlOptions(JobInfo *info) const {
1174 54032 CURL *curl_handle = info->curl_handle;
1175 CURLcode retval;
1176
1177
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 retval = curl_easy_setopt(curl_handle, CURLOPT_CONNECTTIMEOUT,
1178 config_.opt_timeout_sec);
1179
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 54032 times.
 54032 assert(retval == CURLE_OK);
1180
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 retval = curl_easy_setopt(curl_handle, CURLOPT_LOW_SPEED_LIMIT,
1181 kLowSpeedLimit);
1182
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 54032 times.
 54032 assert(retval == CURLE_OK);
1183
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 retval = curl_easy_setopt(curl_handle, CURLOPT_LOW_SPEED_TIME,
1184 config_.opt_timeout_sec);
1185
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 54032 times.
 54032 assert(retval == CURLE_OK);
1186
1187
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 54032 times.
 54032 if (is_curl_debug_) {
1188 ✗ retval = curl_easy_setopt(curl_handle, CURLOPT_VERBOSE, 1);
1189 ✗ assert(retval == CURLE_OK);
1190 }
1191
1192
1/2
✓ Branch 1 taken 54032 times.
✗ Branch 2 not taken.
 54032 string url = MkUrl(info->object_key);
1193
2/2
✓ Branch 0 taken 308 times.
✓ Branch 1 taken 53724 times.
 54032 if (info->request == JobInfo::kReqDeleteMulti)
1194
1/2
✓ Branch 1 taken 308 times.
✗ Branch 2 not taken.
 308 url += "?delete";
1195
1/2
✓ Branch 2 taken 54032 times.
✗ Branch 3 not taken.
 54032 retval = curl_easy_setopt(curl_handle, CURLOPT_URL, url.c_str());
1196
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 54032 times.
 54032 assert(retval == CURLE_OK);
1197
1198
1/2
✓ Branch 2 taken 54032 times.
✗ Branch 3 not taken.
 54032 retval = curl_easy_setopt(curl_handle, CURLOPT_PROXY, config_.proxy.c_str());
1199
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 54032 times.
 54032 assert(retval == CURLE_OK);
1200 54032 }
1201
1202
1203 /**
1204 * Adds transfer time and uploaded bytes to the global counters.
1205 */
1206 54208 void S3FanoutManager::UpdateStatistics(CURL *handle) {
1207 curl_off_t val;
1208
1209
2/4
✓ Branch 1 taken 54208 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 54208 times.
✗ Branch 4 not taken.
 54208 if (curl_easy_getinfo(handle, CURLINFO_SIZE_UPLOAD_T, &val) == CURLE_OK)
1210 54208 statistics_->transferred_bytes += val;
1211 54208 }
1212
1213
1214 /**
1215 * Retry if possible and if not already done too often.
1216 */
1217 264 bool S3FanoutManager::CanRetry(const JobInfo *info) {
1218 264 return (info->error_code == kFailHostConnection
1219
1/2
✓ Branch 0 taken 264 times.
✗ Branch 1 not taken.
 264 || info->error_code == kFailHostResolve
1220
1/2
✓ Branch 0 taken 264 times.
✗ Branch 1 not taken.
 264 || info->error_code == kFailServiceUnavailable
1221
2/2
✓ Branch 0 taken 176 times.
✓ Branch 1 taken 88 times.
 264 || info->error_code == kFailRetry)
1222
2/4
✓ Branch 0 taken 264 times.
✗ Branch 1 not taken.
✓ Branch 2 taken 176 times.
✗ Branch 3 not taken.
 528 && (info->num_retries < config_.opt_max_retries);
1223 }
1224
1225
1226 /**
1227 * Backoff for retry to introduce a jitter into a upload sequence.
1228 *
1229 * \return true if backoff has been performed, false otherwise
1230 */
1231 176 void S3FanoutManager::Backoff(JobInfo *info) {
1232
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 176 times.
 176 if (info->error_code != kFailRetry)
1233 ✗ info->num_retries++;
1234 176 statistics_->num_retries++;
1235
1236
1/2
✓ Branch 0 taken 176 times.
✗ Branch 1 not taken.
 176 if (info->throttle_ms > 0) {
1237 176 LogCvmfs(kLogS3Fanout, kLogDebug, "throttling for %d ms",
1238 info->throttle_ms);
1239 176 const uint64_t now = platform_monotonic_time();
1240
1/2
✓ Branch 0 taken 176 times.
✗ Branch 1 not taken.
 176 if ((info->throttle_timestamp + (info->throttle_ms / 1000)) >= now) {
1241
2/2
✓ Branch 0 taken 44 times.
✓ Branch 1 taken 132 times.
 176 if ((now - timestamp_last_throttle_report_)
1242 > kThrottleReportIntervalSec) {
1243 44 LogCvmfs(kLogS3Fanout, kLogStdout,
1244 "Warning: S3 backend throttling %ums "
1245 "(total backoff time so far %lums)",
1246 44 info->throttle_ms, statistics_->ms_throttled);
1247 44 timestamp_last_throttle_report_ = now;
1248 }
1249 176 statistics_->ms_throttled += info->throttle_ms;
1250 176 SafeSleepMs(info->throttle_ms);
1251 }
1252 } else {
1253 ✗ if (info->backoff_ms == 0) {
1254 // Must be != 0
1255 ✗ info->backoff_ms = prng_.Next(config_.opt_backoff_init_ms + 1);
1256 } else {
1257 ✗ info->backoff_ms *= 2;
1258 }
1259 ✗ if (info->backoff_ms > config_.opt_backoff_max_ms)
1260 ✗ info->backoff_ms = config_.opt_backoff_max_ms;
1261
1262 ✗ LogCvmfs(kLogS3Fanout, kLogDebug, "backing off for %d ms",
1263 info->backoff_ms);
1264 ✗ SafeSleepMs(info->backoff_ms);
1265 }
1266 176 }
1267
1268
1269 /**
1270 * Checks the result of a curl request and implements the failure logic
1271 * and takes care of cleanup.
1272 *
1273 * @return true if request should be repeated, false otherwise
1274 */
1275 54208 bool S3FanoutManager::VerifyAndFinalize(const int curl_error, JobInfo *info) {
1276 54208 LogCvmfs(kLogS3Fanout, kLogDebug,
1277 "Verify uploaded/tested object %s "
1278 "(curl error %d, info error %d, info request %d)",
1279 54208 info->object_key.c_str(), curl_error, info->error_code,
1280 54208 info->request);
1281 54208 UpdateStatistics(info->curl_handle);
1282
1283 // Verification and error classification
1284
1/6
✓ Branch 0 taken 54208 times.
✗ Branch 1 not taken.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
 54208 switch (curl_error) {
1285 54208 case CURLE_OK:
1286
2/2
✓ Branch 0 taken 54032 times.
✓ Branch 1 taken 176 times.
 54208 if ((info->error_code != kFailRetry)
1287
2/2
✓ Branch 0 taken 27192 times.
✓ Branch 1 taken 26840 times.
 54032 && (info->error_code != kFailNotFound)) {
1288 27192 info->error_code = kFailOk;
1289 }
1290 54208 break;
1291 ✗ case CURLE_UNSUPPORTED_PROTOCOL:
1292 case CURLE_URL_MALFORMAT:
1293 ✗ info->error_code = kFailBadRequest;
1294 ✗ break;
1295 ✗ case CURLE_COULDNT_RESOLVE_HOST:
1296 ✗ info->error_code = kFailHostResolve;
1297 ✗ break;
1298 ✗ case CURLE_COULDNT_CONNECT:
1299 case CURLE_OPERATION_TIMEDOUT:
1300 case CURLE_SEND_ERROR:
1301 case CURLE_RECV_ERROR:
1302 // Server or proxy dropped the connection mid-response (e.g. a reaped
1303 // keep-alive connection); the request body is rewound before retrying.
1304 case CURLE_PARTIAL_FILE:
1305 case CURLE_GOT_NOTHING:
1306 ✗ info->error_code = kFailHostConnection;
1307 ✗ break;
1308 ✗ case CURLE_ABORTED_BY_CALLBACK:
1309 case CURLE_WRITE_ERROR:
1310 // Error set by callback
1311 ✗ break;
1312 ✗ default: {
1313 ✗ const string what = (info->request == JobInfo::kReqDeleteMulti)
1314 ? "multi-delete of "
1315 ✗ + StringifyUint(
1316 info->multi_delete_keys.size())
1317 + " objects"
1318 ✗ : "upload of '" + info->object_key + "'";
1319 ✗ LogCvmfs(kLogS3Fanout, kLogStderr | kLogSyslogErr,
1320 "unexpected curl error (%d) during %s: %s", curl_error,
1321 what.c_str(), info->errorbuffer);
1322 ✗ info->error_code = kFailOther;
1323 ✗ break;
1324 }
1325 }
1326
1327 // Transform HEAD to PUT request
1328
2/2
✓ Branch 0 taken 26840 times.
✓ Branch 1 taken 27368 times.
 54208 if ((info->error_code == kFailNotFound)
1329
2/2
✓ Branch 0 taken 26752 times.
✓ Branch 1 taken 88 times.
 26840 && (info->request == JobInfo::kReqHeadPut)) {
1330 26752 LogCvmfs(kLogS3Fanout, kLogDebug, "not found: %s, uploading",
1331 info->object_key.c_str());
1332 26752 info->request = JobInfo::kReqPutCas;
1333 26752 curl_slist_free_all(info->http_headers);
1334 26752 info->http_headers = NULL;
1335 26752 const s3fanout::Failures init_failure = InitializeRequest(
1336 info, info->curl_handle);
1337
1338
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 26752 times.
 26752 if (init_failure != s3fanout::kFailOk) {
1339 ✗ PANIC(kLogStderr,
1340 "Failed to initialize CURL handle "
1341 "(error: %d - %s | errno: %d)",
1342 init_failure, Code2Ascii(init_failure), errno);
1343 }
1344 26752 SetUrlOptions(info);
1345 // Reset origin
1346 26752 info->origin->Rewind();
1347 26752 return true; // Again, Put
1348 }
1349
1350 // Determination if failed request should be repeated
1351 27456 bool try_again = false;
1352
2/2
✓ Branch 0 taken 264 times.
✓ Branch 1 taken 27192 times.
 27456 if (info->error_code != kFailOk) {
1353 264 try_again = CanRetry(info);
1354 }
1355
2/2
✓ Branch 0 taken 176 times.
✓ Branch 1 taken 27280 times.
 27456 if (try_again) {
1356
1/2
✓ Branch 0 taken 176 times.
✗ Branch 1 not taken.
 176 if (info->request == JobInfo::kReqPutCas
1357
1/2
✓ Branch 0 taken 176 times.
✗ Branch 1 not taken.
 176 || info->request == JobInfo::kReqPutDotCvmfs
1358
1/2
✓ Branch 0 taken 176 times.
✗ Branch 1 not taken.
 176 || info->request == JobInfo::kReqPutHtml
1359
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 176 times.
 176 || info->request == JobInfo::kReqDeleteMulti) {
1360 ✗ LogCvmfs(kLogS3Fanout, kLogDebug, "Trying again to upload %s",
1361 info->object_key.c_str());
1362 // Reset origin
1363 ✗ info->origin->Rewind();
1364 ✗ if (info->request == JobInfo::kReqDeleteMulti)
1365 ✗ info->response_body.clear();
1366 }
1367 176 Backoff(info);
1368 176 info->error_code = kFailOk;
1369 176 info->http_error = 0;
1370 176 info->throttle_ms = 0;
1371 176 info->backoff_ms = 0;
1372 176 info->throttle_timestamp = 0;
1373 176 return true; // try again
1374 }
1375
1376 // Cleanup opened resources
1377 27280 info->origin.reset();
1378
1379
3/4
✓ Branch 0 taken 88 times.
✓ Branch 1 taken 27192 times.
✓ Branch 2 taken 88 times.
✗ Branch 3 not taken.
 27280 if ((info->error_code != kFailOk) && (info->http_error != 0)
1380
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 88 times.
 88 && (info->http_error != 404)) {
1381 ✗ LogCvmfs(kLogS3Fanout, kLogStderr, "S3: HTTP failure %d", info->http_error);
1382 }
1383 27280 return false; // stop transfer
1384 }
1385
1386
2/4
✓ Branch 2 taken 572 times.
✗ Branch 3 not taken.
✓ Branch 8 taken 572 times.
✗ Branch 9 not taken.
 572 S3FanoutManager::S3FanoutManager(const S3Config &config) : config_(config) {
1387 572 multi_threaded_.store(0);
1388
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 MakePipe(pipe_terminate_);
1389
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 MakePipe(pipe_jobs_);
1390
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 MakePipe(pipe_completed_);
1391
1392 int retval;
1393 572 jobs_todo_lock_ = reinterpret_cast<pthread_mutex_t *>(
1394 572 smalloc(sizeof(pthread_mutex_t)));
1395 572 retval = pthread_mutex_init(jobs_todo_lock_, NULL);
1396
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 572 times.
 572 assert(retval == 0);
1397 572 curl_handle_lock_ = reinterpret_cast<pthread_mutex_t *>(
1398 572 smalloc(sizeof(pthread_mutex_t)));
1399 572 retval = pthread_mutex_init(curl_handle_lock_, NULL);
1400
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 572 times.
 572 assert(retval == 0);
1401
1402
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 active_requests_ = new set<JobInfo *>;
1403
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 pool_handles_idle_ = new set<CURL *>;
1404
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 pool_handles_inuse_ = new set<CURL *>;
1405
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 curl_sharehandles_ = new map<CURL *, S3FanOutDnsEntry *>;
1406
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 sharehandles_ = new set<S3FanOutDnsEntry *>;
1407 572 watch_fds_max_ = 4 * config_.pool_max_handles;
1408 572 max_available_jobs_ = 4 * config_.pool_max_handles;
1409
2/4
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 572 times.
✗ Branch 5 not taken.
 572 available_jobs_ = new Semaphore(max_available_jobs_);
1410
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 572 times.
 572 assert(NULL != available_jobs_);
1411
1412
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 statistics_ = new Statistics();
1413
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 user_agent_ = new string();
1414
2/4
✓ Branch 2 taken 572 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 572 times.
✗ Branch 6 not taken.
 572 *user_agent_ = "User-Agent: cvmfs " + string(CVMFS_VERSION);
1415
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 complete_hostname_ = MkCompleteHostname();
1416
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 dot_cvmfs_cache_control_header = MkDotCvmfsCacheControlHeader();
1417
1418
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 const CURLcode cretval = curl_global_init(CURL_GLOBAL_ALL);
1419
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 572 times.
 572 assert(cretval == CURLE_OK);
1420
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 curl_multi_ = curl_multi_init();
1421
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 572 times.
 572 assert(curl_multi_ != NULL);
1422 CURLMcode mretval;
1423
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 mretval = curl_multi_setopt(curl_multi_, CURLMOPT_SOCKETFUNCTION,
1424 CallbackCurlSocket);
1425
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 572 times.
 572 assert(mretval == CURLM_OK);
1426
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 mretval = curl_multi_setopt(curl_multi_, CURLMOPT_SOCKETDATA,
1427 static_cast<void *>(this));
1428
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 572 times.
 572 assert(mretval == CURLM_OK);
1429
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 mretval = curl_multi_setopt(curl_multi_, CURLMOPT_MAX_TOTAL_CONNECTIONS,
1430 config_.pool_max_handles);
1431
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 572 times.
 572 assert(mretval == CURLM_OK);
1432
1433 572 prng_.InitLocaltime();
1434
1435 572 thread_upload_ = 0;
1436 572 timestamp_last_throttle_report_ = 0;
1437 572 is_curl_debug_ = (getenv("_CVMFS_CURL_DEBUG") != NULL);
1438
1439 // Parsing environment variables
1440 572 if ((getenv("CVMFS_IPV4_ONLY") != NULL)
1441
2/6
✗ Branch 0 not taken.
✓ Branch 1 taken 572 times.
✗ Branch 3 not taken.
✗ Branch 4 not taken.
✗ Branch 5 not taken.
✓ Branch 6 taken 572 times.
 572 && (strlen(getenv("CVMFS_IPV4_ONLY")) > 0)) {
1442 ✗ opt_ipv4_only_ = true;
1443 } else {
1444 572 opt_ipv4_only_ = false;
1445 }
1446
1447
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 resolver_ = dns::CaresResolver::Create(opt_ipv4_only_, 2, 2000);
1448
1449 572 watch_fds_ = static_cast<struct pollfd *>(smalloc(4 * sizeof(struct pollfd)));
1450 572 watch_fds_size_ = 4;
1451 572 watch_fds_inuse_ = 0;
1452
1453
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 572 ssl_certificate_store_.UseSystemCertificatePath();
1454 572 }
1455
1456 1144 S3FanoutManager::~S3FanoutManager() {
1457 572 pthread_mutex_destroy(jobs_todo_lock_);
1458 572 free(jobs_todo_lock_);
1459 572 pthread_mutex_destroy(curl_handle_lock_);
1460 572 free(curl_handle_lock_);
1461
1462
1/2
✓ Branch 1 taken 572 times.
✗ Branch 2 not taken.
 1144 if (multi_threaded_.load() == 1) {
1463 // Shutdown I/O thread
1464 572 char buf = 'T';
1465 572 WritePipe(pipe_terminate_[1], &buf, 1);
1466 572 pthread_join(thread_upload_, NULL);
1467 }
1468 572 ClosePipe(pipe_terminate_);
1469 572 ClosePipe(pipe_jobs_);
1470 572 ClosePipe(pipe_completed_);
1471
1472 572 set<CURL *>::iterator i = pool_handles_idle_->begin();
1473 572 const set<CURL *>::const_iterator iEnd = pool_handles_idle_->end();
1474
2/2
✓ Branch 2 taken 1144 times.
✓ Branch 3 taken 572 times.
 1716 for (; i != iEnd; ++i) {
1475 1144 curl_easy_cleanup(*i);
1476 }
1477
1478 572 set<S3FanOutDnsEntry *>::iterator is = sharehandles_->begin();
1479 572 const set<S3FanOutDnsEntry *>::const_iterator isEnd = sharehandles_->end();
1480
2/2
✓ Branch 2 taken 484 times.
✓ Branch 3 taken 572 times.
 1056 for (; is != isEnd; ++is) {
1481 484 curl_share_cleanup((*is)->sharehandle);
1482 484 curl_slist_free_all((*is)->clist);
1483
1/2
✓ Branch 1 taken 484 times.
✗ Branch 2 not taken.
 484 delete *is;
1484 }
1485 572 pool_handles_idle_->clear();
1486 572 curl_sharehandles_->clear();
1487 572 sharehandles_->clear();
1488
1/2
✓ Branch 0 taken 572 times.
✗ Branch 1 not taken.
 572 delete active_requests_;
1489
1/2
✓ Branch 0 taken 572 times.
✗ Branch 1 not taken.
 572 delete pool_handles_idle_;
1490
1/2
✓ Branch 0 taken 572 times.
✗ Branch 1 not taken.
 572 delete pool_handles_inuse_;
1491
1/2
✓ Branch 0 taken 572 times.
✗ Branch 1 not taken.
 572 delete curl_sharehandles_;
1492
1/2
✓ Branch 0 taken 572 times.
✗ Branch 1 not taken.
 572 delete sharehandles_;
1493
1/2
✓ Branch 0 taken 572 times.
✗ Branch 1 not taken.
 572 delete user_agent_;
1494 572 curl_multi_cleanup(curl_multi_);
1495
1496
1/2
✓ Branch 0 taken 572 times.
✗ Branch 1 not taken.
 572 delete statistics_;
1497
1498
1/2
✓ Branch 0 taken 572 times.
✗ Branch 1 not taken.
 572 delete available_jobs_;
1499
1500 572 curl_global_cleanup();
1501 572 }
1502
1503 /**
1504 * Spawns the I/O worker thread. No way back except ~S3FanoutManager.
1505 */
1506 572 void S3FanoutManager::Spawn() {
1507 572 LogCvmfs(kLogS3Fanout, kLogDebug, "S3FanoutManager spawned");
1508
1509 572 const int retval = pthread_create(&thread_upload_, NULL, MainUpload,
1510 static_cast<void *>(this));
1511
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 572 times.
 572 assert(retval == 0);
1512
1513 572 multi_threaded_.fetch_add(1);
1514 572 }
1515
1516 132 const Statistics &S3FanoutManager::GetStatistics() { return *statistics_; }
1517
1518 /**
1519 * Push new job to be uploaded to the S3 cloud storage.
1520 */
1521 27280 void S3FanoutManager::PushNewJob(JobInfo *info) {
1522 27280 available_jobs_->Increment();
1523 27280 WritePipe(pipe_jobs_[1], &info, sizeof(info));
1524 27280 }
1525
1526 /**
1527 * Push completed job to list of completed jobs
1528 */
1529 27852 void S3FanoutManager::PushCompletedJob(JobInfo *info) {
1530 27852 WritePipe(pipe_completed_[1], &info, sizeof(info));
1531 27852 }
1532
1533 /**
1534 * Pop completed job
1535 */
1536 27852 JobInfo *S3FanoutManager::PopCompletedJob() {
1537 JobInfo *info;
1538
1/2
✓ Branch 1 taken 27852 times.
✗ Branch 2 not taken.
 27852 ReadPipe(pipe_completed_[0], &info, sizeof(info));
1539 27852 return info;
1540 }
1541
1542 //------------------------------------------------------------------------------
1543
1544
1545 ✗ string Statistics::Print() const {
1546 ✗ return "Transferred Bytes: " + StringifyInt(uint64_t(transferred_bytes))
1547 ✗ + "\n" + "Transfer duration: " + StringifyInt(uint64_t(transfer_time))
1548 ✗ + " s\n" + "Number of requests: " + StringifyInt(num_requests) + "\n"
1549 ✗ + "Number of retries: " + StringifyInt(num_retries) + "\n";
1550 }
1551
1552 } // namespace s3fanout
1553