GCC Code Coverage Report

Directory: cvmfs/
File: cvmfs/session_context.cc
Date: 2026-09-27 02:40:09
Exec Total Coverage
Lines: 140 230 60.9%
Branches: 79 309 25.6%
Line Branch Exec Source
1 /**
2 * This file is part of the CernVM File System.
3 */
4
5 #include "session_context.h"
6
7 #include <algorithm>
8 #include <limits>
9 #include <memory>
10
11 #include "curl/curl.h"
12 #include "gateway_util.h"
13 #include "json_document.h"
14 #include "json_document_write.h"
15 #include "swissknife_lease_curl.h"
16 #include "util/exception.h"
17 #include "util/string.h"
18
19 namespace {
20 // Maximum number of jobs during a session. No limit, for practical
21 // purposes. Note that we use uint32_t so that the Tube code works
22 // correctly with this limit on 32bit systems.
23 const uint32_t kMaxNumJobs = std::numeric_limits<uint32_t>::max();
24 } // namespace
25
26 namespace upload {
27
28 156 size_t SendCB(void *ptr, size_t size, size_t nmemb, void *userp) {
29 156 CurlSendPayload *payload = static_cast<CurlSendPayload *>(userp);
30
31 156 const size_t max_chunk_size = size * nmemb;
32
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 156 times.
 156 if (max_chunk_size < 1) {
33 ✗ return 0;
34 }
35
36 156 size_t current_chunk_size = 0;
37
2/2
✓ Branch 0 taken 208 times.
✓ Branch 1 taken 104 times.
 312 while (current_chunk_size < max_chunk_size) {
38
2/2
✓ Branch 1 taken 26 times.
✓ Branch 2 taken 182 times.
 208 if (payload->index < payload->json_message->size()) {
39 // Can add a chunk from the JSON message
40 26 const size_t read_size = std::min(
41 52 max_chunk_size - current_chunk_size,
42 26 payload->json_message->size() - payload->index);
43 26 current_chunk_size += read_size;
44 26 std::memcpy(ptr, payload->json_message->data() + payload->index,
45 read_size);
46 26 payload->index += read_size;
47 } else {
48 // Can add a chunk from the payload
49 182 const size_t max_read_size = max_chunk_size - current_chunk_size;
50 182 const unsigned nbytes = payload->pack_serializer->ProduceNext(
51 max_read_size,
52 static_cast<unsigned char *>(ptr) + current_chunk_size);
53 182 current_chunk_size += nbytes;
54
55
2/2
✓ Branch 0 taken 52 times.
✓ Branch 1 taken 130 times.
 182 if (!nbytes) {
56 52 break;
57 }
58 }
59 }
60
61 156 return current_chunk_size;
62 }
63
64 ✗ size_t RecvCB(void *buffer, size_t size, size_t nmemb, void *userp) {
65 ✗ std::string *my_buffer = static_cast<std::string *>(userp);
66
67 ✗ if (size * nmemb < 1) {
68 ✗ return 0;
69 }
70
71 ✗ *my_buffer = static_cast<char *>(buffer);
72
73 ✗ return my_buffer->size();
74 }
75
76 219 SessionContextBase::SessionContextBase()
77 219 : upload_results_(kMaxNumJobs)
78 219 , api_url_()
79 219 , session_token_()
80 219 , key_id_()
81 219 , secret_()
82 219 , max_pack_size_(ObjectPack::kDefaultLimit)
83 219 , active_handles_()
84 219 , current_pack_(NULL)
85 219 , current_pack_mtx_()
86 219 , bytes_committed_(0)
87 219 , bytes_dispatched_(0)
88 219 , initialized_(false) { }
89
90 438 SessionContextBase::~SessionContextBase() { }
91
92 219 bool SessionContextBase::Initialize(const std::string &api_url,
93 const std::string &session_token,
94 const std::string &key_id,
95 const std::string &secret,
96 uint64_t max_pack_size,
97 uint64_t max_queue_size) {
98 219 bool ret = true;
99
100 // Initialize session context lock
101 pthread_mutexattr_t attr;
102 219 if (pthread_mutexattr_init(&attr)
103
1/2
✓ Branch 1 taken 219 times.
✗ Branch 2 not taken.
 219 || pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_RECURSIVE)
104
1/2
✓ Branch 1 taken 219 times.
✗ Branch 2 not taken.
 219 || pthread_mutex_init(&current_pack_mtx_, &attr)
105
3/6
✓ Branch 0 taken 219 times.
✗ Branch 1 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 219 times.
✗ Branch 5 not taken.
✓ Branch 6 taken 219 times.
 438 || pthread_mutexattr_destroy(&attr)) {
106 ✗ LogCvmfs(kLogUploadGateway, kLogStderr,
107 "Could not initialize SessionContext lock.");
108 ✗ return false;
109 }
110
111 // Set upstream URL and session token
112
1/2
✓ Branch 1 taken 219 times.
✗ Branch 2 not taken.
 219 api_url_ = api_url;
113
1/2
✓ Branch 1 taken 219 times.
✗ Branch 2 not taken.
 219 session_token_ = session_token;
114
1/2
✓ Branch 1 taken 219 times.
✗ Branch 2 not taken.
 219 key_id_ = key_id;
115
1/2
✓ Branch 1 taken 219 times.
✗ Branch 2 not taken.
 219 secret_ = secret;
116 219 max_pack_size_ = max_pack_size;
117
118 219 bytes_committed_ = 0u;
119 219 bytes_dispatched_ = 0u;
120
121
2/4
✓ Branch 1 taken 219 times.
✗ Branch 2 not taken.
✗ Branch 3 not taken.
✓ Branch 4 taken 219 times.
 219 assert(upload_results_.IsEmpty());
122
123 // Ensure that there are not open object packs
124
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 219 times.
 219 if (current_pack_) {
125 ✗ LogCvmfs(
126 kLogUploadGateway, kLogStderr,
127 "Could not initialize SessionContext - Existing open object packs.");
128 ✗ ret = false;
129 }
130
131
3/6
✓ Branch 1 taken 219 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 219 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 219 times.
✗ Branch 6 not taken.
 219 ret = InitializeDerived(max_queue_size) && ret;
132
133 219 initialized_ = true;
134
135 219 return ret;
136 }
137
138 219 bool SessionContextBase::Finalize(bool commit, const std::string &old_root_hash,
139 const std::string &new_root_hash,
140 const RepositoryTag &tag) {
141
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 219 times.
 219 assert(active_handles_.empty());
142
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 219 times.
 219 if (!initialized_) {
143 ✗ assert(!commit);
144 ✗ return true;
145 }
146
147 {
148 219 const MutexLockGuard lock(current_pack_mtx_);
149
150
5/6
✓ Branch 0 taken 104 times.
✓ Branch 1 taken 115 times.
✓ Branch 3 taken 104 times.
✗ Branch 4 not taken.
✓ Branch 5 taken 104 times.
✓ Branch 6 taken 115 times.
 219 if (current_pack_ && current_pack_->GetNoObjects() > 0) {
151
1/2
✓ Branch 1 taken 104 times.
✗ Branch 2 not taken.
 104 Dispatch();
152 104 current_pack_ = NULL;
153 }
154 219 }
155
156 219 bool results = true;
157
2/2
✓ Branch 1 taken 806 times.
✓ Branch 2 taken 219 times.
 1025 while (!upload_results_.IsEmpty()) {
158 806 Future<bool> *future = upload_results_.PopBack();
159
2/4
✓ Branch 1 taken 806 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 806 times.
✗ Branch 4 not taken.
 806 results = future->Get() && results;
160
1/2
✓ Branch 0 taken 806 times.
✗ Branch 1 not taken.
 806 delete future;
161 }
162
163
2/2
✓ Branch 0 taken 182 times.
✓ Branch 1 taken 37 times.
 219 if (commit) {
164
3/6
✓ Branch 1 taken 182 times.
✗ Branch 2 not taken.
✗ Branch 4 not taken.
✓ Branch 5 taken 182 times.
✗ Branch 6 not taken.
✓ Branch 7 taken 182 times.
 182 if (old_root_hash.empty() || new_root_hash.empty()) {
165 ✗ return false;
166 }
167 182 const bool commit_result = Commit(old_root_hash, new_root_hash, tag);
168
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 182 times.
 182 if (!commit_result) {
169 ✗ LogCvmfs(kLogUploadGateway, kLogStderr,
170 "SessionContext: could not commit session. Aborting.");
171 ✗ FinalizeDerived();
172 ✗ pthread_mutex_destroy(&current_pack_mtx_);
173 ✗ initialized_ = false;
174 ✗ return false;
175 }
176 }
177
178
2/4
✓ Branch 1 taken 219 times.
✗ Branch 2 not taken.
✓ Branch 3 taken 219 times.
✗ Branch 4 not taken.
 219 results &= FinalizeDerived() && (bytes_committed_ == bytes_dispatched_);
179
180 219 pthread_mutex_destroy(&current_pack_mtx_);
181
182 219 initialized_ = false;
183
184 219 return results;
185 }
186
187 1352 ObjectPack::BucketHandle SessionContextBase::NewBucket() {
188 1352 const MutexLockGuard lock(current_pack_mtx_);
189
2/2
✓ Branch 0 taken 416 times.
✓ Branch 1 taken 936 times.
 1352 if (!current_pack_) {
190
2/4
✓ Branch 1 taken 416 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 416 times.
✗ Branch 5 not taken.
 416 current_pack_ = new ObjectPack(max_pack_size_);
191 }
192
1/2
✓ Branch 1 taken 1352 times.
✗ Branch 2 not taken.
 1352 ObjectPack::BucketHandle hd = current_pack_->NewBucket();
193
1/2
✓ Branch 1 taken 1352 times.
✗ Branch 2 not taken.
 1352 active_handles_.push_back(hd);
194 1352 return hd;
195 1352 }
196
197 1768 bool SessionContextBase::CommitBucket(const ObjectPack::BucketContentType type,
198 const shash::Any &id,
199 const ObjectPack::BucketHandle handle,
200 const std::string &name,
201 const bool force_dispatch) {
202 1768 const MutexLockGuard lock(current_pack_mtx_);
203
204
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 1768 times.
 1768 if (!current_pack_) {
205 ✗ LogCvmfs(kLogUploadGateway, kLogStderr,
206 "Error: Called SessionBaseContext::CommitBucket without an open "
207 "ObjectPack.");
208 ✗ return false;
209 }
210
211 1768 const uint64_t size0 = current_pack_->size();
212
1/2
✓ Branch 1 taken 1768 times.
✗ Branch 2 not taken.
 1768 const bool committed = current_pack_->CommitBucket(type, id, handle, name);
213
214
2/2
✓ Branch 0 taken 1352 times.
✓ Branch 1 taken 416 times.
 1768 if (committed) { // Current pack is still not full
215
1/2
✓ Branch 3 taken 1352 times.
✗ Branch 4 not taken.
 2704 active_handles_.erase(
216
1/2
✓ Branch 3 taken 1352 times.
✗ Branch 4 not taken.
 1352 std::remove(active_handles_.begin(), active_handles_.end(), handle),
217 1352 active_handles_.end());
218 1352 const uint64_t size1 = current_pack_->size();
219 1352 bytes_committed_ += size1 - size0;
220
2/2
✓ Branch 0 taken 312 times.
✓ Branch 1 taken 1040 times.
 1352 if (force_dispatch) {
221
1/2
✓ Branch 1 taken 312 times.
✗ Branch 2 not taken.
 312 Dispatch();
222 312 current_pack_ = NULL;
223 }
224 } else { // Current pack is full and can be dispatched
225 416 uint64_t new_size = 0;
226
2/2
✓ Branch 0 taken 26 times.
✓ Branch 1 taken 390 times.
 416 if (handle->capacity > max_pack_size_) {
227 26 new_size = handle->capacity + 1;
228 } else {
229 390 new_size = max_pack_size_;
230 }
231
2/4
✓ Branch 1 taken 416 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 416 times.
✗ Branch 5 not taken.
 416 ObjectPack *new_pack = new ObjectPack(new_size);
232
2/2
✓ Branch 1 taken 572 times.
✓ Branch 2 taken 416 times.
 988 for (size_t i = 0u; i < active_handles_.size(); ++i) {
233
1/2
✓ Branch 2 taken 572 times.
✗ Branch 3 not taken.
 572 current_pack_->TransferBucket(active_handles_[i], new_pack);
234 }
235
236
2/2
✓ Branch 1 taken 390 times.
✓ Branch 2 taken 26 times.
 416 if (current_pack_->GetNoObjects() > 0) {
237
1/2
✓ Branch 1 taken 390 times.
✗ Branch 2 not taken.
 390 Dispatch();
238 }
239 416 current_pack_ = new_pack;
240
241
1/2
✓ Branch 1 taken 416 times.
✗ Branch 2 not taken.
 416 CommitBucket(type, id, handle, name, false);
242 }
243
244 1768 return true;
245 1768 }
246
247 806 void SessionContextBase::Dispatch() {
248 806 const MutexLockGuard lock(current_pack_mtx_);
249
250
1/2
✗ Branch 0 not taken.
✓ Branch 1 taken 806 times.
 806 if (!current_pack_) {
251 ✗ return;
252 }
253
254 806 bytes_dispatched_ += current_pack_->size();
255
2/4
✓ Branch 1 taken 806 times.
✗ Branch 2 not taken.
✓ Branch 4 taken 806 times.
✗ Branch 5 not taken.
 806 upload_results_.EnqueueFront(DispatchObjectPack(current_pack_));
256
1/2
✓ Branch 1 taken 806 times.
✗ Branch 2 not taken.
 806 }
257
258 219 SessionContext::SessionContext()
259 219 : SessionContextBase(), upload_jobs_(), worker_() { }
260
261 219 bool SessionContext::InitializeDerived(uint64_t max_queue_size) {
262 // Start worker thread
263 438 upload_jobs_ = std::unique_ptr<Tube<UploadJob> >(
264
1/2
✓ Branch 2 taken 219 times.
✗ Branch 3 not taken.
 438 new Tube<UploadJob>(max_queue_size));
265
266 219 const int retval = pthread_create(&worker_, NULL, UploadLoop,
267 reinterpret_cast<void *>(this));
268
269 219 return !retval;
270 }
271
272 219 bool SessionContext::FinalizeDerived() {
273 // Note: in FinalizedDerived, we know that the worker is running. The
274 // SessionContext is called only from GatewayUploader::FinalizeSession(),
275 // which in turn is from Spooler::FinalizeSession(). The Spooler ensures
276 // that GatewayUploader::Initialize() is called on construction.
277 //
278 // TODO(jblomer): Refactor SessionContext (and Uploader*) classes to
279 // use a factory method for construction.
280 //
281 219 upload_jobs_->EnqueueFront(&terminator_);
282 219 pthread_join(worker_, NULL);
283
284 219 return true;
285 }
286
287 ✗ bool SessionContext::Commit(const std::string &old_root_hash,
288 const std::string &new_root_hash,
289 const RepositoryTag &tag) {
290 ✗ JsonStringGenerator request_input;
291 ✗ request_input.Add("old_root_hash", old_root_hash);
292 ✗ request_input.Add("new_root_hash", new_root_hash);
293 ✗ request_input.Add("tag_name", tag.name());
294 // Channels are no longer supported: send 0 (i.e. kChannelTrunk) for
295 // backwards compatibility with existing gateways
296 //
297 ✗ request_input.Add("tag_channel", 0);
298 ✗ request_input.Add("tag_description", tag.description());
299 // Space-separated list of tags the receiver should remove in the same history
300 // transaction as this commit. Only sent for tag-removal commits; omitted
301 // otherwise for backwards compatibility with older gateways.
302 ✗ if (!tag.delete_tags().empty()) {
303 ✗ request_input.Add("delete_tags", tag.delete_tags());
304 }
305 ✗ if (tag.auto_tag_threshold() > 0) {
306 ✗ request_input.Add("auto_tag_threshold",
307 ✗ static_cast<int64_t>(tag.auto_tag_threshold()));
308 }
309 ✗ const std::string request = request_input.GenerateString();
310 ✗ CurlBuffer buffer;
311 ✗ return MakeEndRequest("POST", key_id_, secret_, session_token_, api_url_,
312 ✗ request, &buffer);
313 }
314
315 806 Future<bool> *SessionContext::DispatchObjectPack(ObjectPack *pack) {
316 806 UploadJob *job = new UploadJob;
317 806 Future<bool> *result = new Future<bool>();
318 806 job->pack = pack;
319 806 job->result = result;
320 806 upload_jobs_->EnqueueFront(job);
321 806 return result;
322 }
323
324 ✗ bool SessionContext::DoUpload(const SessionContext::UploadJob *job) {
325 // Set up the object pack serializer
326 ✗ ObjectPackProducer serializer(job->pack);
327
328 ✗ shash::Any payload_digest(shash::kSha1);
329 ✗ serializer.GetDigest(&payload_digest);
330 ✗ const std::string json_msg = "{\"session_token\" : \"" + session_token_
331 ✗ + "\", \"payload_digest\" : \""
332 ✗ + payload_digest.ToString(false)
333 ✗ + "\", \"header_size\" : \""
334 ✗ + StringifyInt(serializer.GetHeaderSize())
335 ✗ + "\", \"api_version\" : \""
336 ✗ + StringifyInt(gateway::APIVersion()) + "\"}";
337
338 // Compute HMAC
339 ✗ shash::Any hmac(shash::kSha1);
340 ✗ shash::HmacString(secret_, json_msg, &hmac);
341
342 CurlSendPayload payload;
343 ✗ payload.json_message = &json_msg;
344 ✗ payload.pack_serializer = &serializer;
345 ✗ payload.index = 0;
346
347 ✗ const size_t payload_size = json_msg.size() + serializer.GetHeaderSize()
348 ✗ + job->pack->size();
349
350 // Prepare the Curl POST request
351 ✗ CURL *h_curl = curl_easy_init();
352
353 ✗ if (!h_curl) {
354 ✗ return false;
355 }
356
357 // Set HTTP headers (Authorization and Message-Size)
358 ✗ std::string header_str = std::string("Authorization: ") + key_id_ + " "
359 ✗ + Base64(hmac.ToString(false));
360 ✗ struct curl_slist *auth_header = NULL;
361 ✗ auth_header = curl_slist_append(auth_header, header_str.c_str());
362 ✗ header_str = std::string("Message-Size: ") + StringifyInt(json_msg.size());
363 ✗ auth_header = curl_slist_append(auth_header, header_str.c_str());
364 ✗ curl_easy_setopt(h_curl, CURLOPT_HTTPHEADER, auth_header);
365
366 ✗ std::string reply;
367 ✗ curl_easy_setopt(h_curl, CURLOPT_NOPROGRESS, 1L);
368 ✗ curl_easy_setopt(h_curl, CURLOPT_USERAGENT, "cvmfs/" CVMFS_VERSION);
369 ✗ curl_easy_setopt(h_curl, CURLOPT_MAXREDIRS, 50L);
370 ✗ curl_easy_setopt(h_curl, CURLOPT_CUSTOMREQUEST, "POST");
371 ✗ curl_easy_setopt(h_curl, CURLOPT_URL, (api_url_ + "/payloads").c_str());
372 ✗ curl_easy_setopt(h_curl, CURLOPT_POSTFIELDS, NULL);
373 ✗ curl_easy_setopt(h_curl, CURLOPT_POSTFIELDSIZE_LARGE,
374 static_cast<curl_off_t>(payload_size));
375 ✗ curl_easy_setopt(h_curl, CURLOPT_READDATA, &payload);
376 ✗ curl_easy_setopt(h_curl, CURLOPT_READFUNCTION, SendCB);
377 ✗ curl_easy_setopt(h_curl, CURLOPT_WRITEFUNCTION, RecvCB);
378 ✗ curl_easy_setopt(h_curl, CURLOPT_WRITEDATA, &reply);
379
380 // Perform the Curl POST request
381 ✗ const CURLcode ret = curl_easy_perform(h_curl);
382 ✗ if (ret) {
383 ✗ LogCvmfs(kLogUploadGateway, kLogStderr,
384 "SessionContext::DoUpload - curl_easy_perform failed: %d", ret);
385 }
386
387 ✗ const std::unique_ptr<JsonDocument> reply_json(JsonDocument::Create(reply));
388 ✗ const JSON *reply_status = JsonDocument::SearchInObject(
389 reply_json->root(), "status", JSON_STRING);
390 const bool ok = (reply_status != NULL
391 ✗ && std::string(reply_status->get<std::string>()) == "ok");
392 ✗ if (!ok) {
393 ✗ LogCvmfs(kLogUploadGateway, kLogStderr,
394 "SessionContext::DoUpload - error reply: %s", reply.c_str());
395 }
396
397 ✗ curl_easy_cleanup(h_curl);
398 ✗ h_curl = NULL;
399
400 ✗ return ok && !ret;
401 }
402
403 219 void *SessionContext::UploadLoop(void *data) {
404 219 SessionContext *ctx = reinterpret_cast<SessionContext *>(data);
405 UploadJob *job;
406
407 while (true) {
408 1025 job = ctx->upload_jobs_->PopBack();
409
2/2
✓ Branch 0 taken 219 times.
✓ Branch 1 taken 806 times.
 1025 if (job == &terminator_)
410 219 return NULL;
411
1/2
✗ Branch 1 not taken.
✓ Branch 2 taken 806 times.
 806 if (!ctx->DoUpload(job)) {
412 ✗ PANIC(kLogStderr, "SessionContext: could not submit payload. Aborting.");
413 }
414 806 job->result->Set(true);
415
1/2
✓ Branch 0 taken 806 times.
✗ Branch 1 not taken.
 806 delete job->pack;
416
1/2
✓ Branch 0 taken 806 times.
✗ Branch 1 not taken.
 806 delete job;
417 }
418 }
419
420 SessionContext::UploadJob SessionContext::terminator_;
421
422 } // namespace upload
423