Skip to content
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.

Commit 06e96f6

Browse files
mdouaihymdouaihy
committedSep 10, 2019
Add an HTTP Sender (jaegertracing#165)
Support an HTTP Sender that allows to connect directly to a Collector. For instance, authentication is not supported. Signed-off-by: FR-MUREX-COM\mchaikhadouaihy <mehrez.douaihy@gmail.com>
1 parent 2b5e4fb commit 06e96f6

27 files changed

+1361
-890
lines changed
 

‎.gitignore

+4
Original file line numberDiff line numberDiff line change
@@ -42,3 +42,7 @@
4242

4343
# Log files
4444
*.log
45+
46+
# CLion folders
47+
cmake-build-debug
48+
.idea

‎CMakeLists.txt

+487-484
Large diffs are not rendered by default.

‎README.md

+12
Original file line numberDiff line numberDiff line change
@@ -66,6 +66,18 @@ repoter:
6666
6767
NOTE: It is not recommended to use a remote host for UDP connections.
6868
69+
### Connecting directly to the Collector
70+
71+
In case the client should connect directly to the collector instead of going through an agent, it's necessary update the reporter configuration
72+
73+
```yml
74+
repoter:
75+
endpoint: http://collectorhost:collectorport/api/traces
76+
```
77+
where collectorhost and collectorport are placeholders respectively for your Collector host and port.
78+
79+
Note that if both localAgentHostPort and endpoint are specified, the endpoint will be used.
80+
6981
### Updating Sampling Server URL
7082
7183
The default sampling collector URL is `http://127.0.0.1:5778`. Similar to UDP address above, you can use a different URL by updating the sampler configuration.

‎src/jaegertracing/UDPTransport.cpp ‎src/jaegertracing/ThriftTransport.cpp

+152-142
Original file line numberDiff line numberDiff line change
@@ -1,142 +1,152 @@
1-
/*
2-
* Copyright (c) 2017-2018 Uber Technologies, Inc.
3-
*
4-
* Licensed under the Apache License, Version 2.0 (the "License");
5-
* you may not use this file except in compliance with the License.
6-
* You may obtain a copy of the License at
7-
*
8-
* http://www.apache.org/licenses/LICENSE-2.0
9-
*
10-
* Unless required by applicable law or agreed to in writing, software
11-
* distributed under the License is distributed on an "AS IS" BASIS,
12-
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13-
* See the License for the specific language governing permissions and
14-
* limitations under the License.
15-
*/
16-
17-
#include "jaegertracing/UDPTransport.h"
18-
19-
#include "jaegertracing/Span.h"
20-
#include "jaegertracing/Tag.h"
21-
#include "jaegertracing/Tracer.h"
22-
#include <algorithm>
23-
#include <cstdint>
24-
#include <iostream>
25-
#include <iterator>
26-
#include <string>
27-
#include <thrift/protocol/TCompactProtocol.h>
28-
#include <thrift/transport/TBufferTransports.h>
29-
30-
#ifdef _MSC_VER
31-
#pragma warning(disable : 4267) // Conversion from unsigned to signed. It should not be a problem here.
32-
#endif
33-
34-
namespace jaegertracing {
35-
namespace net {
36-
class IPAddress;
37-
} // namespace net
38-
39-
namespace {
40-
41-
constexpr auto kEmitBatchOverhead = 30;
42-
43-
template <typename ThriftType>
44-
int calcSizeOfSerializedThrift(const ThriftType& base, int maxPacketSize)
45-
{
46-
std::shared_ptr<apache::thrift::transport::TMemoryBuffer> buffer(
47-
new apache::thrift::transport::TMemoryBuffer(maxPacketSize));
48-
apache::thrift::protocol::TCompactProtocolFactory factory;
49-
auto protocol = factory.getProtocol(buffer);
50-
base.write(protocol.get());
51-
uint8_t* data = nullptr;
52-
uint32_t size = 0;
53-
buffer->getBuffer(&data, &size);
54-
return size;
55-
}
56-
57-
} // anonymous namespace
58-
59-
UDPTransport::UDPTransport(const net::IPAddress& ip, int maxPacketSize)
60-
: _client(new utils::UDPClient(ip, maxPacketSize))
61-
, _maxSpanBytes(0)
62-
, _byteBufferSize(0)
63-
, _processByteSize(0)
64-
{
65-
}
66-
67-
int UDPTransport::append(const Span& span)
68-
{
69-
if (_process.serviceName.empty()) {
70-
const auto& tracer = static_cast<const Tracer&>(span.tracer());
71-
_process.serviceName = tracer.serviceName();
72-
73-
const auto& tracerTags = tracer.tags();
74-
std::vector<thrift::Tag> thriftTags;
75-
thriftTags.reserve(tracerTags.size());
76-
std::transform(std::begin(tracerTags),
77-
std::end(tracerTags),
78-
std::back_inserter(thriftTags),
79-
[](const Tag& tag) { return tag.thrift(); });
80-
_process.__set_tags(thriftTags);
81-
82-
_processByteSize =
83-
calcSizeOfSerializedThrift(_process, _client->maxPacketSize());
84-
_maxSpanBytes =
85-
_client->maxPacketSize() - _processByteSize - kEmitBatchOverhead;
86-
}
87-
const auto jaegerSpan = span.thrift();
88-
const auto spanSize =
89-
calcSizeOfSerializedThrift(jaegerSpan, _client->maxPacketSize());
90-
if (spanSize > _maxSpanBytes) {
91-
std::ostringstream oss;
92-
throw Transport::Exception("Span is too large", 1);
93-
}
94-
95-
_byteBufferSize += spanSize;
96-
if (_byteBufferSize <= _maxSpanBytes) {
97-
_spanBuffer.push_back(jaegerSpan);
98-
if (_byteBufferSize < _maxSpanBytes) {
99-
return 0;
100-
}
101-
return flush();
102-
}
103-
104-
// Flush currently full buffer, then append this span to buffer.
105-
const auto flushed = flush();
106-
_spanBuffer.push_back(jaegerSpan);
107-
_byteBufferSize = spanSize + _processByteSize;
108-
return flushed;
109-
}
110-
111-
int UDPTransport::flush()
112-
{
113-
if (_spanBuffer.empty()) {
114-
return 0;
115-
}
116-
117-
thrift::Batch batch;
118-
batch.__set_process(_process);
119-
batch.__set_spans(_spanBuffer);
120-
121-
try {
122-
_client->emitBatch(batch);
123-
} catch (const std::system_error& ex) {
124-
std::ostringstream oss;
125-
oss << "Could not send span " << ex.what()
126-
<< ", code=" << ex.code().value();
127-
throw Transport::Exception(oss.str(), _spanBuffer.size());
128-
} catch (const std::exception& ex) {
129-
std::ostringstream oss;
130-
oss << "Could not send span " << ex.what();
131-
throw Transport::Exception(oss.str(), _spanBuffer.size());
132-
} catch (...) {
133-
throw Transport::Exception("Could not send span, unknown error",
134-
_spanBuffer.size());
135-
}
136-
137-
resetBuffers();
138-
139-
return batch.spans.size();
140-
}
141-
142-
} // namespace jaegertracing
1+
/*
2+
* Copyright (c) 2017-2018 Uber Technologies, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#include "jaegertracing/ThriftTransport.h"
18+
19+
#include "jaegertracing/Span.h"
20+
#include "jaegertracing/Tag.h"
21+
#include "jaegertracing/Tracer.h"
22+
#include <algorithm>
23+
#include <cstdint>
24+
#include <iostream>
25+
#include <iterator>
26+
#include <thrift/transport/TBufferTransports.h>
27+
28+
#ifdef _MSC_VER
29+
#pragma warning(disable : 4267) // Conversion from unsigned to signed. It should not be a problem here.
30+
#endif
31+
32+
namespace jaegertracing {
33+
namespace net {
34+
class IPAddress;
35+
} // namespace net
36+
37+
namespace {
38+
39+
constexpr auto kEmitBatchOverhead = 30;
40+
41+
template <typename ThriftType>
42+
int calcSizeOfSerializedThrift(apache::thrift::protocol::TProtocolFactory& factory, const ThriftType& base, int maxPacketSize)
43+
{
44+
std::shared_ptr<apache::thrift::transport::TMemoryBuffer> buffer(
45+
new apache::thrift::transport::TMemoryBuffer(maxPacketSize));
46+
auto protocol = factory.getProtocol(buffer);
47+
base.write(protocol.get());
48+
uint8_t* data = nullptr;
49+
uint32_t size = 0;
50+
buffer->getBuffer(&data, &size);
51+
return size;
52+
}
53+
54+
} // anonymous namespace
55+
56+
ThriftTransport::ThriftTransport(const net::IPAddress& ip,
57+
int maxPacketSize)
58+
: _sender(new utils::UDPSender(ip, maxPacketSize))
59+
, _maxSpanBytes(0)
60+
, _byteBufferSize(0)
61+
, _processByteSize(0)
62+
, _protocolFactory(_sender->protocolFactory())
63+
{
64+
}
65+
66+
ThriftTransport::ThriftTransport(const net::URI& endpoint,
67+
int maxPacketSize)
68+
: _sender(new utils::HttpSender(endpoint, maxPacketSize))
69+
, _maxSpanBytes(0)
70+
, _byteBufferSize(0)
71+
, _processByteSize(0)
72+
, _protocolFactory(_sender->protocolFactory())
73+
{
74+
}
75+
76+
77+
int ThriftTransport::append(const Span& span)
78+
{
79+
if (_process.serviceName.empty()) {
80+
const auto& tracer = static_cast<const Tracer&>(span.tracer());
81+
_process.serviceName = tracer.serviceName();
82+
83+
const auto& tracerTags = tracer.tags();
84+
std::vector<thrift::Tag> thriftTags;
85+
thriftTags.reserve(tracerTags.size());
86+
std::transform(std::begin(tracerTags),
87+
std::end(tracerTags),
88+
std::back_inserter(thriftTags),
89+
[](const Tag& tag) { return tag.thrift(); });
90+
_process.__set_tags(thriftTags);
91+
92+
_processByteSize =
93+
calcSizeOfSerializedThrift(*_protocolFactory, _process, _sender->maxPacketSize());
94+
_maxSpanBytes =
95+
_sender->maxPacketSize() - _processByteSize - kEmitBatchOverhead;
96+
}
97+
const auto jaegerSpan = span.thrift();
98+
const auto spanSize =
99+
calcSizeOfSerializedThrift(*_protocolFactory, jaegerSpan, _sender->maxPacketSize());
100+
if (spanSize > _maxSpanBytes) {
101+
std::ostringstream oss;
102+
throw Transport::Exception("Span is too large", 1);
103+
}
104+
105+
_byteBufferSize += spanSize;
106+
if (_byteBufferSize <= _maxSpanBytes) {
107+
_spanBuffer.push_back(jaegerSpan);
108+
if (_byteBufferSize < _maxSpanBytes) {
109+
return 0;
110+
}
111+
return flush();
112+
}
113+
114+
// Flush currently full buffer, then append this span to buffer.
115+
const auto flushed = flush();
116+
_spanBuffer.push_back(jaegerSpan);
117+
_byteBufferSize = spanSize + _processByteSize;
118+
return flushed;
119+
}
120+
121+
int ThriftTransport::flush()
122+
{
123+
if (_spanBuffer.empty()) {
124+
return 0;
125+
}
126+
127+
thrift::Batch batch;
128+
batch.__set_process(_process);
129+
batch.__set_spans(_spanBuffer);
130+
131+
try {
132+
_sender->emitBatch(batch);
133+
} catch (const std::system_error& ex) {
134+
std::ostringstream oss;
135+
oss << "Could not send span " << ex.what()
136+
<< ", code=" << ex.code().value();
137+
throw Transport::Exception(oss.str(), _spanBuffer.size());
138+
} catch (const std::exception& ex) {
139+
std::ostringstream oss;
140+
oss << "Could not send span " << ex.what();
141+
throw Transport::Exception(oss.str(), _spanBuffer.size());
142+
} catch (...) {
143+
throw Transport::Exception("Could not send span, unknown error",
144+
_spanBuffer.size());
145+
}
146+
147+
resetBuffers();
148+
149+
return batch.spans.size();
150+
}
151+
152+
} // namespace jaegertracing

‎src/jaegertracing/UDPTransport.h ‎src/jaegertracing/ThriftTransport.h

+67-64
Original file line numberDiff line numberDiff line change
@@ -1,64 +1,67 @@
1-
/*
2-
* Copyright (c) 2017-2018 Uber Technologies, Inc.
3-
*
4-
* Licensed under the Apache License, Version 2.0 (the "License");
5-
* you may not use this file except in compliance with the License.
6-
* You may obtain a copy of the License at
7-
*
8-
* http://www.apache.org/licenses/LICENSE-2.0
9-
*
10-
* Unless required by applicable law or agreed to in writing, software
11-
* distributed under the License is distributed on an "AS IS" BASIS,
12-
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13-
* See the License for the specific language governing permissions and
14-
* limitations under the License.
15-
*/
16-
17-
#ifndef JAEGERTRACING_UDPTRANSPORT_H
18-
#define JAEGERTRACING_UDPTRANSPORT_H
19-
20-
#include "jaegertracing/Compilers.h"
21-
#include "jaegertracing/Span.h"
22-
#include "jaegertracing/Transport.h"
23-
#include "jaegertracing/thrift-gen/jaeger_types.h"
24-
#include "jaegertracing/utils/UDPClient.h"
25-
26-
namespace jaegertracing {
27-
28-
class UDPTransport : public Transport {
29-
public:
30-
UDPTransport(const net::IPAddress& ip, int maxPacketSize);
31-
32-
~UDPTransport() { close(); }
33-
34-
int append(const Span& span) override;
35-
36-
int flush() override;
37-
38-
void close() override { _client->close(); }
39-
40-
protected:
41-
void setClient(std::unique_ptr<utils::UDPClient>&& client)
42-
{
43-
_client = std::move(client);
44-
}
45-
46-
private:
47-
void resetBuffers()
48-
{
49-
_spanBuffer.clear();
50-
_byteBufferSize = _processByteSize;
51-
}
52-
53-
std::unique_ptr<utils::UDPClient> _client;
54-
int _maxSpanBytes;
55-
int _byteBufferSize;
56-
std::vector<thrift::Span> _spanBuffer;
57-
std::shared_ptr<apache::thrift::protocol::TProtocol> _protocol;
58-
thrift::Process _process;
59-
int _processByteSize;
60-
};
61-
62-
} // namespace jaegertracing
63-
64-
#endif // JAEGERTRACING_UDPTRANSPORT_H
1+
/*
2+
* Copyright (c) 2017-2018 Uber Technologies, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#ifndef JAEGERTRACING_THRIFTTRANSPORT_H
18+
#define JAEGERTRACING_THRIFTTRANSPORT_H
19+
20+
#include "jaegertracing/Compilers.h"
21+
#include "jaegertracing/Span.h"
22+
#include "jaegertracing/Transport.h"
23+
#include "jaegertracing/thrift-gen/jaeger_types.h"
24+
#include "jaegertracing/utils/UDPSender.h"
25+
#include "jaegertracing/utils/HttpSender.h"
26+
27+
namespace jaegertracing {
28+
29+
class ThriftTransport : public Transport {
30+
public:
31+
ThriftTransport(const net::IPAddress& ip, int maxPacketSize);
32+
33+
ThriftTransport(const net::URI& endpoint, int maxPacketSize);
34+
35+
~ThriftTransport() { close(); }
36+
37+
int append(const Span& span) override;
38+
39+
int flush() override;
40+
41+
void close() override { _sender->close(); }
42+
43+
protected:
44+
void setClient(std::unique_ptr<utils::UDPSender>&& client)
45+
{
46+
_sender = std::move(client);
47+
}
48+
49+
private:
50+
void resetBuffers()
51+
{
52+
_spanBuffer.clear();
53+
_byteBufferSize = _processByteSize;
54+
}
55+
56+
std::unique_ptr<utils::Sender> _sender;
57+
int _maxSpanBytes;
58+
int _byteBufferSize;
59+
std::vector<thrift::Span> _spanBuffer;
60+
thrift::Process _process;
61+
int _processByteSize;
62+
std::unique_ptr<apache::thrift::protocol::TProtocolFactory> _protocolFactory;
63+
};
64+
65+
} // namespace jaegertracing
66+
67+
#endif //JAEGERTRACING_THRIFTTRANSPORT_H

‎src/jaegertracing/UDPTransportTest.cpp ‎src/jaegertracing/ThriftTransportTest.cpp

+20-21
Original file line numberDiff line numberDiff line change
@@ -16,23 +16,22 @@
1616

1717
#include <gtest/gtest.h>
1818

19-
#include "jaegertracing/Config.h"
2019
#include "jaegertracing/Tracer.h"
21-
#include "jaegertracing/UDPTransport.h"
20+
#include "jaegertracing/ThriftTransport.h"
2221
#include "jaegertracing/testutils/TracerUtil.h"
2322
#include "jaegertracing/utils/ErrorUtil.h"
2423

2524
namespace jaegertracing {
2625
namespace {
2726

28-
class MockUDPClient : public utils::UDPClient {
27+
class MockUDPSender : public utils::UDPSender {
2928
public:
3029
enum class ExceptionType { kSystemError, kException, kString };
3130

32-
MockUDPClient(const net::IPAddress& serverAddr,
31+
MockUDPSender(const net::IPAddress& serverAddr,
3332
int maxPacketSize,
3433
ExceptionType type)
35-
: UDPClient(serverAddr, maxPacketSize)
34+
: UDPSender(serverAddr, maxPacketSize)
3635
, _type(type)
3736
{
3837
}
@@ -55,27 +54,27 @@ class MockUDPClient : public utils::UDPClient {
5554
ExceptionType _type;
5655
};
5756

58-
class MockUDPTransport : public UDPTransport {
57+
class MockThriftTransport : public ThriftTransport {
5958
public:
60-
MockUDPTransport(const net::IPAddress& ip,
59+
MockThriftTransport(const net::IPAddress& ip,
6160
int maxPacketSize,
62-
MockUDPClient::ExceptionType type)
63-
: UDPTransport(ip, maxPacketSize)
61+
MockUDPSender::ExceptionType type)
62+
: ThriftTransport(ip, maxPacketSize)
6463
{
65-
setClient(std::unique_ptr<utils::UDPClient>(
66-
new MockUDPClient(ip, maxPacketSize, type)));
64+
setClient(std::unique_ptr<utils::UDPSender>(
65+
new MockUDPSender(ip, maxPacketSize, type)));
6766
}
6867
};
6968

7069
} // anonymous namespace
7170

72-
TEST(UDPTransport, testManyMessages)
71+
TEST(ThriftTransport, testManyMessages)
7372
{
7473
const auto handle = testutils::TracerUtil::installGlobalTracer();
7574
const auto tracer =
7675
std::static_pointer_cast<const Tracer>(opentracing::Tracer::Global());
7776

78-
UDPTransport sender(handle->_mockAgent->spanServerAddress(), 0);
77+
ThriftTransport sender(handle->_mockAgent->spanServerAddress(), 0);
7978
constexpr auto kNumMessages = 2000;
8079
const auto logger = logging::consoleLogger();
8180
for (auto i = 0; i < kNumMessages; ++i) {
@@ -85,7 +84,7 @@ TEST(UDPTransport, testManyMessages)
8584
}
8685
}
8786

88-
TEST(UDPTransport, testExceptions)
87+
TEST(ThriftTransport, testExceptions)
8988
{
9089
const auto handle = testutils::TracerUtil::installGlobalTracer();
9190
const auto tracer =
@@ -94,15 +93,15 @@ TEST(UDPTransport, testExceptions)
9493
Span span(tracer);
9594
span.SetOperationName("test");
9695

97-
const MockUDPClient::ExceptionType exceptionTypes[] = {
98-
MockUDPClient::ExceptionType::kSystemError,
99-
MockUDPClient::ExceptionType::kException,
100-
MockUDPClient::ExceptionType::kString
96+
const MockUDPSender::ExceptionType exceptionTypes[] = {
97+
MockUDPSender::ExceptionType::kSystemError,
98+
MockUDPSender::ExceptionType::kException,
99+
MockUDPSender::ExceptionType::kString
101100
};
102101
for (auto type : exceptionTypes) {
103-
MockUDPTransport sender(net::IPAddress::v4("localhost", 0), 0, type);
104-
sender.append(span);
105-
ASSERT_THROW(sender.flush(), Transport::Exception);
102+
MockThriftTransport transport(net::IPAddress::v4("localhost", 0), 0, type);
103+
transport.append(span);
104+
ASSERT_THROW(transport.flush(), Transport::Exception);
106105
}
107106
}
108107

‎src/jaegertracing/net/Socket.h

-2
Original file line numberDiff line numberDiff line change
@@ -235,8 +235,6 @@ class Socket {
235235
resolveAddress(const std::string& host, int port, int family, int type);
236236
};
237237

238-
static constexpr auto kUDPPacketMaxLength = 65000;
239-
240238
} // namespace net
241239
} // namespace jaegertracing
242240

‎src/jaegertracing/net/http/Request.cpp

+8
Original file line numberDiff line numberDiff line change
@@ -15,13 +15,21 @@
1515
*/
1616

1717
#include "jaegertracing/net/http/Request.h"
18+
#include "jaegertracing/net/http/SocketReader.h"
1819

1920
#include <regex>
2021

2122
namespace jaegertracing {
2223
namespace net {
2324
namespace http {
2425

26+
Request Request::read(Socket & socket)
27+
{
28+
std::string response = SocketReader::read(socket);
29+
std::istringstream responseStream(response);
30+
return Request::parse(responseStream);
31+
}
32+
2533
Request Request::parse(std::istream& in)
2634
{
2735
const std::regex requestLinePattern(

‎src/jaegertracing/net/http/Request.h

+3
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
#define JAEGERTRACING_NET_HTTP_REQUEST_H
1919

2020
#include "jaegertracing/net/URI.h"
21+
#include "jaegertracing/net/Socket.h"
2122
#include "jaegertracing/net/http/Error.h"
2223
#include "jaegertracing/net/http/Header.h"
2324
#include "jaegertracing/net/http/Method.h"
@@ -30,6 +31,8 @@ class Request {
3031
public:
3132
static Request parse(std::istream& in);
3233

34+
static Request read(Socket& in);
35+
3336
Request()
3437
: _method()
3538
, _target()

‎src/jaegertracing/net/http/Response.cpp

+9-35
Original file line numberDiff line numberDiff line change
@@ -15,21 +15,8 @@
1515
*/
1616

1717
#include "jaegertracing/net/http/Response.h"
18+
#include "jaegertracing/net/http/SocketReader.h"
1819

19-
#ifdef WIN32
20-
#include <winsock2.h>
21-
#include <iphlpapi.h>
22-
#include <ws2tcpip.h>
23-
#include <windows.h>
24-
#elif UNIX
25-
#include <arpa/inet.h>
26-
#include <netdb.h>
27-
#include <sys/socket.h>
28-
#include <sys/types.h>
29-
#include <unistd.h>
30-
#endif
31-
32-
#include <iostream>
3320
#include <regex>
3421
#include <sstream>
3522
#include <stdexcept>
@@ -68,6 +55,13 @@ Response Response::parse(std::istream& in)
6855
return response;
6956
}
7057

58+
Response read(Socket & socket)
59+
{
60+
std::string response = SocketReader::read(socket);
61+
std::istringstream responseStream(response);
62+
return Response::parse(responseStream);
63+
}
64+
7165
Response get(const URI& uri)
7266
{
7367
Socket socket;
@@ -92,27 +86,7 @@ Response get(const URI& uri)
9286
throw std::system_error(errno, std::system_category(), oss.str());
9387
}
9488

95-
constexpr auto kBufferSize = 256;
96-
std::array<char, kBufferSize> buffer;
97-
#ifdef WIN32
98-
auto numRead = ::recv(socket.handle(), &buffer[0], buffer.size(), 0);
99-
#else
100-
auto numRead = ::read(socket.handle(), &buffer[0], buffer.size());
101-
#endif
102-
std::string response;
103-
while (numRead > 0) {
104-
response.append(&buffer[0], numRead);
105-
if (numRead < static_cast<int>(buffer.size())) {
106-
break;
107-
}
108-
#ifdef WIN32
109-
numRead = ::recv(socket.handle(), &buffer[0], buffer.size(), 0);
110-
#else
111-
numRead = ::read(socket.handle(), &buffer[0], buffer.size());
112-
#endif
113-
}
114-
std::istringstream responseStream(response);
115-
return Response::parse(responseStream);
89+
return read(socket);
11690
}
11791

11892
} // namespace http

‎src/jaegertracing/net/http/Response.h

+6
Original file line numberDiff line numberDiff line change
@@ -18,6 +18,7 @@
1818
#define JAEGERTRACING_NET_HTTP_RESPONSE_H
1919

2020
#include "jaegertracing/net/URI.h"
21+
#include "jaegertracing/net/Socket.h"
2122
#include "jaegertracing/net/http/Error.h"
2223
#include "jaegertracing/net/http/Header.h"
2324
#include "jaegertracing/net/http/Method.h"
@@ -57,6 +58,11 @@ class Response {
5758

5859
Response get(const URI& uri);
5960

61+
/**
62+
* Reads http response from a socket
63+
*/
64+
Response read(Socket& socket);
65+
6066
} // namespace http
6167
} // namespace net
6268
} // namespace jaegertracing

‎src/jaegertracing/net/http/SocketReader.h

+48
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
/*
2+
* Copyright (c) 2018, The Jaeger Authors.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#ifndef JAEGERTRACING_NET_HTTP_SOCKET_READER_H
18+
#define JAEGERTRACING_NET_HTTP_SOCKET_READER_H
19+
20+
#include "jaegertracing/net/Socket.h"
21+
22+
namespace jaegertracing {
23+
namespace net {
24+
namespace http {
25+
26+
class SocketReader {
27+
public:
28+
static std::string read(Socket& socket) {
29+
constexpr auto kBufferSize = 1024;
30+
std::array<char, kBufferSize> buffer{};
31+
auto numRead = ::recv(socket.handle(), &buffer[0], buffer.size(), 0);
32+
std::string response;
33+
while (numRead > 0) {
34+
response.append(&buffer[0], numRead);
35+
if (numRead < static_cast<int>(buffer.size())) {
36+
break;
37+
}
38+
numRead = ::recv(socket.handle(), &buffer[0], buffer.size(), 0);
39+
}
40+
return response;
41+
}
42+
};
43+
44+
} // namespace http
45+
} // namespace net
46+
} // namespace jaegertracing
47+
48+
#endif // JAEGERTRACING_NET_HTTP_SOCKET_READER_H

‎src/jaegertracing/reporters/Config.cpp

+1
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ namespace reporters {
2121

2222
constexpr int Config::kDefaultQueueSize;
2323
constexpr const char* Config::kDefaultLocalAgentHostPort;
24+
constexpr const char* Config::kDefaultEndpoint;
2425

2526
} // namespace reporters
2627
} // namespace jaegertracing

‎src/jaegertracing/reporters/Config.h

+18-5
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@
2222
#include <string>
2323

2424
#include "jaegertracing/Logging.h"
25-
#include "jaegertracing/UDPTransport.h"
25+
#include "jaegertracing/ThriftTransport.h"
2626
#include "jaegertracing/metrics/Metrics.h"
2727
#include "jaegertracing/reporters/CompositeReporter.h"
2828
#include "jaegertracing/reporters/LoggingReporter.h"
@@ -39,6 +39,7 @@ class Config {
3939

4040
static constexpr auto kDefaultQueueSize = 100;
4141
static constexpr auto kDefaultLocalAgentHostPort = "127.0.0.1:6831";
42+
static constexpr auto kDefaultEndpoint = "";
4243

4344
static Clock::duration defaultBufferFlushInterval()
4445
{
@@ -62,8 +63,10 @@ class Config {
6263
utils::yaml::findOrDefault<bool>(configYAML, "logSpans", false);
6364
const auto localAgentHostPort = utils::yaml::findOrDefault<std::string>(
6465
configYAML, "localAgentHostPort", "");
66+
const auto endpoint = utils::yaml::findOrDefault<std::string>(
67+
configYAML, "endpoint", "");
6568
return Config(
66-
queueSize, bufferFlushInterval, logSpans, localAgentHostPort);
69+
queueSize, bufferFlushInterval, logSpans, localAgentHostPort, endpoint);
6770
}
6871

6972
#endif // JAEGERTRACING_WITH_YAML_CPP
@@ -73,7 +76,7 @@ class Config {
7376
const Clock::duration& bufferFlushInterval =
7477
defaultBufferFlushInterval(),
7578
bool logSpans = false,
76-
const std::string& localAgentHostPort = kDefaultLocalAgentHostPort)
79+
const std::string& localAgentHostPort = kDefaultLocalAgentHostPort, const std::string& endpoint = kDefaultEndpoint)
7780
: _queueSize(queueSize > 0 ? queueSize : kDefaultQueueSize)
7881
, _bufferFlushInterval(bufferFlushInterval.count() > 0
7982
? bufferFlushInterval
@@ -82,15 +85,19 @@ class Config {
8285
, _localAgentHostPort(localAgentHostPort.empty()
8386
? kDefaultLocalAgentHostPort
8487
: localAgentHostPort)
88+
, _endpoint(endpoint)
8589
{
8690
}
8791

8892
std::unique_ptr<Reporter> makeReporter(const std::string& serviceName,
8993
logging::Logger& logger,
9094
metrics::Metrics& metrics) const
9195
{
92-
std::unique_ptr<UDPTransport> sender(
93-
new UDPTransport(net::IPAddress::v4(_localAgentHostPort), 0));
96+
97+
std::unique_ptr<ThriftTransport> sender(
98+
!_endpoint.empty()
99+
? new ThriftTransport(net::URI::parse(_endpoint), 0)
100+
: new ThriftTransport(net::IPAddress::v4(_localAgentHostPort), 0));
94101
std::unique_ptr<RemoteReporter> remoteReporter(
95102
new RemoteReporter(_bufferFlushInterval,
96103
_queueSize,
@@ -120,11 +127,17 @@ class Config {
120127
return _localAgentHostPort;
121128
}
122129

130+
const std::string& endpoint() const
131+
{
132+
return _endpoint;
133+
}
134+
123135
private:
124136
int _queueSize;
125137
Clock::duration _bufferFlushInterval;
126138
bool _logSpans;
127139
std::string _localAgentHostPort;
140+
std::string _endpoint;
128141
};
129142

130143
} // namespace reporters

‎src/jaegertracing/reporters/ConfigTest.cpp

+52
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,52 @@
1+
/*
2+
* Copyright (c) 2019, The Jaeger Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#include <gtest/gtest.h>
18+
19+
#include <string>
20+
21+
#include <yaml-cpp/yaml.h>
22+
23+
#include "jaegertracing/reporters/Config.h"
24+
25+
namespace jaegertracing {
26+
namespace reporters {
27+
28+
TEST(Config, testLoadConfig)
29+
{
30+
std::string yamlConfig =
31+
"disabled: false\n"
32+
"reporter:\n"
33+
" logSpans: true\n"
34+
" bufferFlushInterval: 88\n"
35+
" localAgentHostPort: ahost:22\n"
36+
" endpoint: http://somehost:33/api/traces\n"
37+
"sampler:\n"
38+
" type: const\n"
39+
" param: 1";
40+
41+
auto configYAML = YAML::Load(yamlConfig);
42+
auto reporterYAML = configYAML["reporter"];
43+
auto config = jaegertracing::reporters::Config::parse(reporterYAML);
44+
45+
ASSERT_EQ(true, config.logSpans());
46+
ASSERT_EQ(std::chrono::seconds(88), config.bufferFlushInterval());
47+
ASSERT_EQ(std::string("ahost:22"), config.localAgentHostPort());
48+
ASSERT_EQ(std::string("http://somehost:33/api/traces"), config.endpoint());
49+
}
50+
51+
} // namespace reporters
52+
} // namespace jaegertracing

‎src/jaegertracing/testutils/MockAgent.cpp

+4-4
Original file line numberDiff line numberDiff line change
@@ -29,7 +29,7 @@
2929
#include "jaegertracing/net/http/Response.h"
3030
#include "jaegertracing/samplers/RemoteSamplingJSON.h"
3131
#include "jaegertracing/utils/ErrorUtil.h"
32-
#include "jaegertracing/utils/UDPClient.h"
32+
#include "jaegertracing/utils/UDPSender.h"
3333

3434
#ifdef _MSC_VER
3535
#pragma warning(disable : 4267)
@@ -111,17 +111,17 @@ void MockAgent::serveUDP(std::promise<void>& started)
111111
agent::thrift::AgentProcessor handler(iface);
112112
TCompactProtocolFactory protocolFactory;
113113
std::shared_ptr<TMemoryBuffer> trans(
114-
new TMemoryBuffer(net::kUDPPacketMaxLength));
114+
new TMemoryBuffer(utils::UDPSender::kUDPPacketMaxLength));
115115

116116
// Notify main thread that setup is done.
117117
_servingUDP = true;
118118
started.set_value();
119119

120-
std::array<uint8_t, net::kUDPPacketMaxLength> buffer;
120+
std::array<uint8_t, utils::UDPSender::kUDPPacketMaxLength> buffer;
121121
while (isServingUDP()) {
122122
try {
123123
const auto numRead =
124-
_transport.read(&buffer[0], net::kUDPPacketMaxLength);
124+
_transport.read(&buffer[0], utils::UDPSender::kUDPPacketMaxLength);
125125
if (numRead > 0) {
126126
trans->write(&buffer[0], numRead);
127127
auto protocol = protocolFactory.getProtocol(trans);

‎src/jaegertracing/testutils/MockAgent.h

+4-4
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,7 @@
2525
#include "jaegertracing/testutils/TUDPTransport.h"
2626
#include "jaegertracing/thrift-gen/Agent.h"
2727
#include "jaegertracing/thrift-gen/jaeger_types.h"
28-
#include "jaegertracing/utils/UDPClient.h"
28+
#include "jaegertracing/utils/UDPSender.h"
2929
#include <atomic>
3030
#include <future>
3131
#include <memory>
@@ -104,10 +104,10 @@ class MockAgent : public agent::thrift::AgentIf,
104104

105105
net::IPAddress spanServerAddress() const { return _transport.addr(); }
106106

107-
std::unique_ptr<agent::thrift::AgentIf> spanServerClient()
107+
std::unique_ptr<utils::Sender> spanServerClient()
108108
{
109-
return std::unique_ptr<agent::thrift::AgentIf>(
110-
new utils::UDPClient(spanServerAddress(), 0));
109+
return std::unique_ptr<utils::Sender>(
110+
new utils::UDPSender(spanServerAddress(), 0));
111111
}
112112

113113
net::IPAddress samplingServerAddress() const { return _httpAddress; }

‎src/jaegertracing/testutils/TUDPTransport.h

+1-1
Original file line numberDiff line numberDiff line change
@@ -27,7 +27,7 @@
2727

2828
#include <thrift/transport/TVirtualTransport.h>
2929

30-
#include "jaegertracing/utils/UDPClient.h"
30+
#include "jaegertracing/utils/UDPSender.h"
3131

3232
namespace jaegertracing {
3333
namespace testutils {

‎src/jaegertracing/testutils/TracerUtil.cpp

+82-62
Original file line numberDiff line numberDiff line change
@@ -1,62 +1,82 @@
1-
/*
2-
* Copyright (c) 2017 Uber Technologies, Inc.
3-
*
4-
* Licensed under the Apache License, Version 2.0 (the "License");
5-
* you may not use this file except in compliance with the License.
6-
* You may obtain a copy of the License at
7-
*
8-
* http://www.apache.org/licenses/LICENSE-2.0
9-
*
10-
* Unless required by applicable law or agreed to in writing, software
11-
* distributed under the License is distributed on an "AS IS" BASIS,
12-
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13-
* See the License for the specific language governing permissions and
14-
* limitations under the License.
15-
*/
16-
17-
#include "jaegertracing/testutils/TracerUtil.h"
18-
#include "jaegertracing/Config.h"
19-
#include "jaegertracing/Logging.h"
20-
#include "jaegertracing/Tracer.h"
21-
#include "jaegertracing/baggage/RestrictionsConfig.h"
22-
#include "jaegertracing/net/IPAddress.h"
23-
#include "jaegertracing/propagation/HeadersConfig.h"
24-
#include "jaegertracing/reporters/Config.h"
25-
#include "jaegertracing/samplers/Config.h"
26-
#include <memory>
27-
#include <ostream>
28-
#include <string>
29-
30-
namespace jaegertracing {
31-
namespace testutils {
32-
namespace TracerUtil {
33-
34-
std::shared_ptr<ResourceHandle> installGlobalTracer()
35-
{
36-
std::unique_ptr<ResourceHandle> handle(new ResourceHandle());
37-
handle->_mockAgent->start();
38-
std::ostringstream samplingServerURLStream;
39-
samplingServerURLStream
40-
<< "http://" << handle->_mockAgent->samplingServerAddress().authority();
41-
Config config(
42-
false,
43-
samplers::Config("const",
44-
1,
45-
samplingServerURLStream.str(),
46-
0,
47-
samplers::Config::Clock::duration()),
48-
reporters::Config(0,
49-
reporters::Config::Clock::duration(),
50-
false,
51-
handle->_mockAgent->spanServerAddress().authority()),
52-
propagation::HeadersConfig(),
53-
baggage::RestrictionsConfig());
54-
55-
auto tracer = Tracer::make("test-service", config, logging::nullLogger());
56-
opentracing::Tracer::InitGlobal(tracer);
57-
return std::move(handle);
58-
}
59-
60-
} // namespace TracerUtil
61-
} // namespace testutils
62-
} // namespace jaegertracing
1+
/*
2+
* Copyright (c) 2017 Uber Technologies, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#include "jaegertracing/testutils/TracerUtil.h"
18+
#include "jaegertracing/Config.h"
19+
#include "jaegertracing/Logging.h"
20+
#include "jaegertracing/Tracer.h"
21+
#include "jaegertracing/baggage/RestrictionsConfig.h"
22+
#include "jaegertracing/net/IPAddress.h"
23+
#include "jaegertracing/propagation/HeadersConfig.h"
24+
#include "jaegertracing/reporters/Config.h"
25+
#include "jaegertracing/samplers/Config.h"
26+
#include <memory>
27+
#include <ostream>
28+
#include <string>
29+
30+
namespace jaegertracing {
31+
namespace testutils {
32+
namespace TracerUtil {
33+
34+
std::shared_ptr<ResourceHandle> installGlobalTracer()
35+
{
36+
std::unique_ptr<ResourceHandle> handle(new ResourceHandle());
37+
handle->_mockAgent->start();
38+
std::ostringstream samplingServerURLStream;
39+
samplingServerURLStream
40+
<< "http://" << handle->_mockAgent->samplingServerAddress().authority();
41+
Config config(
42+
false,
43+
samplers::Config("const",
44+
1,
45+
samplingServerURLStream.str(),
46+
0,
47+
samplers::Config::Clock::duration()),
48+
reporters::Config(0,
49+
reporters::Config::Clock::duration(),
50+
false,
51+
handle->_mockAgent->spanServerAddress().authority()),
52+
propagation::HeadersConfig(),
53+
baggage::RestrictionsConfig());
54+
55+
auto tracer = Tracer::make("test-service", config, logging::nullLogger());
56+
opentracing::Tracer::InitGlobal(tracer);
57+
return std::move(handle);
58+
}
59+
60+
std::shared_ptr<opentracing::Tracer> buildTracer(const std::string& endpoint)
61+
{
62+
std::ostringstream samplingServerURLStream;
63+
Config config(
64+
false,
65+
samplers::Config("const",
66+
1,
67+
samplingServerURLStream.str(),
68+
0,
69+
samplers::Config::Clock::duration()),
70+
reporters::Config(0,
71+
std::chrono::milliseconds(100),
72+
false, "", endpoint),
73+
propagation::HeadersConfig(),
74+
baggage::RestrictionsConfig());
75+
76+
auto tracer = Tracer::make("test-service", config, logging::nullLogger());
77+
return tracer;
78+
}
79+
80+
} // namespace TracerUtil
81+
} // namespace testutils
82+
} // namespace jaegertracing

‎src/jaegertracing/testutils/TracerUtil.h

+50-47
Original file line numberDiff line numberDiff line change
@@ -1,47 +1,50 @@
1-
/*
2-
* Copyright (c) 2017 Uber Technologies, Inc.
3-
*
4-
* Licensed under the Apache License, Version 2.0 (the "License");
5-
* you may not use this file except in compliance with the License.
6-
* You may obtain a copy of the License at
7-
*
8-
* http://www.apache.org/licenses/LICENSE-2.0
9-
*
10-
* Unless required by applicable law or agreed to in writing, software
11-
* distributed under the License is distributed on an "AS IS" BASIS,
12-
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13-
* See the License for the specific language governing permissions and
14-
* limitations under the License.
15-
*/
16-
17-
#ifndef JAEGERTRACING_TESTUTILS_TRACERUTIL_H
18-
#define JAEGERTRACING_TESTUTILS_TRACERUTIL_H
19-
20-
#include "jaegertracing/Tracer.h"
21-
#include "jaegertracing/testutils/MockAgent.h"
22-
23-
namespace jaegertracing {
24-
namespace testutils {
25-
namespace TracerUtil {
26-
27-
struct ResourceHandle {
28-
ResourceHandle()
29-
: _mockAgent(testutils::MockAgent::make())
30-
{
31-
}
32-
33-
~ResourceHandle()
34-
{
35-
opentracing::Tracer::InitGlobal(opentracing::MakeNoopTracer());
36-
}
37-
38-
std::shared_ptr<MockAgent> _mockAgent;
39-
};
40-
41-
std::shared_ptr<ResourceHandle> installGlobalTracer();
42-
43-
} // namespace TracerUtil
44-
} // namespace testutils
45-
} // namespace jaegertracing
46-
47-
#endif // JAEGERTRACING_TESTUTILS_TRACERUTIL_H
1+
/*
2+
* Copyright (c) 2017 Uber Technologies, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#ifndef JAEGERTRACING_TESTUTILS_TRACERUTIL_H
18+
#define JAEGERTRACING_TESTUTILS_TRACERUTIL_H
19+
20+
#include <string>
21+
22+
#include "jaegertracing/Tracer.h"
23+
#include "jaegertracing/testutils/MockAgent.h"
24+
25+
namespace jaegertracing {
26+
namespace testutils {
27+
namespace TracerUtil {
28+
29+
struct ResourceHandle {
30+
ResourceHandle()
31+
: _mockAgent(testutils::MockAgent::make())
32+
{
33+
}
34+
35+
~ResourceHandle()
36+
{
37+
opentracing::Tracer::InitGlobal(opentracing::MakeNoopTracer());
38+
}
39+
40+
std::shared_ptr<MockAgent> _mockAgent;
41+
};
42+
43+
std::shared_ptr<ResourceHandle> installGlobalTracer();
44+
std::shared_ptr<opentracing::Tracer> buildTracer(const std::string& endpoint);
45+
46+
} // namespace TracerUtil
47+
} // namespace testutils
48+
} // namespace jaegertracing
49+
50+
#endif // JAEGERTRACING_TESTUTILS_TRACERUTIL_H

‎src/jaegertracing/utils/HttpSender.cpp

+43
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,43 @@
1+
/*
2+
* Copyright (c) 2019, The Jaeger Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#include "jaegertracing/utils/HttpSender.h"
18+
#include <thrift/protocol/TBinaryProtocol.h>
19+
#include <thrift/protocol/TProtocol.h>
20+
21+
namespace jaegertracing {
22+
namespace utils {
23+
24+
HttpSender::HttpSender(const net::URI& endpoint, int maxPacketSize)
25+
: Sender(maxPacketSize == 0 ? kHttpPacketMaxLength : maxPacketSize)
26+
, _buffer(new apache::thrift::transport::TMemoryBuffer(_maxPacketSize))
27+
, _serverAddr(net::IPAddress::v4(endpoint._host, endpoint._port))
28+
, _httpClient(new ::apache::thrift::transport::THttpClient(
29+
_buffer, endpoint._host, endpoint._path + "?format=jaeger.thrift"))
30+
{
31+
using TProtocolFactory = apache::thrift::protocol::TProtocolFactory;
32+
using TBinaryProtocolFactory =
33+
apache::thrift::protocol::TBinaryProtocolFactory;
34+
35+
_socket.open(AF_INET, SOCK_STREAM);
36+
_socket.connect(_serverAddr);
37+
std::shared_ptr<TProtocolFactory> protocolFactory(
38+
new TBinaryProtocolFactory());
39+
_protocol = protocolFactory->getProtocol(_httpClient);
40+
}
41+
42+
} // namespace utils
43+
} // namespace jaegertracing

‎src/jaegertracing/utils/HttpSender.h

+108
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,108 @@
1+
/*
2+
* Copyright (c) 2019, The Jaeger Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#ifndef JAEGERTRACING_UTILS_UDPCLIENT5_H
18+
#define JAEGERTRACING_UTILS_UDPCLIENT5_H
19+
20+
#include <iostream>
21+
#include <sstream>
22+
#include <stdexcept>
23+
#include <system_error>
24+
25+
#include "jaegertracing/Compilers.h"
26+
27+
#include "jaegertracing/utils/Sender.h"
28+
29+
#include "jaegertracing/net/IPAddress.h"
30+
#include "jaegertracing/net/Socket.h"
31+
#include "jaegertracing/net/http/Response.h"
32+
33+
#include <thrift/protocol/TBinaryProtocol.h>
34+
#include <thrift/transport/TBufferTransports.h>
35+
#include <thrift/transport/THttpClient.h>
36+
37+
namespace jaegertracing {
38+
namespace utils {
39+
40+
class HttpSender : public Sender {
41+
public:
42+
HttpSender(const net::URI& endpoint, int maxPacketSize);
43+
44+
~HttpSender() { close(); }
45+
46+
void emitBatch(const thrift::Batch& batch) override
47+
{
48+
// Resets the buffer to write a new batch
49+
_buffer->resetBuffer();
50+
51+
// Does the serialisation to Thrift
52+
auto oprot = _protocol.get();
53+
batch.write(oprot);
54+
oprot->writeMessageEnd();
55+
oprot->getTransport()->writeEnd();
56+
oprot->getTransport()->flush();
57+
58+
uint8_t* data = nullptr;
59+
uint32_t size = 0;
60+
_buffer->getBuffer(&data, &size);
61+
if (static_cast<int>(size) > _maxPacketSize) {
62+
std::ostringstream oss;
63+
oss << "Data does not fit within one HTTP message"
64+
", size "
65+
<< size << ", max " << _maxPacketSize << ", spans "
66+
<< batch.spans.size();
67+
throw std::logic_error(oss.str());
68+
}
69+
70+
// Sends the HTTP message
71+
const auto numWritten = ::send(_socket.handle(), reinterpret_cast<char*>(data), sizeof(uint8_t) * size, 0);
72+
73+
if (static_cast<unsigned>(numWritten) != size) {
74+
std::ostringstream oss;
75+
oss << "Failed to write message, numWritten=" << numWritten << ", size=" << size;
76+
throw std::system_error(errno, std::system_category(), oss.str());
77+
}
78+
79+
// Waits for response. Check that the server acknowledged
80+
// and returned a green status [200, 201, 202, 203, 204]
81+
net::http::Response response = net::http::read(_socket);
82+
if (response.statusCode() < 200 && response.statusCode() > 204) {
83+
std::ostringstream oss;
84+
oss << "Failed to write message, HTTP error " << response.statusCode() << response.reason();
85+
throw std::system_error(errno, std::system_category(), oss.str());
86+
}
87+
}
88+
89+
std::unique_ptr<apache::thrift::protocol::TProtocolFactory>
90+
protocolFactory() const override
91+
{
92+
return std::unique_ptr<apache::thrift::protocol::TProtocolFactory>(
93+
new apache::thrift::protocol::TBinaryProtocolFactory());
94+
}
95+
96+
private:
97+
std::shared_ptr<apache::thrift::transport::TMemoryBuffer> _buffer;
98+
net::IPAddress _serverAddr;
99+
std::shared_ptr<::apache::thrift::transport::THttpClient> _httpClient;
100+
std::shared_ptr<apache::thrift::protocol::TProtocol> _protocol;
101+
102+
static constexpr auto kHttpPacketMaxLength = 1048576;
103+
};
104+
105+
} // namespace utils
106+
} // namespace jaegertracing
107+
108+
#endif // JAEGERTRACING_UTILS_UDPCLIENT_H

‎src/jaegertracing/utils/HttpSenderTest.cpp

+114
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,114 @@
1+
/*
2+
* Copyright (c) 2017 Uber Technologies, Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#include "jaegertracing/net/IPAddress.h"
18+
#include "jaegertracing/testutils/TracerUtil.h"
19+
#include <gtest/gtest.h>
20+
#include <iterator>
21+
#include <memory>
22+
#include <opentracing/string_view.h>
23+
#include <opentracing/tracer.h>
24+
#include <stdexcept>
25+
#include <string>
26+
#include <utility>
27+
28+
#include "jaegertracing/net/Socket.h"
29+
#include "jaegertracing/net/http/Request.h"
30+
#include "jaegertracing/net/http/Response.h"
31+
#include <future>
32+
#include <thread>
33+
34+
namespace jaegertracing {
35+
36+
namespace utils {
37+
38+
TEST(HttpSender, testBigMessage)
39+
{
40+
net::IPAddress serverAddr;
41+
std::promise<void> started;
42+
43+
std::string target;
44+
net::http::Method method;
45+
std::string contentType;
46+
std::string acceptType;
47+
48+
std::thread serverThread(
49+
[&serverAddr, &started, &target, &method, &contentType, &acceptType]() {
50+
net::Socket socket;
51+
socket.open(AF_INET, SOCK_STREAM);
52+
socket.bind(net::IPAddress::v4("127.0.0.1", 0));
53+
::sockaddr_storage addrStorage;
54+
::socklen_t addrLen = sizeof(addrStorage);
55+
const auto returnCode =
56+
::getsockname(socket.handle(),
57+
reinterpret_cast<::sockaddr*>(&addrStorage),
58+
&addrLen);
59+
ASSERT_EQ(0, returnCode);
60+
serverAddr = net::IPAddress(addrStorage, addrLen);
61+
62+
// Listening to the port
63+
socket.listen();
64+
65+
// Unblocking the client
66+
started.set_value();
67+
68+
// Waiting for the client to connect
69+
auto clientSocket = socket.accept();
70+
71+
net::http::Request request = net::http::Request::read(clientSocket);
72+
73+
target = request.target();
74+
method = request.method();
75+
76+
for (auto tata : request.headers()) {
77+
if (tata.key() == "Content-Type") {
78+
contentType = tata.value();
79+
}
80+
81+
if (tata.key() == "Accept") {
82+
acceptType = tata.value();
83+
}
84+
}
85+
86+
std::string answer(
87+
"HTTP/1.1 202 Accepted\r\nDate: Tue, 10 Sep 2019 09:03:26 "
88+
"GMT\r\nContent-Length: 0\r\n\r\n");
89+
90+
const auto numWritten =
91+
::send(clientSocket.handle(), answer.c_str(), answer.size(), 0);
92+
});
93+
94+
started.get_future().wait();
95+
96+
std::ostringstream oss;
97+
oss << "http://127.0.0.1:" << serverAddr.port() << "/api/traces";
98+
std::shared_ptr<opentracing::Tracer> tracer =
99+
::jaegertracing::testutils::TracerUtil::buildTracer(oss.str());
100+
101+
{
102+
auto span1 = tracer->StartSpan("tracedFunction");
103+
}
104+
105+
serverThread.join();
106+
107+
ASSERT_EQ(std::string("/api/traces?format=jaeger.thrift"), target);
108+
ASSERT_EQ(net::http::Method::POST, method);
109+
ASSERT_EQ(std::string("application/x-thrift"), contentType);
110+
ASSERT_EQ(std::string("application/x-thrift"), acceptType);
111+
}
112+
113+
} // namespace utils
114+
} // namespace jaegertracing

‎src/jaegertracing/utils/Sender.h

+51
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
/*
2+
* Copyright (c) 2019, The Jaeger Authors
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
17+
#ifndef JAEGERTRACING_UTILS_SENDER_H
18+
#define JAEGERTRACING_UTILS_SENDER_H
19+
20+
#include "jaegertracing/net/Socket.h"
21+
#include "jaegertracing/thrift-gen/jaeger_types.h"
22+
23+
namespace jaegertracing {
24+
namespace utils {
25+
26+
class Sender {
27+
public:
28+
Sender(int maxPacketSize)
29+
: _maxPacketSize(maxPacketSize)
30+
{
31+
}
32+
33+
virtual ~Sender() { close(); }
34+
35+
virtual void emitBatch(const thrift::Batch& batch) = 0;
36+
37+
int maxPacketSize() const { return _maxPacketSize; }
38+
39+
void close() { _socket.close(); }
40+
41+
virtual std::unique_ptr< apache::thrift::protocol::TProtocolFactory > protocolFactory() const = 0;
42+
43+
protected:
44+
int _maxPacketSize;
45+
net::Socket _socket;
46+
};
47+
48+
} // namespace utils
49+
} // namespace jaegertracing
50+
51+
#endif // JAEGERTRACING_UTILS_SENDER_H

‎src/jaegertracing/utils/UDPClient.cpp ‎src/jaegertracing/utils/UDPSender.cpp

+3-4
Original file line numberDiff line numberDiff line change
@@ -14,18 +14,17 @@
1414
* limitations under the License.
1515
*/
1616

17-
#include "jaegertracing/utils/UDPClient.h"
17+
#include "jaegertracing/utils/UDPSender.h"
1818
#include <thrift/protocol/TCompactProtocol.h>
1919
#include <thrift/protocol/TProtocol.h>
2020

2121
namespace jaegertracing {
2222
namespace utils {
2323

24-
UDPClient::UDPClient(const net::IPAddress& serverAddr, int maxPacketSize)
25-
: _maxPacketSize(maxPacketSize == 0 ? net::kUDPPacketMaxLength
24+
UDPSender::UDPSender(const net::IPAddress& serverAddr, int maxPacketSize)
25+
: Sender(maxPacketSize == 0 ? kUDPPacketMaxLength
2626
: maxPacketSize)
2727
, _buffer(new apache::thrift::transport::TMemoryBuffer(_maxPacketSize))
28-
, _socket()
2928
, _serverAddr(serverAddr)
3029
, _client()
3130
{

‎src/jaegertracing/utils/UDPClient.h ‎src/jaegertracing/utils/UDPSender.h

+9-10
Original file line numberDiff line numberDiff line change
@@ -27,21 +27,22 @@
2727
#include <thrift/protocol/TCompactProtocol.h>
2828
#include <thrift/transport/TBufferTransports.h>
2929

30+
#include "jaegertracing/utils/Sender.h"
31+
3032
#include "jaegertracing/net/IPAddress.h"
31-
#include "jaegertracing/net/Socket.h"
3233
#include "jaegertracing/thrift-gen/Agent.h"
3334

3435
namespace jaegertracing {
3536
namespace utils {
3637

37-
class UDPClient : public agent::thrift::AgentIf {
38+
class UDPSender : public Sender {
3839
public:
39-
UDPClient(const net::IPAddress& serverAddr, int maxPacketSize);
40+
static constexpr auto kUDPPacketMaxLength = 65000;
4041

41-
~UDPClient() { close(); }
42+
UDPSender(const net::IPAddress& serverAddr, int maxPacketSize);
4243

4344
void emitZipkinBatch(
44-
const std::vector<twitter::zipkin::thrift::Span>& spans) override
45+
const std::vector<twitter::zipkin::thrift::Span>& spans)
4546
{
4647
throw std::logic_error("emitZipkinBatch not implemented");
4748
}
@@ -71,14 +72,12 @@ class UDPClient : public agent::thrift::AgentIf {
7172
}
7273
}
7374

74-
int maxPacketSize() const { return _maxPacketSize; }
75-
76-
void close() { _socket.close(); }
75+
std::unique_ptr< apache::thrift::protocol::TProtocolFactory > protocolFactory() const override {
76+
return std::unique_ptr<apache::thrift::protocol::TProtocolFactory>(new apache::thrift::protocol::TCompactProtocolFactory());
77+
}
7778

7879
private:
79-
int _maxPacketSize;
8080
std::shared_ptr<apache::thrift::transport::TMemoryBuffer> _buffer;
81-
net::Socket _socket;
8281
net::IPAddress _serverAddr;
8382
std::unique_ptr<agent::thrift::AgentClient> _client;
8483
};

‎src/jaegertracing/utils/UDPClientTest.cpp ‎src/jaegertracing/utils/UDPSenderTest.cpp

+5-5
Original file line numberDiff line numberDiff line change
@@ -18,7 +18,7 @@
1818
#include "jaegertracing/net/Socket.h"
1919
#include "jaegertracing/thrift-gen/jaeger_types.h"
2020
#include "jaegertracing/thrift-gen/zipkincore_types.h"
21-
#include "jaegertracing/utils/UDPClient.h"
21+
#include "jaegertracing/utils/UDPSender.h"
2222
#include <future>
2323
#include <gtest/gtest.h>
2424
#include <stdexcept>
@@ -28,7 +28,7 @@
2828
namespace jaegertracing {
2929
namespace utils {
3030

31-
TEST(UDPClient, testZipkinMessage)
31+
TEST(UDPSender, testZipkinMessage)
3232
{
3333
net::IPAddress serverAddr;
3434
std::promise<void> started;
@@ -48,13 +48,13 @@ TEST(UDPClient, testZipkinMessage)
4848
});
4949

5050
started.get_future().wait();
51-
UDPClient udpClient(serverAddr, 0);
51+
UDPSender udpClient(serverAddr, 0);
5252
using ZipkinBatch = std::vector<twitter::zipkin::thrift::Span>;
5353
ASSERT_THROW(udpClient.emitZipkinBatch(ZipkinBatch()), std::logic_error);
5454
serverThread.join();
5555
}
5656

57-
TEST(UDPClient, testBigMessage)
57+
TEST(UDPSender, testBigMessage)
5858
{
5959
net::IPAddress serverAddr;
6060
std::promise<void> started;
@@ -74,7 +74,7 @@ TEST(UDPClient, testBigMessage)
7474
});
7575

7676
started.get_future().wait();
77-
UDPClient udpClient(serverAddr, 1);
77+
UDPSender udpClient(serverAddr, 1);
7878
ASSERT_THROW(udpClient.emitBatch(thrift::Batch()), std::logic_error);
7979
serverThread.join();
8080
}

0 commit comments

Comments
 (0)
Please sign in to comment.