9 #include "XrdVersion.hh"
15 #include <curl/curl.h>
37 uint64_t TPCHandler::m_monid{0};
38 int TPCHandler::m_marker_period = 5;
39 size_t TPCHandler::m_block_size = 16*1024*1024;
40 size_t TPCHandler::m_small_block_size = 1*1024*1024;
49 TPCHandler::TPCLogRecord::~TPCLogRecord()
56 monInfo.
clID = clID.c_str();
58 gettimeofday(&monInfo.
endT, 0);
60 if (log_prefix ==
"PullRequest")
61 {monInfo.
dstURL = local.c_str();
62 monInfo.
srcURL = remote.c_str();
64 monInfo.
dstURL = remote.c_str();
65 monInfo.
srcURL = local.c_str();
69 if (!status) monInfo.
endRC = 0;
70 else if (tpc_status > 0) monInfo.
endRC = tpc_status;
71 else monInfo.
endRC = 1;
72 monInfo.
strm =
static_cast<unsigned char>(streams);
73 monInfo.
fSize = (bytes_transferred < 0 ? 0 : bytes_transferred);
76 tpcMonitor->Report(monInfo);
86 if (curl) curl_easy_cleanup(curl);
101 int TPCHandler::sockopt_callback(
void *clientp, curl_socket_t curlfd, curlsocktype purpose) {
102 TPCLogRecord * rec = (TPCLogRecord *)clientp;
103 if (purpose == CURLSOCKTYPE_IPCXN && rec && rec->pmarkManager.isEnabled()) {
106 return CURL_SOCKOPT_ALREADY_CONNECTED;
108 return CURL_SOCKOPT_OK;
120 int TPCHandler::opensocket_callback(
void *clientp,
121 curlsocktype purpose,
122 struct curl_sockaddr *aInfo)
125 int fd = XrdSysFD_Socket(aInfo->family, aInfo->socktype, aInfo->protocol);
129 return CURL_SOCKET_BAD;
131 TPCLogRecord * rec = (TPCLogRecord *)clientp;
132 if (purpose == CURLSOCKTYPE_IPCXN && clientp)
135 && !thePeer.isMapped());
136 std::stringstream connectErrMsg;
138 if(!rec->pmarkManager.connect(fd, &(aInfo->addr), aInfo->addrlen, CONNECT_TIMEOUT, connectErrMsg)) {
139 rec->m_log->Emsg(rec->log_prefix.c_str(),
"Unable to connect socket:", connectErrMsg.str().c_str());
140 return CURL_SOCKET_BAD;
147 int TPCHandler::closesocket_callback(
void *clientp, curl_socket_t fd) {
148 TPCLogRecord * rec = (TPCLogRecord *)clientp;
153 rec->pmarkManager.endPmark(fd);
173 std::string TPCHandler::prepareURL(
XrdHttpExtReq &req,
bool & hasSetOpaque) {
179 return prepareURL(req,foundHeader);
193 std::stringstream parser(opaque);
194 std::string sequence;
195 std::stringstream output;
197 while (
getline(parser, sequence,
'&')) {
198 if (sequence.empty()) {
continue;}
199 size_t equal_pos = sequence.find(
'=');
201 if (equal_pos != std::string::npos)
202 val = curl_easy_escape(curl, sequence.c_str() + equal_pos + 1, sequence.size() - equal_pos - 1);
204 if (!val && equal_pos != std::string::npos) {
continue;}
206 if (!first) output <<
"&";
208 output << sequence.substr(0, equal_pos);
210 output <<
"=" << val;
222 TPCHandler::ConfigureCurlCA(
CURL *curl)
224 auto ca_filename = m_ca_file ? m_ca_file->CAFilename() :
"";
225 auto crl_filename = m_ca_file ? m_ca_file->CRLFilename() :
"";
226 if (!ca_filename.empty() && !crl_filename.empty()) {
227 curl_easy_setopt(curl, CURLOPT_CAINFO, ca_filename.c_str());
231 std::ifstream in(crl_filename, std::ifstream::ate | std::ifstream::binary);
232 if(in.tellg() > 0 && m_ca_file->atLeastOneValidCRLFound()){
233 curl_easy_setopt(curl, CURLOPT_CRLFILE, crl_filename.c_str());
235 std::ostringstream oss;
236 oss <<
"No valid CRL file has been found in the file " << crl_filename <<
". Disabling CRL checking.";
237 m_log.
Log(
Warning,
"TpcHandler",oss.str().c_str());
240 else if (!m_cadir.empty()) {
241 curl_easy_setopt(curl, CURLOPT_CAPATH, m_cadir.c_str());
243 if (!m_cafile.empty()) {
244 curl_easy_setopt(curl, CURLOPT_CAINFO, m_cafile.c_str());
250 return !strcmp(verb,
"COPY") || !strcmp(verb,
"OPTIONS");
258 if (!strncmp(input.c_str(),
"davs://", 7)) {
259 return "https://" + input.substr(7);
269 if (req.
verb ==
"OPTIONS") {
270 return ProcessOptionsReq(req);
273 if (header != req.
headers.end()) {
274 if (header->second !=
"none") {
275 m_log.
Emsg(
"ProcessReq",
"COPY requested an unsupported credential type: ", header->second.c_str());
276 return req.
SendSimpleResp(400, NULL, NULL,
"COPY requestd an unsupported Credential type", 0);
280 if (header != req.
headers.end()) {
282 return ProcessPullReq(src, req);
285 if (header != req.
headers.end()) {
286 return ProcessPushReq(header->second, req);
288 m_log.
Emsg(
"ProcessReq",
"COPY verb requested but no source or destination specified.");
289 return req.
SendSimpleResp(400, NULL, NULL,
"No Source or Destination specified", 0);
306 m_fixed_route(false),
308 m_first_timeout(120),
309 m_log(log->logger(),
"TPC_"),
312 if (!Configure(config, myEnv)) {
313 throw std::runtime_error(
"Failed to configure the HTTP third-party-copy handler.");
331 return req.
SendSimpleResp(200, NULL, (
char *)
"DAV: 1\r\nDAV: <http://apache.org/dav/propset/fs/1>\r\nAllow: HEAD,GET,PUT,PROPFIND,DELETE,OPTIONS,COPY", NULL, 0);
341 if (authz_header != req.
headers.end()) {
342 char * quoted_url =
quote(authz_header->second.c_str());
343 std::stringstream ss;
344 ss <<
"authz=" << quoted_url;
355 int TPCHandler::RedirectTransfer(
CURL *curl,
const std::string &redirect_resource,
360 if ((ptr == NULL) || (*ptr ==
'\0') || (port == 0)) {
362 std::stringstream ss;
363 ss <<
"Internal error: redirect without hostname";
364 logTransferEvent(
LogMask::Error, rec,
"REDIRECT_INTERNAL_ERROR", ss.str());
365 return req.
SendSimpleResp(rec.status, NULL, NULL, generateClientErr(ss, rec).c_str(), 0);
369 std::string rdr_info = ptr;
370 std::string host, opaque;
371 size_t pos = rdr_info.find(
'?');
372 host = rdr_info.substr(0, pos);
374 if (pos != std::string::npos) {
375 opaque = rdr_info.substr(pos + 1);
378 std::stringstream ss;
379 ss <<
"Location: http" << (m_desthttps ?
"s" :
"") <<
"://" << host <<
":" << port <<
"/" << redirect_resource;
381 if (!opaque.empty()) {
387 return req.
SendSimpleResp(rec.status, NULL,
const_cast<char *
>(ss.str().c_str()),
395 int TPCHandler::OpenWaitStall(
XrdSfsFile &fh,
const std::string &resource,
397 const std::string &authz)
404 size_t pos = resource.find(
'?');
406 std::string path = resource.substr(0, pos);
408 if (pos != std::string::npos) {
409 opaque = resource.substr(pos + 1);
414 opaque += (opaque.empty() ?
"" :
"&");
417 open_result = fh.
open(path.c_str(), mode, openMode, &sec, opaque.c_str());
421 if (open_result ==
SFS_STARTED) {secs_to_stall = secs_to_stall/2 + 5;}
422 std::this_thread::sleep_for (std::chrono::seconds(secs_to_stall));
439 bool &success, TPCLogRecord &rec,
bool shouldReturnErrorToClient) {
441 curl_easy_setopt(curl, CURLOPT_NOBODY, 1);
443 curl_easy_setopt(curl, CURLOPT_TIMEOUT, CONNECT_TIMEOUT);
445 res = curl_easy_perform(curl);
448 curl_easy_setopt(curl, CURLOPT_NOBODY, 0);
450 curl_easy_setopt(curl, CURLOPT_TIMEOUT, 0L);
451 if (res == CURLE_HTTP_RETURNED_ERROR) {
452 std::stringstream ss;
453 ss <<
"Remote server failed request while fetching remote size";
454 std::stringstream ss2;
455 ss2 << ss.str() <<
": " << curl_easy_strerror(res);
458 return shouldReturnErrorToClient ? req.
SendSimpleResp(rec.status, NULL, NULL, generateClientErr(ss, rec, res).c_str(), 0) : -1;
460 std::stringstream ss;
461 ss <<
"Remote side " << req.
clienthost <<
" failed with status code " << state.
GetStatusCode() <<
" while fetching remote size";
464 return shouldReturnErrorToClient ? req.
SendSimpleResp(rec.status, NULL, NULL, generateClientErr(ss, rec).c_str(), 0) : -1;
466 std::stringstream ss;
467 ss <<
"Internal transfer failure while fetching remote size";
468 std::stringstream ss2;
469 ss2 << ss.str() <<
" - HTTP library failed: " << curl_easy_strerror(res);
472 return shouldReturnErrorToClient ? req.
SendSimpleResp(rec.status, NULL, NULL, generateClientErr(ss, rec, res).c_str(), 0) : -1;
474 std::stringstream ss;
475 ss <<
"Successfully determined remote size for pull request: "
482 int TPCHandler::GetContentLengthTPCPull(
CURL *curl,
XrdHttpExtReq &req, uint64_t &contentLength,
bool & success, TPCLogRecord &rec) {
489 if ((result = DetermineXferSize(curl, req, state, success, rec)) || !success) {
501 std::stringstream ss;
502 const std::string crlf =
"\n";
503 ss <<
"Perf Marker" << crlf;
504 ss <<
"Timestamp: " << time(NULL) << crlf;
505 ss <<
"Stripe Index: 0" << crlf;
507 ss <<
"Total Stripe Count: 1" << crlf;
512 ss <<
"RemoteConnections: " << desc << crlf;
517 return req.
ChunkResp(ss.str().c_str(), 0);
524 int TPCHandler::SendPerfMarker(
XrdHttpExtReq &req, TPCLogRecord &rec, std::vector<State*> &state,
525 off_t bytes_transferred)
539 std::stringstream ss;
540 const std::string crlf =
"\n";
541 ss <<
"Perf Marker" << crlf;
542 ss <<
"Timestamp: " << time(NULL) << crlf;
543 ss <<
"Stripe Index: 0" << crlf;
544 ss <<
"Stripe Bytes Transferred: " << bytes_transferred << crlf;
545 ss <<
"Total Stripe Count: 1" << crlf;
549 std::stringstream ss2;
550 for (std::vector<State*>::const_iterator iter = state.begin();
551 iter != state.end(); iter++)
553 std::string desc = (*iter)->GetConnectionDescription();
555 ss2 << (first ?
"" :
",") << desc;
560 ss <<
"RemoteConnections: " << ss2.str() << crlf;
562 rec.bytes_transferred = bytes_transferred;
565 return req.
ChunkResp(ss.str().c_str(), 0);
576 CURLM *multi_handle = curl_multi_init();
580 "Failed to initialize a libcurl multi-handle");
581 std::stringstream ss;
582 ss <<
"Failed to initialize internal server memory";
583 return req.
SendSimpleResp(rec.status, NULL, NULL, generateClientErr(ss, rec).c_str(), 0);
589 mres = curl_multi_add_handle(multi_handle, curl);
592 std::stringstream ss;
593 ss <<
"Failed to add transfer to libcurl multi-handle: HTTP library failure=" << curl_multi_strerror(mres);
594 logTransferEvent(
LogMask::Error, rec,
"CURL_INIT_FAIL", ss.str());
595 curl_multi_cleanup(multi_handle);
596 return req.
SendSimpleResp(rec.status, NULL, NULL, generateClientErr(ss, rec).c_str(), 0);
600 int retval = req.
StartChunkedResp(201,
"Created",
"Content-Type: text/plain");
602 curl_multi_cleanup(multi_handle);
604 "Failed to send the initial response to the TPC client");
608 "Initial transfer response sent to the TPC client");
613 int running_handles = 1;
614 time_t last_marker = 0;
616 off_t last_advance_bytes = 0;
617 time_t last_advance_time = time(NULL);
618 time_t transfer_start = last_advance_time;
619 CURLcode res =
static_cast<CURLcode
>(-1);
621 time_t now = time(NULL);
622 time_t next_marker = last_marker + m_marker_period;
623 if (now >= next_marker) {
625 if (bytes_xfer > last_advance_bytes) {
626 last_advance_bytes = bytes_xfer;
627 last_advance_time = now;
629 if (SendPerfMarker(req, rec, state)) {
630 curl_multi_remove_handle(multi_handle, curl);
631 curl_multi_cleanup(multi_handle);
633 "Failed to send a perf marker to the TPC client");
636 int timeout = (transfer_start == last_advance_time) ? m_first_timeout : m_timeout;
637 if (now > last_advance_time + timeout) {
638 const char *log_prefix = rec.log_prefix.c_str();
639 bool tpc_pull = strncmp(
"Pull", log_prefix, 4) == 0;
642 std::stringstream ss;
643 ss <<
"Transfer failed because no bytes have been "
644 << (tpc_pull ?
"received from the source (pull mode) in "
645 :
"transmitted to the destination (push mode) in ") << timeout <<
" seconds.";
647 curl_multi_remove_handle(multi_handle, curl);
648 curl_multi_cleanup(multi_handle);
654 rec.pmarkManager.startTransfer();
655 mres = curl_multi_perform(multi_handle, &running_handles);
656 if (mres == CURLM_CALL_MULTI_PERFORM) {
660 }
else if (mres != CURLM_OK) {
662 }
else if (running_handles == 0) {
666 rec.pmarkManager.beginPMarks();
673 msg = curl_multi_info_read(multi_handle, &msgq);
674 if (msg && (msg->msg == CURLMSG_DONE)) {
675 CURL *easy_handle = msg->easy_handle;
676 res = msg->data.result;
677 curl_multi_remove_handle(multi_handle, easy_handle);
681 int64_t max_sleep_time = next_marker - time(NULL);
682 if (max_sleep_time <= 0) {
686 #ifdef HAVE_CURL_MULTI_WAIT
687 mres = curl_multi_wait(multi_handle, NULL, 0, max_sleep_time*1000, &fd_count);
691 if (mres != CURLM_OK) {
694 }
while (running_handles);
696 if (mres != CURLM_OK) {
697 std::stringstream ss;
698 ss <<
"Internal libcurl multi-handle error: HTTP library failure=" << curl_multi_strerror(mres);
699 logTransferEvent(
LogMask::Error, rec,
"TRANSFER_CURL_ERROR", ss.str());
701 curl_multi_remove_handle(multi_handle, curl);
702 curl_multi_cleanup(multi_handle);
704 if ((retval = req.
ChunkResp(generateClientErr(ss, rec).c_str(), 0))) {
706 "Failed to send error message to the TPC client");
716 msg = curl_multi_info_read(multi_handle, &msgq);
717 if (msg && (msg->msg == CURLMSG_DONE)) {
718 CURL *easy_handle = msg->easy_handle;
719 res = msg->data.result;
720 curl_multi_remove_handle(multi_handle, easy_handle);
724 if (!state.
GetErrorCode() && res ==
static_cast<CURLcode
>(-1)) {
725 curl_multi_remove_handle(multi_handle, curl);
726 curl_multi_cleanup(multi_handle);
727 std::stringstream ss;
728 ss <<
"Internal state error in libcurl";
729 logTransferEvent(
LogMask::Error, rec,
"TRANSFER_CURL_ERROR", ss.str());
731 if ((retval = req.
ChunkResp(generateClientErr(ss, rec).c_str(), 0))) {
733 "Failed to send error message to the TPC client");
738 curl_multi_cleanup(multi_handle);
752 std::stringstream ss;
753 bool success =
false;
756 std::stringstream ss2;
757 ss2 <<
"Remote side failed with status code " << state.
GetStatusCode();
759 std::replace(err.begin(), err.end(),
'\n',
' ');
760 ss2 <<
"; error message: \"" << err <<
"\"";
762 logTransferEvent(
LogMask::Error, rec,
"TRANSFER_FAIL", ss2.str());
763 ss << generateClientErr(ss2, rec);
766 if (err.empty()) {err =
"(no error message provided)";}
767 else {std::replace(err.begin(), err.end(),
'\n',
' ');}
768 std::stringstream ss2;
769 ss2 <<
"Error when interacting with local filesystem: " << err;
770 logTransferEvent(
LogMask::Error, rec,
"TRANSFER_FAIL", ss2.str());
771 ss << generateClientErr(ss2, rec);
772 }
else if (res != CURLE_OK) {
773 std::stringstream ss2;
774 ss2 <<
"Internal transfer failure";
775 std::stringstream ss3;
776 ss3 << ss2.str() <<
": " << curl_easy_strerror(res);
777 logTransferEvent(
LogMask::Error, rec,
"TRANSFER_FAIL", ss3.str());
778 ss << generateClientErr(ss2, rec, res);
780 ss <<
"success: Created";
784 if ((retval = req.
ChunkResp(ss.str().c_str(), 0))) {
786 "Failed to send last update to remote client");
788 }
else if (success) {
799 int TPCHandler::ProcessPushReq(
const std::string & resource,
XrdHttpExtReq &req) {
800 TPCLogRecord rec(req);
801 rec.log_prefix =
"PushRequest";
803 rec.remote = resource;
807 if (name) rec.name = name;
808 logTransferEvent(
LogMask::Info, rec,
"PUSH_START",
"Starting a push request");
811 auto curl = curlPtr.get();
813 std::stringstream ss;
814 ss <<
"Failed to initialize internal transfer resources";
817 return req.
SendSimpleResp(rec.status, NULL, NULL, generateClientErr(ss, rec).c_str(), 0);
819 curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1);
820 curl_easy_setopt(curl, CURLOPT_HTTP_VERSION, (
long) CURL_HTTP_VERSION_1_1);
823 curl_easy_setopt(curl, CURLOPT_OPENSOCKETFUNCTION, opensocket_callback);
824 curl_easy_setopt(curl, CURLOPT_OPENSOCKETDATA, &rec);
825 curl_easy_setopt(curl, CURLOPT_CLOSESOCKETFUNCTION, closesocket_callback);
826 curl_easy_setopt(curl, CURLOPT_SOCKOPTFUNCTION, sockopt_callback);
827 curl_easy_setopt(curl, CURLOPT_CLOSESOCKETDATA, &rec);
828 curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT, CONNECT_TIMEOUT);
830 std::string redirect_resource = req.
resource;
831 if (query_header != req.
headers.end()) {
832 redirect_resource = query_header->second;
836 uint64_t file_monid =
AtomicInc(m_monid);
838 std::unique_ptr<XrdSfsFile> fh(m_sfs->
newFile(name, file_monid));
841 std::stringstream ss;
842 ss <<
"Failed to initialize internal transfer file handle";
845 return req.
SendSimpleResp(rec.status, NULL, NULL, generateClientErr(ss, rec).c_str(), 0);
847 std::string full_url = prepareURL(req);
849 std::string authz = GetAuthz(req);
851 int open_results = OpenWaitStall(*fh, full_url,
SFS_O_RDONLY, 0644,
854 int result = RedirectTransfer(curl, redirect_resource, req, fh->
error, rec);
856 }
else if (
SFS_OK != open_results) {
858 std::stringstream ss;
860 if (msg == NULL) ss <<
"Failed to open local resource";
863 if (code == EACCES) rec.status = 401;
864 else if (code == EEXIST) rec.status = 412;
866 int resp_result = req.
SendSimpleResp(rec.status, NULL, NULL, generateClientErr(ss, rec).c_str(), 0);
870 ConfigureCurlCA(curl);
871 curl_easy_setopt(curl, CURLOPT_URL, resource.c_str());
873 Stream stream(std::move(fh), 0, 0, m_log);
877 return RunCurlWithUpdates(curl, req, state, rec);
884 int TPCHandler::ProcessPullReq(
const std::string &resource,
XrdHttpExtReq &req) {
885 TPCLogRecord rec(req);
886 rec.log_prefix =
"PullRequest";
888 rec.remote = resource;
892 if (name) rec.name = name;
893 logTransferEvent(
LogMask::Info, rec,
"PULL_START",
"Starting a pull request");
896 auto curl = curlPtr.get();
898 std::stringstream ss;
899 ss <<
"Failed to initialize internal transfer resources";
902 return req.
SendSimpleResp(rec.status, NULL, NULL, generateClientErr(ss, rec).c_str(), 0);
919 std::string host_used;
920 if (host_header != req.
headers.end()) {
921 host_used = host_header->second;
927 ip = (
char *)malloc(ip_size-1);
930 memcpy(ip, buff+1, ip_size-2);
934 curl_easy_setopt(curl, CURLOPT_INTERFACE, ip);
936 curl_easy_setopt(curl, CURLOPT_NOSIGNAL, 1);
937 curl_easy_setopt(curl, CURLOPT_HTTP_VERSION, (
long) CURL_HTTP_VERSION_1_1);
939 curl_easy_setopt(curl, CURLOPT_OPENSOCKETFUNCTION, opensocket_callback);
940 curl_easy_setopt(curl, CURLOPT_OPENSOCKETDATA, &rec);
941 curl_easy_setopt(curl, CURLOPT_SOCKOPTFUNCTION, sockopt_callback);
942 curl_easy_setopt(curl, CURLOPT_SOCKOPTDATA , &rec);
943 curl_easy_setopt(curl, CURLOPT_CLOSESOCKETFUNCTION, closesocket_callback);
944 curl_easy_setopt(curl, CURLOPT_CLOSESOCKETDATA, &rec);
945 curl_easy_setopt(curl, CURLOPT_CONNECTTIMEOUT, CONNECT_TIMEOUT);
946 std::unique_ptr<XrdSfsFile> fh(m_sfs->
newFile(name, m_monid++));
948 std::stringstream ss;
949 ss <<
"Failed to initialize internal transfer file handle";
952 return req.
SendSimpleResp(rec.status, NULL, NULL, generateClientErr(ss, rec).c_str(), 0);
955 std::string redirect_resource = req.
resource;
956 if (query_header != req.
headers.end()) {
957 redirect_resource = query_header->second;
961 if ((overwrite_header == req.
headers.end()) || (overwrite_header->second ==
"T")) {
967 if (streams_header != req.
headers.end()) {
970 stream_req = std::stol(streams_header->second);
973 if (stream_req < 0 || stream_req > 100) {
974 std::stringstream ss;
975 ss <<
"Invalid request for number of streams";
977 logTransferEvent(
LogMask::Info, rec,
"INVALID_REQUEST", ss.str());
978 return req.
SendSimpleResp(rec.status, NULL, NULL, generateClientErr(ss, rec).c_str(), 0);
980 streams = stream_req == 0 ? 1 : stream_req;
983 rec.streams = streams;
984 bool hasSetOpaque =
false;
985 std::string full_url = prepareURL(req, hasSetOpaque);
986 std::string authz = GetAuthz(req);
987 curl_easy_setopt(curl, CURLOPT_URL, resource.c_str());
988 ConfigureCurlCA(curl);
989 uint64_t sourceFileContentLength = 0;
994 GetContentLengthTPCPull(curl, req, sourceFileContentLength, success, rec);
998 full_url += hasSetOpaque ?
"&" :
"?";
999 full_url +=
"oss.asize=" + std::to_string(sourceFileContentLength);
1006 int open_result = OpenWaitStall(*fh, full_url, mode|
SFS_O_WRONLY,
1010 int result = RedirectTransfer(curl, redirect_resource, req, fh->
error, rec);
1012 }
else if (
SFS_OK != open_result) {
1014 std::stringstream ss;
1016 if ((msg == NULL) || (*msg ==
'\0')) ss <<
"Failed to open local resource";
1019 if (code == EACCES) rec.status = 401;
1020 else if (code == EEXIST) rec.status = 412;
1023 generateClientErr(ss, rec).c_str(), 0);
1027 Stream stream(std::move(fh), streams * m_pipelining_multiplier, streams > 1 ? m_block_size : m_small_block_size, m_log);
1033 return RunCurlWithStreams(req, state, streams, rec);
1035 return RunCurlWithUpdates(curl, req, state, rec);
1043 void TPCHandler::logTransferEvent(
LogMask mask,
const TPCLogRecord &rec,
1044 const std::string &event,
const std::string &message)
1048 std::stringstream ss;
1049 ss <<
"event=" <<
event <<
", local=" << rec.local <<
", remote=" << rec.remote;
1050 if (rec.name.empty())
1051 ss <<
", user=(anonymous)";
1053 ss <<
", user=" << rec.name;
1054 if (rec.streams != 1)
1055 ss <<
", streams=" << rec.streams;
1056 if (rec.bytes_transferred >= 0)
1057 ss <<
", bytes_transferred=" << rec.bytes_transferred;
1058 if (rec.status >= 0)
1059 ss <<
", status=" << rec.status;
1060 if (rec.tpc_status >= 0)
1061 ss <<
", tpc_status=" << rec.tpc_status;
1062 if (!message.empty())
1063 ss <<
"; " << message;
1064 m_log.
Log(mask, rec.log_prefix.c_str(), ss.str().c_str());
1067 std::string TPCHandler::generateClientErr(std::stringstream &err_ss,
const TPCLogRecord &rec, CURLcode cCode) {
1068 std::stringstream ssret;
1069 ssret <<
"failure: " << err_ss.str() <<
", local=" << rec.local <<
", remote=" << rec.remote;
1070 if(cCode != CURLcode::CURLE_OK) {
1071 ssret <<
", HTTP library failure=" << curl_easy_strerror(cCode);
1082 if (curl_global_init(CURL_GLOBAL_DEFAULT)) {
1083 log->
Emsg(
"TPCInitialize",
"libcurl failed to initialize");
1089 log->
Emsg(
"TPCInitialize",
"TPC handler requires a config filename in order to load");
1093 log->
Emsg(
"TPCInitialize",
"Will load configuration for the TPC handler from", config);
1095 }
catch (std::runtime_error &re) {
1096 log->
Emsg(
"TPCInitialize",
"Encountered a runtime failure when loading ", re.what());
char * quote(const char *str)
void getline(uchar *buff, int blen)
CURLMcode curl_multi_wait_impl(CURLM *multi_handle, int timeout_ms, int *numfds)
static std::string PrepareURL(const std::string &input)
XrdVERSIONINFO(XrdHttpGetExtHandler, HttpTPC)
XrdHttpExtHandler * XrdHttpGetExtHandler(XrdSysError *log, const char *config, const char *, XrdOucEnv *myEnv)
std::string encode_xrootd_opaque_to_uri(CURL *curl, const std::string &opaque)
int GetStatusCode() const
off_t BytesTransferred() const
void CopyHeaders(XrdHttpExtReq &req)
void SetErrorMessage(const std::string &error_msg)
std::string GetErrorMessage() const
std::string GetConnectionDescription()
void SetContentLength(const off_t content_length)
off_t GetContentLength() const
void SetErrorCode(int error_code)
TPCHandler(XrdSysError *log, const char *config, XrdOucEnv *myEnv)
virtual int ProcessReq(XrdHttpExtReq &req)
virtual bool MatchesPath(const char *verb, const char *path)
Tells if the incoming path is recognized as one of the paths that have to be processed.
int ChunkResp(const char *body, long long bodylen)
Send a (potentially partial) body in a chunked response; invoking with NULL body.
void GetClientID(std::string &clid)
std::map< std::string, std::string > & headers
int StartChunkedResp(int code, const char *desc, const char *header_to_add)
Starts a chunked response; body of request is sent over multiple parts using the SendChunkResp.
const XrdSecEntity & GetSecEntity() const
int SendSimpleResp(int code, const char *desc, const char *header_to_add, const char *body, long long bodylen)
Sends a basic response. If the length is < 0 then it is calculated internally.
static const int noPort
Do not add port number.
int Format(char *bAddr, int bLen, fmtUse fmtType=fmtAuto, int fmtOpts=0)
@ fmtAddr
Address using suitable ipv4 or ipv6 format.
static const char * GetAddrs(const char *hSpec, XrdNetAddr *aListP[], int &aListN, AddrOpts opts=allIPMap, int pNum=PortInSpec)
void * GetPtr(const char *varname)
const char * getErrText()
void setUCap(int ucval)
Set user capabilties.
static std::map< std::string, T >::const_iterator caseInsensitiveFind(const std::map< std::string, T > &m, const std::string &lowerCaseSearchKey)
char * name
Entity's name.
virtual XrdSfsFile * newFile(char *user=0, int MonID=0)=0
virtual int open(const char *fileName, XrdSfsFileOpenMode openMode, mode_t createMode, const XrdSecEntity *client=0, const char *opaque=0)=0
int Emsg(const char *esfx, int ecode, const char *text1, const char *text2=0)
XrdSysLogger * logger(XrdSysLogger *lp=0)
void Log(int mask, const char *esfx, const char *text1, const char *text2=0, const char *text3=0)
static std::string prepareOpenURL(const std::string &reqResource, std::map< std::string, std::string > &reqHeaders, const std::map< std::string, std::string > &hdr2cgimap, bool &hasSetOpaque)
std::unique_ptr< CURL, CurlDeleter > ManagedCurlHandle
void operator()(CURL *curl)
static const int uIPv64
ucap: Supports only IPv4 info