Skip to content

Commit dd51f09

Browse files
mateenali66meastp
andauthored
[BUG] Rewind the curl request body with a seek callback (#4557)
Co-authored-by: Mats Taraldsvik <1156416+meastp@users.noreply.github.com>
1 parent 0cd9e5b commit dd51f09

4 files changed

Lines changed: 162 additions & 1 deletion

File tree

‎CHANGELOG.md‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,11 @@ Increment the:
1515

1616
## [Unreleased]
1717

18+
* [BUG] Install a curl seek callback so an OTLP/HTTP export body can be rewound
19+
when libcurl restarts an upload, instead of failing with
20+
`CURLE_SEND_FAIL_REWIND` and dropping the batch
21+
([#4549](https://github.com/open-telemetry/opentelemetry-cpp/issues/4549))
22+
1823
* [DOC] Fix and clarify the `StartSpanOptions` documentation
1924
[#4526](https://github.com/open-telemetry/opentelemetry-cpp/pull/4526)
2025

‎ext/include/opentelemetry/ext/http/client/curl/http_operation_curl.h‎

Lines changed: 15 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -115,6 +115,20 @@ class HttpOperation
115115

116116
static size_t ReadMemoryCallback(char *buffer, size_t size, size_t nitems, void *userp);
117117

118+
/**
119+
* Reposition the request body for libcurl.
120+
*
121+
* libcurl calls this when it has to restart an upload it already began, for example after a
122+
* connection it had reused was closed before the response arrived. Without it libcurl has no way
123+
* to rewind the body and fails the transfer with CURLE_SEND_FAIL_REWIND.
124+
*
125+
* @param userp The HttpOperation, set through CURLOPT_SEEKDATA
126+
* @param offset Byte offset to seek to, interpreted relative to origin
127+
* @param origin One of SEEK_SET, SEEK_CUR or SEEK_END
128+
* @return CURL_SEEKFUNC_OK on success, CURL_SEEKFUNC_CANTSEEK to tell libcurl to find another way
129+
*/
130+
static int SeekCallback(void *userp, curl_off_t offset, int origin);
131+
118132
static int CurlLoggerCallback(const CURL * /* handle */,
119133
curl_infotype type,
120134
const char *data,
@@ -371,6 +385,7 @@ class HttpOperation
371385
std::future<CURLcode> result_future;
372386
};
373387
friend class HttpOperationAccessor;
388+
friend class HttpOperationTestPeer;
374389
std::unique_ptr<AsyncData> async_data_;
375390
};
376391
} // namespace curl

‎ext/src/http/client/curl/http_operation_curl.cc‎

Lines changed: 35 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,7 @@
1313
#include <chrono>
1414
#include <cmath>
1515
#include <cstdint>
16+
#include <cstdio>
1617
#include <cstring>
1718
#include <functional>
1819
#include <future>
@@ -334,6 +335,27 @@ size_t HttpOperation::ReadMemoryCallback(char *buffer, size_t size, size_t nitem
334335
return nwrite;
335336
}
336337

338+
int HttpOperation::SeekCallback(void *userp, curl_off_t offset, int origin)
339+
{
340+
HttpOperation *self = reinterpret_cast<HttpOperation *>(userp);
341+
if (nullptr == self)
342+
{
343+
return CURL_SEEKFUNC_CANTSEEK;
344+
}
345+
346+
// The body is a fully buffered span owned by the caller, so an absolute seek inside it is just a
347+
// move of the read cursor. Anything else is refused rather than approximated, because reporting
348+
// success without repositioning would resume the upload from the wrong offset and send a
349+
// truncated or misaligned body.
350+
if (origin != SEEK_SET || offset < 0 || static_cast<size_t>(offset) > self->request_body_.size())
351+
{
352+
return CURL_SEEKFUNC_CANTSEEK;
353+
}
354+
355+
self->request_nwrite_ = static_cast<size_t>(offset);
356+
return CURL_SEEKFUNC_OK;
357+
}
358+
337359
#if LIBCURL_VERSION_NUM >= 0x075000
338360
int HttpOperation::PreRequestCallback(void *clientp, char *, char *, int, int)
339361
{
@@ -1336,6 +1358,19 @@ CURLcode HttpOperation::Setup()
13361358
{
13371359
return rc;
13381360
}
1361+
1362+
rc = SetCurlPtrOption(CURLOPT_SEEKFUNCTION,
1363+
reinterpret_cast<void *>(&HttpOperation::SeekCallback));
1364+
if (rc != CURLE_OK)
1365+
{
1366+
return rc;
1367+
}
1368+
1369+
rc = SetCurlPtrOption(CURLOPT_SEEKDATA, this);
1370+
if (rc != CURLE_OK)
1371+
{
1372+
return rc;
1373+
}
13391374
}
13401375
else if (method_ == opentelemetry::ext::http::client::Method::Get)
13411376
{

‎ext/test/http/curl_http_test.cc‎

Lines changed: 107 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,21 +1,23 @@
11
// Copyright The OpenTelemetry Authors
22
// SPDX-License-Identifier: Apache-2.0
33

4+
#include <curl/curl.h>
45
#include <curl/curlver.h>
56
#include "gtest/gtest.h"
67

78
#ifdef ENABLE_OTLP_RETRY_PREVIEW
8-
# include <curl/curl.h>
99
# include "gmock/gmock.h"
1010
#endif // ENABLE_OTLP_RETRY_PREVIEW
1111

1212
#ifdef ENABLE_OTLP_COMPRESSION_PREVIEW
1313
# include <numeric>
1414
#endif // ENABLE_OTLP_COMPRESSION_PREVIEW
1515

16+
#include <algorithm>
1617
#include <atomic>
1718
#include <chrono>
1819
#include <condition_variable>
20+
#include <cstdio>
1921
#include <cstring>
2022
#include <map>
2123
#include <memory>
@@ -57,6 +59,27 @@ class HttpClientTestPeer
5759
public:
5860
static void ResetMultiHandle(HttpClient &client) { client.resetMultiHandle(); }
5961
};
62+
63+
class HttpOperationTestPeer
64+
{
65+
public:
66+
static int Seek(HttpOperation &operation, curl_off_t offset, int origin)
67+
{
68+
return HttpOperation::SeekCallback(&operation, offset, origin);
69+
}
70+
71+
static int SeekNullUserData(curl_off_t offset, int origin)
72+
{
73+
return HttpOperation::SeekCallback(nullptr, offset, origin);
74+
}
75+
76+
static size_t ReadCursor(const HttpOperation &operation) { return operation.request_nwrite_; }
77+
78+
static void SetReadCursor(HttpOperation &operation, size_t value)
79+
{
80+
operation.request_nwrite_ = value;
81+
}
82+
};
6083
} // namespace curl
6184
} // namespace client
6285
} // namespace http
@@ -405,6 +428,89 @@ TEST_F(BasicCurlHttpTests, SendPostRequest)
405428
session_manager->FinishAllSessions();
406429
}
407430

431+
// Send a body large enough to span several read callbacks and check it arrives whole, so a mistake
432+
// in either CURLOPT_READFUNCTION or the seek callback registered beside it shows up as a corrupted
433+
// or short upload.
434+
TEST_F(BasicCurlHttpTests, SendPostRequestWithMultiChunkBody)
435+
{
436+
received_requests_.clear();
437+
auto session_manager = std::make_shared<http_client::curl::HttpCurlClientFactory>()->Create();
438+
EXPECT_TRUE(session_manager != nullptr);
439+
440+
auto session = session_manager->CreateSession("http://127.0.0.1:19000");
441+
auto request = session->CreateRequest();
442+
request->SetUri("post/");
443+
request->SetMethod(http_client::Method::Post);
444+
445+
// Not a round number, so an off-by-one in the read cursor cannot land on a chunk boundary.
446+
constexpr size_t kBodySize = 257u * 1024u + 7u;
447+
http_client::Body body(kBodySize);
448+
for (size_t i = 0; i < kBodySize; ++i)
449+
{
450+
body[i] = static_cast<http_client::Byte>('a' + (i % 26));
451+
}
452+
const http_client::Body expected = body;
453+
454+
request->SetBody(body);
455+
request->AddHeader("Content-Type", "application/octet-stream");
456+
auto handler = std::make_shared<PostEventHandler>();
457+
session->SendRequest(handler);
458+
ASSERT_TRUE(waitForRequests(30, 1));
459+
session->FinishSession();
460+
ASSERT_TRUE(handler->is_called_.load(std::memory_order_acquire));
461+
ASSERT_TRUE(handler->got_response_.load(std::memory_order_acquire));
462+
463+
{
464+
std::unique_lock<std::mutex> lk(mtx_requests);
465+
ASSERT_EQ(received_requests_.size(), 1u);
466+
const auto &received = received_requests_[0].content;
467+
ASSERT_EQ(received.size(), expected.size());
468+
EXPECT_TRUE(std::equal(expected.begin(), expected.end(), received.begin()));
469+
}
470+
471+
session_manager->CancelAllSessions();
472+
session_manager->FinishAllSessions();
473+
}
474+
475+
// Cover both halves of the callback contract, the seeks it honours and the ones it refuses.
476+
TEST_F(BasicCurlHttpTests, SeekCallbackRepositionsTheRequestBody)
477+
{
478+
CustomEventHandler handler;
479+
http_client::HttpSslOptions no_ssl;
480+
http_client::Headers headers;
481+
const char *payload = "0123456789";
482+
http_client::Body body = {payload, payload + std::strlen(payload)};
483+
484+
curl::HttpOperation operation(http_client::Method::Post, "http://127.0.0.1:19000/post/", no_ssl,
485+
&handler, headers, body, http_client::Compression::kNone, false,
486+
curl::kDefaultHttpConnTimeout);
487+
488+
using Peer = curl::HttpOperationTestPeer;
489+
490+
Peer::SetReadCursor(operation, 10);
491+
EXPECT_EQ(CURL_SEEKFUNC_OK, Peer::Seek(operation, 4, SEEK_SET));
492+
EXPECT_EQ(4u, Peer::ReadCursor(operation));
493+
494+
// Rewinding to the start is the case libcurl actually asks for.
495+
EXPECT_EQ(CURL_SEEKFUNC_OK, Peer::Seek(operation, 0, SEEK_SET));
496+
EXPECT_EQ(0u, Peer::ReadCursor(operation));
497+
498+
// Seeking to exactly the end is in range and leaves nothing left to send.
499+
EXPECT_EQ(CURL_SEEKFUNC_OK, Peer::Seek(operation, 10, SEEK_SET));
500+
EXPECT_EQ(10u, Peer::ReadCursor(operation));
501+
502+
// Everything below is refused, and must leave the cursor where it was.
503+
Peer::SetReadCursor(operation, 3);
504+
505+
EXPECT_EQ(CURL_SEEKFUNC_CANTSEEK, Peer::Seek(operation, 11, SEEK_SET));
506+
EXPECT_EQ(CURL_SEEKFUNC_CANTSEEK, Peer::Seek(operation, -1, SEEK_SET));
507+
EXPECT_EQ(CURL_SEEKFUNC_CANTSEEK, Peer::Seek(operation, 0, SEEK_CUR));
508+
EXPECT_EQ(CURL_SEEKFUNC_CANTSEEK, Peer::Seek(operation, 0, SEEK_END));
509+
EXPECT_EQ(3u, Peer::ReadCursor(operation));
510+
511+
EXPECT_EQ(CURL_SEEKFUNC_CANTSEEK, Peer::SeekNullUserData(0, SEEK_SET));
512+
}
513+
408514
TEST_F(BasicCurlHttpTests, RequestTimeout)
409515
{
410516
received_requests_.clear();

0 commit comments

Comments
 (0)