GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/upload_s3.h
Date: 2026-08-30 02:40:36
Exec Total Coverage
Lines: 9 10 90.0%
Branches: 1 4 25.0%
Line Branch Exec Source
1 /**
2 * This file is part of the CernVM File System.
3 */
4
5 #ifndef CVMFS_UPLOAD_S3_H_
6 #define CVMFS_UPLOAD_S3_H_
7
8 #include <pthread.h>
9
10 #include <memory>
11 #include <string>
12 #include <utility>
13 #include <vector>
14
15 #include "network/s3fanout.h"
16 #include "upload_facility.h"
17 #include "util/atomic.h"
18 #include "util/file_backed_buffer.h"
19 #include "util/single_copy.h"
20
21 namespace upload {
22
23 struct S3StreamHandle : public UploadStreamHandle {
24 3030 S3StreamHandle(const CallbackTN *commit_callback,
25 uint64_t in_memory_threshold,
26 const std::string &tmp_dir = "/tmp/")
27 3030 : UploadStreamHandle(commit_callback) {
28
1/2
✓ Branch 1 taken 3030 times.
✗ Branch 2 not taken.
 3030 buffer.reset(FileBackedBuffer::Create(in_memory_threshold, tmp_dir));
29 3030 }
30
31 // Ownership is later transferred to the S3 fanout
32 std::unique_ptr<FileBackedBuffer> buffer;
33 };
34
35 /**
36 * The S3Spooler implements the AbstractSpooler interface to push files
37 * into a S3 CVMFS repository backend.
38 * For a detailed description of the classes interface please have a look into
39 * the AbstractSpooler base class.
40 */
41 class S3Uploader : public AbstractUploader {
42 public:
43 explicit S3Uploader(const SpoolerDefinition &spooler_definition);
44 virtual ~S3Uploader();
45 static bool WillHandle(const SpoolerDefinition &spooler_definition);
46
47 ✗ virtual std::string name() const { return "S3"; }
48
49 virtual bool Create();
50
51 /**
52 * Upload() is not done concurrently in the current implementation of the
53 * S3Spooler, since it is a simple move or copy of a file without CPU
54 * intensive operation
55 * This method calls NotifyListeners and invokes a callback for all
56 * registered listeners (see the Observable template for details).
57 */
58 virtual void DoUpload(const std::string &remote_path,
59 IngestionSource *source,
60 const CallbackTN *callback);
61
62 virtual UploadStreamHandle *InitStreamedUpload(
63 const CallbackTN *callback = NULL);
64 virtual void StreamedUpload(UploadStreamHandle *handle, UploadBuffer buffer,
65 const CallbackTN *callback);
66 virtual void FinalizeStreamedUpload(UploadStreamHandle *handle,
67 const shash::Any &content_hash);
68
69 virtual void DoRemoveAsync(const std::string &file_to_delete);
70 virtual void WaitForUpload() const;
71 virtual bool Peek(const std::string &path);
72 virtual bool Mkdir(const std::string &path);
73 virtual bool PlaceBootstrappingShortcut(const shash::Any &object);
74
75 virtual unsigned int GetNumberOfErrors() const;
76 int64_t DoGetObjectSize(const std::string &file_name);
77
78 // Only for testing
79 90 s3fanout::S3FanoutManager *GetS3FanoutManager() {
80 90 return s3fanout_mgr_.get();
81 }
82
83 private:
84 static const unsigned kDefaultPort = 80;
85 static const unsigned kHttpsPort = 443;
86 static const unsigned kDefaultNumParallelUploads = 16;
87 static const unsigned kDefaultNumRetries = 3;
88 static const unsigned kDefaultTimeoutSec = 60;
89 static const unsigned kDefaultBackoffInitMs = 100;
90 static const unsigned kDefaultBackoffMaxMs = 2000;
91 static const unsigned kInMemoryObjectThreshold = 500 * 1024; // 500KiB
92
93 // Used to make the async HTTP requests synchronous in Peek() Create(),
94 // and Upload() of single bits
95 struct RequestCtrl : SingleCopy {
96 15360 RequestCtrl() : return_code(-1), callback_forward(NULL) {
97 15360 pipe_wait[0] = pipe_wait[1] = -1;
98 15360 }
99
100 void WaitFor();
101
102 int return_code;
103 const CallbackTN *callback_forward;
104 std::string original_path;
105 int pipe_wait[2];
106 };
107
108 void OnReqComplete(const upload::UploaderResults &results, RequestCtrl *ctrl);
109
110 static void *MainCollectResults(void *data);
111
112 bool ParseSpoolerDefinition(const SpoolerDefinition &spooler_definition);
113 void UploadJobInfo(s3fanout::JobInfo *info);
114
115 // Per S3 spec the multi-object DELETE request supports at most 1000 keys.
116 // kDefaultBatchDeleteSize is the default flush threshold; it can be tuned
117 // via CVMFS_S3_BATCH_DELETE_SIZE up to kMaxBatchDeleteSize.
118 static const unsigned kDefaultBatchDeleteSize = 100;
119 static const unsigned kMaxBatchDeleteSize = 1000;
120
121 s3fanout::JobInfo *CreateJobInfo(const std::string &path) const;
122 void FlushDeleteBatch() const;
123
124 mutable std::unique_ptr<s3fanout::S3FanoutManager> s3fanout_mgr_;
125 std::string repository_alias_;
126 std::string host_name_port_;
127 std::string host_name_;
128 std::string region_;
129 std::string flavor_;
130 std::string bucket_;
131 bool dns_buckets_;
132 int num_parallel_uploads_;
133 unsigned num_retries_;
134 unsigned timeout_sec_;
135 std::string access_key_;
136 std::string secret_key_;
137 s3fanout::AuthzMethods authz_method_;
138 bool peek_before_put_;
139 bool use_https_;
140 bool batch_delete_enabled_;
141 unsigned batch_delete_size_;
142 std::string proxy_;
143
144 const std::string temporary_path_;
145 mutable atomic_int32 io_errors_;
146 pthread_t thread_collect_results_;
147
148 std::string x_amz_acl_;
149
150 mutable pthread_mutex_t delete_batch_mutex_;
151 mutable std::vector<std::string> pending_deletes_;
152 }; // S3Uploader
153
154 } // namespace upload
155
156 #endif // CVMFS_UPLOAD_S3_H_
157