« back to all changes in this revision
Viewing changes to plugin/pbms/src/cslib/CSS3Protocol.cc
- browse files at revision 1871
- compare with another revision
- compare with revision 38
- download diff
- download tarball
- view history from revision 1871
- Committer: Lee Bieber
- Date: 2010-10-22 16:47:38 UTC
- mfrom: (1841.1.7 drizzle_pbms)
- Revision ID: kalebral@gmail.com-20101022164738-vv8w22b8towpb307
Merge Barry - fix bug 657830: PBMS build failure in GCC 4.5
- files added:
- plugin/pbms/src/cslib/CSSys.h
- files modified:
- drizzled/plugin/event_observer.cc
added
removed
plugin/pbms/src/cslib/CSS3Protocol.cc
35
35
#ifdef S3_UNIT_TEST
36
36
//#define SHOW_SIGNING
37
37
// Uncomment this line to trace network action during request. Very Usefull!!
38
//#define DEBUG_CURL
38
#define DEBUG_CURL
39
39
#define DUMP_ERRORS
40
40
#endif
41
41
42
//#define DEBUG_CURL
43
#define DUMP_ERRORS
42
//#define DUMP_ERRORS
44
43
//#define SHOW_SIGNING
45
44
46
45
#define HEX_CHECKSUM_VALUE_SIZE (2 *CHECKSUM_VALUE_SIZE)
128
127
129
128
CSMd5 ms_md5;
130
129
char ms_s3Checksum[HEX_CHECKSUM_VALUE_SIZE +1];
130
bool ms_calculate_md5;
131
131
132
132
bool ms_notFound; // True if the object could not be found
133
133
bool ms_retry; // True if the request failed with a retry error.
142
142
bool ms_throw_error; // Gets set if an exception occurs in a callback.
143
143
bool ms_old_libcurl;
144
144
char *ms_safe_url;
145
time_t ms_last_modified;
145
146
146
147
S3ProtocolCon():
147
148
ms_curl(NULL),
148
149
ms_header_list(NULL),
149
150
ms_inputStream(NULL),
150
151
ms_outputStream(NULL),
152
ms_calculate_md5(false),
151
153
ms_notFound(false),
152
154
ms_retry(false),
153
155
ms_slowDown(false),
156
158
ms_replyStatus(0),
157
159
ms_throw_error(false),
158
160
ms_old_libcurl(false),
159
ms_safe_url(NULL)
161
ms_safe_url(NULL),
162
ms_last_modified(0)
160
163
{
161
164
162
165
ms_curl = curl_easy_init();
230
233
//case 307: // Temporary Redirect
231
234
break;
232
235
case 404: // Not Found
236
case 403: // Forbidden (S3 object not found)
233
237
ms_notFound = true;
234
238
break;
235
239
case 500:
367
371
{
368
372
enter_();
369
373
370
push_(output);
374
if (output) {
375
push_(output);
376
THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_HTTPGET, 1L));
377
} else {
378
THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_NOBODY, 1L));
379
}
380
381
//
382
THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_FILETIME, 1L));
371
383
THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_HTTPHEADER, ms_header_list));
372
THROW_CURL_IF(curl_easy_setopt(ms_curl, CURLOPT_HTTPGET, 1L));
384
// Ask curl to parse the Last-Modified header. This is easier than
385
// parsing it ourselves.
373
386
374
387
ms_outputStream = output;
375
388
if (curl_easy_perform(ms_curl) && !ms_throw_error) {
377
390
CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, ms_curl_error);
378
391
}
379
392
ms_outputStream = NULL;
380
release_(output);
393
if (output){
394
release_(output);
395
}
381
396
382
397
if (ms_throw_error)
383
398
throw_();
384
399
385
400
check_reply_status();
401
curl_easy_getinfo(ms_curl, CURLINFO_FILETIME, &ms_last_modified);
386
402
exit_();
387
403
}
388
404
389
inline void ms_execute_put_request(CSInputStream *input, off64_t size, Md5Digest *digest)
405
inline void ms_execute_put_request(CSInputStream *input, off64_t size)
390
406
{
391
407
enter_();
392
408
414
430
415
431
check_reply_status();
416
432
417
// Check that the checksums agree
418
char checksum[HEX_CHECKSUM_VALUE_SIZE +1];
419
420
ms_md5.md5_digest(digest);
421
cs_bin_to_hex(HEX_CHECKSUM_VALUE_SIZE, checksum, CHECKSUM_VALUE_SIZE, digest->val);
422
checksum[HEX_CHECKSUM_VALUE_SIZE] = 0;
423
424
cs_strToUpper(ms_s3Checksum);
425
if (strcmp(checksum, ms_s3Checksum)) {
426
// The request should be restarted in this case.
427
ms_retry = true;
428
CSException::logException(CS_CONTEXT, CS_ERR_CHECKSUM_ERROR, "Calculated checksum did not match S3 checksum");
433
if (ms_calculate_md5) {
434
// If the data was not sent with an md5 checksum then verify
435
// the server's md5 value with the one calculated during the send.
436
char checksum[HEX_CHECKSUM_VALUE_SIZE +1];
437
Md5Digest digest;
438
439
ms_md5.md5_get_digest(&digest);
440
cs_bin_to_hex(HEX_CHECKSUM_VALUE_SIZE, checksum, CHECKSUM_VALUE_SIZE, digest.val);
441
checksum[HEX_CHECKSUM_VALUE_SIZE] = 0;
442
443
cs_strToUpper(ms_s3Checksum);
444
if (strcmp(checksum, ms_s3Checksum)) {
445
// The request should be restarted in this case.
446
ms_retry = true;
447
CSException::logException(CS_CONTEXT, CS_ERR_CHECKSUM_ERROR, "Calculated checksum did not match S3 checksum");
448
}
429
449
}
430
450
431
451
exit_();
432
452
}
433
453
443
463
s3_server(NULL),
444
464
s3_public_key(NULL),
445
465
s3_private_key(NULL),
446
s3_maxRetrys(5)
466
s3_maxRetries(5),
467
s3_sleepTime(0)
447
468
{
448
469
new_(s3_server, CSStringBuffer());
449
470
s3_server->append("s3.amazonaws.com/");
524
545
// CURL callback functions:
525
546
////////////////////////////
526
547
//----------------------
548
//-----------------
549
static bool try_ReadStream(CSThread *self, S3ProtocolCon *con, unsigned char *ptr, size_t buffer_size, size_t *data_sent)
550
{
551
volatile bool rtc = true;
552
try_(a) {
553
*data_sent = con->ms_inputStream->read((char*)ptr, buffer_size);
554
if (*data_sent <= con->ms_data_size) {
555
con->ms_data_size -= *data_sent;
556
if (*data_sent)
557
con->ms_md5.md5_append(ptr, *data_sent); // Calculating the checksum for the data sent.
558
} else if (*data_sent > con->ms_data_size)
559
CSException::RecordException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "Blob larger than expected.");
560
else if (con->ms_data_size && !*data_sent)
561
CSException::RecordException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "Blob smaller than expected.");
562
rtc = false;
563
}
564
565
catch_(a)
566
cont_(a);
567
return rtc;
568
}
569
570
//----------------------
527
571
static size_t send_callback(void *ptr, size_t objs, size_t obj_size, void *v_con)
528
572
{
529
573
S3ProtocolCon *con = (S3ProtocolCon*) v_con;
533
577
return 0;
534
578
535
579
enter_();
536
try_(a) {
537
data_sent = con->ms_inputStream->read((char*)ptr, buffer_size);
538
if (data_sent <= con->ms_data_size) {
539
con->ms_data_size -= data_sent;
540
if (data_sent)
541
con->ms_md5.md5_append((u_char*)ptr, data_sent); // Calculating the checksum for the data sent.
542
} else if (data_sent > con->ms_data_size)
543
CSException::RecordException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "Blob larger than expected.");
544
else if (con->ms_data_size && !data_sent)
545
CSException::RecordException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "Blob smaller than expected.");
580
if (try_ReadStream(self, con, (unsigned char*)ptr, buffer_size, &data_sent)) {
581
con->ms_throw_error = true;
582
data_sent = (size_t)-1;
546
583
}
547
catch_(a);
548
con->ms_throw_error = true;
549
data_sent = SIZE_MAX;
550
551
cont_(a);
552
553
return data_sent;
584
585
return_(data_sent);
554
586
}
555
//----------------------
556
static size_t receive_data(void *vptr, size_t objs, size_t obj_size, void *v_con)
587
588
//-----------------
589
static bool try_WriteStream(CSThread *self, S3ProtocolCon *con, char *ptr, size_t data_len)
557
590
{
558
S3ProtocolCon *con = (S3ProtocolCon*) v_con;
559
size_t data_len = objs * obj_size;
560
561
enter_();
591
volatile bool rtc = true;
562
592
try_(a) {
563
593
if (con->ms_replyStatus >= 400) { // Collect the error reply.
564
594
if (!con->ms_errorReply)
565
595
con->ms_errorReply = new CSStringBuffer(50);
566
con->ms_errorReply->append((char*)vptr, data_len);
596
con->ms_errorReply->append(ptr, data_len);
567
597
} else if ( con->ms_outputStream)
568
con->ms_outputStream->write((char*)vptr, data_len);
598
con->ms_outputStream->write(ptr, data_len);
599
rtc = false;
569
600
}
570
catch_(a);
571
con->ms_throw_error = true;
572
data_len = SIZE_MAX;
573
601
602
catch_(a)
574
603
cont_(a);
604
return rtc;
605
}
606
607
//----------------------
608
static size_t receive_data(void *vptr, size_t objs, size_t obj_size, void *v_con)
609
{
610
S3ProtocolCon *con = (S3ProtocolCon*) v_con;
611
size_t data_len = objs * obj_size;
612
613
enter_();
614
if (try_WriteStream(self, con, (char*)vptr, data_len)) {
615
con->ms_throw_error = true;
616
data_len = (size_t)-1;
617
}
618
575
619
return_(data_len);
576
620
}
577
621
578
622
#define IS_REDIRECT(s) ((s >= 300) && (s < 400))
579
623
//----------------------
624
static bool try_addHeader(CSThread *self, S3ProtocolCon *con, char *name, uint32_t name_len, char *value, uint32_t value_len)
625
{
626
try_(a) {
627
con->ms_reply_headers.addHeader(name, name_len, value, value_len);
628
return false;
629
}
630
631
catch_(a);
632
cont_(a);
633
return true;
634
}
635
636
//----------------------
580
637
static size_t receive_header(void *header, size_t objs, size_t obj_size, void *v_con)
581
638
{
582
639
S3ProtocolCon *con = (S3ProtocolCon*) v_con;
584
641
char *end, *ptr = (char*) header, *name, *value = NULL;
585
642
uint32_t name_len =0, value_len = 0;
586
643
587
CLOBBER_PROTECT(con);
588
CLOBBER_PROTECT(size);
589
CLOBBER_PROTECT(ptr);
590
CLOBBER_PROTECT(value);
591
CLOBBER_PROTECT(value_len);
592
CLOBBER_PROTECT(name_len);
593
594
644
//printf( "receive_header: %s\n", ptr);
595
645
end = ptr + size;
596
646
if (*(end -2) == '\r' && *(end -1) == '\n')
638
688
while (value[value_len-1] == ' ') value_len--;
639
689
640
690
if (!strncasecmp(name, "ETag", 4)) {
641
value++; value_len -=2; // Strip quotation marks from checksum string.
691
if (*value == '"') {
692
value++; value_len -=2; // Strip quotation marks from checksum string.
693
}
642
694
if (value_len == HEX_CHECKSUM_VALUE_SIZE) {
643
695
memcpy(con->ms_s3Checksum, value, value_len);
644
696
con->ms_s3Checksum[value_len] = 0;
646
698
}
647
699
648
700
enter_();
649
try_(a) {
650
con->ms_reply_headers.addHeader(name, name_len, value, value_len);
701
if (try_addHeader(self, con, name, name_len, value, value_len)) {
702
con->ms_throw_error = true;
703
size = (size_t)-1;
651
704
}
652
653
catch_(a);
654
con->ms_throw_error = true;
655
size = (size_t)-1;
656
657
cont_(a);
658
705
return_(size);
659
706
}
660
707
716
763
con_data->ms_execute_delete_request();
717
764
718
765
if (con_data->ms_retry) {
719
if (retry_count == s3_maxRetrys) {
766
if (retry_count == s3_maxRetries) {
720
767
CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "S3 operation aborted after max retries.");
721
768
}
722
printf("RETRY: s3_delete()\n");
769
//printf("RETRY: s3_delete()\n");
723
770
retry_count++;
771
self->sleep(s3_sleepTime);
724
772
goto retry;
725
773
}
726
774
815
863
}
816
864
817
865
if (con_data->ms_retry) {
818
if (retry_count == s3_maxRetrys) {
866
if (retry_count == s3_maxRetries) {
819
867
CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "S3 operation aborted after max retries.");
820
868
}
821
printf("RETRY: s3_copy()\n");
869
//printf("RETRY: s3_copy()\n");
822
870
retry_count++;
871
self->sleep(s3_sleepTime);
823
872
goto retry;
824
873
}
825
874
831
880
832
881
833
882
//-------------------------------
834
CSVector *CSS3Protocol::s3_receive(CSOutputStream *output, const char *bucket, const char *key, bool *found)
883
CSVector *CSS3Protocol::s3_receive(CSOutputStream *output, const char *bucket, const char *key, bool *found, S3RangePtr range, time_t *last_modified)
835
884
{
836
885
CSStringBuffer *s3_buffer;
837
886
char date[64];
839
888
uint32_t retry_count = 0;
840
889
S3ProtocolCon *con_data;
841
890
CSVector *replyHeaders;
891
CSString *range_header = NULL;
892
const char *http_op;
842
893
843
894
enter_();
844
895
845
push_(output);
896
if (output) {
897
push_(output);
898
http_op = "GET";
899
} else
900
http_op = "HEAD";
846
901
847
902
new_(s3_buffer, CSStringBuffer());
848
903
push_(s3_buffer);
872
927
s3_buffer->append(date);
873
928
con_data->ms_setHeader(s3_buffer->getCString());
874
929
930
if (range) {
931
char buffer[80];
932
snprintf(buffer, 80,"Range: bytes=%"PRIu64"-%"PRIu64, range->startByte, range->endByte);
933
934
range_header = CSString::newString(buffer);
935
}
875
936
// Create the authentication signature and add the 'Authorization' header
876
signed_str = s3_getSignature("GET", NULL, NULL, date, bucket, key);
937
if (range_header)
938
con_data->ms_setHeader(range_header->getCString());
939
signed_str = s3_getSignature(http_op, NULL, NULL, date, bucket, key, NULL);
877
940
push_(signed_str);
878
941
s3_buffer->setLength(0);
879
942
s3_buffer->append("Authorization: AWS ");
883
946
release_(signed_str); signed_str = NULL;
884
947
con_data->ms_setHeader(s3_buffer->getCString());
885
948
886
con_data->ms_execute_get_request(RETAIN(output));
949
if (output) output->retain();
950
con_data->ms_execute_get_request(output);
887
951
888
952
if (con_data->ms_retry) {
889
if (retry_count == s3_maxRetrys) {
953
if (retry_count == s3_maxRetries) {
890
954
CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "S3 operation aborted after max retries.");
891
955
}
892
printf("RETRY: s3_receive()\n");
956
//printf("RETRY: s3_receive()\n");
893
957
retry_count++;
894
958
output->reset();
959
self->sleep(s3_sleepTime);
895
960
goto retry;
896
961
}
897
962
963
if (last_modified)
964
*last_modified = con_data->ms_last_modified;
898
965
*found = !con_data->ms_notFound;
899
966
replyHeaders = con_data->ms_reply_headers.takeHeaders();
900
967
release_(con_data);
901
968
release_(s3_buffer);
902
release_(output);
969
if (output)
970
release_(output);
903
971
904
972
return_(replyHeaders);
905
973
}
1000
1068
s3_buffer->append(bucket);
1001
1069
s3_buffer->append(".");
1002
1070
s3_buffer->append(s3_server->getCString());
1071
//s3_buffer->append("/");
1072
//s3_buffer->append(bucket);
1003
1073
if (key_prefix) {
1004
1074
s3_buffer->append("?prefix=");
1005
1075
s3_buffer->append(key_prefix);
1035
1105
con_data->ms_execute_get_request(RETAIN(output));
1036
1106
1037
1107
if (con_data->ms_retry) {
1038
if (retry_count == s3_maxRetrys) {
1108
if (retry_count == s3_maxRetries) {
1039
1109
CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "S3 operation aborted after max retries.");
1040
1110
}
1041
printf("RETRY: s3_list()\n");
1111
//printf("RETRY: s3_list()\n");
1042
1112
retry_count++;
1043
1113
output->reset();
1114
self->sleep(s3_sleepTime);
1044
1115
goto retry;
1045
1116
}
1046
1117
1077
1148
CSStringBuffer *s3_buffer;
1078
1149
char date[64];
1079
1150
CSString *signed_str;
1080
Md5Digest dummy_digest;
1081
1151
uint32_t retry_count = 0;
1082
1152
S3ProtocolCon *con_data;
1083
1153
CSVector *replyHeaders;
1154
char checksum[32], *md5 = NULL;
1084
1155
1085
1156
enter_();
1086
1157
push_(input);
1090
1161
1091
1162
new_(con_data, S3ProtocolCon());
1092
1163
push_(con_data);
1093
1094
if (!digest)
1095
digest = &dummy_digest;
1096
1164
1097
1165
if (!content_type)
1098
1166
content_type = "binary/octet-stream";
1130
1198
s3_buffer->append(content_type);
1131
1199
con_data->ms_setHeader(s3_buffer->getCString());
1132
1200
1201
if (digest) {
1202
// Add the Md5 checksum header
1203
md5 = checksum;
1204
memset(checksum, 0, 32);
1205
base64Encode(digest->val, 16, checksum, 32);
1206
1207
s3_buffer->setLength(0);
1208
s3_buffer->append("Content-MD5: ");
1209
s3_buffer->append(checksum);
1210
con_data->ms_setHeader(s3_buffer->getCString());
1211
con_data->ms_calculate_md5 = false;
1212
} else
1213
con_data->ms_calculate_md5 = true;
1214
1215
1133
1216
// Create the authentication signature and add the 'Authorization' header
1134
1217
if (!s3Authorization)
1135
signed_str = s3_getSignature("PUT", NULL, content_type, date, bucket, key);
1218
signed_str = s3_getSignature("PUT", md5, content_type, date, bucket, key);
1136
1219
else
1137
1220
signed_str = CSString::newString(s3Authorization);
1138
1221
push_(signed_str);
1144
1227
release_(signed_str); signed_str = NULL;
1145
1228
con_data->ms_setHeader(s3_buffer->getCString());
1146
1229
1147
con_data->ms_execute_put_request(RETAIN(input), size, digest);
1230
con_data->ms_execute_put_request(RETAIN(input), size);
1148
1231
1149
1232
if (con_data->ms_retry) {
1150
if (retry_count == s3_maxRetrys) {
1233
if (retry_count == s3_maxRetries) {
1151
1234
CSException::throwException(CS_CONTEXT, CS_ERR_GENERIC_ERROR, "S3 operation aborted after max retries.");
1152
1235
}
1153
printf("RETRY: s3_send()\n");
1236
//printf("RETRY: s3_send()\n");
1154
1237
retry_count++;
1155
1238
input->reset();
1239
self->sleep(s3_sleepTime);
1156
1240
goto retry;
1157
1241
}
1158
1242
1159
1243
replyHeaders = con_data->ms_reply_headers.takeHeaders();
1244
1160
1245
release_(con_data);
1161
1246
release_(s3_buffer);
1162
1247
release_(input);
1211
1296
printf("Delete object:\n\t%s d <bucket> <object_key>\n", cmd);
1212
1297
printf("Delete all object with a given prefix:\n\t%s D <bucket> <object_prefix>\n", cmd);
1213
1298
printf("Get object, data will be written to 'prottest.out':\n\t%s g <bucket> <object_key> <timeout>\n", cmd);
1299
printf("Get object header only:\n\t%s h <bucket> <object_key> <timeout>\n", cmd);
1214
1300
printf("Put (Upload) an object:\n\t%s p <bucket> <object_key> <file>\n", cmd);
1215
1301
printf("List objects in the bucket:\n\t%s l <bucket> [<object_prefix> [max_list_size]]\n", cmd);
1216
1302
printf("Copy object:\n\t%s c <src_bucket> <src_object_key> <dst_bucket> <dst_object_key> \n", cmd);
1240
1326
CSThread *main_thread;
1241
1327
const char *pub_key;
1242
1328
const char *priv_key;
1329
const char *server;
1243
1330
CSS3Protocol *prot = NULL;
1244
1331
1245
1332
if (argc < 3) {
1265
1352
new_(prot, CSS3Protocol());
1266
1353
push_(prot);
1267
1354
1268
prot->s3_setServer("s3.amazonaws.com/");
1355
server = getenv("S3_SERVER");
1356
if ((server == NULL) || (*server == 0))
1357
server = "s3.amazonaws.com/";
1358
prot->s3_setServer(server);
1269
1359
prot->s3_setPublicKey(pub_key);
1270
1360
prot->s3_setPrivateKey(priv_key);
1271
1361
1308
1398
1309
1399
break;
1310
1400
case 'g': // Get the object
1311
if (argc == 4) {
1401
if ((argc == 4) || (argc == 6)) {
1312
1402
CSFile *output;
1313
1403
CSVector *headers;
1314
1404
bool found;
1405
S3RangeRec *range_ptr = NULL, range = {0,0};
1406
1407
if (argc == 6) {
1408
range.startByte = atoi(argv[4]);
1409
range.endByte = atoi(argv[5]);
1410
range_ptr = ⦥
1411
}
1315
1412
1316
1413
output = CSFile::newFile("prottest.out");
1317
1414
push_(output);
1318
1415
output->open(CSFile::CREATE | CSFile::TRUNCATE);
1319
headers = prot->s3_receive(output->getOutputStream(), argv[2], argv[3], &found);
1416
headers = prot->s3_receive(output->getOutputStream(), argv[2], argv[3], &found, range_ptr);
1320
1417
if (!found)
1321
1418
printf("%s/%s could not be found.\n", argv[2], argv[3]);
1322
1419
1328
1425
1329
1426
break;
1330
1427
1428
case 'h': // Get the object header
1429
if (argc == 4) {
1430
CSVector *headers;
1431
bool found;
1432
S3RangeRec range = {0,0};
1433
1434
headers = prot->s3_receive(NULL, argv[2], argv[3], &found);
1435
if (!found)
1436
printf("%s/%s could not be found.\n", argv[2], argv[3]);
1437
1438
dump_headers(headers);
1439
1440
} else
1441
printf("Bad command: h <bucket> <object_key>\n");
1442
1443
break;
1444
1331
1445
case 'p': // Put (Upload) the object
1332
1446
if (argc == 5) {
1333
1447
CSFile *input;
1448
Md5Digest digest;
1334
1449
CSVector *headers;
1335
1450
1336
1451
input = CSFile::newFile(argv[4]);
1337
1452
push_(input);
1338
1453
input->open(CSFile::READONLY);
1339
headers = prot->s3_send(input->getInputStream(), argv[2], argv[3], input->myFilePath->getSize());
1454
input->md5Digest(&digest);
1455
headers = prot->s3_send(input->getInputStream(), argv[2], argv[3], input->myFilePath->getSize(), NULL, &digest);
1340
1456
dump_headers(headers);
1341
1457
release_(input);
1342
1458
} else
Loggerhead is a web-based interface for Breezy
Version: 2.0.1