Directory: | cvmfs/ |
---|---|
File: | cvmfs/upload_s3.h |
Date: | 2025-07-13 02:35:07 |
Exec | Total | Coverage | |
---|---|---|---|
Lines: | 9 | 10 | 90.0% |
Branches: | 3 | 8 | 37.5% |
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 <string> | ||
11 | #include <utility> | ||
12 | #include <vector> | ||
13 | |||
14 | #include "network/s3fanout.h" | ||
15 | #include "upload_facility.h" | ||
16 | #include "util/atomic.h" | ||
17 | #include "util/file_backed_buffer.h" | ||
18 | #include "util/pointer.h" | ||
19 | #include "util/single_copy.h" | ||
20 | |||
21 | namespace upload { | ||
22 | |||
23 | struct S3StreamHandle : public UploadStreamHandle { | ||
24 | 505 | S3StreamHandle(const CallbackTN *commit_callback, | |
25 | uint64_t in_memory_threshold, | ||
26 | const std::string &tmp_dir = "/tmp/") | ||
27 |
1/2✓ Branch 2 taken 505 times.
✗ Branch 3 not taken.
|
505 | : UploadStreamHandle(commit_callback) { |
28 |
2/4✓ Branch 1 taken 505 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 505 times.
✗ Branch 5 not taken.
|
505 | buffer = FileBackedBuffer::Create(in_memory_threshold, tmp_dir); |
29 | 505 | } | |
30 | |||
31 | // Ownership is later transferred to the S3 fanout | ||
32 | UniquePtr<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 bool Peek(const std::string &path); | ||
71 | virtual bool Mkdir(const std::string &path); | ||
72 | virtual bool PlaceBootstrappingShortcut(const shash::Any &object); | ||
73 | |||
74 | virtual unsigned int GetNumberOfErrors() const; | ||
75 | int64_t DoGetObjectSize(const std::string &file_name); | ||
76 | |||
77 | // Only for testing | ||
78 | 15 | s3fanout::S3FanoutManager *GetS3FanoutManager() { | |
79 | 15 | return s3fanout_mgr_.weak_ref(); | |
80 | } | ||
81 | |||
82 | private: | ||
83 | static const unsigned kDefaultPort = 80; | ||
84 | static const unsigned kHttpsPort = 443; | ||
85 | static const unsigned kDefaultNumParallelUploads = 16; | ||
86 | static const unsigned kDefaultNumRetries = 3; | ||
87 | static const unsigned kDefaultTimeoutSec = 60; | ||
88 | static const unsigned kDefaultBackoffInitMs = 100; | ||
89 | static const unsigned kDefaultBackoffMaxMs = 2000; | ||
90 | static const unsigned kInMemoryObjectThreshold = 500 * 1024; // 500KiB | ||
91 | |||
92 | // Used to make the async HTTP requests synchronous in Peek() Create(), | ||
93 | // and Upload() of single bits | ||
94 | struct RequestCtrl : SingleCopy { | ||
95 | 2560 | RequestCtrl() : return_code(-1), callback_forward(NULL) { | |
96 | 2560 | pipe_wait[0] = pipe_wait[1] = -1; | |
97 | 2560 | } | |
98 | |||
99 | void WaitFor(); | ||
100 | |||
101 | int return_code; | ||
102 | const CallbackTN *callback_forward; | ||
103 | std::string original_path; | ||
104 | int pipe_wait[2]; | ||
105 | }; | ||
106 | |||
107 | void OnReqComplete(const upload::UploaderResults &results, RequestCtrl *ctrl); | ||
108 | |||
109 | static void *MainCollectResults(void *data); | ||
110 | |||
111 | bool ParseSpoolerDefinition(const SpoolerDefinition &spooler_definition); | ||
112 | void UploadJobInfo(s3fanout::JobInfo *info); | ||
113 | |||
114 | s3fanout::JobInfo *CreateJobInfo(const std::string &path) const; | ||
115 | |||
116 | UniquePtr<s3fanout::S3FanoutManager> s3fanout_mgr_; | ||
117 | std::string repository_alias_; | ||
118 | std::string host_name_port_; | ||
119 | std::string host_name_; | ||
120 | std::string region_; | ||
121 | std::string flavor_; | ||
122 | std::string bucket_; | ||
123 | bool dns_buckets_; | ||
124 | int num_parallel_uploads_; | ||
125 | unsigned num_retries_; | ||
126 | unsigned timeout_sec_; | ||
127 | std::string access_key_; | ||
128 | std::string secret_key_; | ||
129 | s3fanout::AuthzMethods authz_method_; | ||
130 | bool peek_before_put_; | ||
131 | bool use_https_; | ||
132 | std::string proxy_; | ||
133 | |||
134 | const std::string temporary_path_; | ||
135 | mutable atomic_int32 io_errors_; | ||
136 | pthread_t thread_collect_results_; | ||
137 | |||
138 | std::string x_amz_acl_; | ||
139 | }; // S3Uploader | ||
140 | |||
141 | } // namespace upload | ||
142 | |||
143 | #endif // CVMFS_UPLOAD_S3_H_ | ||
144 |