datadog: HTTP client portion of new tracing library

source/extensions/tracers/datadog/BUILD

@@ -14,11 +14,13 @@ envoy_extension_package()
 envoy_cc_library(
     name = "datadog_tracer_lib",
     srcs = [
+        "agent_http_client.cc",
         "datadog_tracer_impl.cc",
         "dict_util.cc",
         "time_util.cc",
     ],
     hdrs = [
+        "agent_http_client.h",
         "datadog_tracer_impl.h",
         "dict_util.h",
         "time_util.h",

source/extensions/tracers/datadog/agent_http_client.cc

@@ -0,0 +1,144 @@
+#include "source/extensions/tracers/datadog/agent_http_client.h"
+
+#include <cstddef>
+#include <cstdint>
+#include <memory>
+#include <string>
+
+#include "source/common/common/assert.h"
+#include "source/common/http/message_impl.h"
+#include "source/common/http/utility.h"
+#include "source/extensions/tracers/datadog/dict_util.h"
+#include "source/extensions/tracers/datadog/tracer_stats.h"
+
+#include "absl/strings/str_format.h"
+#include "datadog/dict_reader.h"
+#include "datadog/dict_writer.h"
+#include "datadog/error.h"
+#include "datadog/json.hpp"
+
+namespace Envoy {
+namespace Extensions {
+namespace Tracers {
+namespace Datadog {
+
+AgentHTTPClient::AgentHTTPClient(Upstream::ClusterManager& cluster_manager,
+                                 const std::string& cluster, const std::string& reference_host,
+                                 TracerStats& stats)
+    : collector_cluster_(cluster_manager, cluster), cluster_(cluster),
+      reference_host_(reference_host), stats_(stats) {}
+
+AgentHTTPClient::~AgentHTTPClient() {
+  for (const auto& [request_ptr, _] : handlers_) {
+    RELEASE_ASSERT(request_ptr,
+                   "null Http::AsyncClient::Request* in handler map of Datadog::AgentHTTPClient");
+    request_ptr->cancel();
+  }
+}
+
+// datadog::tracing::HTTPClient
+
+datadog::tracing::Expected<void> AgentHTTPClient::post(const URL& url, HeadersSetter set_headers,
+                                                       std::string body,
+                                                       ResponseHandler on_response,
+                                                       ErrorHandler on_error) {
+  ENVOY_LOG(debug, "flushing traces");
+
+  auto message = std::make_unique<Http::RequestMessageImpl>();
+  Http::RequestHeaderMap& headers = message->headers();
+  headers.setReferenceMethod(Http::Headers::get().MethodValues.Post);
+  headers.setReferencePath(url.path);
+  headers.setReferenceHost(reference_host_);
+
+  RequestHeaderWriter writer{message->headers()};
+  set_headers(writer);
+
+  message->body().add(body);
+  ENVOY_LOG(debug, "submitting trace(s) to {} with payload size {}", url.path, body.size());
+
+  if (!collector_cluster_.threadLocalCluster().has_value()) {
+    ENVOY_LOG(debug, "collector cluster '{}' does not exist", cluster_);
+    stats_.reports_skipped_no_cluster_.inc();
+    return datadog::tracing::nullopt;
+  }
+
+  Http::AsyncClient::Request* request =
+      collector_cluster_.threadLocalCluster()->get().httpAsyncClient().send(
+          std::move(message), *this,
+          Http::AsyncClient::RequestOptions().setTimeout(std::chrono::milliseconds(1000)));
+  if (!request) {
+    stats_.reports_failed_.inc();
+    return datadog::tracing::Error{datadog::tracing::Error::ENVOY_HTTP_CLIENT_FAILURE,
+                                   "Failed to create request."};
+  }
+
+  handlers_.emplace(request, Handlers{std::move(on_response), std::move(on_error)});
+  return datadog::tracing::nullopt;
+}
+
+void AgentHTTPClient::drain(std::chrono::steady_clock::time_point) {}
+
+nlohmann::json AgentHTTPClient::config_json() const {
+  return nlohmann::json::object({
+      {"type", "Envoy::Extensions::Tracers::Datadog::AgentHTTPClient"},
+  });
+}
+
+// Http::AsyncClient::Callbacks
+
+void AgentHTTPClient::onSuccess(const Http::AsyncClient::Request& request,
+                                Http::ResponseMessagePtr&& response) {
+  const std::uint64_t status = Http::Utility::getResponseStatus(response->headers());
+  if (status != std::uint64_t(Http::Code::OK)) {
+    stats_.reports_dropped_.inc();
+  } else {
+    ENVOY_LOG(debug, "traces successfully submitted to datadog agent");
+    stats_.reports_sent_.inc();
+  }
+
+  auto found = handlers_.find(const_cast<Http::AsyncClient::Request*>(&request));
+  if (found == handlers_.end()) {
+    ENVOY_LOG(debug, "request at address 0x{} is not in the map of {} Handlers objects",
+              absl::StrFormat("%p", &request), handlers_.size());
+    return;
+  }
+
+  Handlers& handlers = found->second;
+  ResponseHeaderReader reader{response->headers()};
+  handlers.on_response(status, reader, response->bodyAsString());
+
+  handlers_.erase(found);
+}
+
+void AgentHTTPClient::onFailure(const Http::AsyncClient::Request& request,
+                                Http::AsyncClient::FailureReason reason) {
+  auto found = handlers_.find(const_cast<Http::AsyncClient::Request*>(&request));
+  // If the request failed to start, then `post` never even registered the request.
+  if (found == handlers_.end()) {
+    return;
+  }
+
+  stats_.reports_failed_.inc();
+
+  Handlers& handlers = found->second;
+  std::string message = "Failed to send request to Datadog Agent: ";
+  switch (reason) {
+  case Http::AsyncClient::FailureReason::Reset:
+    message += "The stream has been reset.";
+    break;
+  default:
+    message += "Unknown error.";
+  }
+  handlers.on_error(datadog::tracing::Error{datadog::tracing::Error::ENVOY_HTTP_CLIENT_FAILURE,
+                                            std::move(message)});
+
+  handlers_.erase(found);
+}
+
+void AgentHTTPClient::onBeforeFinalizeUpstreamSpan(Tracing::Span&, const Http::ResponseHeaderMap*) {
+}
+
+} // namespace Datadog
+} // namespace Tracers
+} // namespace Extensions
+} // namespace Envoy

source/extensions/tracers/datadog/agent_http_client.h

@@ -0,0 +1,106 @@
+#pragma once
+
+#include <string>
+
+#include "envoy/http/async_client.h"
+
+#include "source/common/common/logger.h"
+#include "source/common/upstream/cluster_update_tracker.h"
+
+#include "absl/container/flat_hash_map.h"
+#include "datadog/http_client.h"
+
+namespace Envoy {
+namespace Extensions {
+namespace Tracers {
+namespace Datadog {
+
+struct TracerStats;
+
+/**
+ * \c datadog::tracing::HTTPClient implementation that uses Envoy's
+ * \c Http::AsyncClient. The class is called \c AgentHTTPClient because it is
+ * not a general-purpose HTTP client. Instead, it sends requests to a specified
+ * cluster only. The idea is that the cluster is configured to point to a
+ * Datadog Agent instance.
+ */
+class AgentHTTPClient : public datadog::tracing::HTTPClient,
+                        public Http::AsyncClient::Callbacks,
+                        private Logger::Loggable<Logger::Id::tracing> {
+public:
+  struct Handlers {
+    ResponseHandler on_response;
+    ErrorHandler on_error;
+  };
+
+public:
+  /**
+   * Create an  \c AgentHTTPClient that uses the specified \p cluster_manager
+   * to make requests to the specified \p cluster, where requests include the
+   * specified \p reference_host as the "Host" header. Use the specified
+   * \p stats to keep track of usage statistics.
+   * @param cluster_manager cluster manager from which the thread local cluster
+   * is retrieved in order to make HTTP requests
+   * @param cluster the name of the cluster to which HTTP requests are made
+   * @param reference_host the value to use for the "Host" HTTP request header
+   * @param stats a collection of counters used to keep track of events, such as
+   * when a request fails
+   */
+  AgentHTTPClient(Upstream::ClusterManager& cluster_manager, const std::string& cluster,
+                  const std::string& reference_host, TracerStats& stats);
+  ~AgentHTTPClient() override;
+
+  // datadog::tracing::HTTPClient
+
+  /**
+   * Send an HTTP POST request to the cluster, where the requested resource is
+   * \p url.path. Invoke \p set_headers immediately to obtain the HTTP request
+   * headers. Send \p body as the request body. Return a
+   * \c datadog::tracing::Error if one occurs, otherwise return
+   * \c datadog::tracing::nullopt. When a complete response is received,
+   * invoke \p on_response with the response status, response headers, and
+   * response body. If an error occurs before a complete response is received,
+   * invoke \p on_error with a \c datadog::tracing::Error.
+   * @param url URL from which the request path is taken
+   * @param set_headers callback invoked immediately to obtain request headers
+   * @param body data to send as POST request body
+   * @param on_response callback to invoke when a complete response is received
+   * @param on_error callback to invoke when an error occurs before a complete
+   * response is received.
+   * @return \c datadog::tracing::Error if an error occurs, or
+   * \c datadog::tracing::nullopt otherwise.
+   */
+  datadog::tracing::Expected<void> post(const URL& url, HeadersSetter set_headers, std::string body,
+                                        ResponseHandler on_response,
+                                        ErrorHandler on_error) override;
+
+  /**
+   * \c drain has no effect. It's a part of the \c datadog::tracing::HTTPClient
+   * that we do not need.
+   */
+  void drain(std::chrono::steady_clock::time_point) override;
+
+  /**
+   * Return a JSON representation of this object's configuration. This function
+   * is used in the startup banner logged by \c dd-trace-cpp.
+   */
+  nlohmann::json config_json() const override;
+
+  // Http::AsyncClient::Callbacks
+
+  void onSuccess(const Http::AsyncClient::Request&, Http::ResponseMessagePtr&&) override;
+  void onFailure(const Http::AsyncClient::Request&, Http::AsyncClient::FailureReason) override;
+  void onBeforeFinalizeUpstreamSpan(Tracing::Span&, const Http::ResponseHeaderMap*) override;
+
+private:
+  absl::flat_hash_map<Http::AsyncClient::Request*, Handlers> handlers_;
+  Upstream::ClusterUpdateTracker collector_cluster_;
+  const std::string cluster_;
+  const std::string reference_host_;
+  TracerStats& stats_;
+};
+
+} // namespace Datadog
+} // namespace Tracers
+} // namespace Extensions
+} // namespace Envoy

source/extensions/tracers/datadog/datadog_tracer_impl.h

@@ -1,7 +1,5 @@
 #pragma once
 
-#include <datadog/opentracing.h>
-
 #include "envoy/config/trace/v3/datadog.pb.h"
 #include "envoy/local_info/local_info.h"
 #include "envoy/runtime/runtime.h"
@@ -14,6 +12,7 @@
 #include "source/common/upstream/cluster_update_tracker.h"
 #include "source/extensions/tracers/common/ot/opentracing_driver_impl.h"
 
+#include "datadog/opentracing.h"
 #include "fmt/ostream.h"
 
 namespace Envoy {

source/extensions/tracers/datadog/dict_util.h

@@ -10,11 +10,11 @@
  * headers, or writing HTTP request headers.
  */
 
-#include <datadog/dict_reader.h>
-#include <datadog/dict_writer.h>
-
 #include <string>
 
+#include "datadog/dict_reader.h"
+#include "datadog/dict_writer.h"
+
 namespace Envoy {
 namespace Tracing {
 class TraceContext;

source/extensions/tracers/datadog/time_util.h

@@ -19,10 +19,10 @@
  * estimateTime function.
  */
 
-#include <datadog/clock.h>
-
 #include "envoy/common/time.h"
 
+#include "datadog/clock.h"
+
 namespace Envoy {
 namespace Extensions {
 namespace Tracers {

test/extensions/tracers/datadog/BUILD

@@ -14,6 +14,7 @@ envoy_package()
 envoy_extension_cc_test(
     name = "datadog_tracer_impl_test",
     srcs = [
+        "agent_http_client_test.cc",
         "datadog_tracer_impl_test.cc",
         "dict_util_test.cc",
         "time_util_test.cc",