ceph/src/jaegertracing/jaeger-client-cpp/crossdock/Server.cpp

   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 "Server.h"
  18
  19 #include <atomic>
  20 #include <cstdlib>
  21 #include <future>
  22 #include <sstream>
  23 #include <thread>
  24
  25 #include <nlohmann/json.hpp>
  26
  27 #include "jaegertracing/Tracer.h"
  28 #include "jaegertracing/net/IPAddress.h"
  29 #include "jaegertracing/net/Socket.h"
  30 #include "jaegertracing/net/http/Request.h"
  31 #include "jaegertracing/net/http/Response.h"
  32
  33 namespace jaegertracing {
  34 namespace crossdock {
  35 namespace thrift {
  36
  37 #define JSON_FROM_FIELD(var, field)                                            \
  38     {                                                                          \
  39         json[#field] = var.field;                                              \
  40     }
  41
  42 #define FIELD_FROM_JSON(var, field)                                            \
  43     {                                                                          \
  44         var.__set_##field(json.at(#field));                                    \
  45     }
  46
  47 void to_json(nlohmann::json& json, const Transport::type& transport)
  48 {
  49     json = _Transport_VALUES_TO_NAMES.at(static_cast<int>(transport));
  50 }
  51
  52 void from_json(const nlohmann::json& json, Transport::type& transport)
  53 {
  54     const auto str = json.get<std::string>();
  55     if (str == "HTTP") {
  56         transport = Transport::HTTP;
  57         return;
  58     }
  59     if (str == "TCHANNEL") {
  60         transport = Transport::TCHANNEL;
  61         return;
  62     }
  63     if (str == "DUMMY") {
  64         transport = Transport::DUMMY;
  65         return;
  66     }
  67     std::ostringstream oss;
  68     oss << "Invalid transport value " << str;
  69     throw std::invalid_argument(oss.str());
  70 }
  71
  72 void to_json(nlohmann::json& json, const Downstream& downstream)
  73 {
  74     JSON_FROM_FIELD(downstream, serviceName);
  75     JSON_FROM_FIELD(downstream, serverRole);
  76     JSON_FROM_FIELD(downstream, host);
  77     JSON_FROM_FIELD(downstream, port);
  78     JSON_FROM_FIELD(downstream, transport);
  79     if (downstream.downstream) {
  80         json["downstream"] = *downstream.downstream;
  81     }
  82 }
  83
  84 void from_json(const nlohmann::json& json, Downstream& downstream)
  85 {
  86     FIELD_FROM_JSON(downstream, serviceName);
  87     FIELD_FROM_JSON(downstream, serverRole);
  88     FIELD_FROM_JSON(downstream, host);
  89     FIELD_FROM_JSON(downstream, port);
  90     downstream.__set_transport(json.at("transport").get<Transport::type>());
  91     auto itr = json.find("downstream");
  92     if (itr != std::end(json) && !itr->is_null()) {
  93         downstream.__set_downstream(
  94             std::make_shared<Downstream>(itr->get<Downstream>()));
  95     }
  96 }
  97
  98 void to_json(nlohmann::json& json, const StartTraceRequest& request)
  99 {
 100     JSON_FROM_FIELD(request, serverRole);
 101     JSON_FROM_FIELD(request, sampled);
 102     JSON_FROM_FIELD(request, baggage);
 103     JSON_FROM_FIELD(request, downstream);
 104 }
 105
 106 void from_json(const nlohmann::json& json, StartTraceRequest& request)
 107 {
 108     FIELD_FROM_JSON(request, serverRole);
 109     FIELD_FROM_JSON(request, sampled);
 110     FIELD_FROM_JSON(request, baggage);
 111     FIELD_FROM_JSON(request, downstream);
 112 }
 113
 114 void to_json(nlohmann::json& json, const JoinTraceRequest& request)
 115 {
 116     JSON_FROM_FIELD(request, serverRole);
 117     if (request.__isset.downstream) {
 118         json["downstream"] = request.downstream;
 119     }
 120 }
 121
 122 void from_json(const nlohmann::json& json, JoinTraceRequest& request)
 123 {
 124     FIELD_FROM_JSON(request, serverRole);
 125     auto itr = json.find("downstream");
 126     if (itr != std::end(json) && !itr->is_null()) {
 127         request.__set_downstream(itr->get<Downstream>());
 128     }
 129 }
 130
 131 void to_json(nlohmann::json& json, const ObservedSpan& observedSpan)
 132 {
 133     JSON_FROM_FIELD(observedSpan, traceId);
 134     JSON_FROM_FIELD(observedSpan, sampled);
 135     JSON_FROM_FIELD(observedSpan, baggage);
 136 }
 137
 138 void from_json(const nlohmann::json& json, ObservedSpan& observedSpan)
 139 {
 140     FIELD_FROM_JSON(observedSpan, traceId);
 141     FIELD_FROM_JSON(observedSpan, sampled);
 142     FIELD_FROM_JSON(observedSpan, baggage);
 143 }
 144
 145 void to_json(nlohmann::json& json, const TraceResponse& response)
 146 {
 147     if (response.__isset.span) {
 148         JSON_FROM_FIELD(response, span);
 149     }
 150     if (response.downstream) {
 151         json["downstream"] = *response.downstream;
 152     }
 153     JSON_FROM_FIELD(response, notImplementedError);
 154 }
 155
 156 void from_json(const nlohmann::json& json, TraceResponse& response)
 157 {
 158     auto itr = json.find("span");
 159     if (itr != std::end(json) && !itr->is_null()) {
 160         response.__set_span(itr->get<ObservedSpan>());
 161     }
 162     itr = json.find("downstream");
 163     if (itr != std::end(json) && !itr->is_null()) {
 164         response.__set_downstream(
 165             std::make_shared<TraceResponse>(itr->get<TraceResponse>()));
 166     }
 167     FIELD_FROM_JSON(response, notImplementedError);
 168 }
 169
 170 #undef FIELD_FROM_JSON
 171 #undef JSON_FROM_FIELD
 172
 173 }  // namespace thrift
 174
 175 namespace {
 176
 177 constexpr auto kBaggageKey = "crossdock-baggage-key";
 178 constexpr auto kDefaultTracerServiceName = "crossdock-cpp";
 179
 180 std::string escape(const std::string& str)
 181 {
 182     std::string result;
 183     result.reserve(str.size());
 184     for (auto&& ch : str) {
 185         switch (ch) {
 186         case '\n': {
 187             result += "\\n";
 188         } break;
 189         case '\r': {
 190             result += "\\r";
 191         } break;
 192         default: {
 193             result += ch;
 194         } break;
 195         }
 196     }
 197     return result;
 198 }
 199
 200 std::string bufferedRead(net::Socket& socket)
 201 {
 202     constexpr auto kBufferSize = 256;
 203     std::array<char, kBufferSize> buffer;
 204     std::string data;
 205     auto numRead = ::read(socket.handle(), &buffer[0], buffer.size());
 206     data.append(&buffer[0], numRead);
 207     while (numRead == kBufferSize) {
 208         numRead = ::read(socket.handle(), &buffer[0], buffer.size());
 209         data.append(&buffer[0], numRead);
 210     }
 211     return data;
 212 }
 213
 214 class RequestReader : public opentracing::HTTPHeadersReader {
 215   public:
 216     explicit RequestReader(const net::http::Request& request)
 217         : _request(request)
 218     {
 219     }
 220
 221     opentracing::expected<void> ForeachKey(
 222         std::function<opentracing::expected<void>(opentracing::string_view,
 223                                                   opentracing::string_view)> f)
 224         const override
 225     {
 226         for (auto&& header : _request.headers()) {
 227             const auto result = f(header.key(), header.value());
 228             if (!result) {
 229                 return result;
 230             }
 231         }
 232         return opentracing::make_expected();
 233     }
 234
 235   private:
 236     const net::http::Request& _request;
 237 };
 238
 239 class RequestWriter : public opentracing::HTTPHeadersWriter {
 240   public:
 241     explicit RequestWriter(std::ostream& requestStream)
 242         : _requestStream(requestStream)
 243     {
 244     }
 245
 246     opentracing::expected<void>
 247     Set(opentracing::string_view key,
 248         opentracing::string_view value) const override
 249     {
 250         _requestStream << key << ": " << value << "\r\n";
 251         return opentracing::make_expected();
 252     }
 253
 254   private:
 255     std::ostream& _requestStream;
 256 };
 257
 258 thrift::ObservedSpan observeSpan(const opentracing::SpanContext& ctx)
 259 {
 260     const auto& sc = static_cast<const SpanContext&>(ctx);
 261     thrift::ObservedSpan observedSpan;
 262     std::ostringstream oss;
 263     oss << sc.traceID();
 264     observedSpan.__set_traceId(oss.str());
 265     observedSpan.__set_sampled(sc.isSampled());
 266     auto itr = sc.baggage().find(kBaggageKey);
 267     if (itr != std::end(sc.baggage())) {
 268         observedSpan.__set_baggage(itr->second);
 269     }
 270     return observedSpan;
 271 }
 272
 273 thrift::TraceResponse callDownstreamHTTP(const opentracing::SpanContext& ctx,
 274                                          const thrift::Downstream& target,
 275                                          opentracing::Tracer& tracer,
 276                                          logging::Logger& logger)
 277 {
 278     thrift::JoinTraceRequest request;
 279     request.__set_serverRole(target.serverRole);
 280     if (target.downstream) {
 281         request.__set_downstream(*target.downstream);
 282     }
 283
 284     const auto requestJSON = nlohmann::json(request).dump();
 285     net::Socket socket;
 286     socket.open(AF_INET, SOCK_STREAM);
 287     const auto authority = target.host + ':' + target.port;
 288     socket.connect("http://" + authority);
 289     std::ostringstream oss;
 290     oss << "POST /join_trace HTTP/1.1\r\n"
 291            "Host: "
 292         << authority << "\r\n";
 293     RequestWriter writer(oss);
 294     tracer.Inject(ctx, writer);
 295     oss << "Connection: close\r\n"
 296            "Content-Type: application/json\r\n"
 297            "Content-Length: "
 298         << requestJSON.size() << "\r\n\r\n"
 299         << requestJSON;
 300     const auto message = oss.str();
 301     logger.info("Sending request downstream: " + escape(message));
 302     const auto numWritten =
 303         ::write(socket.handle(), &message[0], message.size());
 304     (void)numWritten;
 305
 306     const auto responseStr = bufferedRead(socket);
 307     logger.info("Received downstream response: " + escape(responseStr));
 308     std::istringstream iss(responseStr);
 309     auto response = net::http::Response::parse(iss);
 310     return nlohmann::json::parse(response.body());
 311 }
 312
 313 thrift::TraceResponse callDownstream(const opentracing::SpanContext& ctx,
 314                                      const std::string& /* role */,
 315                                      const thrift::Downstream& downstream,
 316                                      opentracing::Tracer& tracer,
 317                                      logging::Logger& logger)
 318 {
 319     thrift::TraceResponse response;
 320
 321     switch (downstream.transport) {
 322     case thrift::Transport::HTTP: {
 323         response = callDownstreamHTTP(ctx, downstream, tracer, logger);
 324     } break;
 325     case thrift::Transport::TCHANNEL: {
 326         response.__set_notImplementedError(
 327             "TCHANNEL transport not implemented");
 328     } break;
 329     case thrift::Transport::DUMMY: {
 330         response.__set_notImplementedError("DUMMY transport not implemented");
 331     } break;
 332     default: {
 333         throw std::invalid_argument("Unrecognized protocol " +
 334                                     std::to_string(downstream.transport));
 335     } break;
 336     }
 337
 338     return response;
 339 }
 340
 341 thrift::TraceResponse prepareResponse(const opentracing::SpanContext& ctx,
 342                                       const std::string& role,
 343                                       const thrift::Downstream* downstream,
 344                                       opentracing::Tracer& tracer,
 345                                       logging::Logger& logger)
 346 {
 347     const auto observedSpan = observeSpan(ctx);
 348     thrift::TraceResponse response;
 349     response.__set_span(observedSpan);
 350     if (downstream) {
 351         response.__set_downstream(std::make_shared<thrift::TraceResponse>(
 352             callDownstream(ctx, role, *downstream, tracer, logger)));
 353     }
 354     return response;
 355 }
 356
 357 struct GenerateTracesRequest {
 358     using StrMap = std::unordered_map<std::string, std::string>;
 359
 360     std::string _type;
 361     std::string _operation;
 362     StrMap _tags;
 363     int _count;
 364 };
 365
 366 void from_json(const nlohmann::json& json, GenerateTracesRequest& request)
 367 {
 368     request._type = json.at("type");
 369     request._operation = json.at("operation");
 370     request._tags = json.at("tags").get<GenerateTracesRequest::StrMap>();
 371     request._count = json.at("count");
 372 }
 373
 374 }  // anonymous namespace
 375
 376 using Handler = std::function<std::string(const net::http::Request&)>;
 377
 378 class Server::SocketListener {
 379   public:
 380     SocketListener(const net::IPAddress& ip,
 381                    const std::shared_ptr<logging::Logger>& logger,
 382                    Handler handler)
 383         : _ip(ip)
 384         , _logger(logger)
 385         , _handler(handler)
 386         , _running(false)
 387     {
 388         assert(_logger);
 389     }
 390
 391     ~SocketListener() { stop(); }
 392
 393     void start()
 394     {
 395         std::promise<void> started;
 396         _thread = std::thread([this, &started]() { start(_ip, started); });
 397         started.get_future().get();
 398     }
 399
 400     void stop() noexcept
 401     {
 402         if (_running) {
 403             _running = false;
 404             _thread.join();
 405             _socket.close();
 406         }
 407     }
 408
 409   private:
 410     void start(const net::IPAddress& ip, std::promise<void>& started)
 411     {
 412         _socket.open(AF_INET, SOCK_STREAM);
 413         const auto enable = 1;
 414         ::setsockopt(_socket.handle(),
 415                      SOL_SOCKET,
 416                      SO_REUSEADDR,
 417                      &enable,
 418                      sizeof(enable));
 419         _socket.bind(ip);
 420         _socket.listen();
 421         _running = true;
 422         started.set_value();
 423
 424         using TaskList = std::deque<std::future<void>>;
 425         TaskList tasks;
 426
 427         while (_running) {
 428             auto client = _socket.accept();
 429             auto future = std::async(
 430                 std::launch::async,
 431                 [this](net::Socket&& socket) {
 432                     net::Socket client(std::move(socket));
 433                     auto requestStr = bufferedRead(client);
 434                     _logger->info("Received request: " + escape(requestStr));
 435
 436                     try {
 437                         std::istringstream iss(requestStr);
 438                         const auto request = net::http::Request::parse(iss);
 439                         const auto responseStr = _handler(request);
 440                         const auto numWritten = ::write(client.handle(),
 441                                                         &responseStr[0],
 442                                                         responseStr.size());
 443                         if (numWritten !=
 444                             static_cast<int>(responseStr.size())) {
 445                             std::ostringstream oss;
 446                             oss << "Unable to write entire response"
 447                                    ", numWritten="
 448                                 << numWritten
 449                                 << ", responseSize=" << responseStr.size();
 450                             _logger->error(oss.str());
 451                         }
 452                     } catch (...) {
 453                         utils::ErrorUtil::logError(*_logger, "Server error");
 454                         constexpr auto message =
 455                             "HTTP/1.1 500 Internal Server Error\r\n\r\n";
 456                         constexpr auto messageSize = sizeof(message) - 1;
 457                         const auto numWritten =
 458                             ::write(client.handle(), message, messageSize);
 459                         (void)numWritten;
 460                     }
 461
 462                     client.close();
 463                 },
 464                 std::move(client));
 465             tasks.emplace_back(std::move(future));
 466         }
 467
 468         std::for_each(std::begin(tasks),
 469                       std::end(tasks),
 470                       [](TaskList::value_type& future) { future.get(); });
 471     }
 472
 473     net::IPAddress _ip;
 474     net::Socket _socket;
 475     std::shared_ptr<logging::Logger> _logger;
 476     Handler _handler;
 477     std::atomic<bool> _running;
 478     std::thread _thread;
 479 };
 480
 481 class Server::EndToEndHandler {
 482   public:
 483     using TracerPtr = std::shared_ptr<opentracing::Tracer>;
 484
 485     EndToEndHandler(const std::string& agentHostPort,
 486                     const std::string& collectorEndpoint,
 487                     const std::string& samplingServerURL)
 488         : _agentHostPort(agentHostPort)
 489         , _collectorEndpoint(collectorEndpoint)
 490         , _samplingServerURL(samplingServerURL)
 491     {
 492     }
 493
 494     TracerPtr findOrMakeTracer(std::string samplerType)
 495     {
 496         if (samplerType.empty()) {
 497             samplerType = kSamplerTypeRemote;
 498         }
 499
 500         std::lock_guard<std::mutex> lock(_mutex);
 501         auto itr = _tracers.find(samplerType);
 502         if (itr != std::end(_tracers)) {
 503             return itr->second;
 504         }
 505         return init(samplerType);
 506     }
 507
 508   private:
 509     Config makeEndToEndConfig(const std::string& samplerType) const
 510     {
 511         return Config(false,
 512                       samplers::Config(samplerType,
 513                                        1.0,
 514                                        _samplingServerURL,
 515                                        samplers::Config::kDefaultMaxOperations,
 516                                        std::chrono::seconds(5)),
 517                       reporters::Config(reporters::Config::kDefaultQueueSize,
 518                                         std::chrono::seconds(1),
 519                                         false,
 520                                         _agentHostPort,
 521                                         _collectorEndpoint));
 522     }
 523
 524     TracerPtr init(const std::string& samplerType)
 525     {
 526         const auto config = makeEndToEndConfig(samplerType);
 527         auto tracer = Tracer::make(kDefaultTracerServiceName, config);
 528         _tracers[config.sampler().type()] = tracer;
 529         return tracer;
 530     }
 531
 532     std::string _agentHostPort;
 533     std::string _collectorEndpoint;
 534     std::string _samplingServerURL;
 535     std::unordered_map<std::string, TracerPtr> _tracers;
 536     std::mutex _mutex;
 537 };
 538
 539 Server::Server(const net::IPAddress& clientIP,
 540                const net::IPAddress& serverIP,
 541                const std::string& agentHostPort,
 542                const std::string& collectorEndpoint,
 543                const std::string& samplingServerURL)
 544     : _logger(logging::consoleLogger())
 545     , _tracer(Tracer::make(kDefaultTracerServiceName, Config(), _logger))
 546     , _clientListener(
 547           new SocketListener(clientIP,
 548                              _logger,
 549                              [this](const net::http::Request& request) {
 550                                  return handleRequest(request);
 551                              }))
 552     , _serverListener(
 553           new SocketListener(serverIP,
 554                              _logger,
 555                              [this](const net::http::Request& request) {
 556                                  return handleRequest(request);
 557                              }))
 558     , _handler(new EndToEndHandler(agentHostPort, collectorEndpoint, samplingServerURL))
 559 {
 560 }
 561
 562 Server::~Server() = default;
 563
 564 void Server::serve()
 565 {
 566     _clientListener->start();
 567     _serverListener->start();
 568 }
 569
 570 template <typename RequestType>
 571 std::string Server::handleJSON(
 572     const net::http::Request& request,
 573     std::function<thrift::TraceResponse(
 574         const RequestType&, const opentracing::SpanContext&)> handler)
 575 {
 576     RequestReader reader(request);
 577     auto result = _tracer->Extract(reader);
 578     if (!result) {
 579         std::ostringstream oss;
 580         oss << "Cannot read request body: opentracing error code "
 581             << result.error().value();
 582         const auto message = oss.str();
 583         oss.str("");
 584         oss.clear();
 585         oss << "HTTP/1.1 400 Bad Request\r\n"
 586                "Content-Length: "
 587             << message.size() << "\r\n\r\n"
 588             << message;
 589     }
 590
 591     std::unique_ptr<opentracing::SpanContext> ctx(result->release());
 592     opentracing::StartSpanOptions options;
 593     options.start_system_timestamp = std::chrono::system_clock::now();
 594     options.start_steady_timestamp = std::chrono::steady_clock::now();
 595     if (ctx) {
 596         options.references.emplace_back(std::make_pair(
 597             opentracing::SpanReferenceType::ChildOfRef, ctx.get()));
 598     }
 599     auto span = _tracer->StartSpanWithOptions("post", options);
 600
 601     RequestType thriftRequest;
 602     try {
 603         thriftRequest = nlohmann::json::parse(request.body());
 604     } catch (const std::exception& ex) {
 605         std::ostringstream oss;
 606         oss << "Cannot parse request JSON: " << ex.what()
 607             << ", json: " << request.body();
 608         const auto message = oss.str();
 609         oss.str("");
 610         oss.clear();
 611         oss << "HTTP/1.1 500 Internal Server Error\r\n"
 612                "Content-Length: "
 613             << message.size() << "\r\n\r\n"
 614             << message;
 615         return oss.str();
 616     } catch (...) {
 617         std::ostringstream oss;
 618         oss << "Cannot parse request JSON, json: " << request.body();
 619         const auto message = oss.str();
 620         oss.str("");
 621         oss.clear();
 622         oss << "HTTP/1.1 500 Internal Server Error\r\n"
 623                "Content-Length: "
 624             << message.size() << "\r\n\r\n"
 625             << message;
 626         return oss.str();
 627     }
 628
 629     const auto thriftResponse = handler(thriftRequest, span->context());
 630     try {
 631         const auto message = nlohmann::json(thriftResponse).dump();
 632         std::ostringstream oss;
 633         oss << "HTTP/1.1 200 OK\r\n"
 634                "Content-Type: application/json\r\n"
 635                "Content-Length: "
 636             << message.size() << "\r\n\r\n"
 637             << message;
 638         return oss.str();
 639     } catch (const std::exception& ex) {
 640         std::ostringstream oss;
 641         oss << "Cannot marshal response to JSON: " << ex.what();
 642         const auto message = oss.str();
 643         oss.str("");
 644         oss.clear();
 645         oss << "HTTP/1.1 500 Internal Server Error\r\n"
 646                "Content-Length: "
 647             << message.size() << "\r\n\r\n"
 648             << message;
 649         return oss.str();
 650     } catch (...) {
 651         std::ostringstream oss;
 652         oss << "Cannot marshal response to JSON";
 653         const auto message = oss.str();
 654         oss.str("");
 655         oss.clear();
 656         oss << "HTTP/1.1 500 Internal Server Error\r\n"
 657                "Content-Length: "
 658             << message.size() << "\r\n\r\n"
 659             << message;
 660         return oss.str();
 661     }
 662 }
 663
 664 std::string Server::handleRequest(const net::http::Request& request)
 665 {
 666     if (request.target() == "/") {
 667         return "HTTP/1.1 200 OK\r\n\r\n";
 668     }
 669     if (request.target() == "/start_trace") {
 670         return handleJSON<thrift::StartTraceRequest>(
 671             request,
 672             [this](const thrift::StartTraceRequest& request,
 673                    const opentracing::SpanContext& /* ctx */) {
 674                 return startTrace(request);
 675             });
 676     }
 677     if (request.target() == "/join_trace") {
 678         return handleJSON<thrift::JoinTraceRequest>(
 679             request,
 680             [this](const thrift::JoinTraceRequest& request,
 681                    const opentracing::SpanContext& ctx) {
 682                 return joinTrace(request, ctx);
 683             });
 684     }
 685     if (request.target() == "/create_traces") {
 686         return generateTraces(request);
 687     }
 688     return "HTTP/1.1 404 Not Found\r\n\r\n";
 689 }
 690
 691 thrift::TraceResponse
 692 Server::startTrace(const crossdock::thrift::StartTraceRequest& request)
 693 {
 694     auto span = _tracer->StartSpan(request.serverRole);
 695     if (request.sampled) {
 696         span->SetTag("sampling.priority", 1);
 697     }
 698     span->SetBaggageItem(kBaggageKey, request.baggage);
 699
 700     return prepareResponse(span->context(),
 701                            request.serverRole,
 702                            &request.downstream,
 703                            *_tracer,
 704                            *_logger);
 705 }
 706
 707 thrift::TraceResponse
 708 Server::joinTrace(const crossdock::thrift::JoinTraceRequest& request,
 709                   const opentracing::SpanContext& ctx)
 710 {
 711     return prepareResponse(ctx,
 712                            request.serverRole,
 713                            request.__isset.downstream ? &request.downstream
 714                                                       : nullptr,
 715                            *_tracer,
 716                            *_logger);
 717 }
 718
 719 std::string Server::generateTraces(const net::http::Request& requestHTTP)
 720 {
 721     GenerateTracesRequest request;
 722     try {
 723         request = nlohmann::json::parse(requestHTTP.body());
 724     } catch (const std::exception& ex) {
 725         std::ostringstream oss;
 726         oss << "JSON payload is invalid: " << ex.what();
 727         const auto message = oss.str();
 728         oss.str("");
 729         oss.clear();
 730         oss << "HTTP/1.1 400 Bad Request\r\n"
 731                "Content-Length: "
 732             << message.size() << "\r\n\r\n"
 733             << message;
 734         return oss.str();
 735     } catch (...) {
 736         const std::string message("JSON payload is invalid");
 737         std::ostringstream oss;
 738         oss << "HTTP/1.1 400 Bad Request\r\n"
 739                "Content-Length: "
 740             << message.size() << "\r\n\r\n"
 741             << message;
 742         return oss.str();
 743     }
 744
 745     auto tracer = _handler->findOrMakeTracer(request._type);
 746     if (!tracer) {
 747         const std::string message("Tracer is not initialized");
 748         std::ostringstream oss;
 749         oss << "HTTP/1.1 500 Internal Server Error\r\n"
 750                "Content-Length: "
 751             << message.size() << "\r\n"
 752             << message;
 753         return oss.str();
 754     }
 755
 756     for (auto i = 0; i < request._count; ++i) {
 757         auto span = tracer->StartSpan(request._operation);
 758         for (auto&& pair : request._tags) {
 759             span->SetTag(pair.first, pair.second);
 760         }
 761         span->Finish();
 762     }
 763
 764     return "HTTP/1.1 200 OK\r\n\r\n";
 765 }
 766
 767 }  // namespace crossdock
 768 }  // namespace jaegertracing
 769
 770 int main()
 771 {
 772     const auto rawSenderType = std::getenv("SENDER");
 773     const std::string senderType(rawSenderType ? rawSenderType : "");
 774
 775     if (senderType.empty()) {
 776       std::cerr << "env SENDER is not specified!\n";
 777       return 1;
 778     }
 779
 780     const auto rawAgentHostPort = std::getenv("AGENT_HOST_PORT");
 781     const std::string agentHostPort(rawAgentHostPort ? rawAgentHostPort : "");
 782
 783     if (agentHostPort.empty() && senderType == "udp") {
 784       std::cerr << "env AGENT_HOST_PORT is not specified!\n";
 785       return 1;
 786     }
 787
 788     const std::string collectorEndpoint(senderType == "http" ? "http://jaeger-collector:14268/api/traces" : "");
 789
 790     const auto rawSamplingServerURL = std::getenv("SAMPLING_SERVER_URL");
 791     const std::string samplingServerURL(
 792         rawSamplingServerURL ? rawSamplingServerURL : "");
 793     if (samplingServerURL.empty()) {
 794         std::cerr << "env SAMPLING_SERVER_URL is not specified!\n";
 795         return 1;
 796     }
 797
 798     jaegertracing::crossdock::Server server(
 799         jaegertracing::net::IPAddress::v4("0.0.0.0:8080"),
 800         jaegertracing::net::IPAddress::v4("0.0.0.0:8081"),
 801         agentHostPort,
 802         collectorEndpoint,
 803         samplingServerURL);
 804     server.serve();
 805
 806     std::this_thread::sleep_for(std::chrono::hours(1));
 807     return 0;
 808 }