GCC Code Coverage Report

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