GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/network/s3fanout.h
Date: 2026-08-30 02:40:36
Exec Total Coverage
Lines: 55 71 77.5%
Branches: 15 36 41.7%
Line Branch Exec Source
1 /**
2 * This file is part of the CernVM File System.
3 */
4
5 #ifndef CVMFS_NETWORK_S3FANOUT_H_
6 #define CVMFS_NETWORK_S3FANOUT_H_
7
8 #include <poll.h>
9 #include <semaphore.h>
10
11 #include <climits>
12 #include <cstdlib>
13 #include <map>
14 #include <memory>
15 #include <set>
16 #include <string>
17 #include <utility>
18 #include <vector>
19
20 #include "dns.h"
21 #include "duplex_curl.h" // IWYU pragma: keep
22 #include "ssl.h"
23 #include "util/concurrency.h"
24 #include "util/file_backed_buffer.h"
25 #include "util/prng.h"
26 #include "util/single_copy.h"
27 #include "util/smalloc.h"
28
29 namespace s3fanout {
30
31 enum AuthzMethods {
32 kAuthzAwsV2 = 0,
33 kAuthzAwsV4,
34 kAuthzAzure
35 };
36
37 /**
38 * Possible return values.
39 */
40 enum Failures {
41 kFailOk = 0,
42 kFailLocalIO,
43 kFailBadRequest,
44 kFailForbidden,
45 kFailHostResolve,
46 kFailHostConnection,
47 kFailNotFound,
48 kFailServiceUnavailable,
49 kFailRetry,
50 kFailInsufficientStorage,
51 kFailOther,
52
53 kFailNumEntries
54 }; // Failures
55
56
57 ✗ inline const char *Code2Ascii(const Failures error) {
58 const char *texts[kFailNumEntries + 1];
59 ✗ texts[0] = "S3: OK";
60 ✗ texts[1] = "S3: local I/O failure";
61 ✗ texts[2] = "S3: malformed URL (bad request)";
62 ✗ texts[3] = "S3: forbidden";
63 ✗ texts[4] = "S3: failed to resolve host address";
64 ✗ texts[5] = "S3: host connection problem";
65 ✗ texts[6] = "S3: not found";
66 ✗ texts[7] = "S3: service not available";
67 ✗ texts[8] = "S3: unknown service error, perhaps wrong authentication protocol";
68 ✗ texts[8] = "S3: too many requests, service asks for backoff and retry";
69 ✗ texts[9] = "S3: Insufficient Storage";
70 ✗ texts[10] = "unclassified failure";
71 ✗ return texts[error];
72 }
73
74
75 struct Statistics {
76 double transferred_bytes;
77 double transfer_time;
78 uint64_t num_requests;
79 uint64_t num_retries;
80 uint64_t ms_throttled; // Total waiting time imposed by HTTP 429 replies
81
82 390 Statistics() {
83 390 transferred_bytes = 0.0;
84 390 transfer_time = 0.0;
85 390 num_requests = 0;
86 390 num_retries = 0;
87 390 ms_throttled = 0;
88 390 }
89
90 std::string Print() const;
91 }; // Statistics
92
93
94 /**
95 * Contains all the information to specify an upload job.
96 */
97 struct JobInfo : SingleCopy {
98 enum RequestType {
99 kReqHeadOnly = 0, // peek
100 kReqHeadPut, // conditional upload of content-addressed objects
101 kReqPutCas, // immutable data object
102 kReqPutDotCvmfs, // one of the /.cvmfs... top level files
103 kReqPutHtml, // HTML file - display instead of downloading
104 kReqPutBucket, // bucket creation
105 kReqDelete,
106 kReqDeleteMulti, // S3 multi-object delete (POST /?delete)
107 };
108
109 const std::string object_key;
110 void *callback; // Callback to be called when job is finished
111 std::unique_ptr<FileBackedBuffer> origin;
112
113 // One constructor per destination
114 18645 JobInfo(const std::string &object_key,
115 void *callback,
116 FileBackedBuffer *origin)
117 18645 : object_key(object_key), origin(origin) {
118 18645 JobInfoInit();
119 18645 this->callback = callback;
120 18645 }
121 18645 void JobInfoInit() {
122 18645 curl_handle = NULL;
123 18645 http_headers = NULL;
124 18645 callback = NULL;
125 18645 request = kReqPutCas;
126 18645 error_code = kFailOk;
127 18645 http_error = 0;
128 18645 num_retries = 0;
129 18645 backoff_ms = 0;
130 18645 throttle_ms = 0;
131 18645 throttle_timestamp = 0;
132 18645 errorbuffer = reinterpret_cast<char *>(
133 18645 smalloc(sizeof(char) * CURL_ERROR_SIZE));
134 18645 }
135 18645 ~JobInfo() { free(errorbuffer); }
136
137 // For kReqDeleteMulti: keys included in the batch, and response body
138 std::vector<std::string> multi_delete_keys;
139 std::string response_body;
140
141 // Internal state, don't touch
142 CURL *curl_handle;
143 struct curl_slist *http_headers;
144 uint64_t payload_size;
145 RequestType request;
146 Failures error_code;
147 int http_error;
148 unsigned char num_retries;
149 // Exponential backoff with cutoff in case of errors
150 unsigned backoff_ms;
151 // Throttle imposed by HTTP 429 reply; mutually exclusive with backoff_ms
152 unsigned throttle_ms;
153 // Remember when the 429 reply came in to only throttle if still necessary
154 uint64_t throttle_timestamp;
155 char *errorbuffer;
156 }; // JobInfo
157
158 struct S3FanOutDnsEntry {
159 330 S3FanOutDnsEntry()
160 330 : counter(0)
161 330 , dns_name()
162 330 , ip()
163
1/2
✓ Branch 2 taken 330 times.
✗ Branch 3 not taken.
 330 , port("80")
164 330 , clist(NULL)
165 330 , sharehandle(NULL) { }
166 unsigned int counter;
167 std::string dns_name;
168 std::string ip;
169 std::string port;
170 struct curl_slist *clist;
171 CURLSH *sharehandle;
172 }; // S3FanOutDnsEntry
173
174
175 class S3FanoutManager : SingleCopy {
176 protected:
177 typedef SynchronizingCounter<uint32_t> Semaphore;
178
179 public:
180 // 250ms pause after HTTP 429 "Too Many Retries"
181 static const unsigned kDefault429ThrottleMs;
182 // Don't throttle for more than a few seconds
183 static const unsigned kMax429ThrottleMs;
184 // Report throttle operations only every so often
185 static const unsigned kThrottleReportIntervalSec;
186 static const unsigned kDefaultHTTPPort;
187 static const unsigned kDefaultHTTPSPort;
188
189 struct S3Config {
190 390 S3Config() {
191 390 authz_method = kAuthzAwsV2;
192 390 dns_buckets = true;
193
1/2
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
 390 protocol = "http";
194 390 pool_max_handles = 0;
195 390 opt_timeout_sec = 20;
196 390 opt_max_retries = 3;
197 390 opt_backoff_init_ms = 100;
198 390 opt_backoff_max_ms = 2000;
199
1/2
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
 390 x_amz_acl = "public-read";
200 390 }
201 std::string access_key;
202 std::string secret_key;
203 std::string hostname_port;
204 AuthzMethods authz_method;
205 std::string region;
206 std::string flavor;
207 std::string bucket;
208 bool dns_buckets;
209 std::string protocol;
210 uint32_t pool_max_handles;
211 unsigned opt_timeout_sec;
212 unsigned opt_max_retries;
213 unsigned opt_backoff_init_ms;
214 unsigned opt_backoff_max_ms;
215 std::string proxy;
216 std::string x_amz_acl;
217 };
218
219 static void DetectThrottleIndicator(const std::string &header, JobInfo *info);
220
221 explicit S3FanoutManager(const S3Config &config);
222
223 ~S3FanoutManager();
224
225 void Spawn();
226
227 void PushNewJob(JobInfo *info);
228 void PushCompletedJob(JobInfo *info);
229 JobInfo *PopCompletedJob();
230
231 const Statistics &GetStatistics();
232
233 private:
234 // Reflects the default Apache configuration of the local backend
235 static const char *kCacheControlCas; // Cache-Control: max-age=259200
236 static const unsigned kLowSpeedLimit = 1024; // Require at least 1kB/s
237
238 std::string MkDotCvmfsCacheControlHeader(unsigned defaultMaxAge = 61,
239 int overrideMaxAge = -1);
240
241 static int CallbackCurlSocket(CURL *easy, curl_socket_t s, int action,
242 void *userp, void *socketp);
243 static void *MainUpload(void *data);
244 std::vector<s3fanout::JobInfo *> jobs_todo_;
245 pthread_mutex_t *jobs_todo_lock_;
246 pthread_mutex_t *curl_handle_lock_;
247
248 CURL *AcquireCurlHandle() const;
249 void ReleaseCurlHandle(JobInfo *info, CURL *handle) const;
250 void InitPipeWatchFds();
251 int InitializeDnsSettings(CURL *handle, std::string remote_host) const;
252 void InitializeDnsSettingsCurl(CURL *handle, CURLSH *sharehandle,
253 curl_slist *clist) const;
254 Failures InitializeRequest(JobInfo *info, CURL *handle) const;
255 void SetUrlOptions(JobInfo *info) const;
256 void UpdateStatistics(CURL *handle);
257 bool CanRetry(const JobInfo *info);
258 void Backoff(JobInfo *info);
259 bool VerifyAndFinalize(const int curl_error, JobInfo *info);
260 std::string GetRequestString(const JobInfo &info) const;
261 std::string GetContentType(const JobInfo &info) const;
262 std::string GetUriEncode(const std::string &val, bool encode_slash) const;
263 std::string GetAwsV4SigningKey(const std::string &date) const;
264 bool MkPayloadHash(const JobInfo &info, std::string *hex_hash) const;
265 bool MkV2Authz(const JobInfo &info, std::vector<std::string> *headers) const;
266 bool MkV4Authz(const JobInfo &info, std::vector<std::string> *headers) const;
267 bool MkAzureAuthz(const JobInfo &info,
268 std::vector<std::string> *headers) const;
269 36840 std::string MkUrl(const std::string &objkey) const {
270
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 36840 times.
 36840 if (config_.dns_buckets) {
271 ✗ return config_.protocol + "://" + complete_hostname_ + "/" + objkey;
272 } else {
273
2/2
✓ Branch 1 taken 210 times.
✓ Branch 2 taken 36630 times.
 36840 if (objkey.empty()) {
274
2/4
✓ Branch 1 taken 210 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 210 times.
✗ Branch 5 not taken.
 420 return config_.protocol + "://" + complete_hostname_ + "/"
275
1/2
✓ Branch 2 taken 210 times.
✗ Branch 3 not taken.
 420 + config_.bucket;
276 }
277
2/4
✓ Branch 1 taken 36630 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 36630 times.
✗ Branch 5 not taken.
 73260 return config_.protocol + "://" + complete_hostname_ + "/"
278
3/6
✓ Branch 2 taken 36630 times.
✗ Branch 3 not taken.
✓ Branch 5 taken 36630 times.
✗ Branch 6 not taken.
✓ Branch 8 taken 36630 times.
✗ Branch 9 not taken.
 109890 + config_.bucket + "/" + objkey;
279 }
280 }
281 390 std::string MkCompleteHostname() {
282
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 390 times.
 390 if (config_.dns_buckets) {
283 ✗ return config_.bucket + "." + config_.hostname_port;
284 } else {
285 390 return config_.hostname_port;
286 }
287 }
288
289 const S3Config config_;
290 std::string complete_hostname_;
291
292 Prng prng_;
293 /**
294 * This is not strictly necessary but it helps the debugging
295 */
296 std::set<JobInfo *> *active_requests_;
297
298 std::set<CURL *> *pool_handles_idle_;
299 std::set<CURL *> *pool_handles_inuse_;
300 std::set<S3FanOutDnsEntry *> *sharehandles_;
301 std::map<CURL *, S3FanOutDnsEntry *> *curl_sharehandles_;
302 dns::CaresResolver *resolver_;
303 CURLM *curl_multi_;
304 std::string *user_agent_;
305
306 /**
307 * AWS4 signing keys are derived from the secret key, a region and a date.
308 * The signing key for current day can be cached.
309 */
310 mutable std::pair<std::string, std::string> last_signing_key_;
311
312 pthread_t thread_upload_;
313 atomic_int32 multi_threaded_;
314
315 struct pollfd *watch_fds_;
316 uint32_t watch_fds_size_;
317 uint32_t watch_fds_inuse_;
318 uint32_t watch_fds_max_;
319
320 // A pipe used to signal termination from S3FanoutManager to MainUpload
321 // thread. Anything written into it results in MainUpload thread exit.
322 int pipe_terminate_[2];
323 // A pipe to used to push jobs from S3FanoutManager to MainUpload thread.
324 // S3FanoutManager writes a JobInfo* pointer. MainUpload then reads the
325 // pointer and processes the job.
326 int pipe_jobs_[2];
327 // A pipe used to collect completed jobs. MainUpload writes in the
328 // pointer to the completed job. PopCompletedJob() used to
329 // retrieve pointer.
330 int pipe_completed_[2];
331
332 bool opt_ipv4_only_;
333
334 unsigned int max_available_jobs_;
335 Semaphore *available_jobs_;
336
337 // Writes and reads should be atomic because reading happens in a different
338 // thread than writing.
339 Statistics *statistics_;
340
341 // Report not every occurrence of throtteling but only every so often
342 uint64_t timestamp_last_throttle_report_;
343
344 bool is_curl_debug_;
345
346 /**
347 * Carries the path settings for SSL certificates
348 */
349 SslCertificateStore ssl_certificate_store_;
350
351 std::string dot_cvmfs_cache_control_header; // Cache-Control: max-age=...
352 }; // S3FanoutManager
353
354 std::string MkV2CanonicalResource(const std::string &bucket,
355 const std::string &object_key,
356 bool dns_buckets, bool multi_delete);
357 std::string ComposeDeleteMultiXml(const std::vector<std::string> &keys);
358 unsigned ParseDeleteMultiResponse(const std::string &response,
359 std::vector<std::string> *error_keys,
360 std::vector<std::string> *error_codes,
361 std::vector<std::string> *error_messages);
362
363 } // namespace s3fanout
364
365 #endif // CVMFS_NETWORK_S3FANOUT_H_
366