db/pgbackrest
1
0
Fork 0
mirror of https://github.com/pgbackrest/pgbackrest.git synced 2026-05-22 10:15:16 +02:00
Code Issues Releases Activity

Add GCS batch delete support.

Browse Source 
The GCS driver sent a single file delete request for each file while deleting a path. Depending on latency this could lead to rather long delete times, especially noticeable during expiration.

Improve GCS delete to use batches, which require multipart HTTP, so also add multipart HTTP infrastructure.
This commit is contained in:
David Steele
2024-04-27 15:42:10 +10:00
committed by GitHub
parent e00e33b528
commit 76bcb740b6
13 changed files with 1159 additions and 220 deletions
Download Patch File Download Diff File Expand all files Collapse all files
test/define.yaml
+1 -1
View File
@@ -393,7 +393,7 @@ unit:
# ----------------------------------------------------------------------------------------------------------------------------
- name: io-http
total: 6
total: 7
coverage:
- common/io/http/client
test/src/module/common/ioHttpTest.c
+123 -1
View File
@@ -309,6 +309,19 @@ testRun(void)
TEST_RESULT_VOID(httpUrlFree(url), "free");
}
// *****************************************************************************************************************************
if (testBegin("HttpResponseMulti"))
{
// -------------------------------------------------------------------------------------------------------------------------
TEST_TITLE("errors");
TEST_ERROR(httpResponseMultiNew(BUFSTRDEF(""), NULL), FormatError, "expected multipart content type");
TEST_ERROR(httpResponseMultiNew(BUFSTRDEF(""), STRDEF("bogus")), FormatError, "expected multipart content type");
TEST_ERROR(
strNewBuf(httpResponseMultiNew(BUFSTRDEF("--YYY"), STRDEF("multipart/mixed; boundary=\"XXX\""))->boundary),
FormatError, "multipart boundary 'XXX' not found");
}
// *****************************************************************************************************************************
if (testBegin("HttpClient"))
{
@@ -842,9 +855,118 @@ testRun(void)
TEST_RESULT_STR_Z(strNewBuf(buffer), "", "check response");
// -----------------------------------------------------------------------------------------------------------------
TEST_TITLE("close connection and end server process");
TEST_TITLE("request with multipart request and response");
ioBufferSizeSet(512);
hrnServerScriptAccept(http);
hrnServerScriptExpectZ(
http,
"GET / HTTP/1.1\r\n" TEST_USER_AGENT
"content-type:multipart/mixed; boundary=QKX4EYg4LARJ\r\n"
"hdr1:1\r\n\r\n"
"\r\n--QKX4EYg4LARJ\r\n"
"content-type:application/http\r\n"
"content-transfer-encoding:binary\r\n"
"content-id:0\r\n\r\n"
"GET / HTTP/1.1\r\n\r\n"
"\r\n--QKX4EYg4LARJ\r\n"
"content-type:application/http\r\n"
"content-transfer-encoding:binary\r\n"
"content-id:1\r\n\r\n"
"POST /ack HTTP/1.1\r\n"
"content-length:3\r\n\r\n"
HTTP_MULTIPART_BOUNDARY_INIT
"\r\n--QKX4EYg4LARJ--\r\n");
hrnServerScriptReplyZ(
http,
"HTTP/1.1 200 OK\r\nConnection:ClosE\r\ncontent-type:multipart/mixed; boundary=XXX\r\n\r\n"
"PREAMBLE JUNK\r\n"
"--XXX\r\n"
"content-type:application/http\r\n"
"content-id:0\r\n\r\n"
"HTTP/1.1 200 OK\r\n\r\n"
"\r\n--XXX\r\n"
"content-type:application/http\r\n"
"content-id:1\r\n"
"\r\n"
"HTTP/1.1 200 OK\r\n"
"content-length:3\r\n"
"\r\n"
"123"
"\r\n--XXX\n"
"EPILOGUE JUNK");
hrnServerScriptClose(http);
HttpRequestMulti *requestMulti = httpRequestMultiNew();
httpRequestMultiAddP(requestMulti, STRDEF("0"), HTTP_VERB_GET_STR, STRDEF("/"), .header = httpHeaderNew(NULL));
httpRequestMultiAddP(
requestMulti, STRDEF("1"), HTTP_VERB_POST_STR, STRDEF("/ack"),
.header = httpHeaderAdd(httpHeaderNew(NULL), HTTP_HEADER_CONTENT_LENGTH_STR, STRDEF("3")),
.content = BUFSTRDEF(HTTP_MULTIPART_BOUNDARY_INIT));
HttpHeader *headerRequest = httpHeaderNew(NULL);
httpHeaderAdd(headerRequest, STRDEF("hdr1"), STRDEF("1"));
httpRequestMultiHeaderAdd(requestMulti, headerRequest);
TEST_ASSIGN(
response,
httpRequestResponse(
httpRequestNewP(
client, STRDEF("GET"), STRDEF("/"), .header = headerRequest,
.content = httpRequestMultiContent(requestMulti)),
false),
"request");
TEST_RESULT_VOID(
FUNCTION_LOG_OBJECT_FORMAT(httpResponseHeader(response), httpHeaderToLog, logBuf, sizeof(logBuf)),
"httpHeaderToLog");
TEST_RESULT_Z(
logBuf, "{connection: 'close', content-type: 'multipart/mixed; boundary=XXX'}", "check response headers");
HttpResponseMulti *responseMulti = NULL;
TEST_ASSIGN(
responseMulti,
httpResponseMultiNew(
httpResponseContent(response), httpHeaderGet(httpResponseHeader(response), HTTP_HEADER_CONTENT_TYPE_STR)),
"response multi");
HttpResponse *responsePart = NULL;
TEST_ASSIGN(responsePart, httpResponseMultiNext(responseMulti), "response part");
TEST_RESULT_UINT(httpResponseCode(responsePart), 200, "response code");
TEST_RESULT_STR_Z(strNewBuf(httpResponseContent(responsePart)), "", "response content");
TEST_ASSIGN(responsePart, httpResponseMultiNext(responseMulti), "response part");
TEST_RESULT_STR_Z(strNewBuf(httpResponseContent(responsePart)), "123", "response content");
TEST_RESULT_PTR(httpResponseMultiNext(responseMulti), NULL, "no more responses");
// -----------------------------------------------------------------------------------------------------------------
TEST_TITLE("error on full boundary in header or content");
requestMulti = httpRequestMultiNew();
TEST_ERROR(
httpRequestMultiAddP(
requestMulti, STRDEF("0"), HTTP_VERB_GET_STR, STRDEF("/"),
.header = httpHeaderAdd(
httpHeaderNew(NULL), HTTP_HEADER_CONTENT_LENGTH_STR,
STRDEF(HTTP_MULTIPART_BOUNDARY_INIT HTTP_MULTIPART_BOUNDARY_EXTRA))),
AssertError, "unable to construct unique boundary");
requestMulti = httpRequestMultiNew();
TEST_ERROR(
httpRequestMultiAddP(
requestMulti, STRDEF("0"), HTTP_VERB_GET_STR, STRDEF("/"), .header = httpHeaderNew(NULL),
.content = BUFSTRDEF(HTTP_MULTIPART_BOUNDARY_INIT HTTP_MULTIPART_BOUNDARY_EXTRA)),
AssertError, "unable to construct unique boundary");
TEST_RESULT_BOOL(
bufEq(requestMulti->boundaryRaw, BUFSTRDEF(HTTP_MULTIPART_BOUNDARY_INIT HTTP_MULTIPART_BOUNDARY_EXTRA)),
true, "max boundary");
// -----------------------------------------------------------------------------------------------------------------
TEST_TITLE("end server process");
hrnServerScriptEnd(http);
}
HRN_FORK_PARENT_END();
test/src/module/common/typeBufferTest.c
+10 -1
View File
@@ -137,11 +137,20 @@ testRun(void)
}
// *****************************************************************************************************************************
if (testBegin("bufDup() and bufEq()"))
if (testBegin("bufDup(), bufEq(), and bufFind()"))
{
TEST_RESULT_BOOL(bufEq(BUFSTRDEF("123"), bufDup(BUFSTRDEF("1234"))), false, "buffer sizes not equal");
TEST_RESULT_BOOL(bufEq(BUFSTR(STRDEF("321")), BUFSTRDEF("123")), false, "buffer sizes equal");
TEST_RESULT_BOOL(bufEq(bufDup(BUFSTRZ("123")), BUF("123", 3)), true, "buffers equal");
const Buffer *haystack = BUFSTRDEF("findsomethinginhere");
TEST_RESULT_PTR(bufFindP(haystack, BUFSTRDEF("xxx")), NULL, "not found");
TEST_RESULT_PTR(bufFindP(haystack, BUFSTRDEF("find")), bufPtrConst(haystack), "found first");
TEST_RESULT_PTR(bufFindP(haystack, BUFSTRDEF("here")), bufPtrConst(haystack) + 15, "found last");
TEST_RESULT_PTR(bufFindP(haystack, BUFSTRDEF("thing")), bufPtrConst(haystack) + 8, "found middle");
TEST_RESULT_PTR(bufFindP(haystack, BUFSTRDEF("find"), .begin = bufPtrConst(haystack) + 1), NULL, "skipped not found");
TEST_RESULT_PTR(bufFindP(haystack, BUFSTRDEF("findsomethinginhere2")), NULL, "needle longer than haystack");
}
// *****************************************************************************************************************************
test/src/module/storage/gcsTest.c
+134 -9
View File
@@ -72,6 +72,8 @@ typedef struct TestRequestParam
bool noBucket;
bool upload;
bool noAuth;
bool multiPart;
const char *path;
const char *object;
const char *query;
const char *contentRange;
@@ -86,8 +88,10 @@ typedef struct TestRequestParam
static void
testRequest(IoWrite *write, const char *verb, TestRequestParam param)
{
String *request = strCatFmt(
strNew(), "%s %s/storage/v1/b%s", verb, param.upload ? "/upload" : "", param.noBucket ? "" : "/bucket/o");
String *const request =
param.path != NULL ?
strCatFmt(strNew(), "%s %s", verb, param.path) :
strCatFmt(strNew(), "%s %s/storage/v1/b%s", verb, param.upload ? "/upload" : "", param.noBucket ? "" : "/bucket/o");
// Add object
if (param.object != NULL)
@@ -115,6 +119,13 @@ testRequest(IoWrite *write, const char *verb, TestRequestParam param)
if (param.contentType != NULL)
strCatFmt(request, "content-type:%s\r\n", param.contentType);
// Add multipart content-type
if (param.multiPart)
{
ASSERT(param.contentType == NULL);
strCatZ(request, "content-type:multipart/mixed; boundary=" HTTP_MULTIPART_BOUNDARY_INIT "\r\n");
}
// Add host
strCatFmt(request, "host:%s\r\n", strZ(hrnServerHost()));
@@ -139,6 +150,7 @@ typedef struct TestResponseParam
{
VAR_PARAM_HEADER;
unsigned int code;
bool multiPart;
const char *header;
const char *content;
} TestResponseParam;
@@ -174,6 +186,10 @@ testResponse(IoWrite *write, TestResponseParam param)
if (param.header != NULL)
strCatFmt(response, "%s\r\n", param.header);
// Add multipart content-type
if (param.multiPart)
strCatZ(response, "content-type:multipart/mixed; boundary=" HTTP_MULTIPART_BOUNDARY_INIT "\r\n");
// Content
if (param.content != NULL)
{
@@ -940,14 +956,62 @@ testRun(void)
" },"
" {"
" \"name\": \"path1/xxx.zzz\""
" },"
" {"
" \"name\": \"path2/file2\""
" }"
" ]"
"}");
testRequestP(service, HTTP_VERB_DELETE, .object = "path/to/test1.txt");
testResponseP(service);
testRequestP(
service, HTTP_VERB_POST, .path = "/batch/storage/v1", .multiPart = true,
.content =
"\r\n--" HTTP_MULTIPART_BOUNDARY_INIT "\r\n"
"content-type:application/http\r\n"
"content-transfer-encoding:binary\r\n"
"content-id:0\r\n"
"\r\n"
"DELETE /storage/v1/b/bucket/o/path%2Fto%2Ftest1.txt HTTP/1.1\r\n"
"content-length:0\r\n"
"\r\n"
"\r\n--" HTTP_MULTIPART_BOUNDARY_INIT "\r\n"
"content-type:application/http\r\n"
"content-transfer-encoding:binary\r\n"
"content-id:1\r\n"
"\r\n"
"DELETE /storage/v1/b/bucket/o/path1%2Fxxx.zzz HTTP/1.1\r\n"
"content-length:0\r\n"
"\r\n"
"\r\n--" HTTP_MULTIPART_BOUNDARY_INIT "\r\n"
"content-type:application/http\r\n"
"content-transfer-encoding:binary\r\n"
"content-id:2\r\n"
"\r\n"
"DELETE /storage/v1/b/bucket/o/path2%2Ffile2 HTTP/1.1\r\n"
"content-length:0\r\n"
"\r\n"
"\r\n--" HTTP_MULTIPART_BOUNDARY_INIT "--\r\n");
testResponseP(
service, .multiPart = true,
.content =
"\r\n--" HTTP_MULTIPART_BOUNDARY_INIT "\r\n"
"content-type:application/http\r\n"
"content-id:response-0\r\n"
"\r\n"
"HTTP/1.1 404 Missing\r\n\r\n"
"\r\n--" HTTP_MULTIPART_BOUNDARY_INIT "\r\n"
"content-type:application/http\r\n"
"content-id:response-1\r\n"
"\r\n"
"HTTP/1.1 200 OK\r\n\r\n"
"\r\n--" HTTP_MULTIPART_BOUNDARY_INIT "\r\n"
"content-type:application/http\r\n"
"content-id:response-2\r\n"
"\r\n"
"HTTP/1.1 300 Error\r\n\r\n"
"\r\n--" HTTP_MULTIPART_BOUNDARY_INIT "--\r\n");
testRequestP(service, HTTP_VERB_DELETE, .object = "path1/xxx.zzz");
testRequestP(service, HTTP_VERB_DELETE, .object = "path2/file2");
testResponseP(service);
TEST_RESULT_VOID(storagePathRemoveP(storage, STRDEF("/"), .recurse = true), "remove");
@@ -955,6 +1019,8 @@ testRun(void)
// -----------------------------------------------------------------------------------------------------------------
TEST_TITLE("remove files from path");
((StorageGcs *)storageDriver(storage))->deleteMax = 1;
testRequestP(service, HTTP_VERB_GET, .query = "fields=nextPageToken%2Cprefixes%2Citems%28name%29&prefix=path%2F");
testResponseP(
service,
@@ -973,13 +1039,72 @@ testRun(void)
" ]"
"}");
testRequestP(service, HTTP_VERB_DELETE, .object = "path/test1.txt");
testResponseP(service);
testRequestP(
service, HTTP_VERB_POST, .path = "/batch/storage/v1", .multiPart = true,
.content =
"\r\n--" HTTP_MULTIPART_BOUNDARY_INIT "\r\n"
"content-type:application/http\r\n"
"content-transfer-encoding:binary\r\n"
"content-id:0\r\n"
"\r\n"
"DELETE /storage/v1/b/bucket/o/path%2Ftest1.txt HTTP/1.1\r\n"
"content-length:0\r\n"
"\r\n"
"\r\n--" HTTP_MULTIPART_BOUNDARY_INIT "--\r\n");
testResponseP(
service, .multiPart = true,
.content =
"\r\n--" HTTP_MULTIPART_BOUNDARY_INIT "\r\n"
"content-type:application/http\r\n"
"content-id:response-0\r\n"
"\r\n"
"HTTP/1.1 404 OK\r\n\r\n"
"\r\n--" HTTP_MULTIPART_BOUNDARY_INIT "--\r\n");
testRequestP(
service, HTTP_VERB_POST, .path = "/batch/storage/v1", .multiPart = true,
.content =
"\r\n--" HTTP_MULTIPART_BOUNDARY_INIT "\r\n"
"content-type:application/http\r\n"
"content-transfer-encoding:binary\r\n"
"content-id:0\r\n"
"\r\n"
"DELETE /storage/v1/b/bucket/o/path%2Fpath1%2Fxxx.zzz HTTP/1.1\r\n"
"content-length:0\r\n"
"\r\n"
"\r\n--" HTTP_MULTIPART_BOUNDARY_INIT "--\r\n");
testResponseP(
service, .multiPart = true,
.content =
"\r\n--" HTTP_MULTIPART_BOUNDARY_INIT "\r\n"
"content-type:application/http\r\n"
"content-id:response-0\r\n"
"\r\n"
"HTTP/1.1 305 Error\r\n"
"content-length:5\r\n"
"content-type:text\r\n\r\n"
"ERROR"
"\r\n--" HTTP_MULTIPART_BOUNDARY_INIT "--\r\n");
testRequestP(service, HTTP_VERB_DELETE, .object = "path/path1/xxx.zzz");
testResponseP(service);
testResponseP(service, .code = 300, .content = "ERROR2");
TEST_RESULT_VOID(storagePathRemoveP(storage, STRDEF("/path"), .recurse = true), "remove");
TEST_ERROR_FMT(
storagePathRemoveP(storage, STRDEF("/path"), .recurse = true), ProtocolError,
"HTTP request failed with 300:\n"
"*** Path/Query ***:\n"
"DELETE /storage/v1/b/bucket/o/path%%2Fpath1%%2Fxxx.zzz\n"
"*** Request Headers ***:\n"
"authorization: <redacted>\n"
"content-length: 0\n"
"host: %s\n"
"*** Response Headers ***:\n"
"content-length: 6\n"
"*** Response Content ***:\n"
"ERROR2",
strZ(hrnServerHost()));
((StorageGcs *)storageDriver(storage))->deleteMax = STORAGE_GCS_DELETE_MAX;
// -----------------------------------------------------------------------------------------------------------------
TEST_TITLE("remove files in empty subpath (nothing to do)");