wbtiger
/
foundationdb
mirror of https://github.com/apple/foundationdb.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Adding Simulated HTTP Server and refactoring HTTP code (#10112) 

* Adding Simulated HTTP Server and refactoring HTTP code

* fixing formatting

* fixing merge conflicts

* fixing more merge conflicts

* code review feedback

* changing reference counted interface

* more fixes

* fixing ide build i guess
This commit is contained in:
Josh Slocum 2023-05-05 12:19:17 -05:00 committed by GitHub
parent fb2fc6a260
commit a4dffa087a
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
15 changed files with 1116 additions and 407 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

86
fdbclient/RESTClient.actor.cpp
View File

@ -87,32 +87,35 @@ std::unordered_map<std::string, int> RESTClient::getKnobs() const {
return knobs.get(); return knobs.get();
} }
ACTOR Future<Reference<HTTP::Response>> doRequest_impl(Reference<RESTClient> client, ACTOR Future<Reference<HTTP::IncomingResponse>> doRequest_impl(Reference<RESTClient> client,
std::string verb, std::string verb,
HTTP::Headers headers, HTTP::Headers headers,
RESTUrl url, RESTUrl url,
std::set<unsigned int> successCodes) { std::set<unsigned int> successCodes) {
state Reference<HTTP::OutgoingRequest> req = makeReference<HTTP::OutgoingRequest>();
state UnsentPacketQueue content; state UnsentPacketQueue content;
state int contentLen = url.body.size(); req->data.content = &content;
req->data.contentLen = url.body.size();
req->data.headers["Host"] = url.host;
req->verb = verb;
req->resource = url.resource;
if (FLOW_KNOBS->REST_LOG_LEVEL >= RESTLogSeverity::VERBOSE) { if (FLOW_KNOBS->REST_LOG_LEVEL >= RESTLogSeverity::VERBOSE) {
TraceEvent("RESTDoRequestImpl").detail("Url", url.toString()); TraceEvent("RESTDoRequestImpl").detail("Url", url.toString());
} }
if (url.body.size() > 0) { if (url.body.size() > 0) {
PacketWriter pw(content.getWriteBuffer(url.body.size()), nullptr, Unversioned()); PacketWriter pw(req->data.content->getWriteBuffer(url.body.size()), nullptr, Unversioned());
pw.serializeBytes(url.body); pw.serializeBytes(url.body);
} }
std::string statsKey = RESTClient::getStatsKey(url.service, url.service); std::string statsKey = RESTClient::getStatsKey(url.host, url.service);
auto sItr = client->statsMap.find(statsKey); auto sItr = client->statsMap.find(statsKey);
if (sItr == client->statsMap.end()) { if (sItr == client->statsMap.end()) {
client->statsMap.emplace(statsKey, std::make_unique<RESTClient::Stats>(statsKey)); client->statsMap.emplace(statsKey, std::make_unique<RESTClient::Stats>(statsKey));
} }
headers["Content-Length"] = format("%d", contentLen);
headers["Host"] = url.host;
state int maxTries = std::min(client->knobs.request_tries, client->knobs.connect_tries); state int maxTries = std::min(client->knobs.request_tries, client->knobs.connect_tries);
state int thisTry = 1; state int thisTry = 1;
state double nextRetryDelay = 2.0; state double nextRetryDelay = 2.0;
@ -125,7 +128,8 @@ ACTOR Future<Reference<HTTP::Response>> doRequest_impl(Reference<RESTClient> cli
state Optional<Error> err; state Optional<Error> err;
state Optional<NetworkAddress> remoteAddress; state Optional<NetworkAddress> remoteAddress;
state bool connectionEstablished = false; state bool connectionEstablished = false;
state Reference<HTTP::Response> r;
state Reference<HTTP::IncomingResponse> r;
try { try {
// Start connecting // Start connecting
@ -138,21 +142,13 @@ ACTOR Future<Reference<HTTP::Response>> doRequest_impl(Reference<RESTClient> cli
connectionEstablished = true; connectionEstablished = true;
remoteAddress = rconn.conn->getPeerAddress(); remoteAddress = rconn.conn->getPeerAddress();
Reference<HTTP::Response> _r = wait(timeoutError(HTTP::doRequest(rconn.conn, Reference<HTTP::IncomingResponse> _r = wait(timeoutError(
verb, HTTP::doRequest(rconn.conn, req, sendReceiveRate, &statsPtr->bytes_sent, sendReceiveRate), reqTimeout));
url.resource,
headers,
contentLen > 0 ? &content : nullptr,
contentLen,
sendReceiveRate,
&statsPtr->bytes_sent,
sendReceiveRate),
reqTimeout));
r = _r; r = _r;
// Since the response was parsed successfully (which is why we are here) reuse the connection unless we // Since the response was parsed successfully (which is why we are here) reuse the connection unless we
// received the "Connection: close" header. // received the "Connection: close" header.
if (r->headers["Connection"] != "close") { if (r->data.headers["Connection"] != "close") {
client->conectionPool->returnConnection(connectPoolKey, rconn, client->knobs.connection_pool_size); client->conectionPool->returnConnection(connectPoolKey, rconn, client->knobs.connection_pool_size);
} }
rconn.conn.clear(); rconn.conn.clear();
@ -217,8 +213,8 @@ ACTOR Future<Reference<HTTP::Response>> doRequest_impl(Reference<RESTClient> cli
if (retryable) { if (retryable) {
// If r is valid then obey the Retry-After response header if present. // If r is valid then obey the Retry-After response header if present.
if (r) { if (r) {
auto iRetryAfter = r->headers.find("Retry-After"); auto iRetryAfter = r->data.headers.find("Retry-After");
if (iRetryAfter != r->headers.end()) { if (iRetryAfter != r->data.headers.end()) {
event.detail("RetryAfterHeader", iRetryAfter->second); event.detail("RetryAfterHeader", iRetryAfter->second);
char* pEnd; char* pEnd;
double retryAfter = strtod(iRetryAfter->second.c_str(), &pEnd); double retryAfter = strtod(iRetryAfter->second.c_str(), &pEnd);
@ -270,10 +266,10 @@ ACTOR Future<Reference<HTTP::Response>> doRequest_impl(Reference<RESTClient> cli
} }
} }
Future<Reference<HTTP::Response>> RESTClient::doPutOrPost(const std::string& verb, Future<Reference<HTTP::IncomingResponse>> RESTClient::doPutOrPost(const std::string& verb,
Optional<HTTP::Headers> optHeaders, Optional<HTTP::Headers> optHeaders,
RESTUrl& url, RESTUrl& url,
std::set<unsigned int> successCodes) { std::set<unsigned int> successCodes) {
HTTP::Headers headers; HTTP::Headers headers;
if (optHeaders.present()) { if (optHeaders.present()) {
headers = optHeaders.get(); headers = optHeaders.get();
@ -282,17 +278,17 @@ Future<Reference<HTTP::Response>> RESTClient::doPutOrPost(const std::string& ver
return doRequest_impl(Reference<RESTClient>::addRef(this), verb, headers, url, successCodes); return doRequest_impl(Reference<RESTClient>::addRef(this), verb, headers, url, successCodes);
} }
Future<Reference<HTTP::Response>> RESTClient::doPost(const std::string& fullUrl, Future<Reference<HTTP::IncomingResponse>> RESTClient::doPost(const std::string& fullUrl,
const std::string& requestBody, const std::string& requestBody,
Optional<HTTP::Headers> optHeaders) { Optional<HTTP::Headers> optHeaders) {
RESTUrl url(fullUrl, requestBody); RESTUrl url(fullUrl, requestBody);
TRACE_REST_OP("DoPost", url); TRACE_REST_OP("DoPost", url);
return doPutOrPost(HTTP::HTTP_VERB_POST, optHeaders, url, { HTTP::HTTP_STATUS_CODE_OK }); return doPutOrPost(HTTP::HTTP_VERB_POST, optHeaders, url, { HTTP::HTTP_STATUS_CODE_OK });
} }
Future<Reference<HTTP::Response>> RESTClient::doPut(const std::string& fullUrl, Future<Reference<HTTP::IncomingResponse>> RESTClient::doPut(const std::string& fullUrl,
const std::string& requestBody, const std::string& requestBody,
Optional<HTTP::Headers> optHeaders) { Optional<HTTP::Headers> optHeaders) {
RESTUrl url(fullUrl, requestBody); RESTUrl url(fullUrl, requestBody);
TRACE_REST_OP("DoPut", url); TRACE_REST_OP("DoPut", url);
return doPutOrPost( return doPutOrPost(
@ -304,10 +300,10 @@ Future<Reference<HTTP::Response>> RESTClient::doPut(const std::string& fullUrl,
{ HTTP::HTTP_STATUS_CODE_OK, HTTP::HTTP_STATUS_CODE_CREATED, HTTP::HTTP_STATUS_CODE_NO_CONTENT }); { HTTP::HTTP_STATUS_CODE_OK, HTTP::HTTP_STATUS_CODE_CREATED, HTTP::HTTP_STATUS_CODE_NO_CONTENT });
} }
Future<Reference<HTTP::Response>> RESTClient::doGetHeadDeleteOrTrace(const std::string& verb, Future<Reference<HTTP::IncomingResponse>> RESTClient::doGetHeadDeleteOrTrace(const std::string& verb,
Optional<HTTP::Headers> optHeaders, Optional<HTTP::Headers> optHeaders,
RESTUrl& url, RESTUrl& url,
std::set<unsigned int> successCodes) { std::set<unsigned int> successCodes) {
HTTP::Headers headers; HTTP::Headers headers;
if (optHeaders.present()) { if (optHeaders.present()) {
headers = optHeaders.get(); headers = optHeaders.get();
@ -316,19 +312,22 @@ Future<Reference<HTTP::Response>> RESTClient::doGetHeadDeleteOrTrace(const std::
return doRequest_impl(Reference<RESTClient>::addRef(this), HTTP::HTTP_VERB_GET, headers, url, successCodes); return doRequest_impl(Reference<RESTClient>::addRef(this), HTTP::HTTP_VERB_GET, headers, url, successCodes);
} }
Future<Reference<HTTP::Response>> RESTClient::doGet(const std::string& fullUrl, Optional<HTTP::Headers> optHeaders) { Future<Reference<HTTP::IncomingResponse>> RESTClient::doGet(const std::string& fullUrl,
Optional<HTTP::Headers> optHeaders) {
RESTUrl url(fullUrl); RESTUrl url(fullUrl);
TRACE_REST_OP("DoGet", url); TRACE_REST_OP("DoGet", url);
return doGetHeadDeleteOrTrace(HTTP::HTTP_VERB_GET, optHeaders, url, { HTTP::HTTP_STATUS_CODE_OK }); return doGetHeadDeleteOrTrace(HTTP::HTTP_VERB_GET, optHeaders, url, { HTTP::HTTP_STATUS_CODE_OK });
} }
Future<Reference<HTTP::Response>> RESTClient::doHead(const std::string& fullUrl, Optional<HTTP::Headers> optHeaders) { Future<Reference<HTTP::IncomingResponse>> RESTClient::doHead(const std::string& fullUrl,
Optional<HTTP::Headers> optHeaders) {
RESTUrl url(fullUrl); RESTUrl url(fullUrl);
TRACE_REST_OP("DoHead", url); TRACE_REST_OP("DoHead", url);
return doGetHeadDeleteOrTrace(HTTP::HTTP_VERB_HEAD, optHeaders, url, { HTTP::HTTP_STATUS_CODE_OK }); return doGetHeadDeleteOrTrace(HTTP::HTTP_VERB_HEAD, optHeaders, url, { HTTP::HTTP_STATUS_CODE_OK });
} }
Future<Reference<HTTP::Response>> RESTClient::doDelete(const std::string& fullUrl, Optional<HTTP::Headers> optHeaders) { Future<Reference<HTTP::IncomingResponse>> RESTClient::doDelete(const std::string& fullUrl,
Optional<HTTP::Headers> optHeaders) {
RESTUrl url(fullUrl); RESTUrl url(fullUrl);
TRACE_REST_OP("DoDelete", url); TRACE_REST_OP("DoDelete", url);
return doGetHeadDeleteOrTrace( return doGetHeadDeleteOrTrace(
@ -341,7 +340,8 @@ Future<Reference<HTTP::Response>> RESTClient::doDelete(const std::string& fullUr
{ HTTP::HTTP_STATUS_CODE_OK, HTTP::HTTP_STATUS_CODE_NO_CONTENT, HTTP::HTTP_STATUS_CODE_ACCEPTED }); { HTTP::HTTP_STATUS_CODE_OK, HTTP::HTTP_STATUS_CODE_NO_CONTENT, HTTP::HTTP_STATUS_CODE_ACCEPTED });
} }
Future<Reference<HTTP::Response>> RESTClient::doTrace(const std::string& fullUrl, Optional<HTTP::Headers> optHeaders) { Future<Reference<HTTP::IncomingResponse>> RESTClient::doTrace(const std::string& fullUrl,
Optional<HTTP::Headers> optHeaders) {
RESTUrl url(fullUrl); RESTUrl url(fullUrl);
TRACE_REST_OP("DoTrace", url); TRACE_REST_OP("DoTrace", url);
return doGetHeadDeleteOrTrace(HTTP::HTTP_VERB_TRACE, optHeaders, url, { HTTP::HTTP_STATUS_CODE_OK }); return doGetHeadDeleteOrTrace(HTTP::HTTP_VERB_TRACE, optHeaders, url, { HTTP::HTTP_STATUS_CODE_OK });

128
fdbclient/S3BlobStore.actor.cpp
View File

@ -427,7 +427,7 @@ ACTOR Future<bool> bucketExists_impl(Reference<S3BlobStoreEndpoint> b, std::stri
std::string resource = constructResourcePath(b, bucket, ""); std::string resource = constructResourcePath(b, bucket, "");
HTTP::Headers headers; HTTP::Headers headers;
Reference<HTTP::Response> r = wait(b->doRequest("HEAD", resource, headers, nullptr, 0, { 200, 404 })); Reference<HTTP::IncomingResponse> r = wait(b->doRequest("HEAD", resource, headers, nullptr, 0, { 200, 404 }));
return r->code == 200; return r->code == 200;
} }
@ -441,7 +441,7 @@ ACTOR Future<bool> objectExists_impl(Reference<S3BlobStoreEndpoint> b, std::stri
std::string resource = constructResourcePath(b, bucket, object); std::string resource = constructResourcePath(b, bucket, object);
HTTP::Headers headers; HTTP::Headers headers;
Reference<HTTP::Response> r = wait(b->doRequest("HEAD", resource, headers, nullptr, 0, { 200, 404 })); Reference<HTTP::IncomingResponse> r = wait(b->doRequest("HEAD", resource, headers, nullptr, 0, { 200, 404 }));
return r->code == 200; return r->code == 200;
} }
@ -456,7 +456,8 @@ ACTOR Future<Void> deleteObject_impl(Reference<S3BlobStoreEndpoint> b, std::stri
HTTP::Headers headers; HTTP::Headers headers;
// 200 or 204 means object successfully deleted, 404 means it already doesn't exist, so any of those are considered // 200 or 204 means object successfully deleted, 404 means it already doesn't exist, so any of those are considered
// successful // successful
Reference<HTTP::Response> r = wait(b->doRequest("DELETE", resource, headers, nullptr, 0, { 200, 204, 404 })); Reference<HTTP::IncomingResponse> r =
wait(b->doRequest("DELETE", resource, headers, nullptr, 0, { 200, 204, 404 }));
// But if the object already did not exist then the 'delete' is assumed to be successful but a warning is logged. // But if the object already did not exist then the 'delete' is assumed to be successful but a warning is logged.
if (r->code == 404) { if (r->code == 404) {
@ -549,7 +550,8 @@ ACTOR Future<Void> createBucket_impl(Reference<S3BlobStoreEndpoint> b, std::stri
std::string region = b->getRegion(); std::string region = b->getRegion();
if (region.empty()) { if (region.empty()) {
Reference<HTTP::Response> r = wait(b->doRequest("PUT", resource, headers, nullptr, 0, { 200, 409 })); Reference<HTTP::IncomingResponse> r =
wait(b->doRequest("PUT", resource, headers, nullptr, 0, { 200, 409 }));
} else { } else {
UnsentPacketQueue packets; UnsentPacketQueue packets;
StringRef body(format("<CreateBucketConfiguration xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\">" StringRef body(format("<CreateBucketConfiguration xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\">"
@ -559,7 +561,7 @@ ACTOR Future<Void> createBucket_impl(Reference<S3BlobStoreEndpoint> b, std::stri
PacketWriter pw(packets.getWriteBuffer(), nullptr, Unversioned()); PacketWriter pw(packets.getWriteBuffer(), nullptr, Unversioned());
pw.serializeBytes(body); pw.serializeBytes(body);
Reference<HTTP::Response> r = Reference<HTTP::IncomingResponse> r =
wait(b->doRequest("PUT", resource, headers, &packets, body.size(), { 200, 409 })); wait(b->doRequest("PUT", resource, headers, &packets, body.size(), { 200, 409 }));
} }
} }
@ -576,10 +578,10 @@ ACTOR Future<int64_t> objectSize_impl(Reference<S3BlobStoreEndpoint> b, std::str
std::string resource = constructResourcePath(b, bucket, object); std::string resource = constructResourcePath(b, bucket, object);
HTTP::Headers headers; HTTP::Headers headers;
Reference<HTTP::Response> r = wait(b->doRequest("HEAD", resource, headers, nullptr, 0, { 200, 404 })); Reference<HTTP::IncomingResponse> r = wait(b->doRequest("HEAD", resource, headers, nullptr, 0, { 200, 404 }));
if (r->code == 404) if (r->code == 404)
throw file_not_found(); throw file_not_found();
return r->contentLen; return r->data.contentLen;
} }
Future<int64_t> S3BlobStoreEndpoint::objectSize(std::string const& bucket, std::string const& object) { Future<int64_t> S3BlobStoreEndpoint::objectSize(std::string const& bucket, std::string const& object) {
@ -831,22 +833,26 @@ std::string awsCanonicalURI(const std::string& resource, std::vector<std::string
// Do a request, get a Response. // Do a request, get a Response.
// Request content is provided as UnsentPacketQueue *pContent which will be depleted as bytes are sent but the queue // Request content is provided as UnsentPacketQueue *pContent which will be depleted as bytes are sent but the queue
// itself must live for the life of this actor and be destroyed by the caller // itself must live for the life of this actor and be destroyed by the caller
ACTOR Future<Reference<HTTP::Response>> doRequest_impl(Reference<S3BlobStoreEndpoint> bstore, ACTOR Future<Reference<HTTP::IncomingResponse>> doRequest_impl(Reference<S3BlobStoreEndpoint> bstore,
std::string verb, std::string verb,
std::string resource, std::string resource,
HTTP::Headers headers, HTTP::Headers headers,
UnsentPacketQueue* pContent, UnsentPacketQueue* pContent,
int contentLen, int contentLen,
std::set<unsigned int> successCodes) { std::set<unsigned int> successCodes) {
state UnsentPacketQueue contentCopy; state UnsentPacketQueue contentCopy;
state Reference<HTTP::OutgoingRequest> req = makeReference<HTTP::OutgoingRequest>();
req->verb = verb;
req->data.content = &contentCopy;
req->data.contentLen = contentLen;
headers["Content-Length"] = format("%d", contentLen); req->data.headers = headers;
headers["Host"] = bstore->host; req->data.headers["Host"] = bstore->host;
headers["Accept"] = "application/xml"; req->data.headers["Accept"] = "application/xml";
// Merge extraHeaders into headers // Merge extraHeaders into headers
for (const auto& [k, v] : bstore->extraHeaders) { for (const auto& [k, v] : bstore->extraHeaders) {
std::string& fieldValue = headers[k]; std::string& fieldValue = req->data.headers[k];
if (!fieldValue.empty()) { if (!fieldValue.empty()) {
fieldValue.append(","); fieldValue.append(",");
} }
@ -870,7 +876,7 @@ ACTOR Future<Reference<HTTP::Response>> doRequest_impl(Reference<S3BlobStoreEndp
state Optional<Error> err; state Optional<Error> err;
state Optional<NetworkAddress> remoteAddress; state Optional<NetworkAddress> remoteAddress;
state bool connectionEstablished = false; state bool connectionEstablished = false;
state Reference<HTTP::Response> r; state Reference<HTTP::IncomingResponse> r;
state std::string canonicalURI = resource; state std::string canonicalURI = resource;
state UID connID = UID(); state UID connID = UID();
state double reqStartTimer; state double reqStartTimer;
@ -884,7 +890,7 @@ ACTOR Future<Reference<HTTP::Response>> doRequest_impl(Reference<S3BlobStoreEndp
// Make a shallow copy of the queue by calling addref() on each buffer in the chain and then prepending that // Make a shallow copy of the queue by calling addref() on each buffer in the chain and then prepending that
// chain to contentCopy // chain to contentCopy
contentCopy.discardAll(); req->data.content->discardAll();
if (pContent != nullptr) { if (pContent != nullptr) {
PacketBuffer* pFirst = pContent->getUnsent(); PacketBuffer* pFirst = pContent->getUnsent();
PacketBuffer* pLast = nullptr; PacketBuffer* pLast = nullptr;
@ -894,7 +900,7 @@ ACTOR Future<Reference<HTTP::Response>> doRequest_impl(Reference<S3BlobStoreEndp
p->bytes_sent = 0; p->bytes_sent = 0;
pLast = p; pLast = p;
} }
contentCopy.prependWriteBuffer(pFirst, pLast); req->data.content->prependWriteBuffer(pFirst, pLast);
} }
// Finish connecting, do request // Finish connecting, do request
@ -908,11 +914,11 @@ ACTOR Future<Reference<HTTP::Response>> doRequest_impl(Reference<S3BlobStoreEndp
// This must be done AFTER the connection is ready because if credentials are coming from disk they are // This must be done AFTER the connection is ready because if credentials are coming from disk they are
// refreshed when a new connection is established and setAuthHeaders() would need the updated secret. // refreshed when a new connection is established and setAuthHeaders() would need the updated secret.
if (bstore->credentials.present() && !bstore->credentials.get().securityToken.empty()) if (bstore->credentials.present() && !bstore->credentials.get().securityToken.empty())
headers["x-amz-security-token"] = bstore->credentials.get().securityToken; req->data.headers["x-amz-security-token"] = bstore->credentials.get().securityToken;
if (CLIENT_KNOBS->HTTP_REQUEST_AWS_V4_HEADER) { if (CLIENT_KNOBS->HTTP_REQUEST_AWS_V4_HEADER) {
bstore->setV4AuthHeaders(verb, resource, headers); bstore->setV4AuthHeaders(verb, resource, req->data.headers);
} else { } else {
bstore->setAuthHeaders(verb, resource, headers); bstore->setAuthHeaders(verb, resource, req->data.headers);
} }
std::vector<std::string> queryParameters; std::vector<std::string> queryParameters;
@ -927,18 +933,13 @@ ACTOR Future<Reference<HTTP::Response>> doRequest_impl(Reference<S3BlobStoreEndp
canonicalURI = "http://" + bstore->host + ":" + bstore->service + canonicalURI; canonicalURI = "http://" + bstore->host + ":" + bstore->service + canonicalURI;
} }
req->resource = canonicalURI;
remoteAddress = rconn.conn->getPeerAddress(); remoteAddress = rconn.conn->getPeerAddress();
wait(bstore->requestRate->getAllowance(1)); wait(bstore->requestRate->getAllowance(1));
Future<Reference<HTTP::Response>> reqF = HTTP::doRequest(rconn.conn, Future<Reference<HTTP::IncomingResponse>> reqF =
verb, HTTP::doRequest(rconn.conn, req, bstore->sendRate, &bstore->s_stats.bytes_sent, bstore->recvRate);
canonicalURI,
headers,
&contentCopy,
contentLen,
bstore->sendRate,
&bstore->s_stats.bytes_sent,
bstore->recvRate);
// if we reused a connection from the pool, and immediately got an error, retry immediately discarding the // if we reused a connection from the pool, and immediately got an error, retry immediately discarding the
// connection // connection
@ -946,12 +947,12 @@ ACTOR Future<Reference<HTTP::Response>> doRequest_impl(Reference<S3BlobStoreEndp
fastRetry = true; fastRetry = true;
} }
Reference<HTTP::Response> _r = wait(timeoutError(reqF, requestTimeout)); Reference<HTTP::IncomingResponse> _r = wait(timeoutError(reqF, requestTimeout));
r = _r; r = _r;
// Since the response was parsed successfully (which is why we are here) reuse the connection unless we // Since the response was parsed successfully (which is why we are here) reuse the connection unless we
// received the "Connection: close" header. // received the "Connection: close" header.
if (r->headers["Connection"] != "close") { if (r->data.headers["Connection"] != "close") {
bstore->returnConnection(rconn); bstore->returnConnection(rconn);
} else { } else {
++bstore->blobStats->expiredConnections; ++bstore->blobStats->expiredConnections;
@ -1037,8 +1038,8 @@ ACTOR Future<Reference<HTTP::Response>> doRequest_impl(Reference<S3BlobStoreEndp
// If r is valid then obey the Retry-After response header if present. // If r is valid then obey the Retry-After response header if present.
if (r) { if (r) {
auto iRetryAfter = r->headers.find("Retry-After"); auto iRetryAfter = r->data.headers.find("Retry-After");
if (iRetryAfter != r->headers.end()) { if (iRetryAfter != r->data.headers.end()) {
event.detail("RetryAfterHeader", iRetryAfter->second); event.detail("RetryAfterHeader", iRetryAfter->second);
char* pEnd; char* pEnd;
double retryAfter = strtod(iRetryAfter->second.c_str(), &pEnd); double retryAfter = strtod(iRetryAfter->second.c_str(), &pEnd);
@ -1090,12 +1091,12 @@ ACTOR Future<Reference<HTTP::Response>> doRequest_impl(Reference<S3BlobStoreEndp
} }
} }
Future<Reference<HTTP::Response>> S3BlobStoreEndpoint::doRequest(std::string const& verb, Future<Reference<HTTP::IncomingResponse>> S3BlobStoreEndpoint::doRequest(std::string const& verb,
std::string const& resource, std::string const& resource,
const HTTP::Headers& headers, const HTTP::Headers& headers,
UnsentPacketQueue* pContent, UnsentPacketQueue* pContent,
int contentLen, int contentLen,
std::set<unsigned int> successCodes) { std::set<unsigned int> successCodes) {
return doRequest_impl( return doRequest_impl(
Reference<S3BlobStoreEndpoint>::addRef(this), verb, resource, headers, pContent, contentLen, successCodes); Reference<S3BlobStoreEndpoint>::addRef(this), verb, resource, headers, pContent, contentLen, successCodes);
} }
@ -1128,7 +1129,8 @@ ACTOR Future<Void> listObjectsStream_impl(Reference<S3BlobStoreEndpoint> bstore,
HTTP::Headers headers; HTTP::Headers headers;
state std::string fullResource = resource + lastFile; state std::string fullResource = resource + lastFile;
lastFile.clear(); lastFile.clear();
Reference<HTTP::Response> r = wait(bstore->doRequest("GET", fullResource, headers, nullptr, 0, { 200 })); Reference<HTTP::IncomingResponse> r =
wait(bstore->doRequest("GET", fullResource, headers, nullptr, 0, { 200 }));
listReleaser.release(); listReleaser.release();
try { try {
@ -1136,7 +1138,7 @@ ACTOR Future<Void> listObjectsStream_impl(Reference<S3BlobStoreEndpoint> bstore,
xml_document<> doc; xml_document<> doc;
// Copy content because rapidxml will modify it during parse // Copy content because rapidxml will modify it during parse
std::string content = r->content; std::string content = r->data.content;
doc.parse<0>((char*)content.c_str()); doc.parse<0>((char*)content.c_str());
// There should be exactly one node // There should be exactly one node
@ -1307,14 +1309,15 @@ ACTOR Future<std::vector<std::string>> listBuckets_impl(Reference<S3BlobStoreEnd
HTTP::Headers headers; HTTP::Headers headers;
state std::string fullResource = resource + lastName; state std::string fullResource = resource + lastName;
Reference<HTTP::Response> r = wait(bstore->doRequest("GET", fullResource, headers, nullptr, 0, { 200 })); Reference<HTTP::IncomingResponse> r =
wait(bstore->doRequest("GET", fullResource, headers, nullptr, 0, { 200 }));
listReleaser.release(); listReleaser.release();
try { try {
xml_document<> doc; xml_document<> doc;
// Copy content because rapidxml will modify it during parse // Copy content because rapidxml will modify it during parse
std::string content = r->content; std::string content = r->data.content;
doc.parse<0>((char*)content.c_str()); doc.parse<0>((char*)content.c_str());
// There should be exactly one node // There should be exactly one node
@ -1587,10 +1590,10 @@ ACTOR Future<std::string> readEntireFile_impl(Reference<S3BlobStoreEndpoint> bst
std::string resource = constructResourcePath(bstore, bucket, object); std::string resource = constructResourcePath(bstore, bucket, object);
HTTP::Headers headers; HTTP::Headers headers;
Reference<HTTP::Response> r = wait(bstore->doRequest("GET", resource, headers, nullptr, 0, { 200, 404 })); Reference<HTTP::IncomingResponse> r = wait(bstore->doRequest("GET", resource, headers, nullptr, 0, { 200, 404 }));
if (r->code == 404) if (r->code == 404)
throw file_not_found(); throw file_not_found();
return r->content; return r->data.content;
} }
Future<std::string> S3BlobStoreEndpoint::readEntireFile(std::string const& bucket, std::string const& object) { Future<std::string> S3BlobStoreEndpoint::readEntireFile(std::string const& bucket, std::string const& object) {
@ -1616,11 +1619,11 @@ ACTOR Future<Void> writeEntireFileFromBuffer_impl(Reference<S3BlobStoreEndpoint>
headers["Content-MD5"] = contentMD5; headers["Content-MD5"] = contentMD5;
if (!CLIENT_KNOBS->BLOBSTORE_ENCRYPTION_TYPE.empty()) if (!CLIENT_KNOBS->BLOBSTORE_ENCRYPTION_TYPE.empty())
headers["x-amz-server-side-encryption"] = CLIENT_KNOBS->BLOBSTORE_ENCRYPTION_TYPE; headers["x-amz-server-side-encryption"] = CLIENT_KNOBS->BLOBSTORE_ENCRYPTION_TYPE;
state Reference<HTTP::Response> r = state Reference<HTTP::IncomingResponse> r =
wait(bstore->doRequest("PUT", resource, headers, pContent, contentLen, { 200 })); wait(bstore->doRequest("PUT", resource, headers, pContent, contentLen, { 200 }));
// For uploads, Blobstore returns an MD5 sum of uploaded content so check it. // For uploads, Blobstore returns an MD5 sum of uploaded content so check it.
if (!r->verifyMD5(false, contentMD5)) if (!HTTP::verifyMD5(&r->data, false, contentMD5))
throw checksum_failed(); throw checksum_failed();
return Void(); return Void();
@ -1684,15 +1687,16 @@ ACTOR Future<int> readObject_impl(Reference<S3BlobStoreEndpoint> bstore,
std::string resource = constructResourcePath(bstore, bucket, object); std::string resource = constructResourcePath(bstore, bucket, object);
HTTP::Headers headers; HTTP::Headers headers;
headers["Range"] = format("bytes=%lld-%lld", offset, offset + length - 1); headers["Range"] = format("bytes=%lld-%lld", offset, offset + length - 1);
Reference<HTTP::Response> r = wait(bstore->doRequest("GET", resource, headers, nullptr, 0, { 200, 206, 404 })); Reference<HTTP::IncomingResponse> r =
wait(bstore->doRequest("GET", resource, headers, nullptr, 0, { 200, 206, 404 }));
if (r->code == 404) if (r->code == 404)
throw file_not_found(); throw file_not_found();
if (r->contentLen != if (r->data.contentLen !=
r->content.size()) // Double check that this wasn't a header-only response, probably unnecessary r->data.content.size()) // Double check that this wasn't a header-only response, probably unnecessary
throw io_error(); throw io_error();
// Copy the output bytes, server could have sent more or less bytes than requested so copy at most length bytes // Copy the output bytes, server could have sent more or less bytes than requested so copy at most length bytes
memcpy(data, r->content.data(), std::min<int64_t>(r->contentLen, length)); memcpy(data, r->data.content.data(), std::min<int64_t>(r->data.contentLen, length));
return r->contentLen; return r->data.contentLen;
} }
Future<int> S3BlobStoreEndpoint::readObject(std::string const& bucket, Future<int> S3BlobStoreEndpoint::readObject(std::string const& bucket,
@ -1713,12 +1717,12 @@ ACTOR static Future<std::string> beginMultiPartUpload_impl(Reference<S3BlobStore
HTTP::Headers headers; HTTP::Headers headers;
if (!CLIENT_KNOBS->BLOBSTORE_ENCRYPTION_TYPE.empty()) if (!CLIENT_KNOBS->BLOBSTORE_ENCRYPTION_TYPE.empty())
headers["x-amz-server-side-encryption"] = CLIENT_KNOBS->BLOBSTORE_ENCRYPTION_TYPE; headers["x-amz-server-side-encryption"] = CLIENT_KNOBS->BLOBSTORE_ENCRYPTION_TYPE;
Reference<HTTP::Response> r = wait(bstore->doRequest("POST", resource, headers, nullptr, 0, { 200 })); Reference<HTTP::IncomingResponse> r = wait(bstore->doRequest("POST", resource, headers, nullptr, 0, { 200 }));
try { try {
xml_document<> doc; xml_document<> doc;
// Copy content because rapidxml will modify it during parse // Copy content because rapidxml will modify it during parse
std::string content = r->content; std::string content = r->data.content;
doc.parse<0>((char*)content.c_str()); doc.parse<0>((char*)content.c_str());
@ -1756,18 +1760,18 @@ ACTOR Future<std::string> uploadPart_impl(Reference<S3BlobStoreEndpoint> bstore,
HTTP::Headers headers; HTTP::Headers headers;
// Send MD5 sum for content so blobstore can verify it // Send MD5 sum for content so blobstore can verify it
headers["Content-MD5"] = contentMD5; headers["Content-MD5"] = contentMD5;
state Reference<HTTP::Response> r = state Reference<HTTP::IncomingResponse> r =
wait(bstore->doRequest("PUT", resource, headers, pContent, contentLen, { 200 })); wait(bstore->doRequest("PUT", resource, headers, pContent, contentLen, { 200 }));
// TODO: In the event that the client times out just before the request completes (so the client is unaware) then // TODO: In the event that the client times out just before the request completes (so the client is unaware) then
// the next retry will see error 400. That could be detected and handled gracefully by retrieving the etag for the // the next retry will see error 400. That could be detected and handled gracefully by retrieving the etag for the
// successful request. // successful request.
// For uploads, Blobstore returns an MD5 sum of uploaded content so check it. // For uploads, Blobstore returns an MD5 sum of uploaded content so check it.
if (!r->verifyMD5(false, contentMD5)) if (!HTTP::verifyMD5(&r->data, false, contentMD5))
throw checksum_failed(); throw checksum_failed();
// No etag -> bad response. // No etag -> bad response.
std::string etag = r->headers["ETag"]; std::string etag = r->data.headers["ETag"];
if (etag.empty()) if (etag.empty())
throw http_bad_response(); throw http_bad_response();
@ -1809,7 +1813,7 @@ ACTOR Future<Void> finishMultiPartUpload_impl(Reference<S3BlobStoreEndpoint> bst
HTTP::Headers headers; HTTP::Headers headers;
PacketWriter pw(part_list.getWriteBuffer(manifest.size()), nullptr, Unversioned()); PacketWriter pw(part_list.getWriteBuffer(manifest.size()), nullptr, Unversioned());
pw.serializeBytes(manifest); pw.serializeBytes(manifest);
Reference<HTTP::Response> r = Reference<HTTP::IncomingResponse> r =
wait(bstore->doRequest("POST", resource, headers, &part_list, manifest.size(), { 200 })); wait(bstore->doRequest("POST", resource, headers, &part_list, manifest.size(), { 200 }));
// TODO: In the event that the client times out just before the request completes (so the client is unaware) then // TODO: In the event that the client times out just before the request completes (so the client is unaware) then
// the next retry will see error 400. That could be detected and handled gracefully by HEAD'ing the object before // the next retry will see error 400. That could be detected and handled gracefully by HEAD'ing the object before

44
fdbclient/include/fdbclient/RESTClient.h
View File

@ -67,32 +67,32 @@ public:
// RESTConnectionPool is used to leverage cached connection if any for 'host:service' pair. API then leverage // RESTConnectionPool is used to leverage cached connection if any for 'host:service' pair. API then leverage
// HTTP::doRequest to accomplish the specified operation // HTTP::doRequest to accomplish the specified operation
Future<Reference<HTTP::Response>> doGet(const std::string& fullUrl, Future<Reference<HTTP::IncomingResponse>> doGet(const std::string& fullUrl,
Optional<HTTP::Headers> optHeaders = Optional<HTTP::Headers>()); Optional<HTTP::Headers> optHeaders = Optional<HTTP::Headers>());
Future<Reference<HTTP::Response>> doHead(const std::string& fullUrl, Future<Reference<HTTP::IncomingResponse>> doHead(const std::string& fullUrl,
Optional<HTTP::Headers> optHeaders = Optional<HTTP::Headers>()); Optional<HTTP::Headers> optHeaders = Optional<HTTP::Headers>());
Future<Reference<HTTP::Response>> doDelete(const std::string& fullUrl, Future<Reference<HTTP::IncomingResponse>> doDelete(const std::string& fullUrl,
Optional<HTTP::Headers> optHeaders = Optional<HTTP::Headers>()); Optional<HTTP::Headers> optHeaders = Optional<HTTP::Headers>());
Future<Reference<HTTP::Response>> doTrace(const std::string& fullUrl, Future<Reference<HTTP::IncomingResponse>> doTrace(const std::string& fullUrl,
Optional<HTTP::Headers> optHeaders = Optional<HTTP::Headers>()); Optional<HTTP::Headers> optHeaders = Optional<HTTP::Headers>());
Future<Reference<HTTP::Response>> doPut(const std::string& fullUrl, Future<Reference<HTTP::IncomingResponse>> doPut(const std::string& fullUrl,
const std::string& requestBody, const std::string& requestBody,
Optional<HTTP::Headers> optHeaders = Optional<HTTP::Headers>()); Optional<HTTP::Headers> optHeaders = Optional<HTTP::Headers>());
Future<Reference<HTTP::Response>> doPost(const std::string& fullUrl, Future<Reference<HTTP::IncomingResponse>> doPost(const std::string& fullUrl,
const std::string& requestBody, const std::string& requestBody,
Optional<HTTP::Headers> optHeaders = Optional<HTTP::Headers>()); Optional<HTTP::Headers> optHeaders = Optional<HTTP::Headers>());
static std::string getStatsKey(const std::string& host, const std::string& service) { return host + ":" + service; } static std::string getStatsKey(const std::string& host, const std::string& service) { return host + ":" + service; }
private: private:
Future<Reference<HTTP::Response>> doGetHeadDeleteOrTrace(const std::string& verb, Future<Reference<HTTP::IncomingResponse>> doGetHeadDeleteOrTrace(const std::string& verb,
Optional<HTTP::Headers> optHeaders, Optional<HTTP::Headers> optHeaders,
RESTUrl& url, RESTUrl& url,
std::set<unsigned int> successCodes); std::set<unsigned int> successCodes);
Future<Reference<HTTP::Response>> doPutOrPost(const std::string& verb, Future<Reference<HTTP::IncomingResponse>> doPutOrPost(const std::string& verb,
Optional<HTTP::Headers> headers, Optional<HTTP::Headers> headers,
RESTUrl& url, RESTUrl& url,
std::set<unsigned int> successCodes); std::set<unsigned int> successCodes);
}; };
#endif #endif

17
fdbclient/include/fdbclient/S3BlobStore.h
View File

@ -320,20 +320,15 @@ public:
std::string getRegion() const { return region; } std::string getRegion() const { return region; }
// Prepend the HTTP request header to the given PacketBuffer, returning the new head of the buffer chain
static PacketBuffer* writeRequestHeader(std::string const& request,
HTTP::Headers const& headers,
PacketBuffer* dest);
// Do an HTTP request to the Blob Store, read the response. Handles authentication. // Do an HTTP request to the Blob Store, read the response. Handles authentication.
// Every blob store interaction should ultimately go through this function // Every blob store interaction should ultimately go through this function
Future<Reference<HTTP::Response>> doRequest(std::string const& verb, Future<Reference<HTTP::IncomingResponse>> doRequest(std::string const& verb,
std::string const& resource, std::string const& resource,
const HTTP::Headers& headers, const HTTP::Headers& headers,
UnsentPacketQueue* pContent, UnsentPacketQueue* pContent,
int contentLen, int contentLen,
std::set<unsigned int> successCodes); std::set<unsigned int> successCodes);
struct ObjectInfo { struct ObjectInfo {
std::string name; std::string name;

421
fdbrpc/HTTP.actor.cpp
View File

@ -19,12 +19,16 @@
*/ */
#include "fdbrpc/HTTP.h" #include "fdbrpc/HTTP.h"
#include "fdbrpc/simulator.h"
#include "flow/IRandom.h"
#include "flow/Net2Packet.h"
#include "md5/md5.h" #include "md5/md5.h"
#include "libb64/encode.h" #include "libb64/encode.h"
#include "flow/Knobs.h" #include "flow/Knobs.h"
#include <cctype> #include <cctype>
#include "flow/IConnection.h" #include "flow/IConnection.h"
#include <unordered_map>
#include "flow/actorcompiler.h" // has to be last include #include "flow/actorcompiler.h" // has to be last include
@ -62,14 +66,45 @@ std::string urlEncode(const std::string& s) {
return o; return o;
} }
bool Response::verifyMD5(bool fail_if_header_missing, Optional<std::string> content_sum) { template <typename T>
auto i = headers.find("Content-MD5"); std::string ResponseBase<T>::getCodeDescription() {
if (i != headers.end()) { if (code == HTTP_STATUS_CODE_OK) {
return "OK";
} else if (code == HTTP_STATUS_CODE_CREATED) {
return "Created";
} else if (code == HTTP_STATUS_CODE_ACCEPTED) {
return "Accepted";
} else if (code == HTTP_STATUS_CODE_NO_CONTENT) {
return "No Content";
} else if (code == HTTP_STATUS_CODE_UNAUTHORIZED) {
return "Unauthorized";
} else if (code == HTTP_STATUS_CODE_NOT_ACCEPTABLE) {
return "Not Acceptable";
} else if (code == HTTP_STATUS_CODE_TIMEOUT) {
return "Timeout";
} else if (code == HTTP_STATUS_CODE_TOO_MANY_REQUESTS) {
return "Too Many Requests";
} else if (code == HTTP_STATUS_CODE_INTERNAL_SERVER_ERROR) {
return "Internal Server Error";
} else if (code == HTTP_STATUS_CODE_BAD_GATEWAY) {
return "Bad Gateway";
} else if (code == HTTP_STATUS_CODE_SERVICE_UNAVAILABLE) {
return "Service Unavailable";
} else if (code == HTTP_STATUS_CODE_GATEWAY_TIMEOUT) {
return "Gateway Timeout";
} else {
throw internal_error();
}
}
bool verifyMD5(HTTPData<std::string>* data, bool fail_if_header_missing, Optional<std::string> content_sum) {
auto i = data->headers.find("Content-MD5");
if (i != data->headers.end()) {
// If a content sum is not provided, calculate one from the response content // If a content sum is not provided, calculate one from the response content
if (!content_sum.present()) { if (!content_sum.present()) {
MD5_CTX sum; MD5_CTX sum;
::MD5_Init(&sum); ::MD5_Init(&sum);
::MD5_Update(&sum, content.data(), content.size()); ::MD5_Update(&sum, data->content.data(), data->content.size());
std::string sumBytes; std::string sumBytes;
sumBytes.resize(16); sumBytes.resize(16);
::MD5_Final((unsigned char*)sumBytes.data(), &sum); ::MD5_Final((unsigned char*)sumBytes.data(), &sum);
@ -82,26 +117,18 @@ bool Response::verifyMD5(bool fail_if_header_missing, Optional<std::string> cont
return !fail_if_header_missing; return !fail_if_header_missing;
} }
std::string Response::toString() { std::string IncomingResponse::toString() const {
std::string r = format("Response Code: %d\n", code); std::string r = fmt::format("Response Code: {0}\n", code);
r += format("Response ContentLen: %lld\n", contentLen); r += fmt::format("Response ContentLen: {0}\n", data.contentLen);
for (auto h : headers) for (auto h : data.headers)
r += format("Reponse Header: %s: %s\n", h.first.c_str(), h.second.c_str()); r += fmt::format("Reponse Header: {0}: {1}\n", h.first, h.second);
r.append("-- RESPONSE CONTENT--\n"); r.append("-- RESPONSE CONTENT--\n");
r.append(content); r.append(data.content);
r.append("\n--------\n"); r.append("\n--------\n");
return r; return r;
} }
PacketBuffer* writeRequestHeader(std::string const& verb, void writeHeaders(HTTP::Headers const& headers, PacketWriter& writer) {
std::string const& resource,
HTTP::Headers const& headers,
PacketBuffer* dest) {
PacketWriter writer(dest, nullptr, Unversioned());
writer.serializeBytes(verb);
writer.serializeBytes(" ", 1);
writer.serializeBytes(resource);
writer.serializeBytes(" HTTP/1.1\r\n"_sr);
for (auto h : headers) { for (auto h : headers) {
writer.serializeBytes(h.first); writer.serializeBytes(h.first);
writer.serializeBytes(": "_sr); writer.serializeBytes(": "_sr);
@ -109,9 +136,58 @@ PacketBuffer* writeRequestHeader(std::string const& verb,
writer.serializeBytes("\r\n"_sr); writer.serializeBytes("\r\n"_sr);
} }
writer.serializeBytes("\r\n"_sr); writer.serializeBytes("\r\n"_sr);
}
PacketBuffer* writeRequestHeader(Reference<OutgoingRequest> req, PacketBuffer* dest) {
PacketWriter writer(dest, nullptr, Unversioned());
writer.serializeBytes(req->verb);
writer.serializeBytes(" ", 1);
writer.serializeBytes(req->resource);
writer.serializeBytes(" HTTP/1.1\r\n"_sr);
writeHeaders(req->data.headers, writer);
return writer.finish(); return writer.finish();
} }
PacketBuffer* writeResponseHeader(Reference<OutgoingResponse> response, PacketBuffer* dest) {
PacketWriter writer(dest, nullptr, Unversioned());
writer.serializeBytes("HTTP/1.1 "_sr);
writer.serializeBytes(std::to_string(response->code));
writer.serializeBytes(" ", 1);
writer.serializeBytes(response->getCodeDescription());
writer.serializeBytes("\r\n"_sr);
writeHeaders(response->data.headers, writer);
return writer.finish();
}
ACTOR Future<Void> writeResponse(Reference<IConnection> conn, Reference<OutgoingResponse> response) {
// Write headers to a packet buffer chain
ASSERT(response.isValid());
response->data.headers["Content-Length"] = std::to_string(response->data.contentLen);
PacketBuffer* pFirst = PacketBuffer::create();
PacketBuffer* pLast = writeResponseHeader(response, pFirst);
// Prepend headers to content packer buffer chain
response->data.content->prependWriteBuffer(pFirst, pLast);
loop {
int trySend = FLOW_KNOBS->HTTP_SEND_SIZE;
if ((!g_network->isSimulated() || !g_simulator->speedUpSimulation) && BUGGIFY_WITH_PROB(0.01)) {
trySend = deterministicRandom()->randomInt(1, 10);
}
int len = conn->write(response->data.content->getUnsent(), trySend);
response->data.content->sent(len);
if (response->data.content->empty()) {
return Void();
}
wait(conn->onWritable());
wait(yield(TaskPriority::WriteSocket));
}
}
// Read at least 1 bytes from conn and up to maxlen in a single read, append read data into *buf // Read at least 1 bytes from conn and up to maxlen in a single read, append read data into *buf
// Returns the number of bytes read. // Returns the number of bytes read.
ACTOR Future<int> read_into_string(Reference<IConnection> conn, std::string* buf, int maxlen) { ACTOR Future<int> read_into_string(Reference<IConnection> conn, std::string* buf, int maxlen) {
@ -120,6 +196,7 @@ ACTOR Future<int> read_into_string(Reference<IConnection> conn, std::string* buf
int originalSize = buf->size(); int originalSize = buf->size();
// TODO: resize is zero-initializing the space we're about to overwrite, so do something else, which probably // TODO: resize is zero-initializing the space we're about to overwrite, so do something else, which probably
// means not using a string for this buffer // means not using a string for this buffer
// FIXME: buggify read size as well
buf->resize(originalSize + maxlen); buf->resize(originalSize + maxlen);
uint8_t* wptr = (uint8_t*)buf->data() + originalSize; uint8_t* wptr = (uint8_t*)buf->data() + originalSize;
int len = conn->read(wptr, wptr + maxlen); int len = conn->read(wptr, wptr + maxlen);
@ -148,11 +225,13 @@ ACTOR Future<size_t> read_delimited_into_string(Reference<IConnection> conn,
loop { loop {
size_t endPos = buf->find(delim, sPos); size_t endPos = buf->find(delim, sPos);
if (endPos != std::string::npos) if (endPos != std::string::npos) {
return endPos - pos; return endPos - pos;
}
// Next search will start at the current end of the buffer - delim size + 1 // Next search will start at the current end of the buffer - delim size + 1
if (sPos >= lookBack) if (buf->size() >= lookBack) {
sPos -= lookBack; sPos = buf->size() - lookBack;
}
wait(success(read_into_string(conn, buf, FLOW_KNOBS->HTTP_READ_SIZE))); wait(success(read_into_string(conn, buf, FLOW_KNOBS->HTTP_READ_SIZE)));
} }
} }
@ -219,28 +298,18 @@ ACTOR Future<Void> read_http_response_headers(Reference<IConnection> conn,
} }
} }
// Reads an HTTP response from a network connection // FIXME: should this throw a different error for http requests? Or should we rename http_bad_response to
// If the connection fails while being read the exception will emitted // http_bad_<something>?
// If the response is not parsable or complete in some way, http_bad_response will be thrown ACTOR Future<Void> readHTTPData(HTTPData<std::string>* r,
ACTOR Future<Void> read_http_response(Reference<HTTP::Response> r, Reference<IConnection> conn, bool header_only) { Reference<IConnection> conn,
state std::string buf; std::string* buf,
state size_t pos = 0; size_t* pos,
bool content_optional,
// Read HTTP response code and version line bool skipCheckMD5) {
size_t lineLen = wait(read_delimited_into_string(conn, "\r\n", &buf, pos));
int reachedEnd = -1;
sscanf(buf.c_str() + pos, "HTTP/%f %d%n", &r->version, &r->code, &reachedEnd);
if (reachedEnd < 0)
throw http_bad_response();
// Move position past the line found and the delimiter length
pos += lineLen + 2;
// Read headers // Read headers
r->headers.clear(); r->headers.clear();
wait(read_http_response_headers(conn, &r->headers, &buf, &pos)); wait(read_http_response_headers(conn, &r->headers, buf, pos));
auto i = r->headers.find("Content-Length"); auto i = r->headers.find("Content-Length");
if (i != r->headers.end()) if (i != r->headers.end())
@ -255,20 +324,21 @@ ACTOR Future<Void> read_http_response(Reference<HTTP::Response> r, Reference<ICo
r->content.clear(); r->content.clear();
// If this is supposed to be a header-only response and the buffer has been fully processed then stop. Otherwise, // If this is allowed to be a header-only response and the buffer has been fully processed then stop. Otherwise,
// there must be response content. // there must be response content.
if (header_only && pos == buf.size()) if (content_optional && *pos == buf->size()) {
return Void(); return Void();
}
// There should be content (or at least metadata describing that there is no content. // There should be content (or at least metadata describing that there is no content.
// Chunked transfer and 'normal' mode (content length given, data in one segment after headers) are supported. // Chunked transfer and 'normal' mode (content length given, data in one segment after headers) are supported.
if (r->contentLen >= 0) { if (r->contentLen >= 0) {
// Use response content as the buffer so there's no need to copy it later. // Use response content as the buffer so there's no need to copy it later.
r->content = buf.substr(pos); r->content = buf->substr(*pos);
pos = 0; *pos = 0;
// Read until there are at least contentLen bytes available at pos // Read until there are at least contentLen bytes available at pos
wait(read_fixed_into_string(conn, r->contentLen, &r->content, pos)); wait(read_fixed_into_string(conn, r->contentLen, &r->content, *pos));
// There shouldn't be any bytes after content. // There shouldn't be any bytes after content.
if (r->content.size() != r->contentLen) if (r->content.size() != r->contentLen)
@ -277,47 +347,47 @@ ACTOR Future<Void> read_http_response(Reference<HTTP::Response> r, Reference<ICo
// Copy remaining buffer data to content which will now be the read buffer for the chunk encoded data. // Copy remaining buffer data to content which will now be the read buffer for the chunk encoded data.
// Overall this will be fairly efficient since most bytes will only be written once but some bytes will // Overall this will be fairly efficient since most bytes will only be written once but some bytes will
// have to be copied forward in the buffer when removing chunk overhead bytes. // have to be copied forward in the buffer when removing chunk overhead bytes.
r->content = buf.substr(pos); r->content = buf->substr(*pos);
pos = 0; *pos = 0;
loop { loop {
{ {
// Read the line that contains the chunk length as text in hex // Read the line that contains the chunk length as text in hex
size_t lineLen = wait(read_delimited_into_string(conn, "\r\n", &r->content, pos)); size_t lineLen = wait(read_delimited_into_string(conn, "\r\n", &r->content, *pos));
state int chunkLen = strtol(r->content.substr(pos, lineLen).c_str(), nullptr, 16); state int chunkLen = strtol(r->content.substr(*pos, lineLen).c_str(), nullptr, 16);
// Instead of advancing pos, erase the chunk length header line (line length + delimiter size) from the // Instead of advancing pos, erase the chunk length header line (line length + delimiter size) from the
// content buffer // content buffer
r->content.erase(pos, lineLen + 2); r->content.erase(*pos, lineLen + 2);
// If chunkLen is 0 then this marks the end of the content chunks. // If chunkLen is 0 then this marks the end of the content chunks.
if (chunkLen == 0) if (chunkLen == 0)
break; break;
// Read (if needed) until chunkLen bytes are available at pos, then advance pos by chunkLen // Read (if needed) until chunkLen bytes are available at pos, then advance pos by chunkLen
wait(read_fixed_into_string(conn, chunkLen, &r->content, pos)); wait(read_fixed_into_string(conn, chunkLen, &r->content, *pos));
pos += chunkLen; *pos += chunkLen;
} }
{ {
// Read the final empty line at the end of the chunk (the required "\r\n" after the chunk bytes) // Read the final empty line at the end of the chunk (the required "\r\n" after the chunk bytes)
size_t lineLen = wait(read_delimited_into_string(conn, "\r\n", &r->content, pos)); size_t lineLen = wait(read_delimited_into_string(conn, "\r\n", &r->content, *pos));
if (lineLen != 0) if (lineLen != 0)
throw http_bad_response(); throw http_bad_response();
// Instead of advancing pos, erase the empty line from the content buffer // Instead of advancing pos, erase the empty line from the content buffer
r->content.erase(pos, 2); r->content.erase(*pos, 2);
} }
} }
// The content buffer now contains the de-chunked, contiguous content at position 0 to pos. Save this length. // The content buffer now contains the de-chunked, contiguous content at position 0 to pos. Save this length.
r->contentLen = pos; r->contentLen = *pos;
// Next is the post-chunk header block, so read that. // Next is the post-chunk header block, so read that.
wait(read_http_response_headers(conn, &r->headers, &r->content, &pos)); wait(read_http_response_headers(conn, &r->headers, &r->content, pos));
// If the header parsing did not consume all of the buffer then something is wrong // If the header parsing did not consume all of the buffer then something is wrong
if (pos != r->content.size()) if (*pos != r->content.size())
throw http_bad_response(); throw http_bad_response();
// Now truncate the buffer to just the dechunked contiguous content. // Now truncate the buffer to just the dechunked contiguous content.
@ -329,11 +399,11 @@ ACTOR Future<Void> read_http_response(Reference<HTTP::Response> r, Reference<ICo
// If there is actual response content, check the MD5 sum against the Content-MD5 response header // If there is actual response content, check the MD5 sum against the Content-MD5 response header
if (r->content.size() > 0) { if (r->content.size() > 0) {
if (r->code == 206 && FLOW_KNOBS->HTTP_RESPONSE_SKIP_VERIFY_CHECKSUM_FOR_PARTIAL_CONTENT) { if (skipCheckMD5) {
return Void(); return Void();
} }
if (!r->verifyMD5(false)) { // false arg means do not fail if the Content-MD5 header is missing. if (!HTTP::verifyMD5(r, false)) { // false arg means do not fail if the Content-MD5 header is missing.
throw http_bad_response(); throw http_bad_response();
} }
} }
@ -341,35 +411,116 @@ ACTOR Future<Void> read_http_response(Reference<HTTP::Response> r, Reference<ICo
return Void(); return Void();
} }
Future<Void> HTTP::Response::read(Reference<IConnection> conn, bool header_only) { // Reads an HTTP request from a network connection
return read_http_response(Reference<HTTP::Response>::addRef(this), conn, header_only); // If the connection fails while being read the exception will emitted
// If the response is not parsable or complete in some way, http_bad_response will be thrown
ACTOR Future<Void> read_http_request(Reference<HTTP::IncomingRequest> r, Reference<IConnection> conn) {
state std::string buf;
state size_t pos = 0;
// Read HTTP response code and version line
size_t lineLen = wait(read_delimited_into_string(conn, "\r\n", &buf, pos));
// FIXME: this is pretty inefficient with 2 copies, but sscanf isn't the best with strings
std::string requestLine = buf.substr(0, lineLen);
std::stringstream ss(requestLine);
// read verb
ss >> r->verb;
if (ss.fail()) {
throw http_bad_response();
}
// read resource
ss >> r->resource;
if (ss.fail()) {
throw http_bad_response();
}
// read http version
std::string httpVersion;
ss >> httpVersion;
if (ss.fail()) {
throw http_bad_response();
}
if (ss && !ss.eof()) {
throw http_bad_response();
}
float version;
sscanf(httpVersion.c_str(), "HTTP/%f", &version);
if (version < 1.1) {
throw http_bad_response();
}
// Move position past the line found and the delimiter length
pos += lineLen + 2;
wait(readHTTPData(&r->data, conn, &buf, &pos, false, false));
return Void();
}
Future<Void> HTTP::IncomingRequest::read(Reference<IConnection> conn, bool header_only) {
return read_http_request(Reference<HTTP::IncomingRequest>::addRef(this), conn);
}
Future<Void> HTTP::OutgoingResponse::write(Reference<IConnection> conn) {
return writeResponse(conn, Reference<HTTP::OutgoingResponse>::addRef(this));
}
void HTTP::OutgoingResponse::reset() {
data.headers = HTTP::Headers();
data.content->discardAll();
data.contentLen = 0;
}
// Reads an HTTP response from a network connection
// If the connection fails while being read the exception will emitted
// If the response is not parsable or complete in some way, http_bad_response will be thrown
ACTOR Future<Void> read_http_response(Reference<HTTP::IncomingResponse> r,
Reference<IConnection> conn,
bool header_only) {
state std::string buf;
state size_t pos = 0;
// Read HTTP request line
size_t lineLen = wait(read_delimited_into_string(conn, "\r\n", &buf, pos));
int reachedEnd = -1;
sscanf(buf.c_str() + pos, "HTTP/%f %d%n", &r->version, &r->code, &reachedEnd);
if (reachedEnd < 0)
throw http_bad_response();
// Move position past the line found and the delimiter length
pos += lineLen + 2;
bool skipCheckMD5 = r->code == 206 && FLOW_KNOBS->HTTP_RESPONSE_SKIP_VERIFY_CHECKSUM_FOR_PARTIAL_CONTENT;
wait(readHTTPData(&r->data, conn, &buf, &pos, header_only, skipCheckMD5));
return Void();
}
Future<Void> HTTP::IncomingResponse::read(Reference<IConnection> conn, bool header_only) {
return read_http_response(Reference<HTTP::IncomingResponse>::addRef(this), conn, header_only);
} }
// Do a request, get a Response. // Do a request, get a Response.
// Request content is provided as UnsentPacketQueue *pContent which will be depleted as bytes are sent but the queue // Request content is provided as UnsentPacketQueue in req, which will be depleted as bytes are sent but the queue
// itself must live for the life of this actor and be destroyed by the caller // itself must live for the life of this actor and be destroyed by the caller
// TODO: pSent is very hackish, do something better. // TODO: pSent is very hackish, do something better.
ACTOR Future<Reference<HTTP::Response>> doRequest(Reference<IConnection> conn, ACTOR Future<Reference<HTTP::IncomingResponse>> doRequestActor(Reference<IConnection> conn,
std::string verb, Reference<OutgoingRequest> request,
std::string resource, Reference<IRateControl> sendRate,
HTTP::Headers headers, int64_t* pSent,
UnsentPacketQueue* pContent, Reference<IRateControl> recvRate) {
int contentLen,
Reference<IRateControl> sendRate,
int64_t* pSent,
Reference<IRateControl> recvRate,
std::string requestIDHeader) {
state TraceEvent event(SevDebug, "HTTPRequest"); state TraceEvent event(SevDebug, "HTTPRequest");
state UnsentPacketQueue empty;
if (pContent == nullptr)
pContent = &empty;
// There is no standard http request id header field, so either a global default can be set via a knob // There is no standard http request id header field, so either a global default can be set via a knob
// or it can be set per-request with the requestIDHeader argument (which overrides the default) // or it can be set per-request with the requestIDHeader argument (which overrides the default)
if (requestIDHeader.empty()) { state std::string requestIDHeader = FLOW_KNOBS->HTTP_REQUEST_ID_HEADER;
requestIDHeader = FLOW_KNOBS->HTTP_REQUEST_ID_HEADER;
}
state bool earlyResponse = false; state bool earlyResponse = false;
state int total_sent = 0; state int total_sent = 0;
@ -377,9 +528,9 @@ ACTOR Future<Reference<HTTP::Response>> doRequest(Reference<IConnection> conn,
event.detail("DebugID", conn->getDebugID()); event.detail("DebugID", conn->getDebugID());
event.detail("RemoteAddress", conn->getPeerAddress()); event.detail("RemoteAddress", conn->getPeerAddress());
event.detail("Verb", verb); event.detail("Verb", request->verb);
event.detail("Resource", resource); event.detail("Resource", request->resource);
event.detail("RequestContentLen", contentLen); event.detail("RequestContentLen", request->data.contentLen);
try { try {
state std::string requestID; state std::string requestID;
@ -390,52 +541,57 @@ ACTOR Future<Reference<HTTP::Response>> doRequest(Reference<IConnection> conn,
requestID = requestID.insert(12, "-"); requestID = requestID.insert(12, "-");
requestID = requestID.insert(8, "-"); requestID = requestID.insert(8, "-");
headers[requestIDHeader] = requestID; request->data.headers[requestIDHeader] = requestID;
event.detail("RequestIDSent", requestID); event.detail("RequestIDSent", requestID);
} }
request->data.headers["Content-Length"] = std::to_string(request->data.contentLen);
// Write headers to a packet buffer chain // Write headers to a packet buffer chain
PacketBuffer* pFirst = PacketBuffer::create(); PacketBuffer* pFirst = PacketBuffer::create();
PacketBuffer* pLast = writeRequestHeader(verb, resource, headers, pFirst); PacketBuffer* pLast = writeRequestHeader(request, pFirst);
// Prepend headers to content packer buffer chain // Prepend headers to content packer buffer chain
pContent->prependWriteBuffer(pFirst, pLast); request->data.content->prependWriteBuffer(pFirst, pLast);
if (FLOW_KNOBS->HTTP_VERBOSE_LEVEL > 1) if (FLOW_KNOBS->HTTP_VERBOSE_LEVEL > 1)
printf("[%s] HTTP starting %s %s ContentLen:%d\n", printf("[%s] HTTP starting %s %s ContentLen:%d\n",
conn->getDebugID().toString().c_str(), conn->getDebugID().toString().c_str(),
verb.c_str(), request->verb.c_str(),
resource.c_str(), request->resource.c_str(),
contentLen); request->data.contentLen);
if (FLOW_KNOBS->HTTP_VERBOSE_LEVEL > 2) { if (FLOW_KNOBS->HTTP_VERBOSE_LEVEL > 2) {
for (auto h : headers) for (auto h : request->data.headers)
printf("Request Header: %s: %s\n", h.first.c_str(), h.second.c_str()); printf("Request Header: %s: %s\n", h.first.c_str(), h.second.c_str());
} }
state Reference<HTTP::Response> r(new HTTP::Response()); state Reference<HTTP::IncomingResponse> r(new HTTP::IncomingResponse());
state Future<Void> responseReading = r->read(conn, verb == "HEAD" || verb == "DELETE" || verb == "CONNECT"); state Future<Void> responseReading = r->read(conn, request->isHeaderOnlyResponse());
send_start = timer(); send_start = timer();
// too many state things here to refactor this with writing the response
loop { loop {
// If we already got a response, before finishing sending the request, then close the connection, // If we already got a response, before finishing sending the request, then close the connection,
// set the Connection header to "close" as a hint to the caller that this connection can't be used // set the Connection header to "close" as a hint to the caller that this connection can't be used
// again, and break out of the send loop. // again, and break out of the send loop.
if (responseReading.isReady()) { if (responseReading.isReady()) {
conn->close(); conn->close();
r->headers["Connection"] = "close"; r->data.headers["Connection"] = "close";
earlyResponse = true; earlyResponse = true;
break; break;
} }
state int trySend = FLOW_KNOBS->HTTP_SEND_SIZE; state int trySend = FLOW_KNOBS->HTTP_SEND_SIZE;
if ((!g_network->isSimulated() || !g_simulator->speedUpSimulation) && BUGGIFY_WITH_PROB(0.01)) {
trySend = deterministicRandom()->randomInt(1, 10);
}
wait(sendRate->getAllowance(trySend)); wait(sendRate->getAllowance(trySend));
int len = conn->write(pContent->getUnsent(), trySend); int len = conn->write(request->data.content->getUnsent(), trySend);
if (pSent != nullptr) if (pSent != nullptr)
*pSent += len; *pSent += len;
sendRate->returnUnused(trySend - len); sendRate->returnUnused(trySend - len);
total_sent += len; total_sent += len;
pContent->sent(len); request->data.content->sent(len);
if (pContent->empty()) if (request->data.content->empty())
break; break;
wait(conn->onWritable()); wait(conn->onWritable());
@ -446,14 +602,14 @@ ACTOR Future<Reference<HTTP::Response>> doRequest(Reference<IConnection> conn,
double elapsed = timer() - send_start; double elapsed = timer() - send_start;
event.detail("ResponseCode", r->code); event.detail("ResponseCode", r->code);
event.detail("ResponseContentLen", r->contentLen); event.detail("ResponseContentLen", r->data.contentLen);
event.detail("Elapsed", elapsed); event.detail("Elapsed", elapsed);
Optional<Error> err; Optional<Error> err;
if (!requestIDHeader.empty()) { if (!requestIDHeader.empty()) {
std::string responseID; std::string responseID;
auto iid = r->headers.find(requestIDHeader); auto iid = r->data.headers.find(requestIDHeader);
if (iid != r->headers.end()) { if (iid != r->data.headers.end()) {
responseID = iid->second; responseID = iid->second;
} }
event.detail("RequestIDReceived", responseID); event.detail("RequestIDReceived", responseID);
@ -471,11 +627,11 @@ ACTOR Future<Reference<HTTP::Response>> doRequest(Reference<IConnection> conn,
.error(err.get()) .error(err.get())
.detail("DebugID", conn->getDebugID()) .detail("DebugID", conn->getDebugID())
.detail("RemoteAddress", conn->getPeerAddress()) .detail("RemoteAddress", conn->getPeerAddress())
.detail("Verb", verb) .detail("Verb", request->verb)
.detail("Resource", resource) .detail("Resource", request->resource)
.detail("RequestContentLen", contentLen) .detail("RequestContentLen", request->data.contentLen)
.detail("ResponseCode", r->code) .detail("ResponseCode", r->code)
.detail("ResponseContentLen", r->contentLen) .detail("ResponseContentLen", r->data.contentLen)
.detail("RequestIDSent", requestID) .detail("RequestIDSent", requestID)
.detail("RequestIDReceived", responseID); .detail("RequestIDReceived", responseID);
} }
@ -489,17 +645,17 @@ ACTOR Future<Reference<HTTP::Response>> doRequest(Reference<IConnection> conn,
r->code, r->code,
earlyResponse, earlyResponse,
elapsed, elapsed,
verb, request->verb,
resource, request->resource,
contentLen, request->data.contentLen,
total_sent, total_sent,
r->contentLen); r->data.contentLen);
} }
if (FLOW_KNOBS->HTTP_VERBOSE_LEVEL > 2) { if (FLOW_KNOBS->HTTP_VERBOSE_LEVEL > 2) {
printf("[%s] HTTP RESPONSE: %s %s\n%s\n", printf("[%s] HTTP RESPONSE: %s %s\n%s\n",
conn->getDebugID().toString().c_str(), conn->getDebugID().toString().c_str(),
verb.c_str(), request->verb.c_str(),
resource.c_str(), request->resource.c_str(),
r->toString().c_str()); r->toString().c_str());
} }
@ -517,9 +673,9 @@ ACTOR Future<Reference<HTTP::Response>> doRequest(Reference<IConnection> conn,
e.name(), e.name(),
earlyResponse, earlyResponse,
elapsed, elapsed,
verb.c_str(), request->verb.c_str(),
resource.c_str(), request->resource.c_str(),
contentLen, request->data.contentLen,
total_sent); total_sent);
} }
event.errorUnsuppressed(e); event.errorUnsuppressed(e);
@ -527,13 +683,18 @@ ACTOR Future<Reference<HTTP::Response>> doRequest(Reference<IConnection> conn,
} }
} }
// IDE build didn't like the actor conversion i guess
Future<Reference<IncomingResponse>> doRequest(Reference<IConnection> conn,
Reference<OutgoingRequest> request,
Reference<IRateControl> sendRate,
int64_t* pSent,
Reference<IRateControl> recvRate) {
return doRequestActor(conn, request, sendRate, pSent, recvRate);
}
ACTOR Future<Void> sendProxyConnectRequest(Reference<IConnection> conn, ACTOR Future<Void> sendProxyConnectRequest(Reference<IConnection> conn,
std::string remoteHost, std::string remoteHost,
std::string remoteService) { std::string remoteService) {
state Headers headers;
headers["Host"] = remoteHost + ":" + remoteService;
headers["Accept"] = "application/xml";
headers["Proxy-Connection"] = "Keep-Alive";
state int requestTimeout = 60; state int requestTimeout = 60;
state int maxTries = FLOW_KNOBS->RESTCLIENT_CONNECT_TRIES; state int maxTries = FLOW_KNOBS->RESTCLIENT_CONNECT_TRIES;
state int thisTry = 1; state int thisTry = 1;
@ -541,21 +702,23 @@ ACTOR Future<Void> sendProxyConnectRequest(Reference<IConnection> conn,
state Reference<IRateControl> sendReceiveRate = makeReference<Unlimited>(); state Reference<IRateControl> sendReceiveRate = makeReference<Unlimited>();
state int64_t bytes_sent = 0; state int64_t bytes_sent = 0;
state Reference<HTTP::OutgoingRequest> req = makeReference<HTTP::OutgoingRequest>();
req->verb = HTTP_VERB_CONNECT;
req->resource = remoteHost + ":" + remoteService;
req->data.content = nullptr;
req->data.contentLen = 0;
req->data.headers["Host"] = req->resource;
req->data.headers["Accept"] = "application/xml";
req->data.headers["Proxy-Connection"] = "Keep-Alive";
loop { loop {
state Optional<Error> err; state Optional<Error> err;
state Reference<Response> r;
state Reference<HTTP::IncomingResponse> r;
try { try {
Reference<Response> _r = wait(timeoutError(doRequest(conn, Future<Reference<HTTP::IncomingResponse>> f =
"CONNECT", HTTP::doRequest(conn, req, sendReceiveRate, &bytes_sent, sendReceiveRate);
remoteHost + ":" + remoteService, Reference<HTTP::IncomingResponse> _r = wait(timeoutError(f, requestTimeout));
headers,
nullptr,
0,
sendReceiveRate,
&bytes_sent,
sendReceiveRate),
requestTimeout));
r = _r; r = _r;
} catch (Error& e) { } catch (Error& e) {
if (e.code() == error_code_actor_cancelled) if (e.code() == error_code_actor_cancelled)
@ -601,8 +764,8 @@ ACTOR Future<Void> sendProxyConnectRequest(Reference<IConnection> conn,
if (retryable) { if (retryable) {
// If r is valid then obey the Retry-After response header if present. // If r is valid then obey the Retry-After response header if present.
if (r) { if (r) {
auto iRetryAfter = r->headers.find("Retry-After"); auto iRetryAfter = r->data.headers.find("Retry-After");
if (iRetryAfter != r->headers.end()) { if (iRetryAfter != r->data.headers.end()) {
event.detail("RetryAfterHeader", iRetryAfter->second); event.detail("RetryAfterHeader", iRetryAfter->second);
char* pEnd; char* pEnd;
double retryAfter = strtod(iRetryAfter->second.c_str(), &pEnd); double retryAfter = strtod(iRetryAfter->second.c_str(), &pEnd);

315
fdbrpc/HTTPServer.actor.cpp Normal file
View File

@ -0,0 +1,315 @@
/*
* HTTPServer.actor.cpp
*
* This source file is part of the FoundationDB open source project
*
* Copyright 2013-2022 Apple Inc. and the FoundationDB project authors
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#include "fdbrpc/HTTP.h"
#include "flow/IRandom.h"
#include "flow/Trace.h"
#include "fdbrpc/simulator.h"
#include "fdbrpc/SimulatorProcessInfo.h"
#include "flow/actorcompiler.h" // This must be the last #include.
ACTOR Future<Void> callbackHandler(Reference<IConnection> conn,
Future<Void> readRequestDone,
Reference<HTTP::IRequestHandler> requestHandler,
Reference<HTTP::IncomingRequest> req,
FlowMutex* mutex) {
state Reference<HTTP::OutgoingResponse> response = makeReference<HTTP::OutgoingResponse>();
state UnsentPacketQueue content;
response->data.content = &content;
response->data.contentLen = 0;
try {
wait(readRequestDone);
wait(requestHandler->handleRequest(req, response));
} catch (Error& e) {
if (e.code() == error_code_operation_cancelled) {
throw e;
}
// FIXME: other errors?
if (e.code() == error_code_http_request_failed || e.code() == error_code_http_bad_response ||
e.code() == error_code_connection_failed) {
TraceEvent(SevWarn, "HTTPServerConnHandlerInternalError").errorUnsuppressed(e);
// reset to empty error response
response->reset();
response->code = 500;
} else {
TraceEvent(SevWarn, "HTTPServerConnHandlerUnexpectedError").errorUnsuppressed(e);
throw e;
}
}
// take out response mutex to ensure no parallel writers to response connection
// FIXME: is this necessary? I think it is
state FlowMutex::Lock lock = wait(mutex->take());
try {
wait(response->write(conn));
} catch (Error& e) {
lock.release();
if (e.code() == error_code_connection_failed) {
// connection back to client failed, end. They will retry if they still need the response.
TraceEvent("HTTPServerConnHandlerResponseError").errorUnsuppressed(e);
return Void();
}
TraceEvent("HTTPServerConnHandlerResponseUnexpectedError").errorUnsuppressed(e);
throw e;
}
lock.release();
return Void();
}
ACTOR Future<Void> connectionHandler(Reference<HTTP::SimServerContext> server,
Reference<IConnection> conn,
Reference<HTTP::IRequestHandler> requestHandler) {
try {
// TODO do we actually have multiple requests on a connection? how does this work
state FlowMutex responseMutex;
state Future<Void> readPrevRequest = Future<Void>(Void());
wait(conn->acceptHandshake());
loop {
wait(readPrevRequest);
wait(delay(0));
wait(conn->onReadable());
state Reference<HTTP::IncomingRequest> req = makeReference<HTTP::IncomingRequest>();
readPrevRequest = req->read(conn, false);
server->actors.add(callbackHandler(conn, readPrevRequest, requestHandler, req, &responseMutex));
}
} catch (Error& e) {
if (e.code() != error_code_actor_cancelled) {
TraceEvent("HTTPConnectionError", server->dbgid)
.errorUnsuppressed(e)
.suppressFor(1.0)
.detail("ConnID", conn->getDebugID())
.detail("FromAddress", conn->getPeerAddress());
}
conn->close();
}
return Void();
}
ACTOR Future<Void> listenActor(Reference<HTTP::SimServerContext> server,
Reference<HTTP::IRequestHandler> requestHandler,
NetworkAddress addr,
Reference<IListener> listener) {
TraceEvent(SevDebug, "HTTPServerListenStart", server->dbgid).detail("ListenAddress", addr.toString());
wait(requestHandler->init());
TraceEvent(SevDebug, "HTTPServerListenInitialized", server->dbgid).detail("ListenAddress", addr.toString());
try {
loop {
Reference<IConnection> conn = wait(listener->accept());
if (!server->running) {
TraceEvent("HTTPServerExitedAfterAccept", server->dbgid);
break;
}
if (conn) {
server->actors.add(connectionHandler(server, conn, requestHandler));
}
}
} catch (Error& e) {
TraceEvent(SevError, "HTTPListenError", server->dbgid).error(e);
throw;
}
return Void();
}
NetworkAddress HTTP::SimServerContext::newAddress() {
// allocate new addr, assert we have enough addr space
ASSERT(listenAddresses.size() < 1000);
return NetworkAddress(
g_simulator->getCurrentProcess()->address.ip, nextPort++, true /* isPublic*/, false /*isTLS*/);
}
void HTTP::SimServerContext::registerNewServer(NetworkAddress addr, Reference<HTTP::IRequestHandler> requestHandler) {
listenAddresses.push_back(addr);
listeners.push_back(INetworkConnections::net()->listen(addr));
actors.add(listenActor(Reference<HTTP::SimServerContext>::addRef(this), requestHandler, addr, listeners.back()));
}
// unit test stuff
ACTOR Future<Void> helloWorldServerCallback(Reference<HTTP::IncomingRequest> req,
Reference<HTTP::OutgoingResponse> response) {
wait(delay(0));
ASSERT_EQ(req->verb, HTTP::HTTP_VERB_POST);
ASSERT_EQ(req->resource, "/hello-world");
ASSERT_EQ(req->data.headers.size(), 2);
ASSERT(req->data.headers.count("Hello"));
ASSERT_EQ(req->data.headers["Hello"], "World");
ASSERT(req->data.headers.count("Content-Length"));
ASSERT_EQ(req->data.headers["Content-Length"], std::to_string(req->data.content.size()));
ASSERT_EQ(req->data.contentLen, req->data.content.size());
ASSERT_EQ(req->data.content, "Hello World Request!");
response->code = 200;
response->data.headers["Hello"] = "World";
std::string hello = "Hello World Response!";
PacketWriter pw(response->data.content->getWriteBuffer(hello.size()), nullptr, Unversioned());
pw.serializeBytes(hello);
response->data.contentLen = hello.size();
return Void();
}
struct HelloWorldRequestHandler : HTTP::IRequestHandler, ReferenceCounted<HelloWorldRequestHandler> {
Future<Void> handleRequest(Reference<HTTP::IncomingRequest> req,
Reference<HTTP::OutgoingResponse> response) override {
return helloWorldServerCallback(req, response);
}
Reference<HTTP::IRequestHandler> clone() override { return makeReference<HelloWorldRequestHandler>(); }
void addref() override { ReferenceCounted<HelloWorldRequestHandler>::addref(); }
void delref() override { ReferenceCounted<HelloWorldRequestHandler>::delref(); }
};
ACTOR Future<Void> helloErrorServerCallback(Reference<HTTP::IncomingRequest> req,
Reference<HTTP::OutgoingResponse> response) {
wait(delay(0));
if (deterministicRandom()->coinflip()) {
throw http_bad_response();
} else {
throw http_request_failed();
}
}
struct HelloErrorRequestHandler : HTTP::IRequestHandler, ReferenceCounted<HelloErrorRequestHandler> {
Future<Void> handleRequest(Reference<HTTP::IncomingRequest> req,
Reference<HTTP::OutgoingResponse> response) override {
return helloErrorServerCallback(req, response);
}
Reference<HTTP::IRequestHandler> clone() override { return makeReference<HelloErrorRequestHandler>(); }
void addref() override { ReferenceCounted<HelloErrorRequestHandler>::addref(); }
void delref() override { ReferenceCounted<HelloErrorRequestHandler>::delref(); }
};
typedef std::function<Future<Reference<HTTP::IncomingResponse>>(Reference<IConnection> conn)> DoRequestFunction;
// handles retrying on timeout and reinitializing connection like other users of HTTP (S3BlobStore, RestClient)
ACTOR Future<Reference<HTTP::IncomingResponse>> doRequestTest(std::string hostname,
std::string service,
DoRequestFunction reqFunction) {
state Reference<IConnection> conn;
loop {
if (!conn) {
wait(store(conn, INetworkConnections::net()->connect(hostname, service, false)));
ASSERT(conn.isValid());
wait(conn->connectHandshake());
}
try {
Future<Reference<HTTP::IncomingResponse>> f = reqFunction(conn);
Reference<HTTP::IncomingResponse> response = wait(f);
conn->close();
return response;
} catch (Error& e) {
conn->close();
if (e.code() != error_code_timed_out && e.code() != error_code_connection_failed) {
throw e;
}
// request got stuck, close conn and try again
conn.clear();
wait(delay(0.1));
}
}
}
ACTOR Future<Reference<HTTP::IncomingResponse>> doHelloWorldReq(Reference<IConnection> conn) {
state UnsentPacketQueue content;
state Reference<HTTP::OutgoingRequest> req = makeReference<HTTP::OutgoingRequest>();
state Reference<IRateControl> sendReceiveRate = makeReference<Unlimited>();
state int64_t bytes_sent = 0;
req->verb = HTTP::HTTP_VERB_POST;
req->resource = "/hello-world";
req->data.headers["Hello"] = "World";
std::string hello = "Hello World Request!";
req->data.content = &content;
req->data.contentLen = hello.size();
PacketWriter pw(req->data.content->getWriteBuffer(hello.size()), nullptr, Unversioned());
pw.serializeBytes(hello);
Reference<HTTP::IncomingResponse> response =
wait(timeoutError(HTTP::doRequest(conn, req, sendReceiveRate, &bytes_sent, sendReceiveRate), 30.0));
ASSERT_EQ(response->code, 200);
ASSERT_EQ(response->data.headers.size(), 2);
ASSERT(response->data.headers.count("Hello"));
ASSERT_EQ(response->data.headers["Hello"], "World");
ASSERT(response->data.headers.count("Content-Length"));
ASSERT_EQ(response->data.headers["Content-Length"], std::to_string(response->data.content.size()));
ASSERT_EQ(response->data.contentLen, response->data.content.size());
ASSERT_EQ(response->data.content, "Hello World Response!");
return response;
}
ACTOR Future<Reference<HTTP::IncomingResponse>> doHelloWorldErrorReq(Reference<IConnection> conn) {
state UnsentPacketQueue content;
state Reference<HTTP::OutgoingRequest> req = makeReference<HTTP::OutgoingRequest>();
state Reference<IRateControl> sendReceiveRate = makeReference<Unlimited>();
state int64_t bytes_sent = 0;
req->verb = HTTP::HTTP_VERB_GET;
req->resource = "/hello-error";
req->data.content = &content;
req->data.contentLen = 0;
Reference<HTTP::IncomingResponse> response =
wait(timeoutError(HTTP::doRequest(conn, req, sendReceiveRate, &bytes_sent, sendReceiveRate), 30.0));
ASSERT(response->code == 500);
return response;
}
// can't run as regular unit test right now because it needs special setup
TEST_CASE("!/HTTP/Server/HelloWorld") {
ASSERT(g_network->isSimulated());
fmt::print("Registering sim server\n");
wait(g_simulator->registerSimHTTPServer("helloworld", "80", 1, makeReference<HelloWorldRequestHandler>()));
fmt::print("Registered sim server\n");
wait(success(doRequestTest("helloworld", "80", doHelloWorldReq)));
fmt::print("Done\n");
return Void();
}
TEST_CASE("!/HTTP/Server/HelloError") {
ASSERT(g_network->isSimulated());
fmt::print("Registering sim server\n");
wait(g_simulator->registerSimHTTPServer("helloerror", "80", 1, makeReference<HelloErrorRequestHandler>()));
fmt::print("Registered sim server\n");
wait(success(doRequestTest("helloerror", "80", doHelloWorldErrorReq)));
fmt::print("Done\n");
return Void();
}

168
fdbrpc/include/fdbrpc/HTTP.h
View File

@ -3,7 +3,7 @@
* *
* This source file is part of the FoundationDB open source project * This source file is part of the FoundationDB open source project
* *
* Copyright 2013-2022 Apple Inc. and the FoundationDB project authors * Copyright 2013-2023 Apple Inc. and the FoundationDB project authors
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License. * you may not use this file except in compliance with the License.
@ -21,61 +21,20 @@
#ifndef FDBRPC_HTTP_H #ifndef FDBRPC_HTTP_H
#define FDBRPC_HTTP_H #define FDBRPC_HTTP_H
#include "flow/NetworkAddress.h"
#pragma once #pragma once
#include "flow/flow.h" #include "flow/flow.h"
#include "flow/Net2Packet.h" #include "flow/ActorCollection.h"
#include "flow/IConnection.h"
#include "flow/IRateControl.h" #include "flow/IRateControl.h"
#include "flow/Net2Packet.h"
class IConnection;
namespace HTTP { namespace HTTP {
struct is_iless { struct is_iless {
bool operator()(const std::string& a, const std::string& b) const { return strcasecmp(a.c_str(), b.c_str()) < 0; } bool operator()(const std::string& a, const std::string& b) const { return strcasecmp(a.c_str(), b.c_str()) < 0; }
}; };
typedef std::map<std::string, std::string, is_iless> Headers;
std::string urlEncode(const std::string& s);
std::string awsV4URIEncode(const std::string& s, bool encodeSlash);
struct Response : ReferenceCounted<Response> {
Response() {}
Future<Void> read(Reference<IConnection> conn, bool header_only);
std::string toString();
float version;
int code;
Headers headers;
std::string content;
int64_t contentLen;
bool verifyMD5(bool fail_if_header_missing, Optional<std::string> content_sum = Optional<std::string>());
};
// Prepend the HTTP request header to the given PacketBuffer, returning the new head of the buffer chain
PacketBuffer* writeRequestHeader(std::string const& verb,
std::string const& resource,
HTTP::Headers const& headers,
PacketBuffer* dest);
// Do an HTTP request to the blob store, parse the response.
Future<Reference<Response>> doRequest(Reference<IConnection> const& conn,
std::string const& verb,
std::string const& resource,
HTTP::Headers const& headers,
UnsentPacketQueue* const& pContent,
int const& contentLen,
Reference<IRateControl> const& sendRate,
int64_t* const& pSent,
Reference<IRateControl> const& recvRate,
const std::string& requestHeader = std::string());
// Connect to proxy, send CONNECT command, and connect to the remote host.
Future<Reference<IConnection>> proxyConnect(const std::string& remoteHost,
const std::string& remoteService,
const std::string& proxyHost,
const std::string& proxyService);
constexpr int HTTP_STATUS_CODE_OK = 200; constexpr int HTTP_STATUS_CODE_OK = 200;
constexpr int HTTP_STATUS_CODE_CREATED = 201; constexpr int HTTP_STATUS_CODE_CREATED = 201;
constexpr int HTTP_STATUS_CODE_ACCEPTED = 202; constexpr int HTTP_STATUS_CODE_ACCEPTED = 202;
@ -87,7 +46,7 @@ constexpr int HTTP_STATUS_CODE_TOO_MANY_REQUESTS = 429;
constexpr int HTTP_STATUS_CODE_INTERNAL_SERVER_ERROR = 500; constexpr int HTTP_STATUS_CODE_INTERNAL_SERVER_ERROR = 500;
constexpr int HTTP_STATUS_CODE_BAD_GATEWAY = 502; constexpr int HTTP_STATUS_CODE_BAD_GATEWAY = 502;
constexpr int HTTP_STATUS_CODE_SERVICE_UNAVAILABLE = 503; constexpr int HTTP_STATUS_CODE_SERVICE_UNAVAILABLE = 503;
constexpr int HTTP_STATUS_GATEWAY_TIMEOUT = 504; constexpr int HTTP_STATUS_CODE_GATEWAY_TIMEOUT = 504;
constexpr int HTTP_RETRYAFTER_DELAY_SECS = 300; constexpr int HTTP_RETRYAFTER_DELAY_SECS = 300;
@ -97,6 +56,121 @@ const std::string HTTP_VERB_DELETE = "DELETE";
const std::string HTTP_VERB_TRACE = "TRACE"; const std::string HTTP_VERB_TRACE = "TRACE";
const std::string HTTP_VERB_PUT = "PUT"; const std::string HTTP_VERB_PUT = "PUT";
const std::string HTTP_VERB_POST = "POST"; const std::string HTTP_VERB_POST = "POST";
const std::string HTTP_VERB_CONNECT = "CONNECT";
typedef std::map<std::string, std::string, is_iless> Headers;
std::string urlEncode(const std::string& s);
std::string awsV4URIEncode(const std::string& s, bool encodeSlash);
template <class T>
struct HTTPData {
Headers headers;
int64_t contentLen;
T content;
};
// class methods on template type classes are weird
bool verifyMD5(HTTPData<std::string>* data,
bool fail_if_header_missing,
Optional<std::string> content_sum = Optional<std::string>());
template <class T>
struct RequestBase : ReferenceCounted<RequestBase<T>> {
RequestBase() {}
std::string verb;
std::string resource;
HTTPData<T> data;
bool isHeaderOnlyResponse() {
return verb == HTTP_VERB_HEAD || verb == HTTP_VERB_DELETE || verb == HTTP_VERB_CONNECT;
}
};
// TODO: utility for constructing packet buffer from string OutgoingRequest
struct IncomingRequest : RequestBase<std::string> {
Future<Void> read(Reference<IConnection> conn, bool header_only = false);
};
struct OutgoingRequest : RequestBase<UnsentPacketQueue*> {};
template <class T>
struct ResponseBase : ReferenceCounted<ResponseBase<T>> {
ResponseBase() {}
float version;
int code;
HTTPData<T> data;
std::string getCodeDescription();
};
struct IncomingResponse : ResponseBase<std::string> {
std::string toString() const; // for debugging
Future<Void> read(Reference<IConnection> conn, bool header_only = false);
};
struct OutgoingResponse : ResponseBase<UnsentPacketQueue*> {
Future<Void> write(Reference<IConnection> conn);
void reset();
};
// Do an HTTP request to the blob store, parse the response.
Future<Reference<IncomingResponse>> doRequest(Reference<IConnection> conn,
Reference<OutgoingRequest> request,
Reference<IRateControl> sendRate,
int64_t* pSent,
Reference<IRateControl> recvRate);
// Connect to proxy, send CONNECT command, and connect to the remote host.
Future<Reference<IConnection>> proxyConnect(const std::string& remoteHost,
const std::string& remoteService,
const std::string& proxyHost,
const std::string& proxyService);
// HTTP server stuff
// typedef std::function<Future<Void>(Reference<IncomingRequest>, Reference<OutgoingResponse>)> ServerCallback;
// Implementation of http server that handles http requests
// TODO: could change to factory pattern instead of clone pattern
struct IRequestHandler {
// Sets up state for each instance of the handler. Provides default stateless implementation, but a stateful handler
// must override this.
virtual Future<Void> init() { return Future<Void>(Void()); };
// Actual callback implementation. Fills out the response object based on the request.
virtual Future<Void> handleRequest(Reference<IncomingRequest>, Reference<OutgoingResponse>) = 0;
// If each instance has a mix of global state provided in the type-specific construtor, but then also local state
// instantiated in init, the default instance passed to registerSimHTTPServer is cloned for each process to copy the
// global state, but before init is called. You may optionally clone after init, but the contract is that clone must
// not copy or share the non-global state between instances.
virtual Reference<IRequestHandler> clone() = 0;
// for reference counting an interface - don't implement ReferenceCounted<T>
virtual void addref() = 0;
virtual void delref() = 0;
};
struct SimServerContext : ReferenceCounted<SimServerContext>, NonCopyable {
UID dbgid;
bool running;
int nextPort;
ActorCollection actors;
std::vector<NetworkAddress> listenAddresses;
std::vector<Future<Void>> listenBinds;
std::vector<Reference<IListener>> listeners;
SimServerContext() : dbgid(deterministicRandom()->randomUniqueID()), running(true), actors(false), nextPort(5000) {}
NetworkAddress newAddress();
void registerNewServer(NetworkAddress addr, Reference<IRequestHandler> server);
void stop() {
running = false;
actors = ActorCollection(false);
listenAddresses.clear();
listenBinds.clear();
listeners.clear();
}
};
} // namespace HTTP } // namespace HTTP

2
fdbrpc/include/fdbrpc/Locality.h
View File

@ -51,6 +51,7 @@ struct ProcessClass {
EncryptKeyProxyClass, EncryptKeyProxyClass,
ConsistencyScanClass, ConsistencyScanClass,
BlobMigratorClass, BlobMigratorClass,
SimHTTPServerClass,
InvalidClass = -1 InvalidClass = -1
}; };
@ -79,6 +80,7 @@ struct ProcessClass {
static_assert(ProcessClass::EncryptKeyProxyClass == 20); static_assert(ProcessClass::EncryptKeyProxyClass == 20);
static_assert(ProcessClass::ConsistencyScanClass == 21); static_assert(ProcessClass::ConsistencyScanClass == 21);
static_assert(ProcessClass::BlobMigratorClass == 22); static_assert(ProcessClass::BlobMigratorClass == 22);
static_assert(ProcessClass::SimHTTPServerClass == 23);
static_assert(ProcessClass::InvalidClass == -1); static_assert(ProcessClass::InvalidClass == -1);
enum Fitness { enum Fitness {

11
fdbrpc/include/fdbrpc/simulator.h
View File

@ -37,6 +37,7 @@
#include "fdbrpc/Locality.h" #include "fdbrpc/Locality.h"
#include "flow/IAsyncFile.h" #include "flow/IAsyncFile.h"
#include "flow/TDMetric.actor.h" #include "flow/TDMetric.actor.h"
#include "fdbrpc/HTTP.h"
#include "fdbrpc/FailureMonitor.h" #include "fdbrpc/FailureMonitor.h"
#include "fdbrpc/Locality.h" #include "fdbrpc/Locality.h"
#include "fdbrpc/ReplicationPolicy.h" #include "fdbrpc/ReplicationPolicy.h"
@ -281,6 +282,12 @@ public:
virtual void destroyProcess(ProcessInfo* p) = 0; virtual void destroyProcess(ProcessInfo* p) = 0;
virtual void destroyMachine(Optional<Standalone<StringRef>> const& machineId) = 0; virtual void destroyMachine(Optional<Standalone<StringRef>> const& machineId) = 0;
virtual void addSimHTTPProcess(Reference<HTTP::SimServerContext> serverContext) = 0;
virtual Future<Void> registerSimHTTPServer(std::string hostname,
std::string service,
int numAddresses,
Reference<HTTP::IRequestHandler> requestHandler) = 0;
int desiredCoordinators; int desiredCoordinators;
int physicalDatacenters; int physicalDatacenters;
int processesPerMachine; int processesPerMachine;
@ -353,6 +360,10 @@ public:
// 'plaintext marker' is present. // 'plaintext marker' is present.
Optional<std::string> dataAtRestPlaintextMarker; Optional<std::string> dataAtRestPlaintextMarker;
std::set<std::string> httpServerHostnames;
std::vector<std::pair<ProcessInfo*, Reference<HTTP::SimServerContext>>> httpServerProcesses;
std::set<IPAddress> httpServerIps;
flowGlobalType global(int id) const final; flowGlobalType global(int id) const final;
void setGlobal(size_t id, flowGlobalType v) final; void setGlobal(size_t id, flowGlobalType v) final;

78
fdbrpc/sim2.actor.cpp
View File

@ -1099,6 +1099,10 @@ public:
} }
Future<Reference<IConnection>> connectExternal(NetworkAddress toAddr) override { Future<Reference<IConnection>> connectExternal(NetworkAddress toAddr) override {
// If sim http connection, do connect instead of external connect
if (httpServerIps.count(toAddr.ip)) {
return connect(toAddr);
}
return SimExternalConnection::connect(toAddr); return SimExternalConnection::connect(toAddr);
} }
@ -2411,6 +2415,80 @@ public:
machines.erase(machineId); machines.erase(machineId);
} }
// add a simulated http server process. New http servers called by registerHTTPServer will run on this process
void addSimHTTPProcess(Reference<HTTP::SimServerContext> context) override {
ProcessInfo* p = getCurrentProcess();
// make sure this process isn't already added
for (int i = 0; i < httpServerProcesses.size(); i++) {
ASSERT(p != httpServerProcesses[i].first);
}
httpServerProcesses.push_back({ p, context });
httpServerIps.insert(p->address.ip);
}
ACTOR static Future<Void> registerSimHTTPServerActor(Sim2* self,
std::string hostname,
std::string service,
int numAddresses,
Reference<HTTP::IRequestHandler> requestHandler) {
// handle race where test client tries to register server before all processes are up, but time out eventually
// FIXME: make this so a server that starts after this or a server that restarts will automatically re-add
// itself, register the handler, and register with dns
state int checks = 0;
while (self->httpServerProcesses.empty()) {
TraceEvent(SevWarn, "NoAvailableHTTPServerProcesses").detail("Checks", checks);
checks++;
ASSERT(checks < 10);
wait(self->delay(1.0, TaskPriority::DefaultDelay));
}
ASSERT(!self->httpServerProcesses.empty());
ASSERT(!self->httpServerHostnames.count(hostname));
ASSERT(numAddresses > 0);
self->httpServerHostnames.insert(hostname);
state std::vector<NetworkAddress> addresses;
addresses.reserve(numAddresses);
// randomize order and round-robin servers among numAddresses
deterministicRandom()->randomShuffle(self->httpServerProcesses);
state ProcessInfo* callingProcess = self->getCurrentProcess();
state int i = 0;
for (; i < numAddresses; i++) {
state ProcessInfo* serverProcess = self->httpServerProcesses[i % self->httpServerProcesses.size()].first;
wait(self->onProcess(serverProcess, TaskPriority::DefaultYield));
try {
auto& proc = self->httpServerProcesses[i % self->httpServerProcesses.size()].second;
NetworkAddress addr = proc->newAddress();
addresses.push_back(addr);
serverProcess->listenerMap[addr] = Reference<IListener>(new Sim2Listener(serverProcess, addr));
self->addressMap[addr] = serverProcess;
proc->registerNewServer(addr, requestHandler->clone());
} catch (Error& e) {
// this should never happen, but would cause weird behavior if it did like unintentionally switching
// processes, so just fail
TraceEvent(SevError, "UnexpectedErrorRegisteringHTTPServer").errorUnsuppressed(e);
ASSERT(false);
}
}
wait(self->onProcess(callingProcess, TaskPriority::DefaultYield));
INetworkConnections::net()->addMockTCPEndpoint(hostname, service, addresses);
return Void();
}
// starts a numAddresses http servers with the dns alias hostname:service with the provided server callback
Future<Void> registerSimHTTPServer(std::string hostname,
std::string service,
int numAddresses,
Reference<HTTP::IRequestHandler> requestHandler) override {
return registerSimHTTPServerActor(this, hostname, service, numAddresses, requestHandler);
}
Sim2(bool printSimTime) Sim2(bool printSimTime)
: time(0.0), timerTime(0.0), currentTaskID(TaskPriority::Zero), yielded(false), yield_limit(0), : time(0.0), timerTime(0.0), currentTaskID(TaskPriority::Zero), yielded(false), yield_limit(0),
printSimTime(printSimTime) { printSimTime(printSimTime) {

105
fdbserver/RESTKmsConnector.actor.cpp
View File

@ -206,7 +206,7 @@ bool shouldRefreshKmsUrls(Reference<RESTKmsConnectorCtx> ctx) {
void extractKmsUrls(Reference<RESTKmsConnectorCtx> ctx, void extractKmsUrls(Reference<RESTKmsConnectorCtx> ctx,
const rapidjson::Document& doc, const rapidjson::Document& doc,
Reference<HTTP::Response> httpResp) { Reference<HTTP::IncomingResponse> httpResp) {
// Refresh KmsUrls cache // Refresh KmsUrls cache
dropCachedKmsUrls(ctx); dropCachedKmsUrls(ctx);
ASSERT(ctx->kmsUrlHeap.empty()); ASSERT(ctx->kmsUrlHeap.empty());
@ -356,7 +356,7 @@ void checkResponseForError(Reference<RESTKmsConnectorCtx> ctx,
} }
void checkDocForNewKmsUrls(Reference<RESTKmsConnectorCtx> ctx, void checkDocForNewKmsUrls(Reference<RESTKmsConnectorCtx> ctx,
Reference<HTTP::Response> resp, Reference<HTTP::IncomingResponse> resp,
const rapidjson::Document& doc) { const rapidjson::Document& doc) {
if (doc.HasMember(KMS_URLS_TAG) && !doc[KMS_URLS_TAG].IsNull()) { if (doc.HasMember(KMS_URLS_TAG) && !doc[KMS_URLS_TAG].IsNull()) {
try { try {
@ -369,7 +369,7 @@ void checkDocForNewKmsUrls(Reference<RESTKmsConnectorCtx> ctx,
} }
Standalone<VectorRef<EncryptCipherKeyDetailsRef>> parseEncryptCipherResponse(Reference<RESTKmsConnectorCtx> ctx, Standalone<VectorRef<EncryptCipherKeyDetailsRef>> parseEncryptCipherResponse(Reference<RESTKmsConnectorCtx> ctx,
Reference<HTTP::Response> resp) { Reference<HTTP::IncomingResponse> resp) {
// Acceptable response payload json format: // Acceptable response payload json format:
// //
// response_json_payload { // response_json_payload {
@ -405,7 +405,7 @@ Standalone<VectorRef<EncryptCipherKeyDetailsRef>> parseEncryptCipherResponse(Ref
} }
rapidjson::Document doc; rapidjson::Document doc;
doc.Parse(resp->content.data()); doc.Parse(resp->data.content.data());
checkResponseForError(ctx, doc, IsCipherType::True); checkResponseForError(ctx, doc, IsCipherType::True);
@ -500,7 +500,7 @@ Standalone<VectorRef<EncryptCipherKeyDetailsRef>> parseEncryptCipherResponse(Ref
} }
Standalone<VectorRef<BlobMetadataDetailsRef>> parseBlobMetadataResponse(Reference<RESTKmsConnectorCtx> ctx, Standalone<VectorRef<BlobMetadataDetailsRef>> parseBlobMetadataResponse(Reference<RESTKmsConnectorCtx> ctx,
Reference<HTTP::Response> resp) { Reference<HTTP::IncomingResponse> resp) {
// Acceptable response payload json format: // Acceptable response payload json format:
// (baseLocation and partitions follow the same properties as described in BlobMetadataUtils.h) // (baseLocation and partitions follow the same properties as described in BlobMetadataUtils.h)
// //
@ -536,7 +536,7 @@ Standalone<VectorRef<BlobMetadataDetailsRef>> parseBlobMetadataResponse(Referenc
} }
rapidjson::Document doc; rapidjson::Document doc;
doc.Parse(resp->content.data()); doc.Parse(resp->data.content.data());
checkResponseForError(ctx, doc, IsCipherType::False); checkResponseForError(ctx, doc, IsCipherType::False);
@ -768,10 +768,11 @@ StringRef getEncryptKeysByKeyIdsRequestBody(Reference<RESTKmsConnectorCtx> ctx,
} }
ACTOR template <class T> ACTOR template <class T>
Future<T> kmsRequestImpl(Reference<RESTKmsConnectorCtx> ctx, Future<T> kmsRequestImpl(
std::string urlSuffix, Reference<RESTKmsConnectorCtx> ctx,
StringRef requestBodyRef, std::string urlSuffix,
std::function<T(Reference<RESTKmsConnectorCtx>, Reference<HTTP::Response>)> parseFunc) { StringRef requestBodyRef,
std::function<T(Reference<RESTKmsConnectorCtx>, Reference<HTTP::IncomingResponse>)> parseFunc) {
state UID requestID = deterministicRandom()->randomUniqueID(); state UID requestID = deterministicRandom()->randomUniqueID();
// Follow 2-phase scheme: // Follow 2-phase scheme:
@ -804,7 +805,7 @@ Future<T> kmsRequestImpl(Reference<RESTKmsConnectorCtx> ctx,
headers["Content-type"] = "application/json"; headers["Content-type"] = "application/json";
headers["Accept"] = "application/json"; headers["Accept"] = "application/json";
Reference<HTTP::Response> resp = Reference<HTTP::IncomingResponse> resp =
wait(ctx->restClient.doPost(kmsEncryptionFullUrl, requestBodyRef.toString(), headers)); wait(ctx->restClient.doPost(kmsEncryptionFullUrl, requestBodyRef.toString(), headers));
curUrl->nRequests++; curUrl->nRequests++;
@ -859,7 +860,7 @@ ACTOR Future<Void> fetchEncryptionKeysByKeyIds(Reference<RESTKmsConnectorCtx> ct
bool refreshKmsUrls = shouldRefreshKmsUrls(ctx); bool refreshKmsUrls = shouldRefreshKmsUrls(ctx);
StringRef requestBodyRef = getEncryptKeysByKeyIdsRequestBody(ctx, req, refreshKmsUrls, req.arena); StringRef requestBodyRef = getEncryptKeysByKeyIdsRequestBody(ctx, req, refreshKmsUrls, req.arena);
std::function<Standalone<VectorRef<EncryptCipherKeyDetailsRef>>(Reference<RESTKmsConnectorCtx>, std::function<Standalone<VectorRef<EncryptCipherKeyDetailsRef>>(Reference<RESTKmsConnectorCtx>,
Reference<HTTP::Response>)> Reference<HTTP::IncomingResponse>)>
f = &parseEncryptCipherResponse; f = &parseEncryptCipherResponse;
wait(store( wait(store(
reply.cipherKeyDetails, reply.cipherKeyDetails,
@ -941,7 +942,7 @@ ACTOR Future<Void> fetchEncryptionKeysByDomainIds(Reference<RESTKmsConnectorCtx>
StringRef requestBodyRef = getEncryptKeysByDomainIdsRequestBody(ctx, req, refreshKmsUrls, req.arena); StringRef requestBodyRef = getEncryptKeysByDomainIdsRequestBody(ctx, req, refreshKmsUrls, req.arena);
std::function<Standalone<VectorRef<EncryptCipherKeyDetailsRef>>(Reference<RESTKmsConnectorCtx>, std::function<Standalone<VectorRef<EncryptCipherKeyDetailsRef>>(Reference<RESTKmsConnectorCtx>,
Reference<HTTP::Response>)> Reference<HTTP::IncomingResponse>)>
f = &parseEncryptCipherResponse; f = &parseEncryptCipherResponse;
wait(store(reply.cipherKeyDetails, wait(store(reply.cipherKeyDetails,
@ -1026,7 +1027,7 @@ ACTOR Future<Void> fetchBlobMetadata(Reference<RESTKmsConnectorCtx> ctx, KmsConn
// for some reason the compiler can't handle just passing &parseBlobMetadata, so you have to explicitly // for some reason the compiler can't handle just passing &parseBlobMetadata, so you have to explicitly
// declare its templated return type as part of an std::function first // declare its templated return type as part of an std::function first
std::function<Standalone<VectorRef<BlobMetadataDetailsRef>>(Reference<RESTKmsConnectorCtx>, std::function<Standalone<VectorRef<BlobMetadataDetailsRef>>(Reference<RESTKmsConnectorCtx>,
Reference<HTTP::Response>)> Reference<HTTP::IncomingResponse>)>
f = &parseBlobMetadataResponse; f = &parseBlobMetadataResponse;
wait( wait(
store(reply.metadataDetails, store(reply.metadataDetails,
@ -1385,7 +1386,7 @@ void addFakeKmsUrls(const rapidjson::Document& reqDoc, rapidjson::Document& resD
void getFakeEncryptCipherResponse(StringRef jsonReqRef, void getFakeEncryptCipherResponse(StringRef jsonReqRef,
const bool baseCipherIdPresent, const bool baseCipherIdPresent,
Reference<HTTP::Response> httpResponse) { Reference<HTTP::IncomingResponse> httpResponse) {
rapidjson::Document reqDoc; rapidjson::Document reqDoc;
reqDoc.Parse(jsonReqRef.toString().data()); reqDoc.Parse(jsonReqRef.toString().data());
@ -1436,14 +1437,14 @@ void getFakeEncryptCipherResponse(StringRef jsonReqRef,
rapidjson::StringBuffer sb; rapidjson::StringBuffer sb;
rapidjson::Writer<rapidjson::StringBuffer> writer(sb); rapidjson::Writer<rapidjson::StringBuffer> writer(sb);
resDoc.Accept(writer); resDoc.Accept(writer);
httpResponse->content.resize(sb.GetSize(), '\0'); httpResponse->data.content.resize(sb.GetSize(), '\0');
memcpy(httpResponse->content.data(), sb.GetString(), sb.GetSize()); memcpy(httpResponse->data.content.data(), sb.GetString(), sb.GetSize());
httpResponse->contentLen = sb.GetSize(); httpResponse->data.contentLen = sb.GetSize();
} }
void getFakeBlobMetadataResponse(StringRef jsonReqRef, void getFakeBlobMetadataResponse(StringRef jsonReqRef,
const bool baseCipherIdPresent, const bool baseCipherIdPresent,
Reference<HTTP::Response> httpResponse) { Reference<HTTP::IncomingResponse> httpResponse) {
rapidjson::Document reqDoc; rapidjson::Document reqDoc;
reqDoc.Parse(jsonReqRef.toString().data()); reqDoc.Parse(jsonReqRef.toString().data());
@ -1499,8 +1500,8 @@ void getFakeBlobMetadataResponse(StringRef jsonReqRef,
rapidjson::StringBuffer sb; rapidjson::StringBuffer sb;
rapidjson::Writer<rapidjson::StringBuffer> writer(sb); rapidjson::Writer<rapidjson::StringBuffer> writer(sb);
resDoc.Accept(writer); resDoc.Accept(writer);
httpResponse->content.resize(sb.GetSize(), '\0'); httpResponse->data.content.resize(sb.GetSize(), '\0');
memcpy(httpResponse->content.data(), sb.GetString(), sb.GetSize()); memcpy(httpResponse->data.content.data(), sb.GetString(), sb.GetSize());
} }
void validateKmsUrls(Reference<RESTKmsConnectorCtx> ctx) { void validateKmsUrls(Reference<RESTKmsConnectorCtx> ctx) {
@ -1526,10 +1527,10 @@ void testGetEncryptKeysByKeyIdsRequestBody(Reference<RESTKmsConnectorCtx> ctx, A
StringRef requestBodyRef = getEncryptKeysByKeyIdsRequestBody(ctx, req, refreshKmsUrls, arena); StringRef requestBodyRef = getEncryptKeysByKeyIdsRequestBody(ctx, req, refreshKmsUrls, arena);
TraceEvent("FetchKeysByKeyIds", ctx->uid).setMaxFieldLength(100000).detail("JsonReqStr", requestBodyRef.toString()); TraceEvent("FetchKeysByKeyIds", ctx->uid).setMaxFieldLength(100000).detail("JsonReqStr", requestBodyRef.toString());
Reference<HTTP::Response> httpResp = makeReference<HTTP::Response>(); Reference<HTTP::IncomingResponse> httpResp = makeReference<HTTP::IncomingResponse>();
httpResp->code = HTTP::HTTP_STATUS_CODE_OK; httpResp->code = HTTP::HTTP_STATUS_CODE_OK;
getFakeEncryptCipherResponse(requestBodyRef, true, httpResp); getFakeEncryptCipherResponse(requestBodyRef, true, httpResp);
TraceEvent("FetchKeysByKeyIds", ctx->uid).setMaxFieldLength(100000).detail("HttpRespStr", httpResp->content); TraceEvent("FetchKeysByKeyIds", ctx->uid).setMaxFieldLength(100000).detail("HttpRespStr", httpResp->data.content);
Standalone<VectorRef<EncryptCipherKeyDetailsRef>> cipherDetails = parseEncryptCipherResponse(ctx, httpResp); Standalone<VectorRef<EncryptCipherKeyDetailsRef>> cipherDetails = parseEncryptCipherResponse(ctx, httpResp);
ASSERT_EQ(cipherDetails.size(), keyMap.size()); ASSERT_EQ(cipherDetails.size(), keyMap.size());
@ -1559,10 +1560,10 @@ void testGetEncryptKeysByDomainIdsRequestBody(Reference<RESTKmsConnectorCtx> ctx
StringRef jsonReqRef = getEncryptKeysByDomainIdsRequestBody(ctx, req, refreshKmsUrls, arena); StringRef jsonReqRef = getEncryptKeysByDomainIdsRequestBody(ctx, req, refreshKmsUrls, arena);
TraceEvent("FetchKeysByDomainIds", ctx->uid).detail("JsonReqStr", jsonReqRef.toString()); TraceEvent("FetchKeysByDomainIds", ctx->uid).detail("JsonReqStr", jsonReqRef.toString());
Reference<HTTP::Response> httpResp = makeReference<HTTP::Response>(); Reference<HTTP::IncomingResponse> httpResp = makeReference<HTTP::IncomingResponse>();
httpResp->code = HTTP::HTTP_STATUS_CODE_OK; httpResp->code = HTTP::HTTP_STATUS_CODE_OK;
getFakeEncryptCipherResponse(jsonReqRef, false, httpResp); getFakeEncryptCipherResponse(jsonReqRef, false, httpResp);
TraceEvent("FetchKeysByDomainIds", ctx->uid).detail("HttpRespStr", httpResp->content); TraceEvent("FetchKeysByDomainIds", ctx->uid).detail("HttpRespStr", httpResp->data.content);
Standalone<VectorRef<EncryptCipherKeyDetailsRef>> cipherDetails = parseEncryptCipherResponse(ctx, httpResp); Standalone<VectorRef<EncryptCipherKeyDetailsRef>> cipherDetails = parseEncryptCipherResponse(ctx, httpResp);
ASSERT_EQ(domainIds.size(), cipherDetails.size()); ASSERT_EQ(domainIds.size(), cipherDetails.size());
@ -1592,10 +1593,10 @@ void testGetBlobMetadataRequestBody(Reference<RESTKmsConnectorCtx> ctx) {
TraceEvent("FetchBlobMetadataStart", ctx->uid); TraceEvent("FetchBlobMetadataStart", ctx->uid);
StringRef jsonReqRef = getBlobMetadataRequestBody(ctx, req, refreshKmsUrls); StringRef jsonReqRef = getBlobMetadataRequestBody(ctx, req, refreshKmsUrls);
TraceEvent("FetchBlobMetadataReq", ctx->uid).detail("JsonReqStr", jsonReqRef.toString()); TraceEvent("FetchBlobMetadataReq", ctx->uid).detail("JsonReqStr", jsonReqRef.toString());
Reference<HTTP::Response> httpResp = makeReference<HTTP::Response>(); Reference<HTTP::IncomingResponse> httpResp = makeReference<HTTP::IncomingResponse>();
httpResp->code = HTTP::HTTP_STATUS_CODE_OK; httpResp->code = HTTP::HTTP_STATUS_CODE_OK;
getFakeBlobMetadataResponse(jsonReqRef, false, httpResp); getFakeBlobMetadataResponse(jsonReqRef, false, httpResp);
TraceEvent("FetchBlobMetadataResp", ctx->uid).detail("HttpRespStr", httpResp->content); TraceEvent("FetchBlobMetadataResp", ctx->uid).detail("HttpRespStr", httpResp->data.content);
Standalone<VectorRef<BlobMetadataDetailsRef>> details = parseBlobMetadataResponse(ctx, httpResp); Standalone<VectorRef<BlobMetadataDetailsRef>> details = parseBlobMetadataResponse(ctx, httpResp);
@ -1641,15 +1642,15 @@ void testMissingOrInvalidVersion(Reference<RESTKmsConnectorCtx> ctx, bool isCiph
versionValue.SetInt(version); versionValue.SetInt(version);
doc.AddMember(versionKey, versionValue, doc.GetAllocator()); doc.AddMember(versionKey, versionValue, doc.GetAllocator());
Reference<HTTP::Response> httpResp = makeReference<HTTP::Response>(); Reference<HTTP::IncomingResponse> httpResp = makeReference<HTTP::IncomingResponse>();
httpResp->code = HTTP::HTTP_STATUS_CODE_OK; httpResp->code = HTTP::HTTP_STATUS_CODE_OK;
httpResp->contentLen = 0; httpResp->data.contentLen = 0;
httpResp->content = ""; httpResp->data.content = "";
rapidjson::StringBuffer sb; rapidjson::StringBuffer sb;
rapidjson::Writer<rapidjson::StringBuffer> writer(sb); rapidjson::Writer<rapidjson::StringBuffer> writer(sb);
doc.Accept(writer); doc.Accept(writer);
httpResp->content.resize(sb.GetSize(), '\0'); httpResp->data.content.resize(sb.GetSize(), '\0');
memcpy(httpResp->content.data(), sb.GetString(), sb.GetSize()); memcpy(httpResp->data.content.data(), sb.GetString(), sb.GetSize());
try { try {
if (isCipher) { if (isCipher) {
@ -1671,14 +1672,14 @@ void testMissingDetailsTag(Reference<RESTKmsConnectorCtx> ctx, bool isCipher) {
refreshUrl.SetBool(true); refreshUrl.SetBool(true);
doc.AddMember(key, refreshUrl, doc.GetAllocator()); doc.AddMember(key, refreshUrl, doc.GetAllocator());
Reference<HTTP::Response> httpResp = makeReference<HTTP::Response>(); Reference<HTTP::IncomingResponse> httpResp = makeReference<HTTP::IncomingResponse>();
httpResp->code = HTTP::HTTP_STATUS_CODE_OK; httpResp->code = HTTP::HTTP_STATUS_CODE_OK;
rapidjson::StringBuffer sb; rapidjson::StringBuffer sb;
rapidjson::Writer<rapidjson::StringBuffer> writer(sb); rapidjson::Writer<rapidjson::StringBuffer> writer(sb);
doc.Accept(writer); doc.Accept(writer);
httpResp->content.resize(sb.GetSize(), '\0'); httpResp->data.content.resize(sb.GetSize(), '\0');
memcpy(httpResp->content.data(), sb.GetString(), sb.GetSize()); memcpy(httpResp->data.content.data(), sb.GetString(), sb.GetSize());
httpResp->contentLen = sb.GetSize(); httpResp->data.contentLen = sb.GetSize();
try { try {
if (isCipher) { if (isCipher) {
@ -1704,14 +1705,14 @@ void testMalformedDetails(Reference<RESTKmsConnectorCtx> ctx, bool isCipher) {
addVersionToDoc(doc, 1); addVersionToDoc(doc, 1);
Reference<HTTP::Response> httpResp = makeReference<HTTP::Response>(); Reference<HTTP::IncomingResponse> httpResp = makeReference<HTTP::IncomingResponse>();
httpResp->code = HTTP::HTTP_STATUS_CODE_OK; httpResp->code = HTTP::HTTP_STATUS_CODE_OK;
rapidjson::StringBuffer sb; rapidjson::StringBuffer sb;
rapidjson::Writer<rapidjson::StringBuffer> writer(sb); rapidjson::Writer<rapidjson::StringBuffer> writer(sb);
doc.Accept(writer); doc.Accept(writer);
httpResp->content.resize(sb.GetSize(), '\0'); httpResp->data.content.resize(sb.GetSize(), '\0');
memcpy(httpResp->content.data(), sb.GetString(), sb.GetSize()); memcpy(httpResp->data.content.data(), sb.GetString(), sb.GetSize());
httpResp->contentLen = sb.GetSize(); httpResp->data.contentLen = sb.GetSize();
try { try {
if (isCipher) { if (isCipher) {
@ -1743,14 +1744,14 @@ void testMalformedDetailNotObj(Reference<RESTKmsConnectorCtx> ctx, bool isCipher
addVersionToDoc(doc, 1); addVersionToDoc(doc, 1);
Reference<HTTP::Response> httpResp = makeReference<HTTP::Response>(); Reference<HTTP::IncomingResponse> httpResp = makeReference<HTTP::IncomingResponse>();
httpResp->code = HTTP::HTTP_STATUS_CODE_OK; httpResp->code = HTTP::HTTP_STATUS_CODE_OK;
rapidjson::StringBuffer sb; rapidjson::StringBuffer sb;
rapidjson::Writer<rapidjson::StringBuffer> writer(sb); rapidjson::Writer<rapidjson::StringBuffer> writer(sb);
doc.Accept(writer); doc.Accept(writer);
httpResp->content.resize(sb.GetSize(), '\0'); httpResp->data.content.resize(sb.GetSize(), '\0');
memcpy(httpResp->content.data(), sb.GetString(), sb.GetSize()); memcpy(httpResp->data.content.data(), sb.GetString(), sb.GetSize());
httpResp->contentLen = sb.GetSize(); httpResp->data.contentLen = sb.GetSize();
try { try {
if (isCipher) { if (isCipher) {
@ -1782,14 +1783,14 @@ void testMalformedDetailObj(Reference<RESTKmsConnectorCtx> ctx, bool isCipher) {
addVersionToDoc(doc, 1); addVersionToDoc(doc, 1);
Reference<HTTP::Response> httpResp = makeReference<HTTP::Response>(); Reference<HTTP::IncomingResponse> httpResp = makeReference<HTTP::IncomingResponse>();
httpResp->code = HTTP::HTTP_STATUS_CODE_OK; httpResp->code = HTTP::HTTP_STATUS_CODE_OK;
rapidjson::StringBuffer sb; rapidjson::StringBuffer sb;
rapidjson::Writer<rapidjson::StringBuffer> writer(sb); rapidjson::Writer<rapidjson::StringBuffer> writer(sb);
doc.Accept(writer); doc.Accept(writer);
httpResp->content.resize(sb.GetSize(), '\0'); httpResp->data.content.resize(sb.GetSize(), '\0');
memcpy(httpResp->content.data(), sb.GetString(), sb.GetSize()); memcpy(httpResp->data.content.data(), sb.GetString(), sb.GetSize());
httpResp->contentLen = sb.GetSize(); httpResp->data.contentLen = sb.GetSize();
try { try {
if (isCipher) { if (isCipher) {
@ -1833,14 +1834,14 @@ void testKMSErrorResponse(Reference<RESTKmsConnectorCtx> ctx, bool isCipher) {
key.SetString(ERROR_TAG, doc.GetAllocator()); key.SetString(ERROR_TAG, doc.GetAllocator());
doc.AddMember(key, errorTag, doc.GetAllocator()); doc.AddMember(key, errorTag, doc.GetAllocator());
Reference<HTTP::Response> httpResp = makeReference<HTTP::Response>(); Reference<HTTP::IncomingResponse> httpResp = makeReference<HTTP::IncomingResponse>();
httpResp->code = HTTP::HTTP_STATUS_CODE_OK; httpResp->code = HTTP::HTTP_STATUS_CODE_OK;
rapidjson::StringBuffer sb; rapidjson::StringBuffer sb;
rapidjson::Writer<rapidjson::StringBuffer> writer(sb); rapidjson::Writer<rapidjson::StringBuffer> writer(sb);
doc.Accept(writer); doc.Accept(writer);
httpResp->content.resize(sb.GetSize(), '\0'); httpResp->data.content.resize(sb.GetSize(), '\0');
memcpy(httpResp->content.data(), sb.GetString(), sb.GetSize()); memcpy(httpResp->data.content.data(), sb.GetString(), sb.GetSize());
httpResp->contentLen = sb.GetSize(); httpResp->data.contentLen = sb.GetSize();
try { try {
if (isCipher) { if (isCipher) {

127
fdbserver/SimulatedCluster.actor.cpp
View File

@ -358,6 +358,9 @@ class TestConfig : public BasicTestConfig {
if (attrib == "blobGranulesEnabled") { if (attrib == "blobGranulesEnabled") {
blobGranulesEnabled = strcmp(value.c_str(), "true") == 0; blobGranulesEnabled = strcmp(value.c_str(), "true") == 0;
} }
if (attrib == "simHTTPServerEnabled") {
simHTTPServerEnabled = strcmp(value.c_str(), "true") == 0;
}
if (attrib == "allowDefaultTenant") { if (attrib == "allowDefaultTenant") {
allowDefaultTenant = strcmp(value.c_str(), "true") == 0; allowDefaultTenant = strcmp(value.c_str(), "true") == 0;
} }
@ -433,6 +436,7 @@ public:
Optional<std::string> remoteConfig; Optional<std::string> remoteConfig;
bool blobGranulesEnabled = false; bool blobGranulesEnabled = false;
bool randomlyRenameZoneId = false; bool randomlyRenameZoneId = false;
bool simHTTPServerEnabled = false; // TODO default to true
bool allowDefaultTenant = true; bool allowDefaultTenant = true;
bool allowCreatingTenants = true; bool allowCreatingTenants = true;
@ -511,6 +515,7 @@ public:
.add("configDB", &configDBType) .add("configDB", &configDBType)
.add("extraMachineCountDC", &extraMachineCountDC) .add("extraMachineCountDC", &extraMachineCountDC)
.add("blobGranulesEnabled", &blobGranulesEnabled) .add("blobGranulesEnabled", &blobGranulesEnabled)
.add("simHTTPServerEnabled", &simHTTPServerEnabled)
.add("allowDefaultTenant", &allowDefaultTenant) .add("allowDefaultTenant", &allowDefaultTenant)
.add("allowCreatingTenants", &allowCreatingTenants) .add("allowCreatingTenants", &allowCreatingTenants)
.add("randomlyRenameZoneId", &randomlyRenameZoneId) .add("randomlyRenameZoneId", &randomlyRenameZoneId)
@ -645,7 +650,35 @@ ACTOR Future<Void> runDr(Reference<IClusterConnectionRecord> connRecord) {
throw internal_error(); throw internal_error();
} }
enum AgentMode { AgentNone = 0, AgentOnly = 1, AgentAddition = 2 }; ACTOR Future<Void> runSimHTTPServer() {
state Reference<HTTP::SimServerContext> context = makeReference<HTTP::SimServerContext>();
g_simulator->addSimHTTPProcess(context);
try {
wait(context->actors.getResult());
} catch (Error& e) {
TraceEvent(SevError, "SimHTTPServerDied").error(e);
context->stop();
throw e;
}
throw internal_error();
}
// enum AgentMode { AgentNone = 0, AgentOnly = 1, AgentAddition = 2 };
// FIXME: could do this as bit flags of (fdbd) (backup agent) (http) etc... if the space gets more complicated
enum ProcessMode { FDBDOnly = 0, BackupAgentOnly = 1, FDBDAndBackupAgent = 2, SimHTTPServer = 3 };
bool processRunBackupAgent(ProcessMode mode) {
return mode == BackupAgentOnly || mode == FDBDAndBackupAgent;
}
bool processRunFDBD(ProcessMode mode) {
return mode == FDBDOnly || mode == FDBDAndBackupAgent;
}
bool processRunHTTPServer(ProcessMode mode) {
return mode == SimHTTPServer;
}
// SOMEDAY: when a process can be rebooted in isolation from the other on that machine, // SOMEDAY: when a process can be rebooted in isolation from the other on that machine,
// a loop{} will be needed around the waiting on simulatedFDBD(). For now this simply // a loop{} will be needed around the waiting on simulatedFDBD(). For now this simply
@ -663,7 +696,7 @@ ACTOR Future<ISimulator::KillType> simulatedFDBDRebooter(Reference<IClusterConne
ClusterConnectionString connStr, ClusterConnectionString connStr,
ClusterConnectionString otherConnStr, ClusterConnectionString otherConnStr,
bool useSeedFile, bool useSeedFile,
AgentMode runBackupAgents, ProcessMode processMode,
std::string whitelistBinPaths, std::string whitelistBinPaths,
ProtocolVersion protocolVersion, ProtocolVersion protocolVersion,
ConfigDBType configDBType, ConfigDBType configDBType,
@ -716,7 +749,8 @@ ACTOR Future<ISimulator::KillType> simulatedFDBDRebooter(Reference<IClusterConne
.detail("DataHall", localities.dataHallId()) .detail("DataHall", localities.dataHallId())
.detail("Address", process->address.toString()) .detail("Address", process->address.toString())
.detail("Excluded", process->excluded) .detail("Excluded", process->excluded)
.detail("UsingSSL", sslEnabled); .detail("UsingSSL", sslEnabled)
.detail("ProcessMode", processMode);
TraceEvent("ProgramStart") TraceEvent("ProgramStart")
.detail("Cycles", cycles) .detail("Cycles", cycles)
.detail("RandomId", randomId) .detail("RandomId", randomId)
@ -734,10 +768,9 @@ ACTOR Future<ISimulator::KillType> simulatedFDBDRebooter(Reference<IClusterConne
try { try {
// SOMEDAY: test lower memory limits, without making them too small and causing the database to stop // SOMEDAY: test lower memory limits, without making them too small and causing the database to stop
// making progress // making progress
FlowTransport::createInstance(processClass == ProcessClass::TesterClass || runBackupAgents == AgentOnly, bool client = processClass == ProcessClass::TesterClass || processMode == BackupAgentOnly ||
1, processMode == SimHTTPServer;
WLTOKEN_RESERVED_COUNT, FlowTransport::createInstance(client, 1, WLTOKEN_RESERVED_COUNT, &allowList);
&allowList);
for (const auto& p : g_simulator->authKeys) { for (const auto& p : g_simulator->authKeys) {
FlowTransport::transport().addPublicKey(p.first, p.second.toPublic()); FlowTransport::transport().addPublicKey(p.first, p.second.toPublic());
} }
@ -748,7 +781,7 @@ ACTOR Future<ISimulator::KillType> simulatedFDBDRebooter(Reference<IClusterConne
NetworkAddress n(ip, listenPort, true, sslEnabled && listenPort == port); NetworkAddress n(ip, listenPort, true, sslEnabled && listenPort == port);
futures.push_back(FlowTransport::transport().bind(n, n)); futures.push_back(FlowTransport::transport().bind(n, n));
} }
if (runBackupAgents != AgentOnly) { if (processRunFDBD(processMode)) {
futures.push_back(fdbd(connRecord, futures.push_back(fdbd(connRecord,
localities, localities,
processClass, processClass,
@ -763,10 +796,14 @@ ACTOR Future<ISimulator::KillType> simulatedFDBDRebooter(Reference<IClusterConne
{}, {},
configDBType)); configDBType));
} }
if (runBackupAgents != AgentNone) { if (processRunBackupAgent(processMode)) {
futures.push_back(runBackup(connRecord)); futures.push_back(runBackup(connRecord));
futures.push_back(runDr(connRecord)); futures.push_back(runDr(connRecord));
} }
if (processRunHTTPServer(processMode)) {
fmt::print("Process run http server\n");
futures.push_back(runSimHTTPServer());
}
futures.push_back(success(onShutdown)); futures.push_back(success(onShutdown));
if (!g_simulator->globalHasSwitchedCluster() && g_simulator->hasSwitchedCluster(process->address)) { if (!g_simulator->globalHasSwitchedCluster() && g_simulator->hasSwitchedCluster(process->address)) {
@ -927,7 +964,7 @@ ACTOR Future<Void> simulatedMachine(ClusterConnectionString connStr,
std::string baseFolder, std::string baseFolder,
bool restarting, bool restarting,
bool useSeedFile, bool useSeedFile,
AgentMode runBackupAgents, ProcessMode processMode,
bool sslOnly, bool sslOnly,
std::string whitelistBinPaths, std::string whitelistBinPaths,
ProtocolVersion protocolVersion, ProtocolVersion protocolVersion,
@ -985,9 +1022,9 @@ ACTOR Future<Void> simulatedMachine(ClusterConnectionString connStr,
useSeedFile ? new ClusterConnectionFile(path, connStr.toString()) useSeedFile ? new ClusterConnectionFile(path, connStr.toString())
: new ClusterConnectionFile(path)); : new ClusterConnectionFile(path));
const int listenPort = i * listenPerProcess + 1; const int listenPort = i * listenPerProcess + 1;
AgentMode agentMode = ProcessMode ipProcessMode =
runBackupAgents == AgentOnly ? (i == ips.size() - 1 ? AgentOnly : AgentNone) : runBackupAgents; processMode == BackupAgentOnly ? (i == ips.size() - 1 ? BackupAgentOnly : FDBDOnly) : processMode;
if (g_simulator->hasDiffProtocolProcess && !g_simulator->setDiffProtocol && agentMode == AgentNone) { if (g_simulator->hasDiffProtocolProcess && !g_simulator->setDiffProtocol && ipProcessMode == FDBDOnly) {
processes.push_back(simulatedFDBDRebooter(clusterFile, processes.push_back(simulatedFDBDRebooter(clusterFile,
ips[i], ips[i],
sslEnabled, sslEnabled,
@ -1001,7 +1038,7 @@ ACTOR Future<Void> simulatedMachine(ClusterConnectionString connStr,
connStr, connStr,
otherConnStr, otherConnStr,
useSeedFile, useSeedFile,
agentMode, ipProcessMode,
whitelistBinPaths, whitelistBinPaths,
protocolVersion, protocolVersion,
configDBType, configDBType,
@ -1021,7 +1058,7 @@ ACTOR Future<Void> simulatedMachine(ClusterConnectionString connStr,
connStr, connStr,
otherConnStr, otherConnStr,
useSeedFile, useSeedFile,
agentMode, ipProcessMode,
whitelistBinPaths, whitelistBinPaths,
g_network->protocolVersion(), g_network->protocolVersion(),
configDBType, configDBType,
@ -1391,7 +1428,7 @@ ACTOR Future<Void> restartSimulatedSystem(std::vector<Future<Void>>* systemActor
baseFolder, baseFolder,
true, true,
i == useSeedForMachine, i == useSeedForMachine,
AgentAddition, FDBDAndBackupAgent,
usingSSL && (listenersPerProcess == 1 || processClass == ProcessClass::TesterClass), usingSSL && (listenersPerProcess == 1 || processClass == ProcessClass::TesterClass),
whitelistBinPaths, whitelistBinPaths,
protocolVersion, protocolVersion,
@ -2412,12 +2449,18 @@ void setupSimulatedSystem(std::vector<Future<Void>>* systemActors,
// TODO: caching disabled for this merge // TODO: caching disabled for this merge
int storageCacheMachines = dc == 0 ? 1 : 0; int storageCacheMachines = dc == 0 ? 1 : 0;
int blobWorkerMachines = 0; int blobWorkerMachines = 0;
int simHTTPMachines = 0;
if (testConfig.blobGranulesEnabled) { if (testConfig.blobGranulesEnabled) {
int blobWorkerProcesses = 1 + deterministicRandom()->randomInt(0, NUM_EXTRA_BW_MACHINES + 1); int blobWorkerProcesses = 1 + deterministicRandom()->randomInt(0, NUM_EXTRA_BW_MACHINES + 1);
blobWorkerMachines = std::max(1, blobWorkerProcesses / processesPerMachine); blobWorkerMachines = std::max(1, blobWorkerProcesses / processesPerMachine);
} }
if (testConfig.simHTTPServerEnabled) {
// FIXME: more eventually?
fmt::print("sim http machines = 1\n");
simHTTPMachines = 1;
}
int totalMachines = machines + storageCacheMachines + blobWorkerMachines; int totalMachines = machines + storageCacheMachines + blobWorkerMachines + simHTTPMachines;
int useSeedForMachine = deterministicRandom()->randomInt(0, totalMachines); int useSeedForMachine = deterministicRandom()->randomInt(0, totalMachines);
Standalone<StringRef> zoneId; Standalone<StringRef> zoneId;
Standalone<StringRef> newZoneId; Standalone<StringRef> newZoneId;
@ -2448,9 +2491,10 @@ void setupSimulatedSystem(std::vector<Future<Void>>* systemActors,
} }
} }
// FIXME: hack to add machines specifically to test storage cache and blob workers // FIXME: hack to add machines specifically to test storage cache and blob workers and http server
// TODO: caching disabled for this merge
// `machines` here is the normal (non-temporary) machines that totalMachines comprises of // `machines` here is the normal (non-temporary) machines that totalMachines comprises of
int processCount = processesPerMachine;
ProcessMode processMode = requiresExtraDBMachines ? BackupAgentOnly : FDBDAndBackupAgent;
if (machine >= machines) { if (machine >= machines) {
if (storageCacheMachines > 0 && dc == 0) { if (storageCacheMachines > 0 && dc == 0) {
processClass = ProcessClass(ProcessClass::StorageCacheClass, ProcessClass::CommandLineSource); processClass = ProcessClass(ProcessClass::StorageCacheClass, ProcessClass::CommandLineSource);
@ -2458,12 +2502,17 @@ void setupSimulatedSystem(std::vector<Future<Void>>* systemActors,
} else if (blobWorkerMachines > 0) { // add blob workers to every DC } else if (blobWorkerMachines > 0) { // add blob workers to every DC
processClass = ProcessClass(ProcessClass::BlobWorkerClass, ProcessClass::CommandLineSource); processClass = ProcessClass(ProcessClass::BlobWorkerClass, ProcessClass::CommandLineSource);
blobWorkerMachines--; blobWorkerMachines--;
} else if (simHTTPMachines > 0) {
processClass = ProcessClass(ProcessClass::SimHTTPServerClass, ProcessClass::CommandLineSource);
processCount = 1;
processMode = SimHTTPServer;
simHTTPMachines--;
} }
} }
std::vector<IPAddress> ips; std::vector<IPAddress> ips;
ips.reserve(processesPerMachine); ips.reserve(processesPerMachine);
for (int i = 0; i < processesPerMachine; i++) { for (int i = 0; i < processCount; i++) {
ips.push_back( ips.push_back(
makeIPAddressForSim(useIPv6, { 2, dc, deterministicRandom()->randomInt(1, i + 2), machine })); makeIPAddressForSim(useIPv6, { 2, dc, deterministicRandom()->randomInt(1, i + 2), machine }));
} }
@ -2485,7 +2534,7 @@ void setupSimulatedSystem(std::vector<Future<Void>>* systemActors,
baseFolder, baseFolder,
false, false,
machine == useSeedForMachine, machine == useSeedForMachine,
requiresExtraDBMachines ? AgentOnly : AgentAddition, processMode,
sslOnly, sslOnly,
whitelistBinPaths, whitelistBinPaths,
protocolVersion, protocolVersion,
@ -2507,23 +2556,23 @@ void setupSimulatedSystem(std::vector<Future<Void>>* systemActors,
LocalityData localities(Optional<Standalone<StringRef>>(), newZoneId, newMachineId, dcUID); LocalityData localities(Optional<Standalone<StringRef>>(), newZoneId, newMachineId, dcUID);
localities.set("data_hall"_sr, dcUID); localities.set("data_hall"_sr, dcUID);
systemActors->push_back( systemActors->push_back(reportErrors(
reportErrors(simulatedMachine(ClusterConnectionString(extraDatabase), simulatedMachine(ClusterConnectionString(extraDatabase),
conn, conn,
extraIps, extraIps,
sslEnabled, sslEnabled,
localities, localities,
processClass, processClass,
baseFolder, baseFolder,
false, false,
machine == useSeedForMachine, machine == useSeedForMachine,
testConfig.extraDatabaseBackupAgents ? AgentAddition : AgentNone, testConfig.extraDatabaseBackupAgents ? FDBDAndBackupAgent : FDBDOnly,
sslOnly, sslOnly,
whitelistBinPaths, whitelistBinPaths,
protocolVersion, protocolVersion,
configDBType, configDBType,
true), true),
"SimulatedMachine")); "SimulatedMachine"));
++cluster; ++cluster;
} }
} }
@ -2565,7 +2614,7 @@ void setupSimulatedSystem(std::vector<Future<Void>>* systemActors,
baseFolder, baseFolder,
false, false,
i == useSeedForMachine, i == useSeedForMachine,
AgentNone, FDBDOnly,
sslOnly, sslOnly,
whitelistBinPaths, whitelistBinPaths,
protocolVersion, protocolVersion,

4
flow/Knobs.cpp
View File

@ -191,8 +191,8 @@ void FlowKnobs::initialize(Randomize randomize, IsSimulated isSimulated) {
init( LOW_PRIORITY_MAX_DELAY, 5.0 ); init( LOW_PRIORITY_MAX_DELAY, 5.0 );
// HTTP // HTTP
init( HTTP_READ_SIZE, 128*1024 ); init( HTTP_READ_SIZE, 128*1024 ); if (randomize && BUGGIFY) HTTP_READ_SIZE = deterministicRandom()->randomSkewedUInt32(1024, 2 * HTTP_READ_SIZE);
init( HTTP_SEND_SIZE, 32*1024 ); init( HTTP_SEND_SIZE, 32*1024 ); if (randomize && BUGGIFY) HTTP_SEND_SIZE = deterministicRandom()->randomSkewedUInt32(1024, 2 * HTTP_SEND_SIZE);
init( HTTP_VERBOSE_LEVEL, 0 ); init( HTTP_VERBOSE_LEVEL, 0 );
init( HTTP_REQUEST_ID_HEADER, "" ); init( HTTP_REQUEST_ID_HEADER, "" );
init( HTTP_RESPONSE_SKIP_VERIFY_CHECKSUM_FOR_PARTIAL_CONTENT, false ); init( HTTP_RESPONSE_SKIP_VERIFY_CHECKSUM_FOR_PARTIAL_CONTENT, false );

1
tests/CMakeLists.txt
View File

@ -170,6 +170,7 @@ if(WITH_PYTHON)
add_fdb_test(TEST_FILES fast/EncryptionOps.toml) add_fdb_test(TEST_FILES fast/EncryptionOps.toml)
add_fdb_test(TEST_FILES fast/EncryptionUnitTests.toml) add_fdb_test(TEST_FILES fast/EncryptionUnitTests.toml)
add_fdb_test(TEST_FILES fast/EncryptKeyProxyTest.toml) add_fdb_test(TEST_FILES fast/EncryptKeyProxyTest.toml)
add_fdb_test(TEST_FILES fast/HTTPServerUnit.toml)
add_fdb_test(TEST_FILES fast/FuzzApiCorrectness.toml) add_fdb_test(TEST_FILES fast/FuzzApiCorrectness.toml)
add_fdb_test(TEST_FILES fast/FuzzApiCorrectnessClean.toml) add_fdb_test(TEST_FILES fast/FuzzApiCorrectnessClean.toml)
add_fdb_test(TEST_FILES fast/IncrementalBackup.toml) add_fdb_test(TEST_FILES fast/IncrementalBackup.toml)

16
tests/fast/HTTPServerUnit.toml Normal file
View File

@ -0,0 +1,16 @@
# This test can't run as part of RandomUnitTests.toml yet, so run it less frequently
testPriority = '100'
[configuration]
simHTTPServerEnabled = true
[[test]]
testTitle = 'HTTPServerUnit'
startDelay = 0
useDB = false
timeout = 1000
[[test.workload]]
testName = 'UnitTests'
maxTestCases = 1
testsMatching = '!/HTTP/Server'